RAGFlow 深度解析：从 DeepDoc 文档理解到混合检索引擎——2026 年最值得研读的开源 RAG 系统完整实战指南

一、为什么 RAGFlow 值得单独拿出来讲？

2026 年的 RAG（Retrieval-Augmented Generation）赛道已经不缺产品了。Dify 做的是全栈应用平台，FastGPT 专注知识库问答，Coze 走低代码路线。但如果你真正深入过生产级 RAG 系统的落地，你会发现一个绕不开的事实：文档解析质量决定了整个系统的上限。

你用再多的向量模型、再精巧的 rerank 策略，如果上游的文档切块是一坨乱码，下游的一切优化都是在垃圾上建高楼。

RAGFlow 做对了一件事：它把文档理解当成了一等公民。

这不是一句营销话。打开 RAGFlow 的源码仓库，你会看到一个叫 deepdoc 的模块——它不是简单的文本提取器，而是一个完整的文档结构理解引擎。PDF 版面分析、表格识别、图片 OCR、公式解析，这些在其他 RAG 产品里通常被外包给第三方 API 的能力，RAGFlow 全部自研实现了。

截至 2026 年 7 月，RAGFlow 在 GitHub 上已经获得了 40k+ Stars，是 Infineon Labs（英飞流）团队的核心开源项目。最新版本 v0.26.3 带来了 MCP Server 集成、SoMark OCR 解析器、BigQuery 数据源接入、Go API 持续扩展等重要更新。

本文将从架构设计、文档解析引擎、混合检索机制、Agent 编排、MCP 协议集成、生产部署六个维度，对 RAGFlow 进行一次彻底的技术拆解。

二、整体架构：不只是"向量检索 + LLM"

很多人对 RAG 的理解停留在一张简洁的流程图：用户提问 → 向量检索 → 拼接 Prompt → LLM 生成回答。

RAGFlow 的架构远比这复杂。用一句话概括它的定位：RAGFlow 是一个面向 LLM 的 Context Layer（上下文层），而不仅仅是一个检索问答工具。

2.1 核心模块划分

打开 RAGFlow 的 main 分支目录结构，你会看到以下核心模块：

ragflow/
├── api/            # RESTful API 服务层
├── deepdoc/        # 文档理解引擎（核心竞争力）
├── rag/            # 检索增强生成核心逻辑
│   ├── nlp/        # NLP 工具链（分词、Embedding）
│   ├── search/     # 混合检索引擎
│   ├── flow/       # 工作流编排
│   └── parser/     # 文档解析流程编排
├── agent/          # Agent 运行时
├── memory/         # 长期记忆模块
├── mcp/            # MCP Server 实现
├── sdk/python/     # Python SDK
├── web/            # 前端（React + TypeScript）
└── docker/         # Docker Compose 部署配置

这个结构清晰地展示了 RAGFlow 的系统边界。它不是一个单体应用，而是一个由多个子系统组成的平台：

离线加工链路：文档上传 → DeepDoc 解析 → 分块策略 → Embedding 向量化 → 写入 Elasticsearch + 向量数据库

在线服务链路：用户查询 → 多路召回（关键词 + 向量 + 知识图谱）→ 融合重排 → Prompt 组装 → LLM 生成 → 引用追溯

运行时扩展层：Agent 编排 + Memory 管理 + MCP Server + 工作流引擎

2.2 技术栈选型

这里最值得注意的设计决策是向量数据库的选择。RAGFlow 没有单独引入 Milvus 或 Qdrant，而是直接使用 Elasticsearch 8.x 的 vector search 功能。这个选择的背后逻辑是：在 RAG 场景中，纯粹的向量相似度搜索远远不够，你需要同时做全文关键词检索、结构化过滤、聚合分析——这些都是 Elasticsearch 的强项。把向量检索和全文检索放在同一个引擎里，天然避免了多数据源同步的问题。

三、DeepDoc：文档理解引擎深度拆解

