📦 RAGFlow rag/ 模块源码分析报告

🏷️ rag/__init__.py 包初始化

最小化的包初始化文件，仅包含版权声明和注释掉的 beartype 类型检查代码。不导出任何公共 API。

⚙️ rag/settings.py 配置占位

空模块，仅包含版权声明。原本可能用于 RAG 应用级配置，目前未被使用（实际配置在 common/settings.py 中）。

📊 rag/benchmark.py 基准测试

主要类：Benchmark(kb_id)

检索基准测试框架，支持 MS MARCO v1.1、TriviaQA、MIRACL 等标准 IR 数据集。

核心方法：

关键依赖：common.*, rag.nlp.search, ranx (评估指标), pandas

🌳 rag/raptor.py RAPTOR 算法

主要类：RecursiveAbstractiveProcessing4TreeOrganizedRetrieval

实现 RAPTOR 递归摘要树算法，支持两种构建策略：

• Classic RAPTOR： UMAP 降维 + GMM/AHC 聚类 → 递归生成摘要层
• Psi Tree： 直接基于成对余弦相似度排名构建合并树，更快且无需降维

核心方法：

关键依赖：umap, sklearn, rag.graphrag.utils (LLM/Embedding 缓存), rag.utils.raptor_utils

🔌 llm/__init__.py — 自动发现与注册 工厂模式

通过 importlib + inspect 自动扫描子模块，提取 _FACTORY_NAME 并注册到工厂字典。

# 核心注册逻辑
MODULE_MAPPING = {
    "chat_model": ChatModel,
    "embedding_model": EmbeddingModel,
    "rerank_model": RerankModel,
    "tts_model": TTSModel,
    "cv_model": CvModel,
    "ocr_model": OcrModel,
    "sequence2txt_model": Seq2txtModel,
}

# 运行时自动发现所有带 _FACTORY_NAME 的类
for mod_name, factory_dict in MODULE_MAPPING.items():
    module = importlib.import_module(f"rag.llm.{mod_name}")
    for name, cls in inspect.getmembers(module, inspect.isclass):
        if hasattr(cls, '_FACTORY_NAME'):
            for n in cls._FACTORY_NAME if isinstance(...) else [cls._FACTORY_NAME]:
                factory_dict[n] = cls  # 注册: ChatModel["OpenAI"] = OpenAI_APIChat

💬 llm/chat_model.py — 对话模型 30+ 提供商

核心类层级：

• Base(ABC) — OpenAI 兼容基类，提供流式/非流式对话、多轮工具调用、错误分类重试、reasoning 内容处理
• LiteLLMBase(ABC) — LiteLLM 路由基类，通过 litellm.acompletion 统一接入 100+ 提供商
• 30+ 具体实现（XinferenceChat, HuggingFaceChat, GoogleChat, BaiduYiyanChat, SparkChat...）

关键机制：

• 错误分类：LLMErrorCode 枚举（RATE_LIMIT、AUTH_ERROR、SERVER_ERROR 等），自动重试
• 模型策略：_apply_model_family_policies() 对 Qwen3/Kimi K2.5/GPT-5 等模型应用参数约束
• 工具调用：async_chat_with_tools() 多轮自动重试工具调用

📐 llm/embedding_model.py — 嵌入模型 30+ 提供商

统一接口：encode(texts) -> (np.ndarray, token_count) 和 encode_queries(text) -> (np.ndarray, token_count)

关键实现：

• JinaMultiVecEmbed — 支持 v4 多向量（mean-pooling）和 v2/v3 单向量
• PerplexityEmbed — 支持 contextualized embeddings (base64 int8 解码)
• SILICONFLOWEmbed — BGE 模型自动截断至 256 tokens
• BuiltinEmbed — 线程安全单例，自动加载 TEI 或 HuggingFace 模型

🎯 llm/rerank_model.py — 重排序模型 15+ 提供商

统一接口：similarity(query, texts) -> (np.ndarray, token_count)

所有实现均提供 _normalize_rank() 静态方法进行 min-max 分数归一化。

👁️ llm/cv_model.py — 视觉模型 20+ 提供商

统一接口：describe(image) -> (text, token_count)

亮点：

• QWenCV — 额外支持视频处理（DashScope MultiModalConversation）
• GeminiCV — 视频处理（20MB 阈值，Files API 处理大文件）
• AnthropicCV — 支持 thinking_delta 流式推理
• 所有基于 GptV4 的实现共享 _form_history 多模态消息构建逻辑

