只能在线观看的电影网站咋么做wordpress教程 菜单
📚 使用 Python + SQLAlchemy 创建知识库数据库（SQLite）—— 构建本地知识库系统的基础
🧠 一、前言
随着大模型技术的发展，越来越多的项目需要构建本地知识库系统来支持 RAG（Retrieval-Augmented Generation）、Agent 记忆管理、文档检索等功能。
本文将介绍如何使用 Python + SQLAlchemy 搭建一个结构清晰、可扩展性强的知识库数据库，适用于 AI 助手、智能问答系统等场景。
🗂️ 二、项目目标
我们希望创建一个 SQLite 数据库，包含以下四张核心表：
| 表名 | 描述 |
|---|---|
knowledge_base | 存储知识库基本信息（名称、简介、向量库类型、嵌入模型等） |
knowledge_file | 存储每个知识库中的文件信息（文件名、大小、修改时间、切分情况等） |
file_doc | 存储文件切片后的文档片段 ID 及其元数据 |
summary_chunk | 存储文档摘要与对应的文档 ID 列表 |
这些表构成了一个完整的知识库管理系统的基础架构，可用于后续的文档加载、向量化、检索和更新操作。
🔧 三、环境依赖
- Python 3.12+
- SQLAlchemy
- sqlite3（Python 自带）
安装命令如下：
pip install sqlalchemy
🗃️ 四、数据库配置与初始化
我们使用 SQLite 作为本地存储方式，路径如下：
DB_PATH = "/Volumes/PSSD/未命名文件夹/donwload/创建知识库数据库/knowledge_base/info.db"
SQLALCHEMY_DATABASE_URI = f"sqlite:///{DB_PATH}"
通过 SQLAlchemy 的 create_engine 和 sessionmaker 初始化数据库连接池。
engine = create_engine(SQLALCHEMY_DATABASE_URI,json_serializer=lambda obj: json.dumps(obj, ensure_ascii=False),
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base: DeclarativeMeta = declarative_base()
📐 五、ORM 模型定义详解
1. KnowledgeBaseModel —— 知识库基础信息表
| 字段 | 类型 | 默认值 | 描述 |
|---|---|---|---|
id | Integer | - | 主键 |
kb_name | String(50) | - | 知识库名称 |
kb_info | String(200) | - | 知识库简介 |
vs_type | String(50) | - | 向量库类型（如 FAISS、Chroma） |
embed_model | String(50) | - | 嵌入模型名称 |
file_count | Integer | 0 | 文件数量 |
create_time | DateTime | func.now() | 创建时间 |
2. KnowledgeFileModel —— 知识库文件信息表
| 字段 | 类型 | 默认值 | 描述 |
|---|---|---|---|
id | Integer | - | 主键 |
file_name | String(255) | - | 文件名 |
file_ext | String(10) | - | 扩展名 |
kb_name | String(50) | - | 所属知识库 |
document_loader_name | String(50) | - | 文档加载器名称 |
text_splitter_name | String(50) | - | 文本分割器名称 |
file_version | Integer | 1 | 文件版本 |
file_mtime | Float | 0.0 | 最后修改时间 |
file_size | Integer | 0 | 文件大小 |
custom_docs | Boolean | False | 是否自定义 docs |
docs_count | Integer | 0 | 切分文档数 |
create_time | DateTime | func.now() | 创建时间 |
3. FileDocModel —— 文件切片文档表
| 字段 | 类型 | 默认值 | 描述 |
|---|---|---|---|
id | Integer | - | 主键 |
kb_name | String(50) | - | 知识库名称 |
file_name | String(255) | - | 文件名 |
doc_id | String(50) | - | 向量库文档 ID |
meta_data | JSON | {} | 元数据（如来源、页码等） |
该表用于记录每个文件被切分后的每一个 chunk 对应的向量 ID。
4. SummaryChunkModel —— 文档摘要与关联表
| 字段 | 类型 | 默认值 | 描述 |
|---|---|---|---|
id | Integer | - | 主键 |
kb_name | String(50) | - | 知识库名称 |
summary_context | String(255) | - | 总结文本 |
summary_id | String(255) | - | 总结内容在向量库中的 ID |
doc_ids | String(1024) | - | 关联的 doc_id 列表 |
meta_data | JSON | {} | 元数据信息 |
用于存储文档的总结信息，并与原始 chunk 进行关联。
🧱 六、表操作函数
我们提供了两个主要函数用于管理数据库表：
✅ 1. 创建所有表
def create_tables():print("📌 正在创建所有表...")Base.metadata.create_all(bind=engine)print("✅ 表创建完毕")
⚠️ 2. 清空并重建所有表
def reset_tables():Base.metadata.drop_all(bind=engine)Base.metadata.create_all(bind=engine)print("⚠️ 数据库表已清空并重新创建")
这两个函数非常适合在初始化或调试阶段使用。
📝 七、命令行执行方式
你可以通过命令行参数控制数据库行为：
# 创建数据库表
python 创建表.py --create-tables# 清空并重建数据库表
python 创建表.py --clear-tables
输出示例：
📦 执行参数:create_tables: Trueclear_tables: False
📌 正在创建所有表...
✅ 表创建完毕
✅ 执行完毕，耗时: 0:00:00.123456
💡 八、详细的完整代码
#!/usr/bin/env python
# coding=utf-8""""""import os
import json
import sys
import argparse
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, Float, Boolean, JSON, func, create_engine
from sqlalchemy.ext.declarative import declarative_base, DeclarativeMeta
from sqlalchemy.orm import sessionmaker# ======== 数据库路径配置 ========SQLALCHEMY_DATABASE_URI = f"sqlite:///{DB_PATH}"# ======== SQLAlchemy 初始化 ========
engine = create_engine(SQLALCHEMY_DATABASE_URI,json_serializer=lambda obj: json.dumps(obj, ensure_ascii=False),
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base: DeclarativeMeta = declarative_base()# ======== ORM 模型定义 ========class KnowledgeBaseModel(Base):__tablename__ = 'knowledge_base'id = Column(Integer, primary_key=True, autoincrement=True, comment='知识库ID')kb_name = Column(String(50), comment='知识库名称')kb_info = Column(String(200), comment='知识库简介(用于Agent)')vs_type = Column(String(50), comment='向量库类型')embed_model = Column(String(50), comment='嵌入模型名称')file_count = Column(Integer, default=0, comment='文件数量')create_time = Column(DateTime, default=func.now(), comment='创建时间')class KnowledgeFileModel(Base):__tablename__ = 'knowledge_file'id = Column(Integer, primary_key=True, autoincrement=True, comment='知识文件ID')file_name = Column(String(255), comment='文件名')file_ext = Column(String(10), comment='文件扩展名')kb_name = Column(String(50), comment='所属知识库名称')document_loader_name = Column(String(50), comment='文档加载器名称')text_splitter_name = Column(String(50), comment='文本分割器名称')file_version = Column(Integer, default=1, comment='文件版本')file_mtime = Column(Float, default=0.0, comment="文件修改时间")file_size = Column(Integer, default=0, comment="文件大小")custom_docs = Column(Boolean, default=False, comment="是否自定义docs")docs_count = Column(Integer, default=0, comment="切分文档数量")create_time = Column(DateTime, default=func.now(), comment='创建时间')class FileDocModel(Base):__tablename__ = 'file_doc'id = Column(Integer, primary_key=True, autoincrement=True, comment='ID')kb_name = Column(String(50), comment='知识库名称')file_name = Column(String(255), comment='文件名称')doc_id = Column(String(50), comment="向量库文档ID")meta_data = Column(JSON, default={})class SummaryChunkModel(Base):__tablename__ = 'summary_chunk'id = Column(Integer, primary_key=True, autoincrement=True, comment='ID')kb_name = Column(String(50), comment='知识库名称')summary_context = Column(String(255), comment='总结文本')summary_id = Column(String(255), comment='总结矢量id')doc_ids = Column(String(1024), comment="向量库id关联列表")meta_data = Column(JSON, default={})# ======== 表操作函数 ========
def create_tables():print("📌 正在创建所有表...")Base.metadata.create_all(bind=engine)print("✅ 表创建完毕")def reset_tables():Base.metadata.drop_all(bind=engine)Base.metadata.create_all(bind=engine)print("⚠️ 数据库表已清空并重新创建")# ======== 命令行入口 ========
if __name__ == "__main__":parser = argparse.ArgumentParser(description="初始化知识库数据库")parser.add_argument("--create-tables",action="store_true",help="创建数据库表")parser.add_argument("--clear-tables",action="store_true",help="清空并重建数据库表")args = parser.parse_args()start_time = datetime.now()print("📦 执行参数:")for arg, value in vars(args).items():print(f" {arg}: {value}")if args.create_tables:create_tables()if args.clear_tables:reset_tables()print(f"✅ 执行完毕，耗时: {datetime.now() - start_time}")
#python 创建表.py --create-tables
#python 创建表.py --clear-tables