DeepDoc 是 RAGFlow 最核心的差异化能力。它解决的是 RAG 系统中最被低估的问题：如何从复杂文档中提取结构化、可检索的知识。

3.1 问题的本质

现实世界中的知识不会乖乖躺在干净的纯文本里。它可能：

• 藏在 PDF 的多栏版面中，需要理解阅读顺序
• 夹在 Word 文档的嵌套表格里，需要保留行列关系
• 以扫描件的形式存在，需要 OCR + 版面分析
• 包含数学公式、代码块、流程图等特殊元素
• 跨页表格需要合并，页眉页脚需要过滤

传统的文本提取方案（比如 PyPDF2、pdfplumber）在面对这些场景时几乎全面崩溃。它们提取出来的文本要么丢失了结构信息，要么把多栏内容混在一起，要么表格变成了无法理解的乱码。

3.2 DeepDoc 的分层架构

DeepDoc 采用了三层架构设计：

第一层：视觉感知层（Vision Module）

对于 PDF 和图片类文档，DeepDoc 首先将每一页渲染为图像，然后使用深度学习模型进行版面分析：

# deepdoc/vision/recognition.py 简化示意
from deepdoc.vision import LayoutRecognizer, TableRecognizer

# 版面分析：识别页面中的文本块、表格、图片、标题等区域
layout_model = LayoutRecognizer()
regions = layout_model(page_image)
# 输出: [{'type': 'text', 'bbox': [x1,y1,x2,y2], 'confidence': 0.95},
#         {'type': 'table', 'bbox': [x1,y1,x2,y2], 'confidence': 0.92},
#         {'type': 'title', 'bbox': [x1,y1,x2,y2], 'confidence': 0.98}]

# 表格识别：从表格区域提取结构化数据
table_model = TableRecognizer()
for region in regions:
    if region['type'] == 'table':
        table_data = table_model(page_image, region['bbox'])
        # 输出: DataFrame 或 HTML table

版面分析模型基于 LayoutLMv3 架构微调，支持识别以下元素类型：

• text：正文文本块
• title：标题（支持多级标题检测）
• table：表格区域
• figure：图片/图表
• header / footer：页眉页脚
• equation：数学公式
• code：代码块
• reference：参考文献

第二层：OCR 引擎层

对于扫描件和图片中的文字，DeepDoc 内置了 OCR 能力。v0.26.3 版本新增的 SoMark OCR 解析器是一个重要升级——它是一个布局感知型 OCR，不是简单地对整页做 OCR，而是先理解版面结构，再对每个区域分别做 OCR，最后按照正确的阅读顺序组装文本。

# SoMark OCR 的核心流程
# 1. 版面检测 → 识别文本区域
# 2. 对每个文本区域独立 OCR
# 3. 按照从上到下、从左到右（或右到左，支持阿拉伯语）的顺序排列
# 4. 保留区域间的层级关系（标题 → 段落 → 列表项）

第三层：解析器层（Parser Layer）

这是 DeepDoc 的最底层，负责不同格式文档的具体解析逻辑。RAGFlow 支持的文档格式包括：

3.3 分块策略：从"暴力切割"到"语义切块"

传统 RAG 系统的分块策略通常是"按固定字数切割"——比如每 512 个 token 切一刀。这种做法简单粗暴，但问题很大：一个表格可能被切成两半，一个完整的论点可能被从中间截断。

RAGFlow 的分块策略要精细得多：

# rag/flow/parser/parser.py 中的分块逻辑（简化示意）

