RAGFlow 文档解析与存储核心流程分析报告

📑 目录

整体架构概览
文档解析层 (deepdoc)
RAG Flow 处理管道
分块策略详解 (TokenChunker / TitleChunker)
分词与向量化 (Tokenizer / Embedding)
Elasticsearch 存储结构详解
Chunk 文档完整字段字典
数据写入流程 (insert_chunks)
检索与重排序流程
更新与删除机制
完整数据流总结

1. 整体架构概览

RAGFlow 的文档处理分为四个核心层次，从原始文件输入到最终可检索的向量化 Chunk，经过一条完整的 Pipeline：

┌──────────────────────────────────────────────────────────────────┐
│                      文档处理完整数据流                            │
├──────────────────────────────────────────────────────────────────┤
│                                                                  │
│  📄 原始文件                                                      │
│    │                                                             │
│    ▼                                                             │
│  ┌──────────┐    ┌───────────────┐    ┌──────────────┐           │
│  │ 1. deepdoc │───▶│ 2. rag/flow   │───▶│ 3. task_exec │───▶ ES  │
│  │   解析层    │    │   Pipeline    │    │   utor.py    │         │
│  └──────────┘    └───────────────┘    └──────────────┘           │
│       │                 │                    │                    │
│       ▼                 ▼                    ▼                    │
│  结构化Sections    Chunks +            插入/更新/删除              │
│  + Tables/Figs     Embeddings          ES 文档操作                 │
│                                                                  │
│  ┌──────────────────────────────────────────────────────────┐    │
│  │                    4. ES 向量存储                          │    │
│  │  • Index: ragflow_{tenant_id}                             │    │
│  │  • 17 动态模板 (Dynamic Templates)                         │    │
│  │  • 支持 KNN 向量检索 + BM25 全文检索 + 混合融合             │    │
│  └──────────────────────────────────────────────────────────┘    │
│                                                                  │
└──────────────────────────────────────────────────────────────────┘

层次	核心模块	关键文件	职责
Layer 1	deepdoc 解析	`deepdoc/parser/*.py`	从 PDF/DOCX/XLSX/PPT/HTML/Markdown 等格式中提取结构化文本、表格和图片
Layer 2	rag/flow Pipeline	`rag/flow/pipeline.py`	将解析结果编排为 File→Parser→Chunker→Tokenizer 的 DAG 流程
Layer 3	Task Executor	`rag/svr/task_executor.py`	执行实际的 chunk 构建、关键词提取、问题生成、向量化、ES 写入
Layer 4	ES 存储	`rag/utils/es_conn.py`	Elasticsearch 的 CRUD、KNN 搜索、混合检索、重排序

2. 文档解析层 (deepdoc)

2.1 支持的解析器一览

RAGFlow 通过 deepdoc/parser/ 目录下的解析器支持 16 种以上的文档格式，并支持多种外部解析服务：

解析器类	文件	输入格式	解析方式
`RAGFlowPdfParser`	`pdf_parser.py`	PDF	DeepDOC pdfplumber + OCR + 版面分析 + 表格识别
`PlainParser`	`pdf_parser.py`	PDF	Plain 纯文本提取
`VisionParser`	`pdf_parser.py`	PDF	Vision 视觉 LLM 直接理解
`RAGFlowDocxParser`	`docx_parser.py`	DOCX	python-docx 段落/表格提取
`RAGFlowExcelParser`	`excel_parser.py`	XLSX, XLS, CSV	openpyxl/pandas 行列解析
`RAGFlowPptParser`	`ppt_parser.py`	PPTX	python-pptx 形状/文本/表格提取
`RAGFlowTxtParser`	`txt_parser.py`	TXT	纯文本读取
`RAGFlowHtmlParser`	`html_parser.py`	HTML	HTML 解析
`RAGFlowJsonParser`	`json_parser.py`	JSON, JSONL	JSON 结构化解析
`RAGFlowMarkdownParser`	`markdown_parser.py`	Markdown	Markdown AST 解析
`RAGFlowEpubParser`	`epub_parser.py`	EPUB	EPUB 电子书解析
`PaddleOCRParser`	`paddleocr_parser.py`	PDF	外部 PaddleOCR API
`DoclingParser`	`docling_parser.py`	PDF	外部 Docling
`MinerUParser`	`mineru_parser.py`	PDF	外部 MinerU API
`OpenDataLoaderParser`	`opendataloader_parser.py`	PDF	外部 OpenDataLoader API
`TCADPParser`	`tcadp_parser.py`	PDF	外部腾讯云文档解析

2.2 PDF 解析核心流程 (最复杂路径)

PDF 通过 RAGFlowPdfParser 的 __call__ 方法执行 8 个阶段的深度解析：

RAGFlowPdfParser.__call__()

