可插拔组件与持久化

30 秒导读: dsRAG 的 KnowledgeBase 不是一坨写死的实现，而是把「用哪个嵌入模型、哪个向量库、哪个 LLM」全部抽成可替换的组件。本章讲两件事:(1) 六大组件都遵守同一套「抽象基类 + 自动注册 + to_dict/from_dict」模式，因此一整个 KB 的配置能序列化成一份 JSON、下次原样重建;(2) 向量库和正文库各存什么、在检索与取回里怎么分工。

本章属于 dsRAG 系列。前几章讲的是流程——摄取管线、AutoContext、RSE 查询;本章讲的是这些流程跑在什么之上的可换零件,以及这些零件的配置如何落盘、重建。总览见 index.md。

---

1. 这是什么(零基础也能懂)

一句话定义: dsRAG 把一个知识库拆成 6 个可插拔的组件,每个组件都是一个「插槽」,你可以插官方默认实现,也可以插备选实现,而 KB 只跟抽象接口打交道。

解决什么问题: RAG 系统里有一堆「外部依赖」——嵌入模型可能用 OpenAI 也可能用本地 Ollama;向量库可能用内存里的 pickle,也可能用 Pinecone、Qdrant。如果把这些写死,换一个就要改一堆代码。dsRAG 的做法是:把每类依赖定义成一个抽象基类,所有实现服从同一个接口,KB 只认接口。

六个插槽分别是:

组件	干什么	抽象基类所在
Embedding	把文本变成向量	dsrag/embedding.py:23
Reranker	对初排结果重新打分	dsrag/reranker.py:7
LLM	AutoContext 里生成标题/摘要	dsrag/llm.py:7
VectorDB	存向量 + 检索元数据	dsrag/database/vector/db.py:6
ChunkDB	存 chunk 正文与页码	dsrag/database/chunk/db.py:7
FileSystem	存页面图片、页面文本等	dsrag/dsparse/file_parsing/file_system.py:10

（还有第 7 个更外围的插槽 MetadataStorage,负责把「KB 配置本身」存到哪——见 §4。）

一句话直觉/类比: 把 KB 想成一台台式机主板,6 个组件就是 6 个标准插槽(内存条、显卡、硬盘……)。只要符合插槽规格,插什么牌子都能开机。而「存 KB 配置」就像把 BIOS 设置导出成一个文件,换机器时导入就还原了。

用起来什么样: 换掉默认组件,只是构造 KnowledgeBase 时多传几个参数——

# 示意，非源码
from dsrag.knowledge_base import KnowledgeBase
from dsrag.embedding import OpenAIEmbedding
from dsrag.reranker import NoReranker

kb = KnowledgeBase(
    kb_id="my_kb",
    embedding_model=OpenAIEmbedding(model="text-embedding-3-large", dimension=1024),
    reranker=NoReranker(),   # 不用重排器
    # vector_db / chunk_db / auto_context_model 不传 → 用默认实现
)

不传的插槽会落到默认实现;传了的就用你给的。默认组合是 OpenAIEmbedding + CohereReranker + OpenAIChatAPI + BasicVectorDB + BasicChunkDB + LocalFileSystem(见 knowledge_base.py:147-160 的 _initialize_components)。

---

2. 顶层全景(它大概怎么转)

怎么读这张图: 上半是「运行期」——KB 持有 6 个组件对象,摄取/查询时调它们;下半是「持久化」——每个组件能把自己序列化成 dict,KB 把这些 dict 拼成一份 JSON 存起来,下次照着 JSON 重建。

运行期(内存里的 KnowledgeBase)
┌──────────────────────────────────────────────┐
│  kb.embedding_model  ─ get_embeddings()       │
│  kb.reranker         ─ rerank_search_results()│
│  kb.auto_context_model ─ make_llm_call()      │
│  kb.vector_db        ─ add_vectors()/search() │
│  kb.chunk_db         ─ add_document()/get_*() │
│  kb.file_system      ─ save_image()/get_files()│
└───────────────┬──────────────────────────────┘
                │  每个组件都有 .to_dict()
                ▼
        components = { "embedding_model": {...}, "reranker": {...}, ... }
                │  KB 拼上 kb_metadata
                ▼
   full_data = { title, language, ..., "components": {...} }
                │  metadata_storage.save(full_data, kb_id)
                ▼
