RAG 数据处理管道

本文引用的源码文件

本文引用的文件

引言

本文件面向 RAG（检索增强生成）数据处理管道的实现，覆盖从多格式文档加载、文本分块、向量嵌入生成到索引构建与查询检索的完整链路。重点说明：

多格式解析路由与纯文本统一输出
中文优化的分块策略与重叠设计
本地 BGE 模型与 OpenAI 兼容远端两种嵌入后端
Vectra 本地向量存储的事务写入与检索接口
混合检索引擎：语义向量 + 关键词命中增强
大规模数据重建、增量更新、崩溃恢复与存储成本优化

项目结构

RAG 子系统位于 agent/src/rag，按职责划分为配置、管理器、文档处理、嵌入、存储与检索等模块：

配置与路径：config.js
知识库生命周期与导入/重建：manager.js
文档加载与解析：document/loader.js 及 parsers/*
文本分块：document/chunker.js
嵌入抽象与后端：embedding/index.js, local.js, openai-compat.js
向量存储适配器：store/vectra.js
检索器与混合检索：searcher.js
运行时检测：detect.js
错误定义：errors.js

graph TB
subgraph "RAG 子系统"
CFG["配置<br/>config.js"]
MGR["知识库管理<br/>manager.js"]
DOC["文档加载<br/>loader.js"]
P_TXT["文本解析<br/>parsers/text.js"]
P_PDF["PDF 解析<br/>parsers/pdf.js"]
P_OFF["Office 解析<br/>parsers/office.js"]
CHUNK["分块<br/>chunker.js"]
EMB_IDX["嵌入工厂<br/>embedding/index.js"]
EMB_LOCAL["本地嵌入<br/>embedding/local.js"]
EMB_OPENAI["OpenAI 兼容嵌入<br/>embedding/openai-compat.js"]
STORE["向量存储<br/>store/vectra.js"]
SEARCH["检索器<br/>searcher.js"]
DETECT["依赖检测<br/>detect.js"]
ERR["错误定义<br/>errors.js"]
end
MGR --> CFG
MGR --> DOC
DOC --> P_TXT
DOC --> P_PDF
DOC --> P_OFF
MGR --> CHUNK
MGR --> EMB_IDX
EMB_IDX --> EMB_LOCAL
EMB_IDX --> EMB_OPENAI
MGR --> STORE
MGR --> SEARCH
SEARCH --> STORE
MGR --> DETECT
MGR --> ERR

图表来源

章节来源

核心组件

配置中心：集中定义 RAG 根目录、集合注册表、默认嵌入模型（BGE 中文）、分块参数、检索阈值、批大小与扩展名路由。
知识库管理器：负责集合增删改查、文档导入（文本/文件/URL）、后台重建索引、原子换入与崩溃恢复、进度上报与并发控制。
文档处理器：按扩展名动态加载解析器，统一输出纯文本；支持 txt/md/json/csv/html/pdf/docx/xlsx/pptx 等。
分块器：段落优先断点 + 字符窗口兜底 + 重叠回退，适配中文 token 近似 1:1 特性并避免静默截断。
嵌入提供者：统一 embedQuery（带前缀）与 embedDocuments（无前缀），支持本地 transformers.js 与 OpenAI 兼容远端。
向量存储：基于 vectra 的 LocalIndex，封装事务批量 upsert、按元数据删除、BM25 旁路参数与结果归一化。
检索器：语义向量检索 + 关键词命中增强（英文词边界/CJK 最长片段）+ 高频词剔除 + 旁路探测。
依赖检测：子进程探测 vectra 与 transformers 可用性，Node 版本前置校验，结果缓存。
错误体系：结构化 RagError 与翻译辅助，便于响应层国际化。

章节来源

架构总览

RAG 管道分为“写入侧”和“读取侧”。写入侧完成文档解析、分块、向量化与索引构建；读取侧执行语义检索与关键词增强，返回候选片段供上层 LLM 使用。

sequenceDiagram
participant Client as "调用方"
participant Manager as "知识库管理器"
participant Loader as "文档加载器"
participant Parser as "解析器(text/pdf/office)"
participant Chunker as "分块器"
participant Embed as "嵌入提供者"
participant Store as "向量存储(Vectra)"
participant Searcher as "检索器"
Note over Client,Manager : 写入流程
Client->>Manager : 导入文本/文件/URL
Manager->>Loader : loadDocument(路径或临时文件)
Loader->>Parser : parse(filePath)
Parser-->>Loader : 纯文本
Loader-->>Manager : {text, fileName}
Manager->>Chunker : chunkText(text, 配置)
Chunker-->>Manager : chunks[{text,start,end}]
Manager->>Embed : embedDocuments(chunks)
Embed-->>Manager : embeddings[][]
Manager->>Store : upsertChunks(docId, chunks, embeddings, metadata)
Store-->>Manager : 成功
Note over Client,Searcher : 读取流程
Client->>Searcher : search(query, options)
Searcher->>Embed : embedQuery(query)
Embed-->>Searcher : queryVec
Searcher->>Store : search(queryVec, query, topK*2)
Store-->>Searcher : 候选集
Searcher->>Searcher : 关键词增强/阈值过滤/旁路探测
Searcher-->>Client : 最终结果

图表来源

详细组件分析

文档加载与多格式解析

扩展名路由：通过 config.js 中的 PARSER_ROUTES 将 .txt/.md/.json/.csv 映射为 text，.html/.htm 映射为 html，.pdf 映射为 pdf，.docx/.pptx/.xlsx/.xls 映射为 office。
动态加载解析器：loader.js 按需 import 对应解析器，减少冷启动开销。
解析器实现：
- text.js：直接读取 UTF-8 并去除 BOM，保证后续 embedding 质量。
- pdf.js：通过 CJS 包 pdf-parse 提取文本，注意 require 包装以避免调试分支问题。
- office.js：docx 走 mammoth 纯文本；表格 xlsx/xls 逐 sheet 转 CSV 保留行列结构；其他 Office 格式走 officeparser。

flowchart TD
Start(["入口: loadDocument"]) --> Ext["解析扩展名"]
Ext --> Route{"是否受支持?"}
Route --> |否| Err["抛出 unsupportedFormat"]
Route --> |是| Load["按需加载解析器"]
Load --> Parse["调用 parse(filePath)"]
Parse --> Text["返回纯文本与文件名"]
Text --> End(["结束"])

图表来源

章节来源

文本分块策略

目标：在 bge-small-zh-v1.5 最大输入 512 token 的安全边界内切分，避免静默截断。
策略：
- 自然断点优先：段落/句号/问号/分号等序列，尽量保持语义完整。
- 字符窗口兜底：当找不到断点时按 chunkSize 切分。
- 重叠回退：下一块起点回退 overlap 字符，防止上下文断裂。
- 安全上限告警：超过 500 字符时打印警告。

flowchart TD
S(["开始: chunkText"]) --> N["规范化换行"]
N --> Params["计算 chunkSize/overlap"]
Params --> Loop{"pos < n ?"}
Loop --> |否| Ret["返回 chunks"]
Loop --> |是| Skip["跳过空白"]
Skip --> EndCalc["end = min(start+chunkSize, n)"]
EndCalc --> Break{"未到文末且找到断点?"}
Break --> |是| Adjust["end=断点后偏移"]
Break --> |否| Keep["保持 end"]
Adjust --> Slice["切片并 trim"]
Keep --> Slice
Slice --> Push{"非空?"}
Push --> |是| Add["加入 chunks"]
Push --> |否| NextPos["计算 next"]
Add --> NextPos
NextPos --> Update["pos=next"]
Update --> Loop

图表来源

章节来源

向量嵌入生成与模型选择

统一接口：EmbeddingProvider 提供 embedQuery（自动附加 queryPrefix）与 embedDocuments（无前缀），确保非对称检索一致性。
本地模式：基于 @huggingface/transformers 的 feature-extraction pipeline，默认模型 Xenova/bge-small-zh-v1.5，q8 量化约 95MB，缓存至 ~/.ai-helper-agent/rag/models，下载端点默认 hf-mirror。
远端模式：OpenAI 兼容 /embeddings 接口，支持 dimensions 透传与 400 错误降级重试，返回向量按 index 排序。
Token 计数：本地后端暴露 countTokens，用于上游发现超窗风险。

classDiagram
class EmbeddingProvider {
+string mode
+string modelName
+number dimensions
+string queryPrefix
+embedQuery(query) number[]
+embedDocuments(texts) number[][]
}
class LocalEmbedding {
+embed(texts) number[][]
+countTokens(texts) object
}
class OpenAICompatEmbedding {
+embed(texts) number[][]
}
EmbeddingProvider --> LocalEmbedding : "本地模式"
EmbeddingProvider --> OpenAICompatEmbedding : "远端模式"

图表来源

章节来源

向量存储与索引构建

VectraStore 封装 LocalIndex：
- ensureCreated：幂等创建索引。
- upsertChunks：beginUpdate/endUpdate 事务包裹逐条 upsertItem，避免频繁落盘。
- search：queryItems 位置参数签名，统一返回结构，score 为余弦相似度 [-1,1]。
- listDocumentChunks/removeDocument/listAllItems/recreate：按文档维度操作与重建。
重建索引：manager._rebuildIndex 使用临时目录全量重放文档备份 → 原子 rename 换入 → 提交注册表 → 清理临时文件；失败时保留现场供恢复。

flowchart TD
A["开始: _rebuildIndex"] --> B["读取 documents/ 备份"]
B --> C["创建临时目录与新索引"]
C --> D["遍历备份: 分块→嵌入→upsertChunks"]
D --> E["写新备份与清单 manifest"]
E --> F{"原子换入成功?"}
F --> |是| G["提交注册表(新配置/计数)"]
F --> |否| H["回滚旧索引(尽力而为)"]
G --> I["删除清单/清理临时目录"]
H --> J["抛错(保留现场)"]
I --> K["结束"]
J --> K

图表来源

章节来源

混合检索引擎与查询优化

语义通道：对 query 调用 embedQuery（自动加前缀），得到 queryVec，调用 store.search(topK*2) 获取候选集。
关键词通道：
- 提取拉丁与 CJK 术语，分别按词边界匹配与最长片段匹配。
- 统计文档频率 df，剔除高频词（>半数 chunk）与低频豁免（df<=2）。
- 对命中 chunk 进行抬分：max(向量分, 保底) + 单元加分，封顶 SCORE_MAX。
- 未命中者分数不变，合并后统一阈值过滤。
旁路探测：若过滤后为空，放宽阈值再检索少量候选，区分“无数据”与“被过滤”。

flowchart TD
Q(["查询: search(query)"]) --> V["embedQuery(query)"]
V --> C["store.search(queryVec, topK*2)"]
C --> K["_boostByKeywords(query, candidates)"]
K --> T["阈值过滤(score>=threshold)"]
T --> Empty{"结果为空?"}
Empty --> |否| R["返回 topK 结果"]
Empty --> |是| F["放宽阈值检索 topK=3"]
F --> Filt["过滤 score>=threshold-0.2"]
Filt --> R

图表来源

章节来源

agent/src/rag/searcher.js#L1-L162

知识库生命周期与导入流程

创建/删除/启用切换：维护 collections.json 注册表，读写串行化，密钥落盘加密。
导入文本/文件/URL：
- 文本：体积预检 → 分块 → 嵌入 → 写入 Vectra → 原文备份。
- 文件：沙箱校验 path → 解析 → 同上；上传 base64 先估算尺寸，临时文件解析后清理。
- URL：遵循 ragUrlIngest 策略，防内网扫描。
进度上报：_reportProgress 提供 phase/current/total/done/error，前端轮询展示。
并发控制：同库导入/重建互斥，_activeOps 占位，避免索引写坏。

sequenceDiagram
participant UI as "前端/工具"
participant M as "RagManager"
participant FS as "文件系统"
participant L as "Loader/Parsers"
participant C as "Chunker"
participant E as "Embedding"
participant S as "VectraStore"
UI->>M : ingestText/ingestFile
M->>M : 体积/沙箱校验
M->>FS : 读取/写入临时文件
M->>L : loadDocument
L-->>M : {text, fileName}
M->>C : chunkText
C-->>M : chunks
M->>E : embedDocuments
E-->>M : embeddings
M->>S : upsertChunks
S-->>M : 成功
M-->>UI : 进度/结果

图表来源

章节来源

依赖关系分析

外部依赖：vectra（向量索引）、@huggingface/transformers（本地嵌入）、pdf-parse/mammoth/xlsx/officeparser（文档解析）。
运行时检测：detect.js 通过子进程 probe-child.mjs 探测 vectra 与 transformers 可用性，Node 版本需 >=22。
耦合关系：
- manager 强依赖 config、embedding、store、loader、chunker、searcher、security、crypto-lite。
- searcher 依赖 keyword-match（复用长期记忆模块）与 store.embedding。
- embedding/index 解耦 local/openai-compat 后端。

graph LR
Detect["detect.js"] --> Transformers["@huggingface/transformers"]
Detect --> Vectra["vectra"]
Manager["manager.js"] --> Config["config.js"]
Manager --> EmbedIdx["embedding/index.js"]
Manager --> Store["store/vectra.js"]
Manager --> Loader["document/loader.js"]
Manager --> Chunker["document/chunker.js"]
Manager --> Searcher["searcher.js"]
EmbedIdx --> Local["embedding/local.js"]
EmbedIdx --> OpenAI["embedding/openai-compat.js"]
Loader --> Parsers["parsers/*"]

图表来源

章节来源

性能与成本优化

分块与嵌入批大小：
- 默认 chunkSize=400、overlap=80，适配中文 1 字≈1 token 与模型 512 上限。
- EMBED_BATCH_SIZE=10，兼容部分平台单次批量上限。
内存与体积限制：
- INGEST_MAX_TEXT_BYTES=10MB，INGEST_MAX_FILE_BYTES=50MB，避免超大内容驻留内存导致 OOM。
- 搜索 topK 上界 SEARCH_TOPK_MAX=50，防止候选集放大。
本地模型缓存：
- transformers.js 模型缓存在 ~/.ai-helper-agent/rag/models，q8 量化降低内存占用。
- pipeline 单例与加载锁避免重复加载。
重建与原子换入：
- 临时目录构建新索引，rename 原子换入，失败回滚旧索引，保障线上可用。
关键词通道保护：
- KEYWORD_SCAN_LIMIT=20000，大库跳过关键词扫描，避免检索卡顿。
存储布局：
- collections.json 注册表 0600 权限，密钥落盘混淆；documents/ 备份原始文档便于重建。

[本节为通用指导，不直接分析具体文件]

故障排查指南

依赖不可用：
- detectRagAvailable 返回 false 时查看 errorSummary，常见原因为 Node 版本过低或 vectra/transformers 缺失。
- 子进程探测超时：PROBE_TIMEOUT_MS=60s，原生库加载卡死会触发 SIGKILL。
维度不匹配：
- Searcher 在 embedQuery 后校验维度，若不匹配抛出 embeddingDimensionMismatch，需重建索引。
导入失败：
- 体积超限：contentTooLarge/fileTooLarge；路径不允许：pathNotAllowed；缺少文件名：missingFileName。
- 解析失败：unsupportedFormat；URL 导入受 ragUrlIngest 策略限制。
重建中断恢复：
- recoverInterruptedRebuilds 扫描 .rebuild_* 目录，依据 REBUILD_MANIFEST 补提交注册表或回滚旧索引。
错误翻译：
- 业务错误统一为 RagError，响应层 translateRagError 转为可翻译消息。

章节来源

结论

该 RAG 数据处理管道以“配置集中、职责清晰、容错完备”为目标，实现了从多格式文档到向量索引的全链路能力。通过非对称检索的前缀机制、中文友好的分块策略、本地与远端双嵌入后端、Vectra 事务写入与混合检索增强，系统在准确性与性能之间取得平衡。重建与崩溃恢复机制保障了大规模数据场景下的稳定性，体积与并发限制则有效控制了资源消耗。建议在部署时关注 Node 版本与依赖安装状态，并根据业务语料调整分块与阈值参数以获得更优召回效果。

消息流与工具调用

RAG 数据处理管道

工具面板集成

聊天管理系统

主题与国际化系统

UI 组件库

Markdown 渲染引擎

RAG 数据处理管道

目录

引言

项目结构

核心组件

架构总览

详细组件分析

文档加载与多格式解析

文本分块策略

向量嵌入生成与模型选择

向量存储与索引构建

混合检索引擎与查询优化

知识库生命周期与导入流程

依赖关系分析

性能与成本优化

故障排查指南

结论

Markdown 渲染引擎

RAG 数据处理管道 ​

目录 ​

引言 ​

项目结构 ​

核心组件 ​

架构总览 ​

详细组件分析 ​

文档加载与多格式解析 ​

文本分块策略 ​

向量嵌入生成与模型选择 ​

向量存储与索引构建 ​

混合检索引擎与查询优化 ​

知识库生命周期与导入流程 ​

依赖关系分析 ​

性能与成本优化 ​

故障排查指南 ​

结论 ​

RAG 数据处理管道

目录

引言

项目结构

核心组件

架构总览

详细组件分析

文档加载与多格式解析

文本分块策略

向量嵌入生成与模型选择

向量存储与索引构建

混合检索引擎与查询优化

知识库生命周期与导入流程

依赖关系分析

性能与成本优化

故障排查指南

结论