📥 PDF 文件
  │
  ├── Step 1: __images__()
  │   ├── pdfplumber 页面渲染 (72*zoomin DPI, 默认 zoomin=3)
  │   ├── pdfplumber 字符级元数据提取 (chars, dedupe_chars)
  │   ├── 乱码检测 (PUA字符比例 > 0.3 或 字体编码映射异常)
  │   └── OCR 文本检测 + 识别 (ONNX 模型: det.onnx + rec.onnx)
  │       └── 多GPU并行: asyncio + PARALLEL_DEVICES
  │
  ├── Step 2: _layouts_rec()
  │   └── LayoutRecognizer (ONNX: layout.onnx / YOLOv10)
  │       ├── 10种版面类型分类: Text, Title, Figure, Figure Caption,
  │       │   Table, Table Caption, Header, Footer, Reference, Equation
  │       └── 垃圾版面过滤 (header/footer/reference 默认丢弃)
  │
  ├── Step 3: _table_transformer_job()
  │   ├── 表格自动旋转矫正 (auto_rotate_tables, 4方向OCR评分)
  │   ├── TableStructureRecognizer (TSR): 表格行列结构识别
  │   │   ├── 列检测 / 行检测 / 表头检测 / 跨单元格检测
  │   │   └── 每个文本框标记: R(row), C(col), H(header), SP(span)
  │   └── XGBoost 上下行拼接判断 (updown_concat_xgb.model)
  │
  ├── Step 4: _text_merge()
  │   ├── K-Means 多栏识别 (x0 坐标聚类, 轮廓系数选最优K)
  │   └── 同行水平合并 (同页/同栏/同版面类型/垂直距离近)
  │
  ├── Step 5: _concat_downward()
  │   └── Y坐标排序 (实际逻辑较简单)
  │
  ├── Step 6: _filter_forpages()
  │   └── 目录页检测与过滤 (点线页码密度判断)
  │
  ├── Step 7: _extract_table_figure()
  │   ├── 分离 Table/Figure 与正文流
  │   ├── 标题-图表匹配 (欧几里得距离最近)
  │   ├── 图表图片裁剪
  │   └── TableStructureRecognizer.construct_table() → HTML/NL描述
  │
  └── Step 8: __filterout_scraps()
      ├── 碎片过滤 (窄/矮片段去除)
      ├── 合并剩余行
      └── 添加位置标签: @@page\tx0\tx1\ttop\tbottom##

📤 输出: (sections_text, [(table_image, table_html_or_desc), ...])

2.3 OCR 系统

位于 deepdoc/vision/ocr.py，基于 PaddleOCR 的 ONNX 模型：

TextDetector: DB (Differentiable Binarization) 文本检测，阈值 0.3，输出候选框
TextRecognizer: CTC 文本识别，输入 48×320，批量推理 (batch=16)，字符字典来自 ocr.res
OCR: 编排检测+识别，支持旋转文字矫正 (长宽比≥1.5 尝试90°旋转)

2.4 输出数据结构

所有解析器统一输出两个部分：

sections: 文本内容列表，每个元素可能是纯文本字符串或 (text, doc_type_kwd, positions) 元组
tables: (PIL.Image, [html_or_text]) 元组列表，保存表格/图片及其结构化描述

DeepDOC PDF 解析器的 parse_into_bboxes() 方法额外返回每个 box 的完整坐标和版面类型。

3. RAG Flow 处理管道

3.1 管道组件链

rag/flow/pipeline.py 中的 Pipeline 类继承自 Graph，将一组组件按 DAG 顺序执行。标准文档处理流程包含5 个组件：

┌────────┐    ┌────────┐    ┌──────────────┐    ┌───────────┐    ┌────────────┐
│  File  │───▶│ Parser │───▶│ TokenChunker  │───▶│ Tokenizer │───▶│ Extractor  │
│        │    │        │    │ (或TitleChunker)│    │           │    │  (可选)    │
└────────┘    └────────┘    └──────────────┘    └───────────┘    └────────────┘
   │              │                │                   │               │
   ▼              ▼                ▼                   ▼               ▼
 获取文档      解析为JSON/     切分为Chunk          分词+向量化     LLM生成元数据
 元数据+blob   Markdown/Text   (512 tokens)        (双路检索准备)  (关键词/问题)

组件	类	核心配置	职责
File	`rag/flow/file.py`	—	从数据库获取文档 name 和 blob
Parser	`rag/flow/parser/parser.py`	parse_method, output_format, remove_toc, remove_header_footer	按文件类型分发到 deepdoc 解析器
TokenChunker	`rag/flow/chunker/token_chunker.py`	chunk_token_size=512, delimiter_mode, overlapped_percent, children_delimiters	按 Token 数量/分隔符切分文本
TitleChunker	`rag/flow/chunker/title_chunker/`	levels (标题正则), hierarchy, include_heading_content	按文档标题层级切分 (替代方案)
Tokenizer	`rag/flow/tokenizer/tokenizer.py`	search_method, filename_embd_weight=0.1, fields=["text"]	全文分词 (tks/ltks) + Embedding 向量生成
Extractor	`rag/flow/extractor/extractor.py`	—	LLM 为每个 Chunk 生成结构化元数据 (可选)

