Article

模型工作流 RAGFlow

更新于：2026-07-20

第一章：RAGFlow 概述与核心理念

1.1 什么是 RAGFlow

概念名称	说明	注意事项
RAGFlow	一种基于检索增强生成（RAG）架构的低代码/可视化工作流引擎，支持灵活编排文档处理、检索、生成等环节	RAGFlow 并非单一模型，而是一套端到端的系统框架
检索增强生成（RAG）	通过从外部知识库中检索相关信息，并将其作为上下文输入给大语言模型，以提升回答准确性与事实性	RAG 的效果高度依赖检索质量与上下文整合方式
可视化工作流	用户可通过图形界面拖拽组件定义数据处理流程，无需编写完整代码即可构建复杂 RAG 系统	部分高级功能仍需通过代码或配置文件扩展
模块化设计	RAGFlow 将加载、分块、嵌入、检索、生成等步骤拆分为独立可替换模块，便于定制与调试	模块间需保持接口兼容，避免版本错配

1.2 RAGFlow 与传统 RAG 的区别

概念名称	说明	注意事项
传统 RAG	通常指固定流程的检索-生成管道，如”加载→分块→嵌入→检索→LLM生成”，流程硬编码、难以调整	难以适应多源异构数据或复杂业务逻辑
RAGFlow 的动态编排	支持用户自定义节点顺序、条件分支、循环等控制结构，实现灵活的数据流与控制流	需理解工作流语义，避免逻辑死循环或数据丢失
组件可插拔性	所有处理单元（如分块器、嵌入模型、重排序器）均可独立替换或升级	替换组件时需验证输入输出格式一致性
内置评估与调试工具	提供检索结果可视化、中间变量查看、延迟/准确率监控等功能	调试模式可能影响性能，生产环境建议关闭
元数据驱动	支持在文档分块时注入来源、章节、时间等元数据，并在检索/生成阶段利用这些信息	元数据需结构化且语义清晰，否则难以有效利用

1.3 RAGFlow 的典型应用场景

场景名称	说明	注意事项
企业知识库问答系统	基于内部文档（如手册、合同、邮件）构建智能客服或员工助手，确保回答基于权威资料	需做好权限控制与敏感信息过滤
学术文献智能分析	自动解析论文 PDF，支持跨文献检索与综述生成	学术文本结构复杂，需定制分块与解析策略
法律/合规咨询助手	结合法律法规、判例库进行精准问答，辅助律师或合规人员	对引用准确性要求极高，需启用高精度检索与溯源
多轮对话式产品支持	在用户交互中动态检索产品文档，生成连贯、上下文相关的解答	需维护对话历史并合理压缩上下文长度
实时新闻摘要与溯源	接入新闻流，自动聚合事件、生成摘要并标注信息来源	数据时效性强，需支持增量索引更新与去重

第二章：环境搭建与项目初始化

2.1 系统依赖与环境要求

依赖项名称	说明	注意事项
Python 版本	推荐 Python 3.9–3.11	不兼容 Python <3.8 或 ≥3.12，可能引发依赖冲突
操作系统	支持 Linux（Ubuntu/CentOS）、macOS，Windows 需 WSL2	Windows 原生环境未官方支持，建议使用 WSL2
GPU 支持（可选）	若使用本地嵌入模型或 LLM（如 bge、ChatGLM），需 CUDA 11.8+ 及 NVIDIA 驱动	CPU 模式可运行但速度较慢，仅适合开发测试
内存要求	最低 8GB RAM，推荐 16GB+（尤其加载大模型或处理大量文档时）	内存不足可能导致 OOM 或进程崩溃
磁盘空间	至少 10GB 可用空间（用于模型缓存、向量数据库、日志等）	向量索引和原始文档会持续占用存储
Docker（可选）	官方提供 Docker 镜像，简化部署	使用 Docker 时需确保版本 ≥20.10，并启用 BuildKit
网络环境	首次安装需访问 Hugging Face、PyPI 等源下载模型与包	企业内网需配置代理或离线包

2.2 安装 RAGFlow

方法名称	语法 / 操作命令	用途	注意事项
pip 安装（推荐）	`pip install ragflow`	从 PyPI 安装最新稳定版	确保虚拟环境已激活，避免全局污染
源码安装	`git clone https://github.com/infiniflow/ragflow.git` `cd ragflow` `pip install -e .`	安装开发版，便于调试或贡献代码	需手动安装额外依赖（如 requirements.txt）
Docker 安装	`docker pull infiniflow/ragflow:latest` `docker run -p 9380:9380 infiniflow/ragflow`	快速启动服务，无需配置本地环境	首次运行会自动下载模型，耗时较长
指定版本安装	`pip install ragflow==0.12.0`	固定版本用于生产环境一致性	查看 GitHub Releases 获取可用版本号
安装带 GPU 支持	`pip install ragflow[torch-gpu]`	启用 CUDA 加速嵌入与推理	需提前安装对应版本的 PyTorch with CUDA
安装中文支持组件	`pip install ragflow[chinese]`	包含中文分词、排版处理等扩展	适用于处理中文 PDF/Word 文档

2.3 初始化项目结构

步骤名称	操作细节	注意事项
创建项目目录	`mkdir my_rag_project && cd my_rag_project`	建议使用独立目录避免文件混杂
初始化配置文件	`ragflow init`	自动生成 config.yaml、.env、workflow/ 等基础结构
配置环境变量	在 `.env` 文件中设置： `LLM_API_KEY=your_key` `EMBEDDING_MODEL=bge-large-zh` `VECTOR_DB=chroma`	敏感信息勿提交至版本控制
目录结构说明	自动生成： - `config.yaml`：主配置 - `data/`：原始文档存放 - `workflow/`：工作流定义 - `logs/`：日志输出 - `models/`：本地模型缓存	可手动调整路径，但需同步修改 config.yaml
验证初始化成功	`ragflow check`	检查依赖、模型、数据库连接是否正常
启动本地服务	`ragflow serve --port 9380`	默认启动 Web UI 与 API 服务

