跳到主要内容

数据截至 (上游 commit 9ad3e4e49aa3)

把文档装进库:索引与切分管线

30 秒导读: 这一章讲 Kotaemon 的摄入路径——你上传一个 PDF/Word/网页,它是怎么被读成文本、切成小块(chunk)、再同时写进两套存储(一套供全文检索、一套供向量检索),并在 SQL 里记下"哪个文件产生了哪些块"的。读完你能完整讲清"一份文件从上传到可被检索"的每一步。检索和问答本身不在这章(见 03-hybrid-retrieval 和 04-qa-citation)。

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

一句话定义: 索引管线(indexing pipeline)是把"人能看的文件"变成"机器能检索的数据"的流水线。

它解决什么问题? RAG(检索增强生成,先查资料再让大模型答)要能工作,前提是资料已经被切成小块、算好向量、存进库里。你不可能每次提问都现场去读一遍 100 页的 PDF。所以必须有一步"离线预处理":读文件 → 切块 → 入库。这一步就是本章的主角。

给谁用? Kotaemon 是一个开源的 RAG 问答 UI。终端用户在网页上点"上传文件",背后就触发这条管线。开发者也能通过配置换掉其中任意一环(换 PDF 解析器、换切块大小、换向量库)。

用起来什么样? 用户视角其实很朴素——拖一个文件进去,界面上滚动打印:

Indexing [1/1]: report.pdf
 => Converting report.pdf to text
 => Converted report.pdf to text
 => [report.pdf] Processed 42 chunks
 => [report.pdf] Created embedding for 42 chunks
 => Finished indexing report.pdf

这些正是管线在 stream() 里一步步 yield 出来的调试消息(libs/ktem/ktem/index/file/pipelines.py:648-655)。文件从此"进了库",可以被提问检索到。

一句话直觉: 把索引想成图书馆入库——先把书拆成一页页(切块),既把书放进书架(向量库,按"意思相近"找),又在目录卡片盒里登记(全文库 + SQL 表,按"字面词"找、按"哪本书"找)。同一批内容,两种索引方式并存。

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

整条路径由两层管线组成:上层 IndexDocumentPipeline 是工厂/调度,负责"这份文件该用哪套处理法";下层 IndexPipeline 是干活的,负责单个文件的读、切、存。

                      上传的文件路径 / URL
                              │
              ┌───────────────▼────────────────┐
              │   IndexDocumentPipeline (工厂)   │   selectreader_mode 选 reader
              │   route(file_path)              │   按扩展名/URL 分派
              └───────────────┬────────────────┘
                              │  为每个文件造一个
                              ▼
              ┌────────────────────────────────┐
              │   IndexPipeline (单文件流水线)    │
              │                                │
              │  ① loader.load_data() 读成文本   │
              │  ② splitter 切成 chunk          │
              │  ③ 双写入库                      │
              └───────┬───────────────┬────────┘
                      │               │
         ┌────────────▼───┐   ┌───────▼───────────┐
         │  doc_store(DS) │   │ vector_store(VS)  │
         │  存原文,供全文  │   │ 存 embedding,供   │
         │  检索           │   │ 向量检索           │
         └────────────────┘   └───────────────────┘
                      │               │
                      └───────┬───────┘
                              ▼
                  SQL: Source 表(登记文件)
                       Index 表(记 file↔chunk 关系)

怎么读这张图: 从上到下就是一份文件的生命周期。左右两个存储是同一批 chunk 的两份索引,最底下的 SQL 表是把它们串起来的"账本"。

各部件一句话职责:

部件干什么在哪
IndexDocumentPipeline工厂:按 reader_mode 和文件类型选管线libs/ktem/ktem/index/file/pipelines.py:659
IndexPipeline单文件流水线:读→切→双写libs/ktem/ktem/index/file/pipelines.py:330
TokenSplitter按 token 数把文本切成 chunklibs/kotaemon/kotaemon/indices/splitters/__init__.py:10
VectorIndexing真正执行"写 doc_store + 写 vector_store"libs/kotaemon/kotaemon/indices/vectorindex.py:21
BaseFileIndexIndexing索引管线的抽象契约,声明能拿到哪些资源libs/ktem/ktem/index/file/base.py:36
Source / Index 表SQL 里登记文件、记 chunk 归属libs/ktem/ktem/index/file/index.py:91,115

主线走一遍(高层): 文件路径进来 → 工厂 route() 挑出合适的 reader 和 splitter,拼出一个 IndexPipeline → 这个管线读文件成文本、切成 chunk → 把每个 chunk 同时写进 doc_store 和 vector_store,并在 Index 表记下"这个 chunk 属于这个文件" → 收尾统计 token 数 → 文件可被检索。