3.2 输出格式

Parser 支持 4 种输出格式：json (结构化列表)、markdown (Markdown 文本)、text (纯文本)、html (HTML)。不同格式决定下游 Chunker 的分支逻辑。

3.3 PDF Chunk 元数据处理

rag/flow/parser/pdf_chunk_metadata.py 负责将 DeepDOC 解析出的 BBox 坐标转换为 ES 可索引字段：

normalize_pdf_items_metadata() → 将 BBox 位置转换为统一的 _pdf_positions 二维数组
reorder_multi_column_bboxes() → 多栏重排序 (按阅读顺序)
finalize_pdf_chunk() → 转换为 position_int, page_num_int, top_int
restore_pdf_text_previews() → 生成文本块的 PDF 裁剪预览图

4. 分块策略详解

4.1 TokenChunker — 基于 Token 数量的切分 (默认)

rag/flow/chunker/token_chunker.py 实现了 3 种切分模式，由 delimiter_mode 控制：

模式	行为	适用场景
`"token_size"`	按 `chunk_token_size` (默认512) 切分，使用句子边界感知的 `naive_merge()` 算法	通用文本
`"delimiter"`	按 `delimiters` 配置的正则分隔符切分 (默认 `["\n"]`)	有明确分隔符的文本
`"one"`	整个文档作为一个 Chunk	短文档

TokenChunkerParam 完整参数

参数	默认值	说明
`chunk_token_size`	512	每个 Chunk 的目标 Token 数
`delimiter_mode`	"token_size"	切分策略: token_size / delimiter / one
`delimiters`	["\n"]	自定义切分分隔符 (反引号包裹的视为正则)
`overlapped_percent`	0	相邻 Chunk 之间的重叠比例 (0-100)
`children_delimiters`	[]	子级切分符，启用 Parent-Child Chunk 层次结构
`table_context_size`	0	表格 Chunk 附加上下文 Token 数
`image_context_size`	0	图片 Chunk 附加上下文 Token 数

json 格式处理流程 (标准路径)

Parser 输出的 JSON items
  │
  ├── _build_json_chunks()
  │   ├── Text item → 可选的 delimiter 子切分
  │   └── Table/Image item → 独立 Chunk，含 context_above/context_below
  │
  ├── _attach_context_to_media_chunks()
  │   └── 为图片/表格 Chunk 附加前后文文本 (table_context_size/image_context_size)
  │
  ├── _merge_text_chunks_by_token_size()
  │   └── 合并相邻 Text Chunk: 累计超过 chunk_token_size*(100-overlap)/100 时新建
  │
  ├── _split_chunk_docs_by_children()
  │   └── 应用 children_delimiters 做二次切分 (产生 mom_id 父子关系)
  │
  ├── restore_pdf_text_previews()
  │   └── 为 PDF 文本 Chunk 生成位置预览图
  │
  └── _finalize_json_chunks()
      └── 转换为最终输出格式 (text, doc_type_kwd, _pdf_positions, ...)

4.2 TitleChunker — 基于标题层级的切分

位于 rag/flow/chunker/title_chunker/，利用文档自身的结构 (标题层级) 而非 Token 数量来切分：

GroupTitleChunker: 按 section ID 分组，Token 数约束 32-1024。小 section 自动合并。
HierarchyTitleChunker: 构建标题树 (_ChunkNode)，按路径 (root→leaf) 提取 Chunk。支持 include_heading_content 将标题文本附加到 Chunk 中。

标题级别解析：

resolve_outline_levels() — 匹配 PDF 大纲 (书签)
resolve_frequency_levels() — 匹配正则模式 + 频率统计

5. 分词与向量化

5.1 Tokenizer 组件

rag/flow/tokenizer/tokenizer.py 为双路检索 (全文+向量) 准备数据，通过 search_method 控制：

全文检索路径 (`"full_text" in search_method`)

为每个 Chunk 生成以下分词字段：

字段	分词器	来源
`title_tks`	rag_tokenizer (粗粒度)	文件名 (去扩展名)
`title_sm_tks`	rag_tokenizer (细粒度)	文件名 (去扩展名)
`content_ltks`	rag_tokenizer (粗粒度)	Chunk 文本内容
`content_sm_ltks`	rag_tokenizer (细粒度)	Chunk 文本内容
`question_tks`	rag_tokenizer	自动生成的问题
`important_tks`	rag_tokenizer	自动提取的关键词
`chunk_order_int`	—	Chunk 在文档中的序号

向量检索路径 (`"embedding" in search_method`)

Embedding 生成流程 (_embedding 方法):

1. 解析 Embedding 模型配置
   └── kb.tenant_embd_id → kb.embd_id → tenant 默认模型

2. 提取文本
   └── fields=["text"] (默认) → 清洗 HTML table 标签 → truncate(max_length-10)