第三章：文档加载与预处理

3.1 支持的文档格式与加载器

方法名称（加载器）	语法 / 调用方式	用途	代码示例	注意事项
PDFLoader	`from ragflow.loaders import PDFLoader` `loader = PDFLoader(file_path)`	加载 PDF 文档，保留文本与基础布局	`loader = PDFLoader("manual.pdf")` `docs = loader.load()`	扫描版 PDF 需 OCR 支持（需额外配置 Tesseract）
DocxLoader	`from ragflow.loaders import DocxLoader` `loader = DocxLoader(file_path)`	解析 .docx 文件，提取正文、标题、列表等	`loader = DocxLoader("report.docx")` `docs = loader.load()`	不支持旧版 .doc 格式
TxtLoader	`from ragflow.loaders import TxtLoader` `loader = TxtLoader(file_path)`	读取纯文本文件	`loader = TxtLoader("notes.txt")` `docs = loader.load()`	默认编码 UTF-8，非 UTF 文件需指定 encoding 参数
HtmlLoader	`from ragflow.loaders import HtmlLoader` `loader = HtmlLoader(file_path)`	提取 HTML 中的正文内容，去除标签	`loader = HtmlLoader("page.html")` `docs = loader.load()`	可通过 select 参数指定 CSS 选择器保留特定区域
CSVLoader	`from ragflow.loaders import CSVLoader` `loader = CSVLoader(file_path, columns=["Q", "A"])`	将 CSV 行转为问答对或段落	`loader = CSVLoader("faq.csv", columns=["question", "answer"])` `docs = loader.load()`	必须指定参与加载的列名
DirectoryLoader	`from ragflow.loaders import DirectoryLoader` `loader = DirectoryLoader(dir_path, glob="*.pdf")`	批量加载目录下指定类型文件	`loader = DirectoryLoader("docs/", glob="*.docx")` `docs = loader.load()`	支持通配符，但不递归子目录（除非设置 recursive=True）
UnstructuredLoader	`from ragflow.loaders import UnstructuredLoader` `loader = UnstructuredLoader(file_path, strategy="fast")`	通用加载器，支持 PPT、Excel、图像等	`loader = UnstructuredLoader("slides.pptx")` `docs = loader.load()`	依赖 unstructured 库，首次使用需安装额外依赖

3.2 文本分块策略

方法名称（分块器）	语法 / 调用方式	用途	代码示例	注意事项
RecursiveCharacterTextSplitter	`from ragflow.splitters import RecursiveCharacterTextSplitter` `splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)`	按字符递归切分，优先保留段落完整性	`splitter = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=60)` `chunks = splitter.split_documents(docs)`	默认分隔符：[“\n\n”, “\n”, ” ”, ""]，适合多数文本
SemanticChunker	`from ragflow.splitters import SemanticChunker` `splitter = SemanticChunker(embedding_model="bge-small")`	基于语义相似度动态合并句子形成语义块	`splitter = SemanticChunker(embedding_model="text-embedding-ada-002")` `chunks = splitter.split_documents(docs)`	计算开销大，适合高精度场景；需预加载嵌入模型
MarkdownHeaderTextSplitter	`from ragflow.splitters import MarkdownHeaderTextSplitter` `splitter = MarkdownHeaderTextSplitter(headers_to_split_on=[("#", "Header1"), ("##", "Header2")])`	按 Markdown 标题层级切分，保留结构信息	`splitter = MarkdownHeaderTextSplitter(headers_to_split_on=[("#", "Chapter")])` `chunks = splitter.split_text(md_text)`	仅适用于 Markdown 源文件
TokenTextSplitter	`from ragflow.splitters import TokenTextSplitter` `splitter = TokenTextSplitter(encoding_name="cl100k_base", chunk_size=400)`	按 token 数精确切分，适配 LLM 上下文限制	`splitter = TokenTextSplitter(chunk_size=800, chunk_overlap=100)` `chunks = splitter.split_text(long_text)`	encoding_name 需匹配目标 LLM（如 gpt-4 用 cl100k_base）
CustomRuleSplitter	`from ragflow.splitters import CustomRuleSplitter` `splitter = CustomRuleSplitter(separators=["\nQ:", "\nA:"], keep_separator=True)`	用户自定义分隔符规则	`splitter = CustomRuleSplitter(separators=["---", "【案例】"])` `chunks = splitter.split_text(text)`	分隔符顺序影响切分结果，建议从粗到细排列

3.3 元数据注入与管理