class ChunkingStrategy:
    def __init__(self, method="smart", chunk_token_num=512, delimiter="\n"):
        self.method = method
        self.chunk_token_num = chunk_token_num
        self.delimiter = delimiter

    def chunk(self, parsed_doc):
        if self.method == "smart":
            return self._smart_chunk(parsed_doc)
        elif self.method == "naive":
            return self._naive_chunk(parsed_doc)
        elif self.method == "paper":
            return self._paper_chunk(parsed_doc)
        elif self.method == "book":
            return self._book_chunk(parsed_doc)
        elif self.method == "qa":
            return self._qa_chunk(parsed_doc)
        elif self.method == "table":
            return self._table_chunk(parsed_doc)

    def _smart_chunk(self, parsed_doc):
        """
        智能分块：
        1. 按文档结构（标题/段落/列表）切分
        2. 表格作为独立 chunk，不切割
        3. 代码块作为独立 chunk
        4. 段落过长时按语义边界（句号、换行）二次切分
        5. 保留 chunk 间的父子关系（用于层级检索）
        """
        chunks = []
        current_chunk = []
        current_token_count = 0

        for element in parsed_doc.elements:
            # 表格和代码块直接作为独立 chunk
            if element.type in ('table', 'code', 'equation'):
                if current_chunk:
                    chunks.append(self._make_chunk(current_chunk))
                    current_chunk = []
                    current_token_count = 0
                chunks.append(self._make_chunk([element], standalone=True))
                continue

            token_count = self._count_tokens(element.text)

            # 标题总是作为新 chunk 的开始
            if element.type == 'title':
                if current_chunk:
                    chunks.append(self._make_chunk(current_chunk))
                current_chunk = [element]
                current_token_count = token_count
                continue

            # 普通段落：追加到当前 chunk 或开启新 chunk
            if current_token_count + token_count > self.chunk_token_num:
                if current_chunk:
                    chunks.append(self._make_chunk(current_chunk))
                current_chunk = [element]
                current_token_count = token_count
            else:
                current_chunk.append(element)
                current_token_count += token_count

        if current_chunk:
            chunks.append(self._make_chunk(current_chunk))

        return chunks

RAGFlow 提供了六种内置分块模式：

1. Smart（智能模式）：默认模式，按文档结构语义切分
2. Naive（朴素模式）：按固定 token 数切割，适合纯文本
3. Paper（论文模式）：识别摘要、引言、方法、实验等论文结构
4. Book（书籍模式）：按章节、小节层级切分
5. QA（问答模式）：自动提取问答对
6. Table（表格模式）：表格转自然语言描述

3.4 可视化标注：让解析结果可审计

RAGFlow 有一个非常实用的功能：文档解析结果可视化。在 Web 界面中，你可以看到每一页文档被解析后的标注结果——哪些区域被识别为文本、哪些是表格、哪些是图片，分块的边界在哪里。

这个功能看似简单，但在生产环境中极其重要。当检索结果不理想时，你可以直接查看文档的解析和分块结果，快速定位问题出在哪个环节——是 OCR 识别错了？还是表格被切割了？还是分块边界不合理？

四、混合检索引擎：不止是向量搜索

RAGFlow 的检索引擎是另一个值得深入分析的模块。它不是简单的"把用户问题做 Embedding 然后算余弦相似度"，而是一个多路召回 + 融合重排的混合系统。

4.1 三路召回架构

# rag/search/search.py 简化示意