3. 文件名向量化
   └── embed(filename) → vts[0] → tile 到 (n_chunks, dim)

4. 内容批量向量化
   └── batch_size = EMBEDDING_BATCH_SIZE (默认16)
   └── thread_pool_exec + embed_limiter (semaphore 并发控制)
   └── async 编码: mdl.encode([truncated_texts])

5. 加权融合
   └── vects = title_w * tts + (1 - title_w) * cnts
   └── title_w = filename_embd_weight (默认0.1)

6. 存储
   └── d["q_{dim}_vec"] = v.tolist()  例如: q_768_vec, q_1536_vec

5.2 Embedding 参数

参数	默认值	来源	说明
`EMBEDDING_BATCH_SIZE`	16	环境变量	每批编码的文本数量
`filename_embd_weight`	0.1	TokenizerParam	文件名向量在最终向量中的权重
`fields`	["text"]	TokenizerParam	参与编码的字段
`embed_limiter`	Semaphore(1)	全局	并发编码控制 (默认串行)
向量维度	模型决定	模型输出	支持 512/768/1024/1536 维
相似度	cosine	mapping.json	ES dense_vector 索引使用的相似度度量

6. Elasticsearch 存储结构详解

6.1 索引命名规则

Index Name = "ragflow_" + tenant_id

每个租户 (Tenant) 拥有一个独立 ES 索引，该租户下所有知识库 (KB/Dataset) 的 Chunk 共享此索引。KB 级别的数据隔离通过每个文档上的 kb_id 字段实现。

设计决策：不按 KB 分索引，因为 Elasticsearch 的索引数量有实际限制 (集群状态开销)。通过 kb_id 过滤条件在查询时实现逻辑隔离。

6.2 索引设置

{
  "number_of_shards": 2,
  "number_of_replicas": 0,
  "refresh_interval": "1000ms"
}

6.3 自定义相似度算法

ES 索引定义了一个名为 scripted_sim 的自定义相似度，用于 *_tks 字段：

scripted_sim: {
  "type": "scripted",
  "script": {
    "source": "double idf = Math.log(1+(field.docCount-term.docFreq+0.5)/(term.docFreq + 0.5))
               / Math.log(1+((field.docCount-0.5)/1.5));
               return query.boost * idf * Math.min(doc.freq, 1);"
  }
}

关键：Math.min(doc.freq, 1) 实现了布尔式 TF 加权——词项在文档中出现 1 次或 100 次，贡献相同。这避免了长文档中的词频膨胀问题。

6.4 动态模板系统 (Dynamic Templates)

ES 映射不定义显式字段，而是通过 17 个动态模板按字段名后缀/前缀匹配自动分配类型。这使得不同维度的 Embedding 向量和不同类型的字段可以共存于同一索引。

模板	匹配模式	ES 类型	关键属性	用途
`dense_vector (512)`	`*_512_vec`	dense_vector	dims:512, similarity:cosine, index:true	512 维 Embedding 向量
`dense_vector (768)`	`*_768_vec`	dense_vector	dims:768, similarity:cosine, index:true	768 维 Embedding 向量 (常见)
`dense_vector (1024)`	`*_1024_vec`	dense_vector	dims:1024, similarity:cosine, index:true	1024 维 Embedding 向量
`dense_vector (1536)`	`*_1536_vec`	dense_vector	dims:1536, similarity:cosine, index:true	1536 维 Embedding 向量 (OpenAI)
`tks`	`*_tks`	text	analyzer:whitespace, similarity:scripted_sim, store:true	粗粒度全文检索 (布尔TF-IDF)
`ltks`	`*_ltks`	text	analyzer:whitespace, store:true	轻量分词全文检索 (标准相似度)
`kwd`	`^(.*_(kwd\|id\|ids\|uid\|uids)\|uid\|id)$`	keyword	similarity:boolean, store:true	精确匹配字段 (ID、标签等)
`int`	`*_int`	integer	store:true	整数字段
`numeric`	`*_flt`	float	store:true	浮点数字段
`dt`	`^.*(_dt\|_time\|_at)$`	date	format: yyyy-MM-dd HH:mm:ss\|\|yyyy-MM-dd\|\|yyyy-MM-dd_HH:mm:ss, store:true	日期字段
`string`	`^.*_(with_weight\|list)$`	text	index:false, store:true	存储但不索引的文本
`ulong`	`*_ulong`	unsigned_long	store:true	无符号长整数
`long`	`*_long`	long	store:true	长整数
`short`	`*_short`	short	store:true	短整数
`rank_feature`	`*_fea`	rank_feature	—	单值排序特征 (PageRank)
`rank_features`	`*_feas`	rank_features	—	多值排序特征 (标签评分)
`object`	`*_obj`	object	dynamic:true	动态 JSON 对象
`nested`	`*_nst`	nested	—	嵌套文档
`binary`	`*_bin`	binary	—	二进制数据