方法名称 / 操作	语法 / 操作细节	用途	代码示例	注意事项
手动添加元数据	`doc.metadata.update({"source": "policy_v2.pdf", "section": "3.1"})`	为单个文档片段附加来源、章节等信息	`docs[0].metadata["author"] = "HR Dept"`	metadata 为 dict 类型，键值需为字符串或简单类型
加载时自动注入	`loader = PDFLoader("guide.pdf", extract_metadata=True)`	自动提取文件名、页码、创建时间等	`loader = DocxLoader("plan.docx", metadata={"department": "Finance"})`	并非所有加载器都支持 extract_metadata
分块时继承元数据	`chunks = splitter.split_documents(docs)`	分块后每个 chunk 自动继承原始 doc 的 metadata	—	若原始 doc 无 metadata，则 chunk 为空 dict
动态元数据生成	`for doc in docs:` `doc.metadata["ingest_time"] = datetime.now().isoformat()`	在预处理流水线中注入时间戳、版本号等	`doc.metadata["doc_id"] = hashlib.md5(doc.page_content.encode()).hexdigest()`	避免在 metadata 中存入大对象（如图片）
元数据过滤检索	`retriever.search(query, filter={"department": "Legal"})`	检索时限定特定元数据条件	`results = db.similarity_search(query, k=3, filter={"source_type": "contract"})`	向量数据库需支持元数据过滤（如 Chroma、Qdrant）
元数据持久化	向量数据库自动存储 metadata 字段	确保检索结果可溯源	—	导出/备份时需包含 metadata 字段，否则丢失上下文

第四章：向量嵌入与索引构建

4.1 嵌入模型选择与配置

方法名称（嵌入模型）	语法 / 配置方式	用途	代码示例	注意事项
HuggingFaceEmbeddings	`from ragflow.embeddings import HuggingFaceEmbeddings` `embedder = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5")`	使用开源 Hugging Face 模型生成嵌入	`embedder = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")` `vectors = embedder.embed_documents(texts)`	首次运行自动下载模型；需磁盘空间 ≥1GB；支持本地路径
OpenAIEmbeddings	`from ragflow.embeddings import OpenAIEmbeddings` `embedder = OpenAIEmbeddings(api_key="sk-xxx", model="text-embedding-ada-002")`	调用 OpenAI API 生成高质量嵌入	`embedder = OpenAIEmbeddings(model="text-embedding-3-large")` `vec = embedder.embed_query("What is RAG?")`	需有效 API Key；产生调用费用；注意速率限制
LocalEmbeddingWrapper	`from ragflow.embeddings import LocalEmbeddingWrapper` `embedder = LocalEmbeddingWrapper(model_path="/models/bge-small")`	加载本地 ONNX 或 GGUF 格式的量化模型	`embedder = LocalEmbeddingWrapper(model_path="./bge-m3-onnx")`	需提前转换模型格式；适合离线/私有部署
CohereEmbeddings	`from ragflow.embeddings import CohereEmbeddings` `embedder = CohereEmbeddings(cohere_api_key="xxx", model="embed-multilingual-v3.0")`	支持多语言嵌入，适用于国际化场景	`embedder = CohereEmbeddings(model="embed-english-v3.0")`	需注册 Cohere 账号；部分模型需指定 input_type（如 “search_document”）
EmbeddingConfig（全局）	在 `config.yaml` 中设置： `embedding:` `provider: "huggingface"` `model_name: "BAAI/bge-reranker-base"`	统一配置项目默认嵌入模型	—	修改后需重启服务或重新初始化 pipeline

4.2 向量数据库集成

数据库名称	语法 / 初始化方式	用途	代码示例	注意事项
Chroma	`from ragflow.vectorstores import Chroma` `db = Chroma(persist_directory="./chroma_db", embedding_function=embedder)`	轻量级本地向量库，适合开发与小规模部署	`db = Chroma.from_documents(docs, embedder, persist_directory="./db")`	默认内存模式，需显式调用 persist() 保存；不支持生产级高并发
Qdrant	`from ragflow.vectorstores import Qdrant` `db = Qdrant(host="localhost", port=6333, collection_name="rag_docs", embedding_dim=1024)`	高性能开源向量数据库，支持过滤与分片	`client = QdrantClient(host="qdrant.example.com")` `db = Qdrant(client, "my_collection", embedder)`	需单独部署 Qdrant 服务；支持云托管（Qdrant Cloud）
Milvus	`from ragflow.vectorstores import Milvus` `db = Milvus(embedding_function=embedder, connection_args={"host": "localhost", "port": "19530"})`	企业级分布式向量数据库	`db = Milvus.from_documents(docs, embedder, collection_name="kb_v1")`	部署复杂，需 etcd、MinIO 等依赖；适合超大规模场景
FAISS	`from ragflow.vectorstores import FAISS` `db = FAISS.from_documents(docs, embedder)`	Facebook 开源，纯内存索引，检索极快	`db.save_local("./faiss_index")` `loaded_db = FAISS.load_local("./faiss_index", embedder)`	不支持动态更新；仅适合静态知识库；无元数据过滤（除非用 FAISS+HNSW）
Elasticsearch	`from ragflow.vectorstores import ElasticsearchStore` `db = ElasticsearchStore(es_url="http://localhost:9200", index_name="ragflow", embedding=embedder)`	结合全文检索与向量检索	`db.add_documents(docs)`	需启用 dense_vector 字段；适合混合检索（关键词+语义）

4.3 构建与更新索引

操作名称	操作细节	用途	代码示例	注意事项
从文档构建索引	`vectorstore = VectorStore.from_documents(documents, embedding_model)`	初次加载文档并生成向量索引	`db = Chroma.from_documents(chunks, embedder, persist_directory="./index")`	自动调用 embedder；耗时取决于文档量与模型速度
增量添加文档	`vectorstore.add_documents(new_docs)`	动态扩展知识库，无需重建全量索引	`db.add_documents([new_doc])`	所有向量数据库均支持；FAISS 需使用 IndexFlatIP 或特殊策略
删除指定文档	`vectorstore.delete(ids=["doc_id_123"])`	移除过期或错误内容	`db.delete(ids=["policy_2023_v1"])`	并非所有后端都支持（如基础 FAISS 不支持）
更新文档（先删后增）	`vectorstore.delete(ids=[old_id])` `vectorstore.add_documents([updated_doc])`	实现文档内容更新	`db.delete(ids=["faq_001"])` `db.add_documents([revised_faq])`	需确保 updated_doc 的 metadata 包含新 ID
异步批量构建	`ragflow index build --dir ./data --batch-size 100 --workers 4`	命令行工具支持后台批量索引	`ragflow index update --collection legal_kb`	适合处理 GB 级文档；避免阻塞主进程
索引持久化	`vectorstore.persist()` （Chroma）或自动保存（Qdrant/Milvus）	确保重启后索引不丢失	`db.persist()`	Chroma 必须显式调用 persist()；否则仅内存存在
索引版本管理	通过 `collection_name` 或 `index_tag` 区分不同版本	支持 A/B 测试或多环境隔离	`prod_db = Qdrant(..., collection_name="kb_v2_prod")`	建议结合 CI/CD 流程自动化版本切换