3. 核心原理(逐个机制,由浅入深)​

3.1 双层管线:工厂选料,流水线干活​

要解决的小问题: 不同文件类型(PDF、Word、图片、网页)需要不同的读取器;而且同一种 PDF,用户可能想用不同引擎解析(开源默认 vs Adobe vs Azure)。把"选择逻辑"和"执行逻辑"混在一起会很乱。

思路: 分成两层。上层 IndexDocumentPipeline 只做决策——它自己不读文件,而是像工厂一样造出一个配好 reader 和 splitter 的下层 IndexPipeline。文档注释里说得很直白:

"This method is essentially a factory to decide which indexing pipeline to use." —— libs/ktem/ktem/index/file/pipelines.py:660-661

上层入口 stream() 遍历所有待索引文件,对每个文件先 route() 出一条管线,再委托它干活(pipelines.py:836-841):

# 示意,贴近源码。重点看:route 造管线,再 yield from 委托
pipeline = self.route(file_path)                       # 选料:造一条单文件管线
file_id, docs = yield from pipeline.stream(            # 干活:委托它读→切→存
    file_path, reindex=reindex, **kwargs
)

用 yield from 把下层管线的进度消息透传给 UI——这就是你在界面上看到那些 => Processed N chunks 滚动打印的原因。整个 stream() 用异常包裹每个文件,单个失败不会中断整批(pipelines.py:852-864),失败的文件在返回列表里对应位置记为 None + 错误消息。

3.2 选 reader:reader_mode + 按扩展名分派​

要解决的小问题: ".pdf" 该交给谁读?".docx" 呢?URL 又该怎么办?

两步决策:

第一步——reader_mode 决定 PDF 用哪个引擎。 IndexDocumentPipeline 有个 reader_mode 参数(默认 "default"),由一个 @Param.auto 惰性构建的 readers 字典承接(pipelines.py:675-711)。它先拷贝一份默认扩展名映射 KH_DEFAULT_FILE_EXTRACTORS,再按 reader_mode 覆盖 PDF(及图片)对应的 reader:

reader_mode效果
default用开源默认(PDF 走 PDFThumbnailReader)
adobePDF 换成 adobe_reader(带图/表抽取)
azure-diPDF 换成 azure_reader(Azure 文档智能)
doclingPDF 换成 docling_reader
paddle-structPDF + 各图片格式换成 paddle_struct_reader
paddle-vlPDF + 各图片格式换成 paddle_vl_reader(视觉大模型解析)

@Param.auto(depends_on="reader_mode") 是组件模型里的机制:当 reader_mode 变了,readers 会自动重算,不用手写刷新逻辑。

默认映射本身覆盖常见办公格式(libs/kotaemon/kotaemon/indices/ingests/files.py:48-64):.xlsx→PandasExcelReader、.docx/.pptx/.doc/图片→unstructured、.html→HtmlReader、.pdf→PDFThumbnailReader、.txt/.md→TxtReader 等。

第二步——route() 按扩展名/URL 取 reader。 真正分派在 route()(pipelines.py:757-803):

# 示意,贴近源码。重点看:URL 走 web_reader,否则按扩展名查表,查不到兜底 unstructured
if self.is_url(file_path):
    reader = web_reader                          # http(s):// → 网页读取器
else:
    ext = file_path.suffix.lower()               # 取扩展名 .pdf/.docx…
    reader = self.readers.get(ext, unstructured) # 查表,查不到就用 unstructured 兜底

选好 reader 后,route() 顺手把 TokenSplitter 一起装配进新造的 IndexPipeline,并把所有资源(Source/Index 表、VS、DS、FSPath、user_id、embedding)透传下去(pipelines.py:784-801)。

3.3 切块:TokenSplitter 怎么把文本变成 chunk​

要解决的小问题: 一份长文档不能整块塞进检索——太大,向量表达不精确,也超模型上下文。要切成大小合适、还带一点重叠的小块。

思路: 按 token 数(不是字符数、不是行数)切,让每块大小可控。默认参数在 route() 里(pipelines.py:786-791):

参数默认值含义
chunk_size1024每块目标 token 数
chunk_overlap256相邻块重叠 token 数(避免切断语义)
separator"\n\n"首选切分点(段落)
backup_separators["\n", ".", "​"]段落切不动时的退路

chunk_size / chunk_overlap 可被 index 配置或 flowsettings.py 里的开发者设置覆盖(pipelines.py:763-766,回退链见 dev_settings() at pipelines.py:57-76)。