持久化   ~/dsRAG/metadata/<kb_id>.json   ← 一份纯 JSON 的「配方」
                │  下次 KnowledgeBase(kb_id=...) 启动
                ▼
        Embedding.from_dict(...) / VectorDB.from_dict(...) ... ← 照配方重建每个组件

主线走一遍(高层):

1. 首次创建 KB → _initialize_components 把没传的插槽填成默认对象(knowledge_base.py:133)。
2. _save 调每个组件的 to_dict(),拼成 components,再和 kb_metadata 合并成 full_data,交给 metadata_storage.save(knowledge_base.py:166-185)。
3. 下次打开同一个 kb_id → _load 从 JSON 读回,调各基类的 from_dict() 把 dict 变回组件对象(knowledge_base.py:187-261)。

关键点:KB 从不 import 具体实现来重建,它只调基类的 from_dict;具体是哪个子类,由 JSON 里的一个字符串字段决定。这套魔法就是下一节。

---

3. 核心原理

3.1 统一的「自动注册 + 工厂序列化」模式(全书精华)

它要解决的小问题: 存 JSON 时只能存字符串和数字,存不了「一个 Python 对象」。那重建时,光凭一份 dict,怎么知道该 new 哪个类?

思路/直觉: 每类组件做两件事——

1. 每个子类一出现就把自己登记到基类的一张「花名册」里(类名 → 类对象)。
2. 序列化时把自己的类名写进 dict;反序列化时按这个类名去花名册里查类,再把剩下的字段当构造参数喂进去。

这样 JSON 里存的不是对象,而是「类名 + 构造参数」这份配方,谁都能照着 new 回来。

第一步:自动注册。 每个基类都定义了 __init_subclass__——这是 Python 的一个钩子,任何子类被定义时自动触发,dsRAG 用它把子类塞进 subclasses 字典:

真实实现(embedding.py:23-31,Embedding.__init_subclass__):

class Embedding(ABC):
    subclasses = {}

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls.subclasses[cls.__name__] = cls   # 子类名 → 子类,自动登记

只要 class OpenAIEmbedding(Embedding) 这行代码被加载,Embedding.subclasses["OpenAIEmbedding"] 就自动有了——不需要任何手工注册表。六大基类一字不差地都用这套写法(reranker.py:10、llm.py:10、database/vector/db.py:9、database/chunk/db.py:10、file_parsing/file_system.py:16)。

第二步:to_dict 把类名写进配方。 基类的 to_dict 只吐一个 subclass_name,子类在其上追加自己的构造参数:

真实实现(embedding.py:33-34 基类 + embedding.py:72-75 子类覆写):

def to_dict(self):   # Embedding 基类
    return {"subclass_name": self.__class__.__name__, "dimension": self.dimension}

def to_dict(self):   # OpenAIEmbedding
    base_dict = super().to_dict()
    base_dict.update({"model": self.model})   # 追加自己特有的 model
    return base_dict

所以一个 OpenAIEmbedding 序列化后长这样:{"subclass_name": "OpenAIEmbedding", "dimension": 768, "model": "text-embedding-3-small"}——恰好是把它 __init__ 需要的参数原样记下来。

第三步:from_dict 照配方重建。 这是工厂:摘掉 subclass_name,查花名册,把剩下的 dict 当关键字参数喂给构造函数:

真实实现(embedding.py:36-45,Embedding.from_dict):

@classmethod
def from_dict(cls, config) -> "Embedding":
    subclass_name = config.pop("subclass_name", None)  # 取出并移除类名
    subclass = cls.subclasses.get(subclass_name)       # 查花名册
    if subclass:
        return subclass(**config)   # 剩下的字段就是构造参数
    raise ValueError(f"Unknown subclass: {subclass_name}")

config.pop 先把 subclass_name 摘掉,是因为构造函数不接受这个参数——剩下的 {"dimension":768, "model":"..."} 正好对上 OpenAIEmbedding.__init__ 的形参。