第五章：检索机制详解

5.1 检索器类型与配置

检索器名称	语法 / 初始化方式	用途	代码示例	注意事项
VectorStoreRetriever	`from ragflow.retrievers import VectorStoreRetriever` `retriever = VectorStoreRetriever(vectorstore=db, search_type="similarity", k=4)`	基于向量相似度的标准检索器	`retriever = VectorStoreRetriever(db, search_kwargs={"k": 5})`	search_type 支持 “similarity”、“mmr” 等
BM25Retriever	`from ragflow.retrievers import BM25Retriever` `retriever = BM25Retriever.from_documents(docs, k=3)`	基于关键词频率的传统稀疏检索	`retriever = BM25Retriever.from_texts(texts, metadatas=metas, k=4)`	适合处理术语匹配（如法律条文编号）；不依赖嵌入模型
HybridRetriever	`from ragflow.retrievers import HybridRetriever` `retriever = HybridRetriever(vector_retriever=vs_ret, bm25_retriever=bm25_ret, weight=0.7)`	融合向量与关键词检索结果	`retriever = HybridRetriever(vs_ret, bm25_ret, weight=0.6)`	weight 表示向量检索权重（0~1），需调优
ParentDocumentRetriever	`from ragflow.retrievers import ParentDocumentRetriever` `retriever = ParentDocumentRetriever(vectorstore=db, docstore=doc_store, child_splitter=child_splitter)`	先检索小块，再返回其所属完整父文档	`retriever = ParentDocumentRetriever(db, full_doc_store, child_splitter)`	需预先存储父文档；适用于需要上下文完整的场景
ContextualCompressionRetriever	`from ragflow.retrievers import ContextualCompressionRetriever` `retriever = ContextualCompressionRetriever(base_retriever=base_ret, compressor=compressor)`	对检索结果进行相关性压缩，去除冗余	`compressor = LLMChainExtractor.from_llm(llm)` `retriever = ContextualCompressionRetriever(base_ret, compressor)`	依赖 LLM，增加延迟；适合减少 token 消耗

5.2 多路召回与重排序

方法名称	语法 / 调用方式	用途	代码示例	注意事项
MultiRetriever	`from ragflow.retrievers import MultiRetriever` `retriever = MultiRetriever(retrievers=[ret1, ret2], weights=[0.6, 0.4])`	并行调用多个检索器并加权融合结果	`retriever = MultiRetriever([bm25_ret, vs_ret], weights=[0.3, 0.7])`	结果自动去重；weights 总和建议为 1
RerankWithModel	`from ragflow.rerankers import RerankWithModel` `reranker = RerankWithModel(model_name="BAAI/bge-reranker-v2-m3")`	使用专用重排序模型对初检结果重新打分	`reranked = reranker.rerank(query, docs, top_k=3)`	重排序模型输入为 (query, doc) 对；计算开销较大
CohereReranker	`from ragflow.rerankers import CohereReranker` `reranker = CohereReranker(api_key="xxx", model="rerank-english-v3.0")`	调用 Cohere API 进行高精度重排序	`results = reranker.compress_documents(docs, query)`	需网络连接；按 token 计费
ReciprocalRankFusion	`from ragflow.fusion import ReciprocalRankFusion` `fuser = ReciprocalRankFusion(k=60)`	无监督融合多路召回结果（基于排名倒数）	`fused = ReciprocalRankFusion().fuse([bm25_res, vs_res])` `final_docs = fuser.fuse([results1, results2])`	无需训练；对不同检索器输出鲁棒性强
LLMReranker	`from ragflow.rerankers import LLMReranker` `reranker = LLMReranker(llm=llm, prompt_template="...")`	利用大模型判断文档与问题的相关性	`reranker = LLMReranker(llm, prompt="Is this doc relevant to: {query}?")`	可定制判断逻辑；但延迟高、成本高，仅用于关键场景

5.3 检索结果过滤与评分

操作名称	操作细节	用途	代码示例	注意事项
元数据过滤	`retriever.get_relevant_documents(query, filter={"department": "Finance"})`	限定检索范围（如部门、文档类型、时间）	`results = db.similarity_search(query, filter={"year": {"$gte": "2023"}})`	向量库需支持过滤（Chroma/Qdrant/Milvus 支持，FAISS 不支持）
相似度阈值截断	设置 `score_threshold` 参数	排除低相关性结果	`retriever = VectorStoreRetriever(db, search_kwargs={"score_threshold": 0.6})`	余弦相似度通常在 0~1 之间；阈值需根据模型校准
动态评分调整	自定义 `post_process` 函数对 scores 加权	结合业务规则调整排序（如优先内部文档）	`def adjust(doc): return doc.score * (1.2 if doc.metadata.get("internal") else 1.0)`	需在检索后手动实现；不影响原始向量分数
时间衰减加权	在 metadata 中包含 timestamp，并在融合时降低旧文档权重	使新文档在排序中占优	`weight = exp(-lambda * (now - doc_time))`	适用于新闻、公告等时效性强的内容
检索结果去重	使用 `unique=True` 或 post-hoc deduplication	避免相同内容多次出现	`from ragflow.utils import dedup_by_content` `unique_docs = dedup_by_content(docs)`	可基于文本哈希或语义嵌入去重；注意性能开销