class HybridSearchEngine:
    def __init__(self, kb_id, embedding_model, rerank_model):
        self.kb_id = kb_id
        self.embedding_model = embedding_model
        self.rerank_model = rerank_model

    def search(self, query, top_k=20):
        # 第一路：关键词检索（BM25）
        keyword_results = self._keyword_search(query, top_k=top_k * 2)

        # 第二路：语义向量检索
        query_embedding = self.embedding_model.encode(query)
        vector_results = self._vector_search(query_embedding, top_k=top_k * 2)

        # 第三路：知识图谱检索（可选）
        kg_results = self._knowledge_graph_search(query, top_k=top_k)

        # 融合：RRF（Reciprocal Rank Fusion）
        fused_results = self._rrf_fusion(
            [keyword_results, vector_results, kg_results],
            weights=[0.3, 0.5, 0.2]
        )

        # 重排：使用 Rerank 模型精排
        if self.rerank_model:
            final_results = self._rerank(query, fused_results[:top_k * 2], top_k=top_k)
        else:
            final_results = fused_results[:top_k]

        return final_results

    def _keyword_search(self, query, top_k):
        """BM25 全文检索，利用 Elasticsearch 的中文分词能力"""
        es_query = {
            "bool": {
                "should": [
                    {"match": {"content": {"query": query, "boost": 1.0}}},
                    {"match": {"content_with_weight": {"query": query, "boost": 0.5}}}
                ]
            }
        }
        return self.es.search(index=f"ragflow_{self.kb_id}", body=es_query, size=top_k)

    def _vector_search(self, query_embedding, top_k):
        """Elasticsearch 8.x kNN 向量检索"""
        es_query = {
            "knn": {
                "field": "q_dense_vec",
                "query_vector": query_embedding,
                "k": top_k,
                "num_candidates": top_k * 10
            }
        }
        return self.es.search(index=f"ragflow_{self.kb_id}", body=es_query, size=top_k)

    def _rrf_fusion(self, result_lists, weights, k=60):
        """
        Reciprocal Rank Fusion 算法
        RRF_score(d) = Σ weight_i * 1 / (k + rank_i(d))
        """
        scores = {}
        for results, weight in zip(result_lists, weights):
            for rank, doc in enumerate(results):
                doc_id = doc['id']
                if doc_id not in scores:
                    scores[doc_id] = 0.0
                scores[doc_id] += weight * 1.0 / (k + rank + 1)

        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return [doc_map[doc_id] for doc_id, _ in sorted_docs]

4.2 为什么需要混合检索？

纯粹的向量检索有一个致命弱点：它不擅长精确匹配。

举个例子：用户问"张三的工号是多少？"。向量检索可能会找到所有提到"员工信息"的文档段落，但不一定能找到包含"张三"和具体工号的那个段落。而关键词检索（BM25）可以精确匹配"张三"和"工号"这两个关键词，直接定位到目标文档。

反过来，用户问"公司对远程办公的态度是什么？"，关键词检索可能找不到，因为文档里写的是"灵活办公政策"——语义相近但用词不同。这时候向量检索就能发挥作用。

RAGFlow 的混合检索策略是：两路同时召回，然后用 RRF（Reciprocal Rank Fusion）算法融合排序，最后用 Rerank 模型精排。

4.3 Rerank 的关键作用

Rerank（重排序）是 RAGFlow 检索链路的最后一环，也是提升准确率的关键步骤。

# 使用 Rerank 模型对候选文档精排
from rag.nlp.rerank import RerankModel

reranker = RerankModel("BAAI/bge-reranker-v2-m3")

# 候选文档（来自多路召回融合）
candidates = [
    "公司规定每周可远程办公2天，需提前向直属领导报备...",
    "灵活办公政策适用于全体正式员工，试用期员工不适用...",
    "远程办公期间需保持企业微信在线，会议准时参加..."
]

query = "公司对远程办公的态度是什么？"

# Rerank 模型会对每个 (query, document) 对打分
scores = reranker.rank(query, candidates)
# 输出: [0.85, 0.92, 0.67]

# 按分数重排
ranked_results = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)

RAGFlow 支持多种 Rerank 模型：

• BAAI/bge-reranker-v2-m3：多语言重排模型，中英文效果好
• BAAI/bge-reranker-large：大参数量版本，精度更高
• Cohere rerank-v3：商业 API，效果稳定
• 自定义模型：支持接入任意 HuggingFace Rerank 模型

4.4 知识图谱增强检索（GraphRAG）

RAGFlow 还支持知识图谱增强的检索方式。在文档处理阶段，系统可以自动从文档中抽取实体和关系，构建知识图谱：

# 知识图谱构建流程（简化示意）
# 1. 使用 LLM 从文档 chunk 中抽取实体和关系
# 2. 实体去重和合并
# 3. 构建图结构，存储到 Neo4j 或内置图存储

from rag.graph import KnowledgeGraphBuilder