为什么这套模式值得记住: 它让「一整个 KB 配置」成为一份声明式、跨进程、可版本化的 JSON;KnowledgeBase 重建组件时只依赖抽象基类,不 import 任何具体实现,新增一个后端(比如新写一个 VectorDB 子类)零改动就能被 from_dict 认出来——只要它被 import 过、触发过 __init_subclass__。

关键细节/坑:

• 子类必须被 import 过,花名册里才有它。 subclasses 是运行时填的;如果某个后端模块从没被加载,from_dict 会抛 Unknown subclass。dsRAG 靠 dsrag/database/vector/__init__.py 之类的聚合导入把内置实现都拉进内存。
• config.pop 会就地改传入的 dict。 from_dict 直接在调用者给的 config 上 pop,属于有副作用的写法——传进去的 dict 会少掉 subclass_name 键。

3.2 Reranker 的 transform:把重排分「拉平」成绝对相关度

它要解决的小问题: RSE 算法靠 chunk 的绝对相关度来拼段落,但各家 reranker 吐的 relevance_score 分布很挤(大量结果都贴近某个值),直接用会让 RSE 失灵。

思路/直觉: 用 Beta 分布的累积分布函数(beta.cdf)做一次单调变换,把原本挤成一团的分数重新摊开到 [0,1] 上更均匀。这就是每个真实 reranker 里都有的 transform:

真实实现(reranker.py:42-48,CohereReranker.transform):

def transform(self, x):
    a, b = 0.4, 0.4          # 形状参数,不同 reranker 取值不同
    return beta.cdf(x, a, b)

VoyageReranker 用的是 a, b = 0.5, 1.8(reranker.py:85)——每家模型的原始分布不同,所以标定的形状参数也不同,这是经验值。

变换后写回哪里: rerank_search_results 把变换后的值写进结果的 similarity 字段,覆盖掉初排相似度:

真实实现(reranker.py:63-64,CohereReranker.rerank_search_results):

for i, result in enumerate(reranked_search_results):
    result['similarity'] = self.transform(reranked_similarity_scores[i])

也就是说,similarity 这个键在整个检索链路里承载的是「归一化后的绝对相关度」,下游 RSE 直接吃它。

不想重排怎么办: NoReranker 是一个「什么都不做」的合法实现。它有个开关 ignore_absolute_relevance——若为 True,给每个结果盖一个固定 similarity = 0.8(代表「中等相关」),用于嵌入模型本身的相似度不可靠时:

真实实现(reranker.py:119-123,NoReranker.rerank_search_results):

def rerank_search_results(self, query, search_results):
    if self.ignore_absolute_relevance:
        for result in search_results:
            result['similarity'] = 0.8   # 统一盖成中等相关
    return search_results

3.3 各插槽的默认实现与备选

同一个接口,dsRAG 内置了多种后端。选型时按「本地/单机 vs 托管/生产」这条线看:

插槽	默认实现	备选实现	文件位置
Embedding	OpenAIEmbedding	Cohere / VoyageAI / Ollama	dsrag/embedding.py
Reranker	CohereReranker	Voyage / NoReranker	dsrag/reranker.py
LLM	OpenAIChatAPI	Anthropic / Gemini / Ollama	dsrag/llm.py
VectorDB	BasicVectorDB	chroma / qdrant / milvus / pinecone / weaviate / postgres	dsrag/database/vector/
ChunkDB	BasicChunkDB	sqlite / postgres / dynamo	dsrag/database/chunk/
FileSystem	LocalFileSystem	S3FileSystem	dsrag/dsparse/file_parsing/file_system.py

几个值得注意的实现细节:

• Embedding 的维度自查表。 embedding.py:8 有一张 dimensionality 字典(模型名 → 维度)。Cohere/Voyage/Ollama 的构造函数在你没显式给 dimension 时会去查表,查不到就报错要你手填(embedding.py:90-96)。这个维度最终被 knowledge_base.py:164 读成 self.vector_dimension。
• LLM 统一入口是 make_llm_call。 抽象方法约定「输入 OpenAI 格式的 chat_messages,输出字符串」(llm.py:28-33)。各家实现负责把这个统一格式翻译成自家 SDK——AnthropicChatAPI 要把 system 消息单独抽出来(llm.py:77-96),GeminiAPI 要把 assistant 角色改名成 model 并把 system 指令拼进首条消息(llm.py:160-207)。接口一致、内部各自适配,是这层的核心价值。
• input_type 语义对齐。 嵌入接口带一个 input_type("query"/"document"),CohereEmbedding 会把它翻成自家的 "search_query"/"search_document"(embedding.py:100-104)——查询向量和文档向量在某些模型里要走不同分支。