第六章：大模型交互与提示工程

6.1 LLM 接入方式

方法名称（LLM 封装类）	语法 / 初始化方式	用途	代码示例	注意事项
OpenAILLM	`from ragflow.llms import OpenAILLM` `llm = OpenAILLM(api_key="sk-xxx", model="gpt-4o", temperature=0.3)`	接入 OpenAI 官方 API	`response = llm.invoke("Explain RAG in one sentence.")`	需有效 API Key；注意费用与速率限制；支持流式输出
HuggingFaceLLM	`from ragflow.llms import HuggingFaceLLM` `llm = HuggingFaceLLM(model_name="meta-llama/Llama-3-8b-chat-hf", device="cuda")`	加载开源 Hugging Face 模型进行本地推理	`llm = HuggingFaceLLM(model_name="Qwen/Qwen2-7B-Instruct", max_new_tokens=512)`	首次运行下载模型；需足够 GPU 显存；可量化加速
OllamaLLM	`from ragflow.llms import OllamaLLM` `llm = OllamaLLM(model="llama3:8b", base_url="http://localhost:11434")`	通过 Ollama 服务调用本地或远程 LLM	`llm = OllamaLLM(model="qwen:7b", temperature=0.5)`	需提前运行 ollama serve；适合快速测试开源模型
TongyiQwenLLM	`from ragflow.llms import TongyiQwenLLM` `llm = TongyiQwenLLM(api_key="sk-xxx", model="qwen-max")`	接入阿里云通义千问 API	`llm = TongyiQwenLLM(model="qwen-plus", top_p=0.8)`	需开通阿里云百炼平台；支持私有化部署版本
CustomAPILLM	`from ragflow.llms import CustomAPILLM` `llm = CustomAPILLM(api_url="https://my-llm/api/v1/chat", headers={"Authorization": "Bearer xxx"})`	对接私有 LLM 服务（符合 OpenAI 兼容协议）	`llm = CustomAPILLM(api_url="http://llm.internal:8000/v1", model="custom-rag-7b")`	要求接口返回格式与 OpenAI Chat Completions 一致

6.2 动态提示模板设计

方法名称	语法 / 使用方式	用途	代码示例	注意事项
PromptTemplate	`from ragflow.prompts import PromptTemplate` `template = PromptTemplate(template="回答问题：{question}\n参考：{context}")`	定义静态占位符模板	`prompt = template.format(question="什么是RAG?", context="RAG是...")`	占位符必须与 format 参数名一致
FewShotPromptTemplate	`from ragflow.prompts import FewShotPromptTemplate` `prompt = FewShotPromptTemplate(examples=examples, suffix="问题：{input}")`	提供少量示例引导模型行为	`examples = [{"input": "X", "output": "Y"}, ...]` `prompt = FewShotPromptTemplate(examples=demo_qa, input_variables=["input"])`	示例质量直接影响效果；避免过长
ConditionalPrompt	自定义函数动态选择模板	根据问题类型切换不同提示策略	`def get_prompt(q): return tech_template if "技术" in q else policy_template`	需在 pipeline 中集成逻辑判断
SystemMessageWrapper	`from ragflow.prompts import SystemMessageWrapper` `wrapper = SystemMessageWrapper(system_message="你是一个企业知识助手，请基于文档回答。")`	注入系统角色指令	`messages = wrapper.wrap(user_query="报销流程？")`	仅适用于支持 system role 的模型（如 GPT、Qwen）
TemplateRegistry	在 `config.yaml` 中定义多个模板并按名称调用	统一管理多场景提示模板	`prompt = registry.get("legal_qa").format(context=c, question=q)`	支持 YAML 配置热加载；便于非开发人员维护

6.3 上下文压缩与注入策略

方法名称	语法 / 调用方式	用途	代码示例	注意事项
LLMChainExtractor	`from ragflow.compressors import LLMChainExtractor` `compressor = LLMChainExtractor.from_llm(llm)`	让 LLM 判断每段是否相关并提取关键句	`extractor = LLMChainExtractor.from_llm(OpenAILLM(model="gpt-3.5-turbo"))` `compressed_docs = compressor.compress_documents(docs, query)`	增加一次 LLM 调用；适合高价值场景
EmbeddingsFilter	`from ragflow.compressors import EmbeddingsFilter` `filter = EmbeddingsFilter(embeddings=embedder, similarity_threshold=0.5)`	基于嵌入相似度过滤低相关文档	`ef = EmbeddingsFilter(embedder, threshold=0.6)` `filtered_docs = filter.filter_documents(docs, query)`	速度快；但无法处理语义复杂性（如否定、对比）
ContextualMaxSize	自定义截断逻辑：按 token 数从后往前保留	确保上下文不超过 LLM 最大窗口	`from ragflow.utils import truncate_context` `ctx = truncate_context(docs, max_tokens=3000, tokenizer=tokenizer)`	优先保留靠前或靠后的 chunk 可配置；默认保留最新/最相关
ReorderRelevantFirst	对检索结果按相关性得分降序排列后再拼接	提高关键信息在上下文中的位置权重	`docs_sorted = sorted(docs, key=lambda x: x.metadata.get("score", 0), reverse=True)`	多数 LLM 对靠前信息更敏感
ChunkSummaryInjection	先对每个 chunk 生成摘要，再将摘要+原始内容注入	平衡信息密度与完整性	`summaries = [llm.invoke(f"Summarize: {d.page_content}") for d in docs]` `context = "\n".join([f"[SUMMARY]{s}\n[DETAIL]{d.page_content}" for s,d in zip(summaries, docs)])`	显著增加延迟和成本；仅用于极长上下文