kg_builder = KnowledgeGraphBuilder(llm="gpt-4o-mini")

# 从文档 chunks 中抽取三元组
triples = kg_builder.extract_triples(chunks)
# 输出: [("公司A", "收购", "公司B"), ("张三", "任职于", "技术部"), ...]

# 构建知识图谱
kg = kg_builder.build(triples)

# 图谱增强检索：当用户提问涉及实体关系时，先从图谱中找到相关实体，
# 再扩展检索范围
def graph_enhanced_search(query, kg, vector_search):
    entities = extract_entities(query)
    related_entities = kg.expand(entities, hops=2)
    expanded_query = query + " " + " ".join(related_entities)
    return vector_search(expanded_query)

这种方式特别适合需要推理的场景，比如"张三的领导是谁？"——系统需要先找到"张三"，再通过"汇报关系"找到他的领导。纯向量检索很难处理这种多跳关系。

五、Agent 编排与工作流引擎

RAGFlow 不仅仅是一个检索引擎，它还内置了 Agent 编排能力。v0.26.x 版本的 Agent 模块已经相当成熟，支持可视化工作流编排。

5.1 工作流节点类型

RAGFlow 的工作流引擎（Flow）基于有向无环图（DAG）设计，每个节点代表一个处理步骤：

# rag/flow/ 节点类型
NODE_TYPES = {
    # 基础节点
    "begin": "开始节点，定义输入变量",
    "answer": "回答节点，输出最终结果",

    # LLM 节点
    "llm": "调用大语言模型",
    "categorize": "使用 LLM 进行意图分类",
    "relevant_judgement": "判断回答是否相关",

    # 检索节点
    "retrieval": "从知识库中检索相关内容",

    # 数据处理节点
    "template": "模板渲染（Jinja2）",
    "json_parser": "JSON 解析和提取",
    "csv_parser": "CSV 数据解析",
    "concat": "字符串拼接",

    # 外部交互节点
    "http_request": "发送 HTTP 请求",
    "code_exec": "执行 Python/JavaScript 代码",
    "email": "发送邮件",

    # 逻辑控制节点
    "switch": "条件分支",
    "variable": "变量赋值",
}

5.2 一个多轮检索 Agent 的实例

下面是一个典型的 RAGFlow 工作流——自适应多轮检索 Agent：

# 工作流定义（JSON 格式，RAGFlow Web UI 可视化编排）
workflow = {
    "nodes": [
        {
            "id": "begin",
            "type": "begin",
            "params": {"query": "{{user_input}}"}
        },
        {
            "id": "classify",
            "type": "categorize",
            "params": {
                "model": "gpt-4o-mini",
                "categories": ["知识库问题", "闲聊", "需要多步推理"],
                "input": "{{begin.query}}"
            }
        },
        {
            "id": "simple_retrieval",
            "type": "retrieval",
            "params": {
                "kb_ids": ["kb_main"],
                "top_k": 5,
                "similarity_threshold": 0.3
            }
        },
        {
            "id": "multi_hop",
            "type": "llm",
            "params": {
                "model": "gpt-4o",
                "prompt": "请将以下问题分解为多个子问题：\n{{begin.query}}",
                "temperature": 0.1
            }
        },
        {
            "id": "judge",
            "type": "relevant_judgement",
            "params": {
                "model": "gpt-4o-mini",
                "question": "{{begin.query}}",
                "answer": "{{simple_retrieval.result}}"
            }
        },
        {
            "id": "answer",
            "type": "answer",
            "params": {
                "input": "{{judge.result}}"
            }
        }
    ],
    "edges": [
        {"from": "begin", "to": "classify"},
        {"from": "classify", "to": "simple_retrieval", "condition": "知识库问题"},
        {"from": "classify", "to": "multi_hop", "condition": "需要多步推理"},
        {"from": "simple_retrieval", "to": "judge"},
        {"from": "judge", "to": "answer"}
    ]
}

5.3 长期记忆模块