---

4. 深入实现:配置怎么落盘、怎么重建

4.1 KB 启动时的三岔路

KnowledgeBase.__init__(knowledge_base.py:90-123)按「KB 是否已存在」分三种情况:

metadata_storage.kb_exists(kb_id)?
├─ 存在 且 exists_ok=True  → _load(...) 读回组件,再 _save() 重新盖章
├─ 存在 且 exists_ok=False → 报错(不许覆盖)
└─ 不存在                  → _initialize_components(...) 建默认组件 → _save()

kb_exists 是问 MetadataStorage——即 §1 提的第 7 个插槽,它决定「KB 配置这份 JSON 存到哪」。默认是 LocalMetadataStorage,把 JSON 写到 <storage_directory>/metadata/<kb_id>.json(metadata.py:32-52);备选是 DynamoDBMetadataStorage,写进一张 DynamoDB 表(metadata.py:98)。

4.2 _save:六个 to_dict 拼成一份配方

真实实现(knowledge_base.py:166-185,KnowledgeBase._save):

components = {
    "embedding_model": self.embedding_model.to_dict(),
    "reranker": self.reranker.to_dict(),
    "auto_context_model": self.auto_context_model.to_dict(),
    "vector_db": self.vector_db.to_dict(),
    "chunk_db": self.chunk_db.to_dict(),
    "file_system": self.file_system.to_dict(),
    "vlm_client": (self.vlm_client.to_dict() if getattr(self, "vlm_client", None) else None),
}
full_data = {**self.kb_metadata, "components": components}
self.metadata_storage.save(full_data, self.kb_id)

注意 full_data 把 KB 元数据(title/description/language/supp_id/created_on,见 knowledge_base.py:103-109)和 组件配方 平铺在一起——前者是「关于这个 KB 的事实」,后者是「重建它需要的零件清单」。

4.3 _load:from_dict 逐个复活 + 可选覆盖

_load(knowledge_base.py:187-261)从 JSON 拆出 components,对每个键调对应基类的 from_dict:

# 示意,截自 _load 逻辑
self.embedding_model = Embedding.from_dict(components.get("embedding_model", {}))
self.reranker = reranker if reranker else Reranker.from_dict(components.get("reranker", {}))
self.vector_db = vector_db if vector_db else VectorDB.from_dict(components.get("vector_db", {}))

一个重要设计:允许「加载时覆盖」但分安全等级。 传给 _load 的 reranker / auto_context_model 若非空,就用传入的、跳过存档里的——这两个被认为可安全替换(knowledge_base.py:199-201 的 note 明说)。而覆盖 vector_db/chunk_db/file_system 会打一条 logging.warning(knowledge_base.py:224-240),因为换掉存储层很可能和已落盘的数据不兼容。

最后一步很关键: _load 结尾重新算 self.vector_dimension = self.embedding_model.dimension(knowledge_base.py:261)——嵌入维度不存在 JSON 顶层,而是从复活出来的嵌入组件身上现取。

4.4 MetadataStorage 里的类型转换坑

DynamoDBMetadataStorage 存 KB 配置时要处理 DynamoDB 不吃 float 的毛病:save 前把所有 int/float 递归转成 Decimal(metadata.py:60-73 的 convert_numbers_to_decimal),load 后再转回来(metadata.py:76-95 的 convert_decimal_to_numbers)。转回来时还要特判:Decimal('0')/Decimal('1') 被还原成 bool——这是个有损的启发式(任何值为 0 或 1 的整数字段都会变成布尔),用它的人要留意。LocalMetadataStorage 用普通 json.dump 就没这问题。

---

5. VectorDB 与 ChunkDB:两个库,各存一半

这是本章第二条主线:为什么正文和向量要分两个库存。 答案是职责不同、访问模式不同。