另有 1 个显式字段：lat_lon (geo_point, store:true)。以及 date_detection: true 用于字符串自动日期识别。

7. Chunk 文档完整字段字典

下面是一个写入 ES 的完整 Chunk 文档所包含的所有字段，按功能分组：

7.1 核心标识字段

字段名	ES 类型	来源	含义
`id`	keyword	task_executor.py:352	Chunk 唯一ID: `xxhash.xxh64(content_with_weight + doc_id)` 的 16 进制摘要。同时作为 ES 的 `_id`，实现幂等写入。
`doc_id`	keyword	Pipeline / task_executor	父文档的 UUID
`kb_id`	keyword	insert() 方法强制写入	所属知识库 ID。所有查询必须携带此过滤条件，实现 KB 级数据隔离。
`docnm_kwd`	keyword	Pipeline	文档名称 (如 "年度报告.pdf")
`mom_id`	keyword	insert_chunks()	父 Chunk ID。当使用 children_delimiters 时，子 Chunk 通过此字段关联到父 Chunk。
`img_id`	keyword	image2id()	MinIO 中存储的图片 ID。用于表格/图片 Chunk。

7.2 文本内容字段

字段名	ES 类型	分词器	含义
`content_with_weight`	text (index:false)	—	原始文本内容。存储但不索引，用于回显和重排序时取回原文。
`content_ltks`	text	whitespace	粗粒度分词的文本内容，用于全文检索。
`content_sm_ltks`	text	whitespace	细粒度分词的文本内容，用于子词级匹配。
`title_tks`	text	whitespace (scripted_sim)	文档标题分词 (布尔TF-IDF)，用于标题匹配。
`title_sm_tks`	text	whitespace	细粒度标题分词。
`question_kwd`	keyword	—	LLM 自动生成的示例问题列表 (启用 auto_questions 时)。
`question_tks`	text	whitespace (scripted_sim)	问题分词，用于 Q2Q 检索 (用问题匹配问题)。
`important_kwd`	keyword	—	LLM 自动提取的关键词列表 (启用 auto_keywords 时)。
`important_tks`	text	whitespace (scripted_sim)	关键词分词，用于关键词匹配检索。
`extra`	text (index:false)	—	额外元数据列表。

7.3 向量字段

字段名	ES 类型	相似度	含义
`q_512_vec`	dense_vector	cosine	512 维语义向量
`q_768_vec`	dense_vector	cosine	768 维语义向量 (如 BGE-Large)
`q_1024_vec`	dense_vector	cosine	1024 维语义向量 (如 Jina AI)
`q_1536_vec`	dense_vector	cosine	1536 维语义向量 (如 OpenAI text-embedding-ada-002)

向量合成公式：vects = 0.1 × embed(文档名).tile(n) + 0.9 × embed(Chunk内容)
文件名向量权重默认 0.1 (filename_embd_weight)，可通过 ParserConfig 调整。

7.4 排序与位置字段

字段名	ES 类型	含义
`chunk_order_int`	integer	Chunk 在文档中的顺序编号 (0, 1, 2, ...)
`page_num_int`	integer	来源 PDF 页码列表，如 `[1, 1, 2]` (跨页 Chunk)
`top_int`	integer	在页面中的垂直位置 (top 坐标)，用于按阅读顺序排序
`position_int`	integer	BBox 坐标列表: `[[page, left, right, top, bottom], ...]`

7.5 时间字段

字段名	ES 类型	含义
`create_time`	date	Chunk 创建时间: `str(datetime.now()).replace("T", " ")[:19]` 格式 yyyy-MM-dd HH:mm:ss
`create_timestamp_flt`	float	Unix 时间戳 (浮点)，用于精确排序

7.6 排名/评分字段

字段名	ES 类型	含义
`pagerank_fea`	rank_feature	Chunk 的 PageRank 评分 (标量)，用于提升重要 Chunk 的检索排名
`tag_feas`	rank_features	标签加权评分 (多值)，如 `{"技术": 3.5, "财务": 1.2}`

7.7 标记与分类字段

字段名	ES 类型	含义
`doc_type_kwd`	keyword	Chunk 类型: `"text"` / `"table"` / `"image"`
`available_int`	integer	可用性标记: 0=不可用 (母 Chunk/内部 Chunk), 1=可检索
`removed_kwd`	keyword	墓碑标记 (软删除)
`raptor_kwd`	keyword	RAPTOR 摘要标记: 值为 `"raptor"` 表示该 Chunk 是 RAPTOR 树节点的摘要
`raptor_layer_int`	integer	RAPTOR 树层级编号 (0=叶子层)
`toc_kwd`	keyword	目录 Chunk 标记: 值为 `"toc"`
`tag_kwd`	keyword	LLM 标注的内容标签
`knowledge_graph_kwd`	keyword	知识图谱关联 ID
`entity_kwd`	keyword	知识图谱实体
`from_entity_kwd`	keyword	知识图谱关系的源实体
`to_entity_kwd`	keyword	知识图谱关系的目标实体