第七章：流程编排与执行引擎

7.1 工作流定义语法

定义方式	语法 / 文件格式示例	用途	示例片段	注意事项
YAML 声明式	`workflow.yaml`	以配置文件形式定义端到端流程	`name: rag_qa_flow` `nodes:` `- id: loader` `type: PDFLoader` `params: {path: "docs/"}`	支持变量插值（如 `${env.DATA_DIR}`）；适合 DevOps 集成
Python DSL	`from ragflow.workflow import Workflow` `wf = Workflow("qa_pipeline")`	通过代码动态构建复杂逻辑	`wf.add_node("embed", HuggingFaceEmbedder())` `wf.add_edge("split", "embed")`	可嵌入条件分支、循环；适合开发调试
JSON Schema	符合 RAGFlow Workflow Schema 的 JSON 对象	供 Web UI 或 API 动态生成流程	`{"nodes": [{"id":"retriever","type":"VectorStoreRetriever",...}]}`	需校验 schema；便于前端可视化编辑
DAG 图形化导出	`ragflow workflow export --format=mermaid`	将工作流导出为 Mermaid 或 Graphviz 可视化	`ragflow workflow visualize my_flow.yaml`	用于文档生成或团队对齐；不支持执行
模板继承机制	`extends: base_rag.yaml`	复用通用流程（如”带重排序的RAG”）	`extends: standard_rag.yaml` `overrides:` `retriever.k: 5`	避免重复定义；支持多层继承

7.2 节点类型与连接规则

节点类型	输入类型	输出类型	连接规则说明	示例节点	注意事项
DocumentLoader	无（或目录路径）	`List[Document]`	必须作为流程起点	PDFLoader, DirectoryLoader	不接受上游输入
TextProcessor	`List[Document]`	`List[Document]`	可链式连接多个处理器	MetadataInjector, Cleaner	输出 metadata 可被后续节点读取
Splitter	`List[Document]`	`List[Chunk]`	通常接在 Loader 或 Processor 后	RecursiveCharacterTextSplitter	每个 Chunk 继承原始 metadata
Embedder	`List[Chunk]`	`List[Vector + Chunk]`	输出需接入向量数据库或检索器	HuggingFaceEmbeddings	耗时操作，建议异步
Retriever	`Query (str)`	`List[Chunk]`	查询入口节点，常由外部触发	VectorStoreRetriever	可配置 k、filter 等参数
LLMNode	`Context + Query`	`str (回答)`	通常位于流程末端	OpenAILLM, QwenLLM	支持流式输出回调
Reranker	`List[Chunk] + Query`	`List[Chunk] (重排后)`	必须接在 Retriever 之后	BGEReranker	输入需包含原始 scores
ConditionalRouter	`Any`	分支输出	根据条件路由到不同下游	LangDetectRouter, TopicRouter	需定义明确的路由规则
连接约束	—	—	- 类型兼容（如 Document → Splitter） - 无环（DAG） - 单入口多出口允许	—	循环依赖将导致执行失败

7.3 异步执行与状态管理

机制名称	实现方式 / API	用途	代码/命令示例	注意事项
异步任务提交	`job_id = workflow.run_async(inputs={"query": "报销流程？"})`	非阻塞执行长流程	`result = await workflow.wait(job_id)`	返回 job_id 用于查询状态
状态持久化	后端自动将中间结果存入 StateStore（如 Redis / PostgreSQL）	支持断点续跑、审计追踪	`state = workflow.get_state(job_id)`	敏感数据需加密存储
进度回调	`workflow.on_progress(lambda step, data: logger.info(f"{step} done"))`	实时监控执行进度	`workflow.run(..., callbacks=[MyCallback()])`	回调函数不得阻塞主线程
错误恢复策略	`retry_policy: {max_retries: 3, backoff: "exponential"}`	自动重试失败节点（如网络超时）	在节点配置中设置：`retries: 2`	不适用于逻辑错误（如格式解析失败）
分布式执行	部署 Worker 节点并连接消息队列（如 Celery + RabbitMQ）	横向扩展高并发请求	`ragflow worker --queue gpu_tasks`	需统一共享存储（如 NFS）存放模型/数据
状态快照导出	`ragflow job export --job-id j123 --output snapshot.json`	用于调试或回放	—	包含所有中间输出与元数据
超时控制	`timeout: 300s` （在 workflow 或节点级别配置）	防止流程无限挂起	`node_config: {timeout: 60}`	超时后标记为 FAILED 并触发告警

第八章：评估、调试与优化

8.1 检索质量评估指标