维度	VectorDB	ChunkDB
存什么	向量 + 检索用的轻量元数据	chunk 正文全文 + 页码 + 标题/摘要
主要方法	add_vectors / search	add_document / get_chunk_text / get_chunk_page_numbers
检索阶段的角色	被查:算相似度、返回 top-k	不参与打分
取回阶段的角色	提供命中 chunk 的定位信息	按定位取正文、拼段落、查页码
抽象基类	database/vector/db.py:6	database/chunk/db.py:7
默认实现	BasicVectorDB(pickle)	BasicChunkDB(pickle)

它们存的元数据不一样——这是分家的关键。 摄取时,add_document.py 把同一批 chunk 写进两个库,但写的内容有别:

• 写进 VectorDB 的 metadata:doc_id、chunk_index、chunk_text、chunk_header、页码,外加用户自定义 metadata(add_document.py:191-207,add_vectors_to_db)。这是「检索命中后拿来定位和重排」的最小信息。
• 写进 ChunkDB 的每条记录:chunk_text、document_title、document_summary、section_title、section_summary、chunk_page_start/end、is_visual(add_document.py:161-178,add_chunks_to_db)。这是「取回时要拼出完整段落和页码」的全量信息。

为什么不合并? 因为检索路径和取回路径的访问方式完全不同:

查询流(见 knowledge_base.py:_search / query)
  query ──embedding──▶ query_vector
                          │
                          ▼
              VectorDB.search(query_vector, top_k)   ← 只有向量库参与打分
                          │ 返回 [{metadata:{doc_id,chunk_index,...}, similarity}]
                          ▼
              Reranker.rerank_search_results          ← 归一化成绝对相关度(§3.2)
                          │
                          ▼
                     RSE 算出「要哪些 chunk 区间」
                          │  拿 (doc_id, chunk_start..chunk_end)
                          ▼
   ChunkDB.get_chunk_text(doc_id, i) / get_chunk_page_numbers(...)  ← 正文库按坐标取回
                          │
                          ▼
                   拼成带页码的完整段落

• 检索时 只碰 VectorDB:_search(knowledge_base.py:787-797)把查询向量交给 vector_db.search,再交给 reranker。ChunkDB 完全不参与打分。
• 取回时 只碰 ChunkDB:RSE 决定了要 doc_id 的第 chunk_start..chunk_end 段后,_get_segment_content_from_database(knowledge_base.py:821-856)逐个 chunk_db.get_chunk_text 取正文、get_chunk_page_numbers 取页码,拼成段落。

契约在基类里写死。 VectorDB.search 的 docstring(database/vector/db.py:46-61)明确要求返回 {'metadata': {...}, 'similarity': ...} 这个形状——所以不管底层是 pickle 还是 Pinecone,reranker 和 RSE 拿到的结构都一样。ChunkDB 则用一堆细粒度 getter(get_chunk_text/get_is_visual/get_chunk_page_numbers/get_document_title/get_document_summary,database/chunk/db.py:44-98)约定「按坐标点取字段」的接口。

默认实现有多朴素: BasicVectorDB 就是把 (vectors, metadata) 两个 list pickle.dump 成一个 .pkl,search 用 sklearn 的 cosine_similarity 全量算一遍再排序(database/vector/basic_db.py:36-64、117-130)。BasicChunkDB 是一个 {doc_id: {chunk_index: {...}}} 的嵌套字典,同样 pickle 落盘(database/chunk/basic_db.py:28-39、133-142)。能跑、易懂、不抗并发——文档明说默认实现非线程安全(knowledge_base.py:645-647)。生产要换成 SQLiteDB(把上述字段建成真实表列,database/chunk/sqlite_db.py:21-36)或托管向量库。

---

6. 巧妙之处(可借鉴的技术)