RAGFlow 的 Memory 模块解决了 RAG 系统中一个常见问题：跨会话的上下文保持。

传统的 RAG 系统是无状态的——每次对话都是独立的，用户上次说过什么、喜欢什么格式的回答，系统完全不记得。RAGFlow 的 Memory 模块通过以下机制实现长期记忆：

# memory/memory.py 简化示意
class MemoryManager:
    def __init__(self, user_id, llm):
        self.user_id = user_id
        self.llm = llm
        self.store = MemoryStore(user_id)  # Redis/ES 存储

    def remember(self, conversation):
        """从对话中提取值得记忆的信息"""
        # 使用 LLM 从对话中提取关键事实
        facts = self.llm.extract_facts(conversation)
        # 例如：["用户偏好简洁的回答", "用户是Python开发者",
        #        "用户上次问过关于Docker的问题"]

        for fact in facts:
            self.store.add(fact, metadata={
                "timestamp": time.time(),
                "source_conversation": conversation.id
            })

    def recall(self, query, top_k=5):
        """在记忆中检索相关信息"""
        memories = self.store.search(query, top_k=top_k)
        return [m.text for m in memories]

    def inject_into_prompt(self, query, base_prompt):
        """将记忆注入到 Prompt 中"""
        memories = self.recall(query)
        if memories:
            memory_text = "\n".join(f"- {m}" for m in memories)
            return f"{base_prompt}\n\n关于用户的已知信息：\n{memory_text}"
        return base_prompt

六、MCP Server：让 RAGFlow 成为 AI Agent 的知识后端

v0.26.3 版本最重要的更新之一是 MCP Server 集成。这意味着 RAGFlow 不再只是一个独立的 Web 应用，而是可以作为任何 AI Agent 的标准化知识后端。

6.1 什么是 MCP？

MCP（Model Context Protocol）是 Anthropic 在 2024 年底开源的协议，定义了 AI 模型与外部工具、数据源的标准化通信方式。到 2026 年，MCP 已经成为行业标准，Claude Desktop、Cursor、VS Code、JetBrains 等主流客户端都原生支持。

6.2 RAGFlow 的 MCP Server 实现

RAGFlow 通过 MCP 协议暴露了以下工具：

# mcp/tools.py 简化示意
@mcp_tool("retrieve")
def retrieve(kb_ids: list[str], query: str, top_k: int = 5) -> dict:
    """从指定知识库中检索相关内容

    Args:
        kb_ids: 知识库 ID 列表
        query: 检索查询
        top_k: 返回结果数量

    Returns:
        检索结果列表，包含文本内容和相关度分数
    """
    results = hybrid_search.search(kb_ids, query, top_k=top_k)
    return {"results": [r.to_dict() for r in results]}

@mcp_tool("list_knowledge_bases")
def list_knowledge_bases() -> dict:
    """列出所有可用的知识库"""
    kbs = KnowledgeBase.list_all()
    return {"knowledge_bases": [kb.to_dict() for kb in kbs]}

@mcp_tool("ask")
def ask(kb_ids: list[str], question: str) -> dict:
    """基于知识库回答问题（RAG 问答）

    Args:
        kb_ids: 知识库 ID 列表
        question: 用户问题

    Returns:
        包含答案和引用来源的响应
    """
    context = hybrid_search.search(kb_ids, question, top_k=5)
    answer = llm.generate(question=question, context=context)
    return {"answer": answer, "sources": context}

6.3 在 Claude Code 中使用 RAGFlow MCP

配置非常简单，只需在 Claude Desktop 或 Claude Code 的 MCP 配置文件中添加：

{
  "mcpServers": {
    "ragflow": {
      "type": "streamable-http",
      "url": "http://your-ragflow-server:9380/mcp",
      "headers": {
        "Authorization": "Bearer YOUR_API_KEY"
      }
    }
  }
}

配置完成后，Claude 就可以直接调用 RAGFlow 的知识库进行检索和问答了。这意味着你可以让 Claude 在回答问题时自动参考你公司的内部文档、技术规范、产品手册等私有知识。