TokenSplitter 本身是个薄封装,底层复用 llama-index 的 TokenTextSplitter(libs/kotaemon/kotaemon/indices/splitters/__init__.py:10-28)。它继承自 DocTransformer,所以是一个可直接当函数调用的组件——self.splitter(text_docs) 就切完了。

切块发生在 handle_docs()(pipelines.py:374-385)。这里有个关键细节:只有文本类文档被切,图片/缩略图/表格等非文本内容原样保留:

# 示意,贴近源码。重点看:文本才切,其余原样,最后合并成待入库列表
if self.splitter:
    all_chunks = self.splitter(text_docs)   # 只切纯文本
else:
    all_chunks = text_docs
# …给每个 chunk 关联同页缩略图的 doc_id(供检索时展示原图)…
to_index_chunks = all_chunks + non_text_docs + thumbnail_docs   # 合并

handle_docs() 先按 doc.metadata["type"] 把读出来的文档分成 text / thumbnail / 其它三类(pipelines.py:360-367),再只对 text 那一坨切块。

3.4 双写入库:同一批 chunk,两套存储​

这是整章最核心的一步:同一批 chunk 会被写进两个地方,分别服务两种检索方式。

为什么要两套? 检索章(03)会讲"混合检索":既要按语义找(向量库),又要按关键词字面找(全文库)。所以入库时就得两份都备好。

谁来写? IndexPipeline 委托 VectorIndexing(vectorindex.py:21)。它由 @Node.auto 惰性构建,把 VS/DS/embedding 一并注入(pipelines.py:348-352)。VectorIndexing 暴露两个独立方法:

  • add_to_docstore(docs) —— 把原文 chunk 加进 doc_store,供全文检索(vectorindex.py:79-82)。
  • add_to_vectorstore(docs) —— 先对 chunk 算 embedding,再把向量加进 vector_store,ids 用 chunk 的 doc_id(vectorindex.py:84-93):
# 示意,贴近源码。重点看:先算 embedding,再连同 doc_id 一起写向量库
embeddings = self.embedding(docs)                 # 一批 chunk → 一批向量
self.vector_store.add(
    embeddings=embeddings,
    ids=[t.doc_id for t in docs],                 # 用 chunk 自己的 id 当向量 id
)

注意 doc_store 存的是完整原文(检索命中后要拿回原文给 LLM 看),vector_store 存的只是向量 + id;两边靠同一个 doc_id 对齐。这就是为什么向量库命中后还要回 doc_store 拿原文(见 03)。

独立方法给了并发的余地。 IndexPipeline.handle_docs() 把两次写入拆成两趟(先全部写 docstore、再分批写 vectorstore),并允许把向量写入丢到后台线程跑(pipelines.py:414-421,由 run_embedding_in_thread / "quick index mode" 控制),这样计算 embedding 的耗时不阻塞 docstore 落库。

VectorIndexing.run()(vectorindex.py:95-113)则是"一把梭"的入口:接收 str 或 Document 列表,依次 add_to_vectorstore → add_to_docstore → write_chunk_to_file。ktem 的文件管线没直接用 run(),而是调更细粒度的 add_to_* 方法以便分批和记账。

3.5 SQL 账本:Source 表与 Index 表​

要解决的小问题: 向量库里躺着一堆 chunk 向量,doc_store 里躺着一堆 chunk 原文——但"哪些 chunk 属于 report.pdf?""哪些是向量、哪些是原文?"这些关系存哪?答案是 SQL。

两张表(每个 index 一套,表名带 index id):

表一行代表关键列
Source一个上传的文件id(文件 id)、name、path(文件内容的 sha256)、size、note(JSON)
Index一条 file→chunk 关系source_id(文件 id)、target_id(chunk 的 doc_id)、relation_type

表定义在 libs/ktem/ktem/index/file/index.py:Source at :91-114(private 变体 at :63-89),Index at :115-126。

relation_type 是精髓。 同一个 chunk 会在 Index 表里产生两行:一行 relation_type="document"(对应 doc_store),一行 relation_type="vector"(对应 vector_store)。写入分别发生在:

  • handle_chunks_docstore():写 doc_store 后,记 relation_type="document"(pipelines.py:426-443)。
  • handle_chunks_vectorstore():写 vector_store 后,记 relation_type="vector"(pipelines.py:445-464)。
# 示意,贴近源码。重点看:每个 chunk 记一行 file→chunk 关系,类型区分它在哪个库
nodes.append(
    self.Index(
        source_id=file_id,          # 属于哪个文件
        target_id=chunk.doc_id,     # 是哪个 chunk
        relation_type="document",   # 在 doc_store(vector 分支则为 "vector")
    )
)

