使用 Python + SQLAlchemy 创建知识库数据库（SQLite）—— 构建本地知识库系统的基础《一》

📚 使用 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