7.8 一个典型 Chunk 文档示例 (JSON)

{
  "id": "a1b2c3d4e5f6a7b8",
  "doc_id": "d4e5f6a7-b8c9-40d1-a2b3-c4d5e6f7a8b9",
  "kb_id": "1234567890",
  "docnm_kwd": "年度财务报告.pdf",
  "doc_type_kwd": "text",
  "available_int": 1,
  "chunk_order_int": 5,
  "content_with_weight": "2024年度公司总营收达到12.5亿元，同比增长23.8%...",
  "content_ltks": "2024年度 公司 总 营收 达到 12.5亿 元 同比 增长 23.8% ...",
  "content_sm_ltks": "2024 年度 公司 总 营收 达到 12.5 亿 元 同比 增长 23.8% ...",
  "title_tks": "年度 财务 报告",
  "title_sm_tks": "年度 财务 报告",
  "page_num_int": [3],
  "top_int": [450],
  "position_int": [[3, 72, 540, 450, 480]],
  "create_time": "2026-06-25 14:30:00",
  "create_timestamp_flt": 1750842600.0,
  "q_768_vec": [0.023, -0.145, 0.887, ..., 0.034],
  "pagerank_fea": 2.5,
  "tag_feas": {"财务": 3.2, "年报": 1.8},
  "important_kwd": ["营收", "增长", "财务报告"],
  "important_tks": "营收 增长 财务 报告",
  "question_kwd": ["2024年公司营收是多少？", "营收增长了多少？"],
  "question_tks": "2024年 公司 营收 是 多少 营收 增长 了 多少",
  "img_id": "",
  "kb_id": "1234567890"
}

8. 数据写入流程 (insert_chunks)

8.1 写入链路

do_handle_task()  [task_executor.py]
  │
  ├── build_chunks()       ← 调用 rag/flow Pipeline 生成 Chunk 列表
  │
  ├── 并行异步后处理 (asyncio.gather):
  │   ├── image2id()       ← 将图片上传到 MinIO，获取 img_id
  │   ├── keyword_extraction()  ← LLM 提取关键词 (auto_keywords > 0)
  │   ├── question_proposal()   ← LLM 生成问题 (auto_questions > 0)
  │   ├── gen_metadata()        ← LLM 生成结构化元数据 (enable_metadata)
  │   └── content_tagging()     ← LLM 内容标签 (tag_kb_ids)
  │
  ├── build_TOC()          ← LLM 生成文档目录 (可选)
  │
  ├── embedding()          ← 向量化: title*0.1 + content*0.9
  │   └── d["q_{dim}_vec"] = v.tolist()
  │
  └── insert_chunks()      ← ES Bulk 写入
      │
      ├── dedup + 写入 mothers (母 Chunk, available_int=0)
      ├── 分批写入 chunks (DOC_BULK_SIZE 每批)
      │   └── settings.docStoreConn.insert(chunks_batch, index_name, kb_id)
      └── TaskService.update_chunk_ids()  ← 更新任务进度

8.2 ES Bulk Insert 详解

ESConnection.insert() 方法 (rag/utils/es_conn.py:307)：

def insert(self, documents, index_name, knowledgebase_id):
    operations = []
    for d in documents:
        d_copy = deepcopy(d)
        d_copy["kb_id"] = knowledgebase_id        # 强制写入 kb_id
        meta_id = d_copy.get("id", "")
        operations.append({"index": {"_index": index_name, "_id": meta_id}})
        operations.append(d_copy)

    r = self.es.bulk(
        index=index_name,
        operations=operations,
        refresh="wait_for",    # 等待 refresh 后返回 (实时可搜索)
        timeout="60s"          # 60秒超时
    )
    # 错误处理: 检查 r["errors"] 并返回失败列表

幂等性保证：Chunk ID 是 xxhash.xxh64(content_with_weight + doc_id) 的 hex 摘要。相同内容的文档重新处理时产生相同 ID，Bulk API 的 index action 会覆盖已存在的 _id，实现天然的幂等写入。

8.3 批量写入参数

参数	默认值	说明
`DOC_BULK_SIZE`	环境变量	每批 Bulk 写入的 Chunk 数量
`ATTEMPT_TIME`	2	连接超时重试次数
`refresh`	"wait_for"	等待索引 refresh 完成后返回
`timeout`	"60s"	Bulk 请求超时时间

8.4 Mother-Child 结构

当使用 children_delimiters 时，产生两层结构：

Mother Chunk (available_int=0): 不参与检索，仅保存完整父级文本
Child Chunk: 通过 mom_id 指向母 Chunk，mom/mom_with_weight 字段在写入后从子 Chunk 中移除

9. 检索与重排序流程

9.1 Search Query 构建

ESConnection.search() (rag/utils/es_conn.py:141) 支持三种匹配表达式：