🛠️ llm/tool_decorator.py — 工具装饰器 轻量级

@tool
def search_web(query: str, max_results: int = 5) -> list[dict]:
    """Search the web for information.
    :param query: The search query string.
    :param max_results: Maximum number of results to return.
    """
    ...

# 自动生成 OpenAI function schema:
# {"type": "function", "function": {"name": "search_web",
#  "description": "Search the web...", "parameters": {...}}}

FunctionToolSession 将装饰函数桥接到对话模型的工具调用循环。

🔎 nlp/search.py — 核心检索引擎 核心

主要类：Dealer(dataStore) — 多阶段检索引擎

检索流程：

1. 初检：search() — 文本/向量/混合搜索 + 融合表达式
2. 过滤：_prune_deleted_chunks() — 移除已删除文档的 chunks
3. 重排序：rerank_by_model() — 外部 Cross-Encoder 精排
4. 排序：按相似度阈值过滤，分页
5. 聚合：group_docs — 按文档分组结果

核心方法一览：

📝 nlp/query.py — 全文查询构建 查询理解

主要类：FulltextQueryer(QueryBase)

将自然语言 query 转换为结构化全文查询：

• 英文路径：词权重计算 → 同义词扩展 → Bigram 生成 → 字段加权
• 中文路径：细粒度分词 → 同义词扩展 → 子词 n-gram → 位置加权

提供 hybrid_similarity() 和 token_similarity() 用于检索后重排序。

✂️ nlp/rag_tokenizer.py — 分词器 Infinity 适配

主要类：RagTokenizer (继承 Infinity 的 C++ 分词器)

当使用 Infinity 引擎时，tokenize() 和 fine_grained_tokenize() 为透传（Infinity 自行处理分词）。

模块级导出：tokenize, fine_grained_tokenize, tag, freq, tradi2simp。

🔤 nlp/term_weight.py — 词权重 排序信号

类：Dealer — 组合多种信号计算词权重：NER 类型得分、词性标签权重、语料内词频、文档频率 IDF。

权重归一化至总和 1.0，用于查询构建时的字段加权。

🏭 flow/__init__.py — 自动组件发现 反射机制

通过 pkgutil.walk_packages 遍历所有子包，自动导入并注册所有公共类到 rag.flow 命名空间。

🧱 flow/base.py — 流水线基类 抽象基类

核心类：

• ProcessParamBase(ComponentParamBase) — 参数基类，默认 timeout=100000000
• ProcessBase(ComponentBase) — 组件基类，提供 async _invoke(**kwargs) 抽象方法

执行模型：超时保护 + 异常捕获 + 进度回调 + 耗时统计

@timeout(int(os.environ.get("COMPONENT_EXEC_TIMEOUT", 60*60)))
async def _invoke(self, **kwargs):
    ...

🚀 flow/pipeline.py — 流水线编排器 核心

主要类：Pipeline(Graph) — 继承 Agent 框架的 Graph 类

从 DSL JSON 解析组件图，沿 DAG 路径串行执行组件。每个组件通过 invoke() 异步调用，进度通过 Redis 日志记录。

执行路径：File → Parser → Chunker → Tokenizer → Extractor（线性 DAG）

取消机制：通过 Redis 检查取消标记，抛出 TaskCanceledException。

📄 flow/parser/parser.py — 文档解析分发器 最大模块

类：Parser(ProcessBase) — 支持 12+ 文件格式

✂️ flow/chunker/ — 文本分块 两种策略

TokenChunker：按 token 数或自定义分隔符分块，支持重叠。对表格/图片自动附加上下文窗口。

TitleChunker：基于标题层级的分块：

• HierarchyTitleChunker — 树形层级分块（DFS 路径提取）
• GroupTitleChunker — 扁平分组分块（Token 约束合并）
• BaseTitleChunker — 共享基类，标题级别识别（PDF 书签/正则匹配）

🔢 flow/tokenizer/tokenizer.py — 分词与向量化

对每个 chunk 执行全文分词（标题/内容/问题关键词/重要关键词）并生成向量嵌入。支持文件名加权向量计算和批量处理速率限制。

🤖 flow/extractor/extractor.py — LLM 信息提取

Extractor(ProcessBase, LLM) — 双重继承自流水线基类和 Agent LLM 组件。

两种模式：

• TOC 生成：field_name == "toc" 时，调用 run_toc_from_text() 生成目录
• 元数据提取：自定义 LLM Prompt 提取指定字段