七、Python SDK 实战：从零构建一个企业知识库

7.1 安装和初始化

pip install ragflow-sdk

from ragflow_sdk import RAGFlow

# 初始化客户端
rag = RAGFlow(
    base_url="http://localhost:9380",
    auth_token="your_api_key"
)

7.2 创建知识库并上传文档

# 创建知识库
kb = rag.create_knowledge_base(
    name="公司技术文档",
    description="包含所有内部技术规范和开发文档",
    embedding_model="BAAI/bge-large-zh-v1.5",  # 中文 Embedding 模型
    chunk_method="smart",  # 智能分块模式
    parser_config={
        "chunk_token_num": 512,
        "delimiter": "\n",
        "layout_recognize": True,  # 启用版面分析
        "html4excel": True,  # Excel 转 HTML 表格
    }
)

print(f"知识库创建成功: {kb.id}")

# 上传文档
documents = kb.upload_documents([
    {"name": "API设计规范.pdf", "blob": open("api_spec.pdf", "rb").read()},
    {"name": "数据库设计文档.docx", "blob": open("db_design.docx", "rb").read()},
    {"name": "部署手册.md", "blob": open("deploy.md", "rb").read()},
])

# 异步解析文档（后台任务）
for doc in documents:
    doc.async_parse()
    print(f"文档 {doc.name} 开始解析...")

# 等待解析完成
import time
while True:
    all_done = all(doc.progress == 1.0 for doc in documents)
    if all_done:
        break
    time.sleep(5)
    for doc in documents:
        doc.update_progress()
        print(f"  {doc.name}: {doc.progress*100:.0f}%")

7.3 检索和问答

# 纯检索模式
results = kb.retrieve(
    query="API 接口的认证方式有哪些？",
    top_k=5,
    similarity_threshold=0.3
)

for r in results:
    print(f"[{r.similarity:.2f}] {r.document_name}")
    print(f"  {r.content[:200]}...")
    print()

# RAG 问答模式
session = rag.create_session("技术问答助手")

for chunk in session.ask(
    question="我们的 API 有哪些认证方式？各有什么特点？",
    kb_ids=[kb.id],
    stream=True  # 流式输出
):
    print(chunk.content, end="", flush=True)
    if chunk.reference:
        print(f"\n[引用: {chunk.reference}]")

7.4 构建一个简单的对话机器人

from ragflow_sdk import RAGFlow
import json

class KnowledgeBot:
    def __init__(self, base_url, token, kb_name):
        self.rag = RAGFlow(base_url=base_url, auth_token=token)
        self.kb = self.rag.list_knowledge_bases(name=kb_name)[0]
        self.session = self.rag.create_session("知识库助手")

    def chat(self, user_input: str) -> str:
        """单轮对话"""
        answer_parts = []
        for chunk in self.session.ask(
            question=user_input,
            kb_ids=[self.kb.id],
            stream=True
        ):
            answer_parts.append(chunk.content)

        return "".join(answer_parts)

    def chat_with_history(self, messages: list) -> str:
        """多轮对话（RAGFlow 自动管理上下文）"""
        user_input = messages[-1]["content"]
        return self.chat(user_input)

# 使用示例
bot = KnowledgeBot(
    base_url="http://localhost:9380",
    token="your_api_key",
    kb_name="公司技术文档"
)

# 交互式对话
while True:
    question = input("用户: ")
    if question.lower() in ("exit", "quit", "q"):
        break
    answer = bot.chat(question)
    print(f"助手: {answer}\n")

八、生产部署指南

8.1 Docker Compose 一键部署

RAGFlow 提供了完整的 Docker Compose 配置，一个命令即可启动所有服务：

# 克隆仓库
git clone https://github.com/infiniflow/ragflow.git
cd ragflow/docker

# 启动所有服务
docker compose -f docker-compose.yml up -d

docker-compose.yml 定义了以下服务：