表达式	ES 实现	用途
`MatchTextExpr`	`query_string` (best_fields) 在 `fields` 上执行full-text search，boost = 1.0 - vector_similarity_weight	BM25 全文检索
`MatchDenseExpr`	`knn` 查询，topn*2 候选池 + cosine similarity，filter 用于 KB 隔离	KNN 向量检索
`FusionExpr`	weighted_sum 融合: `weighttext_score + weightvector_score`	混合检索融合

9.2 混合检索流程

用户 Query
  │
  ├── 文本检索路径:
  │   └── query_string (best_fields) on [content_ltks, title_tks, ...]
  │       └── boost = 1.0 - vector_weight
  │
  ├── 向量检索路径:
  │   └── KNN on q_{dim}_vec
  │       ├── topn * 2 (候选池放大)
  │       ├── filter: {kb_id, available_int, ...}  (布尔条件预过滤)
  │       └── similarity: cosine
  │
  └── Fusion: weighted_sum
      └── final_score = w_text * bm25_score + w_vector * cosine_similarity

📊 默认权重: vector_similarity_weight = 0.5 (等权融合)

9.3 检索结果后处理 (Dealer.retrieval)

rag/nlp/search.py 中的 Dealer.retrieval() 方法执行：

消除过期 Chunk: 交叉比对 SQL 数据库，移除父文档已删除的 Chunk
重排序 (ES 路径): rerank_with_knn() 对候选 Chunk 做第二次纯 KNN 查询获取精确余弦相似度，与本地 Token 相似度合并
重排序 (外部模型): 使用 Reranker 模型的 similarity + Token 相似度 + Rank Feature
阈值过滤 + 排序: 按合并得分排序，低于阈值的结果丢弃

9.4 Rank Features

检索时可通过 rank_feature 查询提升特定 Chunk 的得分：

pagerank_fea: 标量 rank_feature，以 linear 函数加入 should 子句
tag_feas.{tag_name}: 多值 rank_features，按标签加权

9.5 大偏移量分页 (Search-After)

当 offset + limit > 10000 时，自动切换为 search_after 分页，按 SEARCH_AFTER_BATCH_SIZE (1000) 批次遍历。

10. 更新与删除机制

10.1 单文档更新

# 条件: condition["id"] 为字符串
self.es.update(index=index_name, id=chunk_id, doc=doc)

# 特殊操作:
# - remove (str)   → ctx._source.remove(field)
# - remove (dict)  → ctx._source.{field}.indexOf(value) 移除数组元素

10.2 批量更新 (Update By Query)

# 条件: condition["id"] 为列表 或 其他条件
self.es.update_by_query(
    index=index_name,
    query=bool_query,
    script=painless_script,     # 动态生成
    refresh=True,
    slices=5,                   # 并行分片
    conflicts="proceed"         # 冲突时继续
)

10.3 删除 (Delete By Query)

self.es.delete_by_query(
    index=index_name,
    query=build_bool_query(condition + kb_id_filter),
    refresh=True
)
# 返回: deleted_count

10.4 PageRank 原子调整

# 原子更新 pagerank_fea 字段
script = """
  double w = ctx._source.pagerank_fea + params.inc_w;
  if (w <= 0) { ctx._source.remove('pagerank_fea'); }
  else if (w >= params.max_w) { ctx._source.pagerank_fea = params.max_w; }
  else { ctx._source.pagerank_fea = w; }
"""
# retry_on_conflict=3 乐观并发控制

10.5 索引生命周期

创建: create_idx() 幂等，仅在索引不存在时创建
删除: delete_idx() 仅在 dataset_id="" 时删除索引。由于多 KB 共享同一索引，当 dataset_id 非空时不会删除索引
无自动 Migration: 依赖动态模板实现字段兼容。新增字段只需匹配已有的命名约定即可自动获得正确的 ES 类型

11. 完整数据流总结