🏭 graphrag/general/index.py — 构建编排器 核心

主函数：run_graphrag_for_kb() — 完整的 GraphRAG 构建流水线

1. 子图生成：generate_subgraph() 逐文档并行提取实体/关系
2. 子图合并：merge_subgraph() 合并到全局 NetworkX 图
3. 实体消歧：resolve_entities() LLM 判断同名实体是否指代同一事物
4. 社区报告：extract_community() Leiden 聚类 + LLM 生成社区摘要

可恢复性：通过 Redis 阶段标记 (phase_markers.py) 支持中断后恢复。

三种提取策略：

• general — 微软 GraphRAG 风格，LLM gleaning 迭代提取
• light — LightRAG 风格，简化 Prompt
• ner — spaCy NER + 共现关系，无需 LLM

🔍 graphrag/search.py — 图谱检索 多源融合

类：KGSearch(Dealer) — 继承自搜索引擎，扩展图谱检索能力

检索流程：查询改写 → 实体关键词检索 → 关系检索 → n-hop 路径扩展 → 社区报告检索

使用 xxhash 进行 LLM 结果缓存，避免重复调用。

🧬 graphrag/general/leiden.py — Leiden 聚类 社区发现

实现层次 Leiden 聚类算法，用于图谱社区发现。使用 graspologic 库生成加权层次社区结构，支持 stable_largest_connected_component 和确定性图排序。

🧠 graphrag/ner/graph_extractor.py — NER 提取 无 LLM

基于 spaCy 的实体/关系提取，无需调用 LLM：

• MGranRAG 关键词堆叠：3 轮叠加（连字符/撇号合并 → 大写词合并 → 名词/数字合并）
• LinearRAG 共现关系：同句共现或依存路径连接

📋 prompts/template.py — 模板加载器

load_prompt(name: str) -> str — 从 prompts/ 目录加载 .md 模板文件，带内存缓存。

🤖 prompts/generator.py — 提示词生成中心 30+ 函数

所有 LLM 交互的提示词构建集中在此模块，使用 Jinja2 沙盒模板引擎。

关键功能分组：

技术细节：gen_json() 使用 json_repair 修复 LLM 输出的畸形 JSON，支持重试。所有异步函数使用 chat_mdl.async_chat() 调用 LLM。

🔌 文档存储连接器

🗄️ utils/redis_conn.py — Redis 工具 核心基础设施

类：RedisDB (单例) — 完整 Redis 操作封装

功能：KV 操作、集合、有序集合、自增 ID、Lua 脚本事务、Stream 消息队列（生产者/消费者）、未确认消息重处理。

类：RedisDistributedLock — 分布式锁（delete-if-equal 安全释放）

🌲 tree_structured_query_decomposition_retrieval.py DeepResearcher

类：TreeStructuredQueryDecompositionRetrieval (别名 DeepResearcher)

实现树形查询分解深度研究：

1. 检索：从 KB + Web (Tavily) + KG 三源检索
2. 充分性检查：LLM 判断信息是否足以回答问题
3. 递归分解：若不充分，将问题分解为子问题，递归深度研究（有深度限制）
4. 去重合并：线程安全的 chunk 信息合并

关键依赖：rag.prompts.generator (充分性检查/多查询生成), rag.utils.tavily_conn, rag.nlp.search

⚡ svr/task_executor.py — 任务执行器 核心

Redis Stream 消费者，处理所有文档摄入任务：

# 并发控制信号量
task_limiter       # 任务级并发限制
chunk_limiter      # 分块并发限制
embed_limiter      # 嵌入生成速率限制
minio_limiter      # MinIO 并发限制
kg_limiter         # 知识图谱并发限制

任务类型分发：

RAPTOR 任务特殊处理：支持检查点（checkpointing）、清理已完成摘要 chunk、取消时的回滚删除。

🔄 svr/sync_data_source.py — 数据源同步 25+ 连接器

类层级：SyncBase → _BlobLikeBase → 具体连接器

支持的数据源：S3, R2, OCI Storage, Google Cloud Storage, RSS, Confluence, Notion, Discord, Gmail, Dropbox, GoogleDrive, Jira, SharePoint, Slack, Teams, WebDAV, Moodle, BOX, Airtable, Asana, Github, Gitlab, Bitbucket, SeaFile, DingTalk, IMAP, Zendesk, MySQL, PostgreSQL, REST API。

同步机制：基于指纹的变更检测 + 定时轮询调度。