指标名称	计算方式 / 定义	用途	代码示例	注意事项
Hit Rate @k	在 top-k 检索结果中至少包含一个相关文档的比例	衡量基础召回能力	`hit_rate = sum(1 for q in queries if any(is_relevant(d, q) for d in retriever.get_relevant_documents(q, k=3))) / len(queries)`	需人工或自动标注”相关”文档
MRR (Mean Reciprocal Rank)	对每个查询，取第一个相关文档的排名倒数，再求平均	关注最相关结果是否排在前面	`mrr = np.mean([1/rank for rank in ranks if rank > 0])`	对排名敏感；适合单答案场景
Recall@k	检索到的相关文档数 / 总相关文档数（上限为 k）	评估覆盖度	`recall = len(retrieved_relevant) / len(ground_truth_relevant)`	需完整相关集（常用于封闭域 QA）
Precision@k	检索到的相关文档数 / k	衡量结果纯净度	`precision = len([d for d in top_k if is_relevant(d)]) / k`	高 precision 可能牺牲 recall
nDCG@k	考虑相关性等级的加权排序指标（如 0/1/2 分）	适用于多级相关性标注	使用 `sklearn.metrics.ndcg_score`	需细粒度标注；比 binary 指标更真实
Embedding Drift	监控嵌入向量分布变化（如 PCA 投影均值偏移）	检测模型更新或数据漂移导致的性能下降	`drift = cosine_similarity(mean_old, mean_new)`	适用于持续监控；非直接效果指标

8.2 端到端效果调试工具

工具名称	调用方式 / 功能描述	用途	示例命令 / 代码	注意事项
RAGDebugger	`from ragflow.debug import RAGDebugger` `debugger = RAGDebugger(workflow)` `debugger.trace(query="...")`	可视化全流程中间输出	`trace = debugger.trace("合同审核要点？")` `print(trace.retrieved_docs)`	自动记录各节点输入/输出及耗时
LLM-as-a-Judge	使用强 LLM（如 GPT-4o）对回答进行自动评分	无需人工即可评估生成质量	`score = llm_judge.evaluate(answer, ground_truth, query)`	prompt 设计关键；存在评估偏差风险
Groundedness Checker	检查生成答案是否可由检索到的上下文支持	识别幻觉（Hallucination）	`is_grounded = check_groundedness(answer, retrieved_contexts)`	可基于 NLI 模型或 LLM 实现
Query Rewriter Analyzer	对比原始查询与重写后查询的检索效果	诊断查询理解模块是否有效	`orig_res = retriever.invoke(q); rw_res = retriever.invoke(rewritten_q)`	常用于多跳问答或模糊查询场景
A/B Test Runner	并行运行两个工作流版本并对比指标	科学验证优化方案有效性	`ab_test.run(variant_a, variant_b, test_queries)`	需足够样本量；避免时间偏差
Log Replay	回放生产日志中的 query-response 对进行离线评估	利用真实用户行为数据做回归测试	`ragflow eval replay --log-file prod.log --metrics hit_rate,mrr`	需脱敏处理；注意隐私合规

8.3 性能优化技巧

优化方向	具体策略	适用场景	实施示例	注意事项
向量索引加速	使用 HNSW 或 IVF 索引替代暴力搜索	大规模向量库（>10万条）	Qdrant: `hnsw_config = {"m": 16, "ef_construct": 100}`	精度 vs 速度需权衡；HNSW 不支持动态删除
嵌入缓存	对重复文本或查询缓存嵌入结果	用户频繁提问相似问题	`embedder = CachedEmbeddings(HuggingFaceEmbeddings(), cache_backend=RedisCache())`	缓存键建议用文本哈希；注意内存占用
异步批处理	将多个查询合并为 batch 请求嵌入或 LLM	高并发 API 服务	使用 `embedder.embed_documents(batch_texts)` 而非逐条调用	需对齐 batch 内最长序列；padding 开销
上下文剪枝	仅保留与查询最相关的句子/段落	减少 LLM token 消耗	结合 `LLMChainExtractor` 或关键词匹配预筛	过度剪枝可能丢失关键信息
模型量化	使用 GGUF / AWQ / GPTQ 格式加载 4-bit/8-bit 量化模型	本地部署资源受限	`llm = HuggingFaceLLM(model_name="TheBloke/Llama-3-8B-GGUF", model_file="llama-3-8b.Q4_K_M.gguf")`	量化可能轻微降低生成质量
检索-生成解耦	先返回检索结果，再异步生成答案	提升首屏响应速度	Web 前端先展示”找到以下资料”，后台继续生成总结	需设计用户体验流程
冷热数据分离	高频文档存入内存向量库（FAISS），低频文档存入磁盘库（Qdrant on SSD）	成本与性能平衡	路由器根据 `metadata["popularity"]` 选择检索路径	需维护热度统计机制

第九章：部署与生产实践

9.1 本地与云部署方案

部署方式	架构组成	适用场景	示例命令 / 配置	注意事项
单机本地部署	Python + FAISS + Ollama（全栈运行于一台机器）	开发测试、POC 验证	`ragflow run --config local.yaml`	资源受限；不支持高可用
Docker Compose	容器化服务：API、Worker、VectorDB（如 Qdrant）、Redis	中小型团队快速上线	`docker-compose -f ragflow-prod.yml up -d`	需配置 volume 持久化数据
Kubernetes (Helm)	Helm Chart 部署：Deployment + Service + Ingress + PVC	企业级弹性伸缩与 CI/CD 集成	`helm install ragflow ./charts/ragflow --set llm.model=qwen-max`	需熟悉 K8s 网络与资源配额
公有云托管方案	阿里云百炼 + PAI + OSS + ApsaraDB for Redis	免运维、合规要求高的场景	在百炼平台选择”RAG 工作流模板”一键部署	成本按量计费；注意区域限制
混合部署	敏感数据本地处理（嵌入/检索），LLM 调用公有云 API	数据不出域 + 利用大模型能力	本地运行 retriever，LLM 节点指向 `https://dashscope.aliyuncs.com`	网络延迟需评估；API 安全认证

9.2 API 服务封装

