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

news2025/12/14 12:03:21

📚 使用 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: True
  clear_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