• 用 __init_subclass__ 做零样板的自动注册。 六个基类共用同一段 3 行代码,子类「一被定义就登记」,新增后端无需改任何注册表——这是全库最值得抄走的模式(embedding.py:29-31)。
• 序列化的是「配方」不是「对象」。 to_dict 存的恰好是构造参数,from_dict 用 subclass(**config) 复活。整个 KB 配置因此变成一份人类可读、可版本化的 JSON(knowledge_base.py:166-185)。
• similarity 字段承载「绝对相关度」这一跨层契约。 reranker 用 beta.cdf 把各家模型挤成团的分数拉平后写回 similarity,下游 RSE 无需知道用了哪个 reranker(reranker.py:42-64)。
• 向量/正文分库,按访问模式切分而非按数据切分。 检索只碰向量库、取回只碰正文库,两条路径互不干扰、可各自替换后端(knowledge_base.py:787-856)。

7. 边界与局限

• 默认组件不抗并发。 BasicVectorDB/BasicChunkDB 是「读全量→改→pickle 全量写回」,add_documents(max_workers>1) 时会丢数据,文档已警告(knowledge_base.py:645-647)。
• from_dict 依赖「子类已被 import」。 花名册是运行时填的;冷启动若没加载某后端模块,from_dict 直接 Unknown subclass 报错。
• 加载时覆盖存储层有风险。 覆盖 vector_db/chunk_db/file_system 只打 warning 不拦截(knowledge_base.py:224-240),换成和已有数据不兼容的实现会静默出问题。
• DynamoDB 元数据的 0/1→bool 特判有损。 任何值恰为 0 或 1 的整数字段,load 时会被还原成布尔(metadata.py:85-87)。

8. 横向对比

• 和 摄取管线 的关系: 摄取产出的 chunk,正是 add_chunks_to_db/add_vectors_to_db 分别写进本章两个库的原料。
• 和 RSE 查询 的关系: RSE 吃的「绝对相关度」由本章 §3.2 的 reranker transform 生产;RSE 定出 chunk 区间后,由本章 §5 的 ChunkDB 取回路径落地成段落。
• 和 对话层 的关系: 对话层带引用的回答,页码来自 ChunkDB 的 get_chunk_page_numbers——即本章正文库存下来的字段。

9. 代码地图(导航索引)

主题	文件路径	符号名
自动注册钩子(六处同款)	dsrag/embedding.py	Embedding.__init_subclass__
工厂反序列化	dsrag/embedding.py	Embedding.from_dict
序列化基类 + 子类追加	dsrag/embedding.py	Embedding.to_dict / OpenAIEmbedding.to_dict
嵌入维度自查表	dsrag/embedding.py	dimensionality
重排分归一化	dsrag/reranker.py	CohereReranker.transform
空重排器 / 固定分	dsrag/reranker.py	NoReranker.rerank_search_results
LLM 统一入口	dsrag/llm.py	LLM.make_llm_call
Anthropic/Gemini 格式适配	dsrag/llm.py	AnthropicChatAPI.make_llm_call / GeminiAPI._convert_messages
组件初始化(填默认)	dsrag/knowledge_base.py	KnowledgeBase._initialize_components
配置落盘	dsrag/knowledge_base.py	KnowledgeBase._save
配置重建 + 覆盖规则	dsrag/knowledge_base.py	KnowledgeBase._load
KB 配置存储插槽	dsrag/metadata.py	MetadataStorage / LocalMetadataStorage / DynamoDBMetadataStorage
DynamoDB 数字↔Decimal	dsrag/metadata.py	convert_numbers_to_decimal / convert_decimal_to_numbers
向量库接口 + 返回契约	dsrag/database/vector/db.py	VectorDB.search
向量库默认实现(pickle)	dsrag/database/vector/basic_db.py	BasicVectorDB.save / search
正文库接口	dsrag/database/chunk/db.py	ChunkDB.get_chunk_text / get_chunk_page_numbers
正文库默认实现(pickle)	dsrag/database/chunk/basic_db.py	BasicChunkDB.add_document
正文库表结构(生产)	dsrag/database/chunk/sqlite_db.py	SQLiteDB.columns
文件系统插槽	dsrag/dsparse/file_parsing/file_system.py	FileSystem / LocalFileSystem / S3FileSystem
分别写两库	dsrag/add_document.py	add_chunks_to_db / add_vectors_to_db
检索只碰向量库	dsrag/knowledge_base.py	KnowledgeBase._search
取回只碰正文库	dsrag/knowledge_base.py	KnowledgeBase._get_segment_content_from_database