封装方式	技术栈 / 接口规范	功能特性	示例请求	注意事项
FastAPI RESTful API	自动生成 OpenAPI 文档，支持异步	标准化输入输出	`POST /query` `{"question": "报销流程？", "user_id": "u123"}`	支持 Pydantic 模型校验
gRPC 服务	Protocol Buffers 定义 `.proto`，高性能二进制通信	低延迟、适合内部微服务调用	`client.Query(QueryRequest(question="..."))`	调试复杂度高于 REST
LangServe 兼容接口	基于 LangChain/LangServe 自动暴露 chain 为 API	快速将工作流转为服务	`from ragflow.serve import serve_workflow; serve_workflow(wf, port=8000)`	仅支持简单输入输出结构
流式响应（SSE）	使用 `StreamingResponse` 返回 token-by-token 结果	实现”打字机”效果，提升用户体验	前端通过 `EventSource("/stream")` 接收	需前端配合处理流
认证与限流	集成 API Key + Rate Limiter（如 slowapi）	防止滥用，保障服务稳定性	请求头：`Authorization: Bearer sk-xxx`	密钥应轮换；限流策略可配置

9.3 监控与日志管理

监控维度	工具 / 实现方式	采集指标	可视化方案	注意事项
系统指标	Prometheus + Node Exporter	CPU、内存、GPU 显存、磁盘 IO	Grafana Dashboard	容器环境需 cAdvisor
应用性能	OpenTelemetry (OTel)	各节点耗时、LLM token 数、缓存命中率	Jaeger / Tempo 查看 trace 链路	需在 workflow 中注入 span
业务指标	自定义埋点	Query 成功率、平均响应时间、无结果率	写入 ClickHouse 或 Loki	关键路径必须覆盖
日志集中管理	ELK（Elasticsearch + Logstash + Kibana）或 Loki + Promtail	结构化日志（含 job_id、user_id、error）	Kibana 查询：`job_id:"j123" AND level:error`	日志需脱敏（如移除 PII）
告警机制	Alertmanager + 企业微信/钉钉 webhook	LLM 调用失败 >5 次/分钟	自动通知值班人员	告警阈值需动态调整，避免噪声

第十章：高级功能与扩展开发

10.1 自定义组件开发

组件类型	开发接口 / 基类	注册方式	示例场景	注意事项
自定义 Loader	继承 `BaseDocumentLoader`，实现 `load()` 方法	`@register_loader("my_pdf")`	解析带水印的 PDF 或内部格式文档	需处理编码与异常
自定义 Retriever	实现 `get_relevant_documents(query: str, **kwargs) -> List[Document]`	通过 `RetrieverFactory.register("my_ret", MyRetriever)`	融合数据库查询与向量检索	必须兼容 metadata 输出
自定义 LLM Wrapper	继承 `BaseLLM`，重写 `_call` 和 `_acall`	在 `config.yaml` 中指定 `type: "my_llm"`	对接私有推理引擎（如 TensorRT-LLM）	需处理 stop token 和 streaming
自定义 Evaluator	实现 `evaluate(predictions, references, queries)`	用于 `ragflow eval --custom my_eval`	领域特定评分（如法律条款匹配度）	返回标准化指标字典
插件热加载	放置 `.py` 文件到 `plugins/` 目录，系统启动时自动扫描	无需修改核心代码	快速集成第三方工具	需遵循命名规范

10.2 插件系统与生态集成

集成目标	插件名称 / 方式	功能描述	配置示例	注意事项
Slack / 钉钉机器人	`ragflow-integrations/slack-bot`	用户在群聊中提问，自动回复	`SLACK_BOT_TOKEN=xoxb-...`	需配置事件订阅与权限
Notion 同步	`NotionDataSourcePlugin`	定期拉取 Notion 页面作为知识源	`notion_token: "secret_..."` `database_id: "..."`	注意 rate limit
Airflow DAG 触发	提供 `RAGFlowOperator`	在 Airflow 中调度 RAG 更新流程	`dag >> RAGFlowOperator(workflow="update_kb")`	适用于定时知识库刷新
LangChain 兼容	所有 retriever/llm 实现 Runnable 接口	可直接嵌入 LangChain Chain	`chain = rag_retriever	llm`
Webhook 回调	配置 `on_complete_webhook: https://myapp.com/rag-callback`	流程结束后推送结果	`POST payload: {"job_id": "...", "answer": "...", "status": "success"}`	需验证签名防伪造

10.3 多模态与多语言支持

能力类型	技术方案	支持内容	示例代码	注意事项
图文混合检索	使用 CLIP 或 BLIP-2 嵌入图像+文本，构建统一向量空间	用户上传图片，系统返回相关文本或图片	`embedder = CLIPImageTextEmbedder(model_name="openai/clip-vit-base-patch32")`	图像需预处理（resize/normalize）
语音输入	集成 Whisper ASR 模块，将语音转文本后再进入 RAG 流程	语音提问 → 文本检索 → 语音回答（TTS）	`text = whisper.transcribe(audio_file)`	需处理方言与噪音
多语言问答	使用 multilingual 模型（如 `intfloat/multilingual-e5-large`）	支持中/英/法/西等 100+ 语言	`retriever = VectorStoreRetriever(embeddings=MultilingualEmbeddings())`	检索与生成语言需一致
跨语言检索	查询翻译 + 目标语料检索（如中文问 → 英文答）	跨语言知识获取	`translated_q = translator.translate(q, src="zh", tgt="en")`	翻译误差可能影响召回
表格/公式理解	使用 Donut、TableFormer 或 GPT-4V 解析非结构化表格	从 PDF 表格中提取结构化数据参与检索	`table_extractor = DonutTableExtractor(model="naver-clova-ix/donut-base-finetuned-cord-v2")`	计算开销大；建议离线预处理