11 双存储架构：SQLite + ChromaDB 协作

🎯 一句话理解: ChromaDB 负责快速语义检索找碎片ID，SQLite 负责回查完整文档和上下文。各司其职，协作完成 RAG。

对应代码: dual_storage_demo.py、models.py

📖 生活类比：图书馆系统#

组件	类比	作用
ChromaDB	图书馆的索引卡片柜	只记录”这本书讲什么主题”，快速找到相关书籍
SQLite	图书馆的藏书仓库	存放完整的书籍内容，供你阅读
Embedding	给每本书贴上的主题标签	把”这本书讲 Python”变成一串数字坐标

工作流程#

1
读者问："有没有讲编程的书？"
2
        ↓
3
查索引卡片（ChromaDB）→ 快速找到《Python入门》的卡片编号
4
        ↓
5
去藏书仓库（SQLite）→ 根据编号取出完整书籍
6
        ↓
7
把书的内容给读者

🤔 为什么需要双存储？#

ChromaDB 的局限性#

1
# 查询返回的只是碎片（80字），不是完整文档
2
results = collection.query(query_embeddings=[query_vec], n_results=3)
3
# → ["Python 是一种高级编程语言...", "Web 开发、数据分析...", "NumPy、Pandas..."]

问题：

碎片只有 80 字，信息不完整
碎片之间没有上下文关联
无法按时间、来源等条件过滤

SQLite 的局限性#

1
SELECT * FROM documents WHERE content LIKE '%Python%'

问题：

只能精确匹配关键词，不能语义搜索
用户问”编程语言”找不到”Python”（因为没有精确包含这个词）

结论：两者互补#

需求	用哪个	原因
语义搜索（找意思相近的）	ChromaDB	向量相似度计算
查看完整文档	SQLite	存完整内容
按时间/来源过滤	SQLite	关系型查询
避免冗余存储	两者协作	ChromaDB 只存碎片

🏗️ 数据结构设计#

SQLite ORM 模型#

1
class Document(Base):
2
    """📚 完整文档（藏书仓库里的整本书）"""
3
    __tablename__ = "documents"
4

5
    id = Column(Integer, primary_key=True)       # 自增ID
6
    title = Column(String(200), nullable=False)  # 文档标题
7
    source = Column(String(500))                  # 来源（文件路径/URL）
8
    content = Column(Text, nullable=False)        # 完整内容
9
    created_at = Column(DateTime, default=datetime.now)
10

11
    # 一本书有多个切片
12
    chunks = relationship("DocumentChunk", back_populates="document",
13
                          cascade="all, delete-orphan")
14

15

16
class DocumentChunk(Base):
17
    """✂️ 文档切片（索引卡片）"""
18
    __tablename__ = "document_chunks"
19

20
    id = Column(Integer, primary_key=True)
21
    document_id = Column(Integer, ForeignKey("documents.id"), nullable=False)
22
    chunk_index = Column(Integer, nullable=False)     # 第几个切片（0, 1, 2...）
23
    content = Column(Text, nullable=False)            # 切片内容（80字左右）
24
    embedding_id = Column(String(100), unique=True)   # ⭐ 绳子①：对应 ChromaDB 的 ID
25

26
    # 一个切片属于一本书
27
    document = relationship("Document", back_populates="chunks")

ChromaDB Collection 结构#

1
Collection "document_chunks"
2
├── ids:        ["doc1_chunk0", "doc1_chunk1", ...]  ← 绳子①
3
├── embeddings: [[0.02, -0.01, ...], ...]            ← 4096维向量
4
├── documents:  ["Python 是一种高级...", ...]         ← 碎片文本
5
└── metadatas:  [                                      ← 绳子②
6
       {"document_id": 1, "chunk_index": 0, "title": "Python..."},
7
       ...
8
    ]

🔗 两条”绳子”连接机制#

核心问题: ChromaDB 和 SQLite 是两个独立的数据库，它们怎么关联？

绳子①：`embedding_id`（字符串ID双向索引）#

1
SQLite document_chunks 表          ChromaDB Collection
2
┌──────────────────────────┐       ┌──────────────────────┐
3
│ embedding_id             │       │ ids                  │
4
│ "doc1_chunk0"  ──────────┼───→───│ "doc1_chunk0"        │
5
│ "doc1_chunk1"  ──────────┼───→───│ "doc1_chunk1"        │
6
│ "doc2_chunk0"  ──────────┼───→───│ "doc2_chunk0"        │
7
└──────────────────────────┘       └──────────────────────┘

格式: "doc{文档ID}_chunk{碎片序号}" 用途: 精确匹配、删除操作

绳子②：`metadatas`（查询时直接返回关联信息）#