这个账本让下游能干两件事:

  • 检索时反查 chunk。 拿到用户选的文件 id,查 Index 表 relation_type=="document" 就得到该文件所有 chunk 的 id,把检索范围限定在这些文件内(检索侧用法见 pipelines.py:148-153,03 详述)。
  • 删除时精确清理。 delete_file() 按 relation_type 把 chunk id 分成 vs_ids / ds_ids 两拨,分别去 vector_store 和 doc_store 删干净,再删 SQL 行(pipelines.py:574-597)。

3.6 一条完整路径:stream() 从头到尾​

把前面几节串起来,IndexPipeline.stream()(pipelines.py:604-656)就是单个文件的完整生命周期:

① 去重检查     get_id_if_exists() —— 按文件名查 Source 表
              已存在且 reindex=False → 报错停下
              已存在且 reindex=True  → delete_file() 清旧,再重存
② 登记文件     store_file() —— 存文件内容(sha256 命名)到 FSPath,
              往 Source 表插一行,拿到 file_id
③ 读成文本     loader.load_data(file_path) —— reader 干活,出 list[Document]
④ 切+双写     handle_docs() —— 切块 → 写 doc_store + 写 vector_store
                            → 往 Index 表记 document/vector 两类关系
⑤ 收尾统计     finish() —— 数 token 数、记下用的 loader 名,写回 Source.note

去重靠内容哈希。 store_file() 读文件内容算 sha256 当存储文件名(pipelines.py:524-527),同时把 sha256 记进 Source 表的 path 列;文件名冲突则靠 get_id_if_exists() 按 Source.name 查(pipelines.py:466-490)。URL 走类似的 store_url(),对 URL 字符串算哈希(pipelines.py:492-513)。

finish() 补统计信息。 索引完再查一遍该文件的所有 document chunk,用 tiktoken 数 token 总数,连同 loader 类名写进 Source.note(JSON 列),供 UI 展示(pipelines.py:541-568)。

4. 深入实现:BaseFileIndexIndexing 契约与资源注入​

IndexDocumentPipeline 继承自 BaseFileIndexIndexing(libs/ktem/ktem/index/file/base.py:36),这个基类定义了"一条文件索引管线该长什么样"的契约:

它承诺实现: run()(批量索引,返回 file_ids + errors)、stream()(流式版,多返回 indexed docs)、以及类方法 get_pipeline()(返回配好的实例)、get_user_settings()(返回 UI 设置项)。

它保证能拿到的资源(都是 Param,base.py:51-59):

资源是什么
SourceSQLAlchemy Source 表类
IndexSQLAlchemy Index 表类
VS向量存储 BaseVectorStore
DS文档存储 BaseDocumentStore
FSPath文件原件的存储目录
user_id / private / chunk_size / chunk_overlap用户与切块配置

资源从哪来? 上层 FileIndex._setup_resources() 为每个 index 创建独立的表、向量库、doc_store、存储目录(index.py:51-163;表名形如 index__{id}__source)。然后 get_indexing_pipeline() 把这些资源注入到管线实例上(index.py:449-458):

# 示意,贴近源码。重点看:工厂造好 obj 后,把该 index 的资源逐一挂上去
obj = self._indexing_pipeline_cls.get_pipeline(stripped_settings, self.config)
obj.Source = self._resources["Source"]     # 这个 index 专属的表/库
obj.Index = self._resources["Index"]
obj.VS = self._vs
obj.DS = self._docstore
obj.FSPath = self._fs_path

这种"基类声明契约 + 外部注入资源"的设计,让同一套索引逻辑能服务多个隔离的知识库(每个 index 一套自己的表和存储),也让开发者能整体替换索引管线类(_setup_indexing_cls() 的回退链,index.py:165-197)。这正是组件模型里 Param / 依赖注入思想的实际落地。

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

  • 工厂 + 流水线两层分离。 决策(选 reader/splitter)与执行(读/切/存)解耦,换解析引擎只动工厂,换存储只动流水线。见 IndexDocumentPipeline.route() pipelines.py:757。
  • 同一批 chunk 双写、用 doc_id 对齐两库。 vector_store 只存向量 + id,doc_store 存原文,靠 chunk 的 doc_id 关联——检索命中向量后回 doc_store 取原文即可。见 vectorindex.py:84-93。
  • relation_type 把"一份 chunk 在两个库"显式记账。 让删除能精确分拣、检索能按库反查,而不用扫存储本身。见 pipelines.py:426-464、574-597。
  • embedding 可丢后台线程。 "quick index mode" 下向量计算不阻塞 docstore 落库,UI 更快返回。见 pipelines.py:414-419。
  • 内容 sha256 去重 + 文件名唯一约束双保险。 既避免同内容重复存盘,又能按文件名判"是否已索引"。见 pipelines.py:515-539。
  • 只切文本、保留图/表/缩略图。 切块只作用于 text 类型,图片和缩略图原样入库并与文本 chunk 关联,为带图检索留出通道。见 pipelines.py:360-385。