╔══════════════════════════════════════════════════════════════════════════════╗
║                  RAGFlow 文档解析 → 存储 完整数据流                           ║
╠══════════════════════════════════════════════════════════════════════════════╣
║                                                                              ║
║  📄 原始文件 (PDF/DOCX/XLSX/PPT/MD/HTML/...)                                ║
║  ═══════════════════════════════════════                                      ║
║           │                                                                  ║
║           ▼                                                                  ║
║  ┌─────────────────────────────────────────┐                                ║
║  │ deepdoc 解析层                            │                                ║
║  │  • 16+ 解析器 (内置 + 外部API)            │                                ║
║  │  • PDF: pdfplumber + OCR + 版面分析       │                                ║
║  │    + 表格结构识别 (TSR)                    │                                ║
║  │  • 输出: Sections + Tables/Figures        │                                ║
║  └────────────────┬────────────────────────┘                                ║
║                   │                                                          ║
║                   ▼                                                          ║
║  ┌─────────────────────────────────────────┐                                ║
║  │ rag/flow Pipeline                        │                                ║
║  │  • File → Parser → Chunker → Tokenizer   │                                ║
║  │  • TokenChunker: 512 tokens/chunk        │                                ║
║  │  • TitleChunker: 文档标题层级切分         │                                ║
║  │  • Tokenizer: 分词(tks/ltks) + 向量化     │                                ║
║  │    向量: 0.1*title + 0.9*content          │                                ║
║  └────────────────┬────────────────────────┘                                ║
║                   │                                                          ║
║                   ▼                                                          ║
║  ┌─────────────────────────────────────────┐                                ║
║  │ Task Executor 后处理                      │                                ║
║  │  • 图片上传 MinIO (image2id)              │                                ║
║  │  • LLM: 关键词提取 + 问题生成 + 元数据     │                                ║
║  │  • LLM: TOC 目录生成 + 内容标签           │                                ║
║  │  • Embedding: 批量向量编码 (batch_size=16) │                                ║
║  └────────────────┬────────────────────────┘                                ║
║                   │                                                          ║
║                   ▼                                                          ║
║  ┌─────────────────────────────────────────┐                                ║
║  │ Elasticsearch 存储                        │                                ║
║  │  • Index: ragflow_{tenant_id}             │                                ║
║  │  • 17 动态模板 (字段后缀自动类型映射)       │                                ║
║  │  • Chunk ID = xxhash(content+doc_id)      │                                ║
║  │  • Bulk Insert (refresh=wait_for)         │                                ║
║  │  • 支持: KNN vector + BM25 full-text      │                                ║
║  │    + weighted_sum 混合融合                 │                                ║
║  └─────────────────────────────────────────┘                                ║
║                                                                              ║
║  🔍 检索时:                                                                  ║
║  ┌─────────────────────────────────────────┐                                ║
║  │ ES Search                                 │                                ║
║  │  • query_string (BM25) on *_tks/*_ltks    │                                ║
║  │  • KNN on q_{dim}_vec (cosine)            │                                ║
║  │  • rank_feature: pagerank_fea + tag_feas  │                                ║
║  └────────────────┬────────────────────────┘                                ║
║                   │                                                          ║
║                   ▼                                                          ║
║  ┌─────────────────────────────────────────┐                                ║
║  │ Dealer.retrieval() 后处理                 │                                ║
║  │  • 消除过期 Chunk (比对SQL)               │                                ║
║  │  • 二次 KNN 精确重排序 (rerank_with_knn)   │                                ║
║  │  • Token 相似度合并                       │                                ║
║  │  • 阈值过滤 + 排序输出                     │                                ║
║  └─────────────────────────────────────────┘                                ║
║                                                                              ║
╚══════════════════════════════════════════════════════════════════════════════╝

关键数字

指标	值
支持解析格式	16+ (PDF, DOCX, XLSX, PPTX, MD, HTML, JSON, EPUB, TXT, 图片, 音频, 视频, 邮件 + 外部 API)
PDF 解析阶段数	8 (images → layouts → table → merge → concat → filter → extract → cleanup)
默认 Chunk 大小	512 tokens
Embedding 批大小	16 (EMBEDDING_BATCH_SIZE)
ES 分片数	2 (number_of_shards)
ES 副本数	0 (number_of_replicas)
ES Refresh 间隔	1000ms
索引命名规则	ragflow_{tenant_id} (每租户一索引)
支持向量维度	512, 768, 1024, 1536
相似度算法	cosine (向量) + scripted_sim (布尔TF-IDF, 全文)
Chunk ID 算法	xxhash.xxh64 (幂等)
Bulk 写入一致性	refresh=wait_for (即时可搜索)
混合检索融合	weighted_sum (默认等权 0.5:0.5)

核心文件索引

文件	职责
`deepdoc/parser/pdf_parser.py`	PDF 深度解析 (核心解析器, 约2000行)
`deepdoc/vision/ocr.py`	PaddleOCR ONNX 文本检测+识别
`deepdoc/vision/layout_recognizer.py`	版面分析 (10种版面类型)
`deepdoc/vision/table_structure_recognizer.py`	表格结构识别 (TSR)
`rag/flow/pipeline.py`	Pipeline DAG 编排器
`rag/flow/parser/parser.py`	Parser 组件 (文件类型分发)
`rag/flow/parser/pdf_chunk_metadata.py`	PDF 坐标归一化与元数据生成
`rag/flow/chunker/token_chunker.py`	Token 数量切分器
`rag/flow/tokenizer/tokenizer.py`	分词 + 向量化组件
`rag/svr/task_executor.py`	Task 执行器 (build/embed/insert/cache)
`rag/utils/es_conn.py`	ES 连接器 (CRUD + Search)
`rag/nlp/search.py`	检索编排器 (Dealer) + 重排序
`conf/mapping.json`	ES 索引映射定义 (17个动态模板)

RAGFlow 文档解析与存储核心流程分析