1
# ChromaDB 查询返回时，metadatas 直接告诉你去 SQLite 找谁
2
{
3
    "metadatas": [[{
4
        "document_id": 1,    # ← 去 SQLite 查 documents.id = 1
5
        "chunk_index": 0,    # ← 去 SQLite 查 document_chunks.chunk_index = 0
6
        "title": "Python编程语言介绍"
7
    }]]
8
}

用途: 查询返回时快速定位 SQLite 数据，无需解析字符串

关键澄清#

metadatas 标签是手动构造的，不是通过 ID 去数据库查出来的！

1
# ✅ 标签来自已有的 Python 变量
2
chroma_metadatas.append({
3
    "document_id": doc.id,   # ← doc 对象早就有了
4
    "chunk_index": idx,       # ← 循环变量
5
    "title": title            # ← 函数参数
6
})
7

8
# ❌ 不是这样：
9
# metadatas = ChromaDB.去SQLite查(doc.id).拿到标题()

💻 完整代码流程#

存入阶段#

1
def add_document(title: str, content: str, source: str = ""):
2
    db = SessionLocal()
3
    try:
4
        # 1. 存入 SQLite：完整文档
5
        doc = Document(title=title, content=content, source=source)
6
        db.add(doc)
7
        db.commit()       # 先提交，拿到 doc.id
8
        db.refresh(doc)   # 刷新，获取自增ID
9

10
        # 2. 文本切片
11
        chunks = split_text(content, chunk_size=80, overlap=10)
12

13
        chroma_ids = []
14
        chroma_embeddings = []
15
        chroma_documents = []
16
        chroma_metadatas = []
17

18
        # 3. 逐片处理
19
        for idx, chunk_text in enumerate(chunks):
20
            # 3a. 存入 SQLite：切片信息
21
            chunk = DocumentChunk(
22
                document_id=doc.id,
23
                chunk_index=idx,
24
                content=chunk_text,
25
                embedding_id=f"doc{doc.id}_chunk{idx}"  # 绳子①
26
            )
27
            db.add(chunk)
28

29
            # 3b. 生成向量
30
            vec = get_embedding(chunk_text)
31

32
            # 3c. 准备 ChromaDB 数据
33
            chroma_ids.append(f"doc{doc.id}_chunk{idx}")
34
            chroma_embeddings.append(vec)
35
            chroma_documents.append(chunk_text)
36
            chroma_metadatas.append({                    # 绳子②
37
                "document_id": doc.id,
38
                "chunk_index": idx,
39
                "title": title
40
            })
41

42
        # 4. 提交 SQLite 切片
43
        db.commit()
44

45
        # 5. 存入 ChromaDB：向量
46
        collection.add(
47
            ids=chroma_ids,
48
            embeddings=chroma_embeddings,
49
            documents=chroma_documents,
50
            metadatas=chroma_metadatas
51
        )
52
    finally:
53
        db.close()

查询阶段#

1
def semantic_search(query: str, n_results: int = 3):
2
    # 1. 查询转成向量
3
    query_vec = get_embedding(query)
4

5
    # 2. ChromaDB 快速检索
6
    chroma_results = collection.query(
7
        query_embeddings=[query_vec],
8
        n_results=n_results,
9
        include=["metadatas", "distances"]
10
    )
11

12
    # 3. 解析结果 → 去 SQLite 回查完整信息
13
    db = SessionLocal()
14
    try:
15
        for i in range(n_results):
16
            metadata = chroma_results["metadatas"][0][i]
17
            doc_id = metadata["document_id"]       # 绳子②给的
18
            chunk_idx = metadata["chunk_index"]     # 绳子②给的
19

20
            # 查 SQLite
21
            chunk = db.query(DocumentChunk).filter(
22
                DocumentChunk.document_id == doc_id,
23
                DocumentChunk.chunk_index == chunk_idx
24
            ).first()
25

26
            document = db.query(Document).filter(
27
                Document.id == doc_id
28
            ).first()
29
    finally:
30
        db.close()

📊 数据流全景图#

存入时#

1
用户上传文档
2
    │
3
    ▼
4
┌─────────────────┐
5
│ 1. 存入 SQLite   │  ← 完整文档（title, content, source）
6
│    Document 表   │
7
└─────────────────┘
8
    │
9
    ▼
10
┌─────────────────┐
11
│ 2. 文本切片      │  ← split_text() 切成 80 字片段（含重叠）
12
└─────────────────┘
13
    │
14
    ▼
15
┌─────────────────┐     ┌─────────────────┐
16
│ 3a. 切片存SQLite │     │ 3b. 向量存Chroma │
17
│  DocumentChunk  │     │  collection.add  │
18
│  (content,text) │     │  (embedding)     │
19
└─────────────────┘     └─────────────────┘
20
         │                       │
21
         └──────────┬────────────┘
22
                    ▼
23
              用 embedding_id 关联
24
              ("doc1_chunk0")