6. 边界与局限​

  • run() 未实现,只能流式跑。 IndexPipeline.run() 和 IndexDocumentPipeline.run() 都直接 raise NotImplementedError(pipelines.py:599-602、805-808);实际索引一律走 stream()。
  • 去重按文件名判定。 非 private 模式下 Source 表的 name 唯一;换句话说,同名不同内容的文件不 reindex 会被判为"已索引"而拒绝(pipelines.py:611-619)。
  • 图/表抽取依赖外部服务。 adobe / azure-di 等 reader_mode 需要相应 API 端点与凭据,默认开源模式不含这些。
  • write_chunk_to_file 是可选副产物。 仅当配置了 KH_CHUNKS_OUTPUT_DIR 才把 chunk 落成 markdown 文件,用于调试/审查,不影响检索(vectorindex.py:45-77)。
  • 本章止于"可检索"。 怎么把这些库里的 chunk 查出来、并行向量+全文再重排,是 03-hybrid-retrieval;怎么组装证据、带引证生成答案是 04-qa-citation。

7. 横向对比​

同 shelf 的其它 RAG/agent 项目在"摄入"这步的取舍各异:有的把切块策略做成可插拔的层级/语义切分,有的只用单一向量库(不做全文并存)。Kotaemon 的特色是双存储 + SQL 关系账本——用一点存储冗余和一张关系表,换来后续"混合检索"和"精确删除/反查"的便利。这与它整体"一切皆可组合的 BaseComponent"哲学一致:每一环(reader/splitter/store)都是可替换的组件。

8. 代码地图(导航索引)​

主题文件路径符号名
索引工厂:选 reader、分派管线libs/ktem/ktem/index/file/pipelines.pyIndexDocumentPipeline
reader_mode → readers 字典libs/ktem/ktem/index/file/pipelines.pyIndexDocumentPipeline.readers
按扩展名/URL 分派libs/ktem/ktem/index/file/pipelines.pyIndexDocumentPipeline.route
批量流式入口libs/ktem/ktem/index/file/pipelines.pyIndexDocumentPipeline.stream
单文件流水线libs/ktem/ktem/index/file/pipelines.pyIndexPipeline
切块 + 分类 + 双写编排libs/ktem/ktem/index/file/pipelines.pyIndexPipeline.handle_docs
写 doc_store + 记 document 关系libs/ktem/ktem/index/file/pipelines.pyIndexPipeline.handle_chunks_docstore
写 vector_store + 记 vector 关系libs/ktem/ktem/index/file/pipelines.pyIndexPipeline.handle_chunks_vectorstore
单文件生命周期(去重→存→读→切写→收尾)libs/ktem/ktem/index/file/pipelines.pyIndexPipeline.stream
内容哈希去重与登记libs/ktem/ktem/index/file/pipelines.pyIndexPipeline.store_file / get_id_if_exists
精确删除(按 relation_type 分拣)libs/ktem/ktem/index/file/pipelines.pyIndexPipeline.delete_file
双写核心:doc_store + vector_storelibs/kotaemon/kotaemon/indices/vectorindex.pyVectorIndexing
算 embedding 并写向量库libs/kotaemon/kotaemon/indices/vectorindex.pyVectorIndexing.add_to_vectorstore
写原文到 doc_storelibs/kotaemon/kotaemon/indices/vectorindex.pyVectorIndexing.add_to_docstore
Token 切块器libs/kotaemon/kotaemon/indices/splitters/__init__.pyTokenSplitter
默认扩展名→reader 映射libs/kotaemon/kotaemon/indices/ingests/files.pyKH_DEFAULT_FILE_EXTRACTORS
索引管线契约与资源声明libs/ktem/ktem/index/file/base.pyBaseFileIndexIndexing
建表/建库/建目录(每 index 一套)libs/ktem/ktem/index/file/index.pyFileIndex._setup_resources
资源注入到管线实例libs/ktem/ktem/index/file/index.pyFileIndex.get_indexing_pipeline