services:
  ragflow:
    image: infiniflow/ragflow:v0.26.3
    ports:
      - "9380:9380"   # API 端口
      - "80:80"       # Web UI 端口
    depends_on:
      - elasticsearch
      - mysql
      - minio
      - redis
    environment:
      - ES_HOSTS=http://elasticsearch:9200
      - MYSQL_HOST=mysql:3306
      - MINIO_ENDPOINT=minio:9000

  elasticsearch:
    image: elasticsearch:8.11.3
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
    volumes:
      - es_data:/usr/share/elasticsearch/data

  mysql:
    image: mysql:8.0
    environment:
      - MYSQL_ROOT_PASSWORD=your_password
    volumes:
      - mysql_data:/var/lib/mysql

  minio:
    image: quay.io/minio/minio
    command: server /data --console-address ":9001"
    volumes:
      - minio_data:/data

  redis:
    image: redis:7-alpine

8.2 资源配置建议

8.3 性能优化建议

1. Embedding 模型选择：中文场景推荐 BAAI/bge-large-zh-v1.5，英文推荐 text-embedding-3-small，多语言推荐 BAAI/bge-m3
2. Elasticsearch 调优：合理设置分片数（建议单个分片 10-50GB），调整 refresh_interval 为 30s（批量导入时）
3. GPU 加速：OCR 和 Embedding 计算都可以用 GPU 加速，生产环境建议至少一块 GPU
4. 缓存策略：对高频查询启用结果缓存，减少 LLM 调用次数

九、RAGFlow vs Dify vs FastGPT：如何选？

选型建议：

• 如果你的核心需求是从大量复杂文档中准确检索信息（比如法律文档、技术手册、研报），选 RAGFlow
• 如果你需要快速搭建各种 AI 应用（不只是 RAG），选 Dify
• 如果你只是需要一个简单的知识库问答机器人，选 FastGPT

十、v0.26.3 新特性速览

RAGFlow v0.26.3 是一个重要的功能更新版本，核心变化包括：

10.1 MCP Server 新增工具

新增两个 MCP 工具，支持枚举数据集和聊天记录，进一步增强了与外部 Agent 的集成能力。

10.2 SoMark OCR 解析器

布局感知型 OCR 解析器，显著提升了复杂文档（扫描件、图文混排）的解析质量。

10.3 BigQuery 数据源接入

支持从 Google BigQuery 直接导入数据，对企业级数据分析场景更友好。

10.4 Go API 持续扩展

用户注册、搜索、聊天、会话、Agent、Mindmap 等核心 API 逐步用 Go 重写，性能和并发能力大幅提升。

10.5 批量上传容错

单文件失败不再拖垮整个批次，支持部分成功模式。

十一、总结与展望

RAGFlow 在 2026 年的 RAG 赛道中，走出了一条差异化的路。当其他产品在比拼"低代码"、"可视化"、"全栈能力"的时候，RAGFlow 选择了一条更硬核的路：把文档理解和检索质量做到极致。

DeepDoc 的存在让 RAGFlow 在处理复杂文档时具有不可替代的优势。混合检索引擎和知识图谱的结合让检索结果更加准确。MCP Server 的集成让它成为了 AI Agent 生态中的重要一环。

对于开发者而言，RAGFlow 的价值在于：它不只是让你"能用"RAG，而是让你的 RAG 系统真正"好用"。当你的文档不是简单的纯文本，而是包含表格、图片、公式、多栏版面的复杂 PDF 时，RAGFlow 几乎是目前唯一一个能从源码层面解决这些问题的开源方案。

未来，随着 RAG 技术从"检索增强生成"向"Context Engineering"演进，RAGFlow 这种重视底层文档理解的思路，可能会成为更多产品的参考范式。毕竟，再高级的 AI 模型，也需要高质量的上下文才能发挥真正的价值。

项目地址：https://github.com/infiniflow/ragflow

官方文档：https://ragflow.io

许可证：Apache 2.0