查询时#

1
用户提问 "什么是Python？"
2
    │
3
    ▼
4
┌─────────────────┐
5
│ 1. 转 Embedding │  ← get_embedding(query) → 4096维向量
6
└─────────────────┘
7
    │
8
    ▼
9
┌─────────────────┐
10
│ 2. ChromaDB检索  │  ← 返回 metadatas（含 document_id, chunk_index）
11
│  (快速语义匹配)  │
12
└─────────────────┘
13
    │
14
    ▼
15
┌─────────────────┐
16
│ 3. SQLite查详情  │  ← 根据 document_id 查 Document（完整文档）
17
│  (获取完整信息)  │  ← 根据 chunk_index 查 DocumentChunk（碎片详情）
18
└─────────────────┘
19
    │
20
    ▼
21
┌─────────────────┐
22
│ 4. 拼Prompt给LLM │  ← "根据以下文档片段回答：[片段1][片段2]"
23
└─────────────────┘

📋 `collection.add()` 四参数速查表#

参数	类型	参与相似度计算？	作用
`ids`	`list[str]`	❌	唯一标识，绳子①连接 SQLite
`embeddings`	`list[list[float]]`	✅ 唯一参与	语义相似度计算（余弦距离）
`documents`	`list[str]`	❌	查询返回时展示给用户看
`metadatas`	`list[dict]`	❌	绳子②，存 document_id 用于回查 SQLite

核心规则: 四个参数按数组下标一一对应

1
ids[0] ←→ embeddings[0] ←→ documents[0] ←→ metadatas[0]  属于同一个碎片
2
ids[1] ←→ embeddings[1] ←→ documents[1] ←→ metadatas[1]  属于同一个碎片

⚠️ 常见错误#

错误	原因	正确做法
ChromaDB 返回了碎片但 SQLite 查不到	embedding_id 不一致	确保两边用同样的 ID 格式
切片内容断裂	chunk_size 太大或没有 overlap	设置合理的 overlap（10-20%）
相似度都很低	查询和文档主题完全不相关	检查文档内容是否覆盖查询主题
删除文档时只删 SQLite	ChromaDB 残留垃圾数据	两边都要删
`metadatas` 放错字段	把 document_id 当 ids 用	metadatas 只是标签，不是主键
碎片太小不够回答	只返回单个碎片	取 N 个碎片 + 前后邻居拼接

📋 速查表#

1
双存储架构 = SQLite（关系型）+ ChromaDB（向量型）
2

3
SQLite 负责：
4
  - 存完整文档（Document.title, .content, .source）
5
  - 存切片内容（DocumentChunk.content）
6
  - 存元数据和关联关系（外键、时间戳）
7
  - 支持关系型查询（按时间、来源过滤）
8

9
ChromaDB 负责：
10
  - 存向量（Embedding，4096维）
11
  - 语义检索（余弦相似度计算）
12
  - 快速返回相关 ID 和 metadatas
13

14
关联方式（两条绳子）：
15
  绳子①: embedding_id = "doc{doc_id}_chunk{chunk_idx}"（同名字符串）
16
  绳子②: metadatas = {"document_id": id, "chunk_index": idx}（查询返回）

✅ 检查点#

你能解释为什么需要双存储（而不是只用一个数据库）吗？
你知道存入时数据怎么分流到两个数据库吗？
你知道查询时两个数据库怎么协作吗？
你能说出两条”绳子”分别在哪里、怎么用的吗？
你理解 metadatas 标签是手动构造的，还是查出来的？
你能修改 split_text() 的 chunk_size 看看效果变化吗？

🚀 下一步#

FastAPI + Chroma 最小原型: 把双存储封装成 API 接口（POST /semantic-search）
手搓最小 RAG 闭环: 切片 → Embedding → 存储 → 检索 → 拼 Prompt → 调 LLM
优化检索策略: 不只用单个碎片，取”碎片 + 前后邻居”增强上下文

📖 生活类比：图书馆系统#

工作流程#

🤔 为什么需要双存储？#

ChromaDB 的局限性#

SQLite 的局限性#

结论：两者互补#

🏗️ 数据结构设计#

SQLite ORM 模型#

ChromaDB Collection 结构#

🔗 两条”绳子”连接机制#

绳子①：embedding_id（字符串ID双向索引）#

绳子②：metadatas（查询时直接返回关联信息）#

关键澄清#

💻 完整代码流程#

存入阶段#

查询阶段#

📊 数据流全景图#

存入时#

查询时#

📋 collection.add() 四参数速查表#

⚠️ 常见错误#

📋 速查表#

✅ 检查点#

🚀 下一步#

绳子①：`embedding_id`（字符串ID双向索引）#

绳子②：`metadatas`（查询时直接返回关联信息）#

📋 `collection.add()` 四参数速查表#