文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布
社区首页 >专栏 >静态 RAG 与动态 RAG 技术全解析

静态 RAG 与动态 RAG 技术全解析

作者头像
tunsuy
发布于 2026-04-09 09:46:18
发布于 2026-04-09 09:46:18
1020
举报
文章被收录于专栏：有文化的技术人有文化的技术人

❝本文系统介绍静态 RAG 与动态 RAG 的核心原理、技术对比、主流实现方案及代码实践，适合技术选型和深入学习参考。❞

目录

  • 一、RAG 技术概述
  • 二、静态 RAG
    • 2.1 核心原理
    • 2.2 优化技术
    • 2.3 主流实践方案
    • 2.4 代码示例
  • 三、动态 RAG
    • 3.1 核心原理
    • 3.2 主流实现方案
  • 四、Self-RAG 详解
    • 4.1 核心原理
    • 4.2 反思令牌机制
    • 4.3 环境配置
    • 4.4 代码实现
  • 五、CRAG 详解
    • 5.1 核心原理
    • 5.2 环境配置
    • 5.3 完整代码实现（LangGraph）
  • 六、RAGFlow 平台
    • 6.1 平台定位
    • 6.2 Agent 工作流机制
    • 6.3 SDK 使用
  • 七、技术对比与选型建议
    • 7.1 静态 vs 动态 RAG
    • 7.2 Self-RAG vs CRAG
    • 7.3 选型建议
  • 八、参考资源

一、RAG 技术概述

RAG（Retrieval-Augmented Generation，检索增强生成）是一种结合信息检索与文本生成的技术架构，通过从外部知识库检索相关信息来增强大语言模型的生成能力。

为什么需要 RAG？

问题

RAG 解决方案

LLM 知识截止日期

检索最新的外部知识

幻觉问题

基于检索到的事实生成

领域知识不足

接入专业知识库

私有数据访问

检索企业内部文档

RAG 基础流程

代码语言：javascript
复制
用户问题 → 向量化 → 相似度检索 → 获取相关文档 → 构建 Prompt → LLM 生成 → 返回答案

二、静态 RAG

2.1 核心原理

静态 RAG 是传统的检索增强生成方法，采用「一次检索、一次生成」的线性流程：

代码语言：javascript
复制
┌─────────────────────────────────────────────────────────────┐
│                      静态 RAG 流程                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────┐    ┌──────┐    ┌──────┐    ┌──────┐    ┌──────┐  │
│  │ 用户 │ →  │ 向量化│ →  │ 检索 │ →  │ 拼接 │ →  │ 生成 │  │
│  │ 问题 │    │ Query │    │ Top-K│    │Prompt│    │ 答案 │  │
│  └──────┘    └──────┘    └──────┘    └──────┘    └──────┘  │
│                                                             │
│  特点：流程固定，检索一次，上下文不再更新                      │
└─────────────────────────────────────────────────────────────┘

「核心特点：」

  • 预处理阶段：文档离线分块、向量化、存入向量数据库
  • 检索一次性：用户查询时一次性检索相关文档
  • 固定上下文：检索到的内容直接拼接到 prompt，不再更新
  • 流程固定：查询 → 检索 → 生成，线性执行

2.2 优化技术

虽然是静态流程，但可以通过多种技术优化检索和生成质量：

技术

描述

适用场景

「HyDE」

先让 LLM 生成假设答案，用假设答案去检索

问题表述模糊时

「Query Expansion」

扩展用户查询为多个变体，提升召回率

提高召回率

「Reranker」

检索后用交叉编码器重排序

提高精确度

「Sentence Window」

检索小块，返回时扩展上下文窗口

需要更多上下文

「Parent Document」

检索小块，返回其父文档

保持文档完整性

「Fusion RAG」

多路检索结果融合（RRF 算法）

多维度召回

「Hybrid Search」

向量检索 + BM25 关键词检索混合

兼顾语义和关键词

2.3 主流实践方案

开源框架
  • 「LangChain」：最流行的 RAG 框架，生态丰富
  • 「LlamaIndex」：专注于数据索引和查询，提供多种索引类型
  • 「Haystack」：模块化的 NLP 管道框架
向量数据库
  • 「Milvus」：高性能分布式向量数据库
  • 「Chroma」：轻量级嵌入式向量数据库
  • 「Pinecone」：全托管云向量数据库
  • 「Weaviate」：支持多模态的向量数据库
  • 「Qdrant」：Rust 编写的高性能向量数据库
企业级产品
  • Azure AI Search + OpenAI
  • Amazon Bedrock Knowledge Bases
  • Google Vertex AI Search

2.4 代码示例

基础静态 RAG（LangChain）
代码语言：javascript
复制
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 1. 加载文档
loader = WebBaseLoader("https://example.com/document")
documents = loader.load()

# 2. 分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
splits = text_splitter.split_documents(documents)

# 3. 创建向量库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

# 5. 定义 Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "基于以下上下文回答问题。如果不知道答案，请说不知道。\n\n上下文：{context}"),
    ("human", "{question}")
])

# 6. 构建 Chain
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def format_docs(docs):
    return"\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 7. 查询
answer = rag_chain.invoke("你的问题")
print(answer)
带 Reranker 的静态 RAG
代码语言：javascript
复制
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# 创建 Reranker
model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")
compressor = CrossEncoderReranker(model=model, top_n=3)

# 包装检索器
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

# 使用带重排序的检索器
rag_chain = (
    {"context": compression_retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

三、动态 RAG

3.1 核心原理

动态 RAG（也称 Agentic RAG）是更智能的检索增强方法，核心特点是「按需检索、迭代优化」：

代码语言：javascript
复制
┌─────────────────────────────────────────────────────────────────┐
│                       动态 RAG 流程                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────┐    ┌──────────┐    ┌──────────┐                      │
│  │ 用户 │ →  │ LLM 分析 │ →  │ 需要检索?│                      │
│  │ 问题 │    │ 问题复杂度│    │          │                      │
│  └──────┘    └──────────┘    └────┬─────┘                      │
│                                   │                             │
│                    ┌──────────────┼──────────────┐              │
│                    ↓ 否           ↓ 是           ↓ 复杂         │
│              ┌──────────┐   ┌──────────┐   ┌──────────┐        │
│              │ 直接生成 │   │ 单次检索 │   │ 多轮迭代 │        │
│              └──────────┘   └────┬─────┘   └────┬─────┘        │
│                                  ↓              ↓               │
│                            ┌──────────┐   ┌──────────┐         │
│                            │ 评估相关性│   │ 分解子问题│         │
│                            └────┬─────┘   └────┬─────┘         │
│                                 ↓              ↓               │
│                    ┌────────────┴────────────┐                 │
│                    │ 不满意？改写查询重新检索  │                 │
│                    └────────────┬────────────┘                 │
│                                 ↓                              │
│                           ┌──────────┐                         │
│                           │ 生成答案 │                         │
│                           └──────────┘                         │
└─────────────────────────────────────────────────────────────────┘

「核心特点：」

  • 「迭代检索」：模型可在生成过程中多次触发检索
  • 「自适应决策」：LLM 判断何时需要检索、检索什么内容
  • 「多源整合」：可动态选择不同知识源（数据库、API、搜索引擎）
  • 「反思与纠错」：可验证检索结果的相关性，必要时重新检索

3.2 主流实现方案

方案

类型

核心特点

是否需要专门训练

「Self-RAG」

端到端

模型内置反思令牌，自动决定是否检索

✅ 需要

「CRAG」

工作流

检索后评估文档质量，支持网络搜索兜底

❌ 不需要

「Adaptive RAG」

工作流

根据问题复杂度动态选择策略

❌ 不需要

「IRCoT」

工作流

交错检索与思维链推理

❌ 不需要

四、Self-RAG 详解

4.1 核心原理

Self-RAG（Self-Reflective RAG）的核心是「训练模型本身具备"反思"能力」，通过特殊的 Reflection Tokens（反思令牌）让模型自己决定检索和评估。

代码语言：javascript
复制
┌─────────────────────────────────────────────────────────────────┐
│                     Self-RAG 工作流程                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  用户问题                                                        │
│      ↓                                                          │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │              Self-RAG 模型（内置反思能力）                 │   │
│  │                                                          │   │
│  │  Step 1: 输出 [Retrieval] 或 [No Retrieval]              │   │
│  │          → 模型自己决定是否需要检索                        │   │
│  │                                                          │   │
│  │  Step 2: 如果检索，输出 [Relevant] 或 [Irrelevant]        │   │
│  │          → 模型自己评估文档相关性                          │   │
│  │                                                          │   │
│  │  Step 3: 生成答案，输出 [Fully Supported] 等              │   │
│  │          → 模型自己评估答案是否有依据                      │   │
│  │                                                          │   │
│  │  Step 4: 输出 [Utility:1-5]                              │   │
│  │          → 模型自己评估整体质量                            │   │
│  └─────────────────────────────────────────────────────────┘   │
│      ↓                                                          │
│  最终答案                                                        │
└─────────────────────────────────────────────────────────────────┘

4.2 反思令牌机制

Self-RAG 定义了四种反思令牌：

令牌类型

作用

可能的值

「Retrieve」

判断是否需要检索

[Retrieval], [No Retrieval]

「ISREL」

评估检索文档是否相关

[Relevant], [Irrelevant]

「ISSUP」

评估生成内容是否有文档支持

[Fully Supported], [Partially Supported], [No Support]

「ISUSE」

评估整体回答质量

[Utility:5] 到 [Utility:1]

「关键优势」：这些判断都是模型在训练时学会的，推理时自动输出，不需要额外的评估 prompt。

4.3 环境配置

代码语言：javascript
复制
# 1. 克隆仓库
git clone https://github.com/AkariAsai/self-rag.git
cd self-rag

# 2. 创建环境
conda create -n selfrag python=3.10
conda activate selfrag

# 3. 安装依赖
pip install vllm transformers torch accelerate

# 4. 模型会从 HuggingFace 自动下载
# 可用模型：
# - selfrag/selfrag_llama2_7b  (7B 版本，需要 ~16GB 显存)
# - selfrag/selfrag_llama2_13b (13B 版本，需要 ~32GB 显存)

4.4 代码实现

基础使用
代码语言：javascript
复制
from vllm import LLM, SamplingParams

# 初始化模型
model = LLM(
    "selfrag/selfrag_llama2_7b",
    dtype="half",
    tensor_parallel_size=1,  # GPU 数量
)

sampling_params = SamplingParams(
    temperature=0.0,
    top_p=1.0,
    max_tokens=200,
    skip_special_tokens=False# 重要：保留反思令牌以便观察
)

def format_prompt(question, paragraph=None):
    """格式化输入提示"""
    prompt = f"### Instruction:\n{question}\n\n### Response:\n"
    if paragraph:
        prompt += f"[Retrieval]<paragraph>{paragraph}</paragraph>"
    return prompt

# 示例：不带检索的生成
question = "法国的首都是哪里？"
prompt = format_prompt(question)
output = model.generate([prompt], sampling_params)
print(output[0].outputs[0].text)
# 可能输出: [No Retrieval]法国的首都是巴黎。[Utility:5]

# 示例：带检索的生成
paragraph = "巴黎是法国的首都和最大城市，位于法国北部塞纳河畔..."
prompt_with_retrieval = format_prompt(question, paragraph)
output = model.generate([prompt_with_retrieval], sampling_params)
print(output[0].outputs[0].text)
# 可能输出: [Relevant]法国的首都是巴黎...[Fully Supported][Utility:5]
完整 Self-RAG 系统
代码语言：javascript
复制
from vllm import LLM, SamplingParams
from typing import List, Optional
import re

class SelfRAGSystem:
    """完整的 Self-RAG 系统实现"""
    
    def __init__(self, model_name: str = "selfrag/selfrag_llama2_7b"):
        self.model = LLM(model_name, dtype="half")
        self.sampling_params = SamplingParams(
            temperature=0.0,
            top_p=1.0,
            max_tokens=300,
            skip_special_tokens=False
        )
        
        # 反思令牌定义
        self.RETRIEVAL_TOKEN = "[Retrieval]"
        self.NO_RETRIEVAL_TOKEN = "[No Retrieval]"
        self.RELEVANT_TOKEN = "[Relevant]"
        self.IRRELEVANT_TOKEN = "[Irrelevant]"
        
    def format_prompt(self, question: str, context: Optional[str] = None) -> str:
        """格式化提示"""
        prompt = f"### Instruction:\n{question}\n\n### Response:\n"
        if context:
            prompt += f"[Retrieval]<paragraph>{context}</paragraph>"
        return prompt
    
    def should_retrieve(self, question: str) -> bool:
        """让模型决定是否需要检索"""
        prompt = self.format_prompt(question)
        output = self.model.generate([prompt], self.sampling_params)[0]
        text = output.outputs[0].text
        
        # 模型会输出 [Retrieval] 或 [No Retrieval]
        return self.RETRIEVAL_TOKEN in text
    
    def evaluate_relevance(self, question: str, document: str) -> bool:
        """让模型评估文档相关性"""
        prompt = self.format_prompt(question, document)
        output = self.model.generate([prompt], self.sampling_params)[0]
        text = output.outputs[0].text
        
        # 模型会输出 [Relevant] 或 [Irrelevant]
        return self.RELEVANT_TOKEN in text
    
    def retrieve_documents(self, query: str, top_k: int = 3) -> List[str]:
        """
        检索文档（需要接入你的向量库）
        这里是示例接口，实际使用时替换为你的检索逻辑
        """
        # 示例：使用 Chroma
        # from langchain_community.vectorstores import Chroma
        # vectorstore = Chroma(persist_directory="./chroma_db", ...)
        # docs = vectorstore.similarity_search(query, k=top_k)
        # return [doc.page_content for doc in docs]
        
        # 占位返回
        return []
    
    def generate_answer(self, question: str, documents: List[str]) -> str:
        """基于文档生成答案"""
        # 筛选相关文档
        relevant_docs = [
            doc for doc in documents 
            if self.evaluate_relevance(question, doc)
        ]
        
        if relevant_docs:
            context = "\n\n".join(relevant_docs)
            prompt = self.format_prompt(question, context)
        else:
            prompt = self.format_prompt(question)
        
        output = self.model.generate([prompt], self.sampling_params)[0]
        return self.clean_output(output.outputs[0].text)
    
    def clean_output(self, text: str) -> str:
        """清理输出中的反思令牌"""
        tokens_to_remove = [
            "[Retrieval]", "[No Retrieval]",
            "[Relevant]", "[Irrelevant]",
            "[Fully supported]", "[Partially supported]", "[No support]",
            r"\[Utility:\d\]"
        ]
        for token in tokens_to_remove:
            text = re.sub(token, "", text)
        
        # 移除 paragraph 标签
        text = re.sub(r"<paragraph>.*?</paragraph>", "", text, flags=re.DOTALL)
        return text.strip()
    
    def run(self, question: str) -> dict:
        """
        完整的 Self-RAG 流程
        返回包含决策过程的详细结果
        """
        result = {
            "question": question,
            "retrieval_needed": False,
            "documents_retrieved": [],
            "relevant_documents": [],
            "answer": ""
        }
        
        # Step 1: 模型决定是否需要检索
        if self.should_retrieve(question):
            result["retrieval_needed"] = True
            print(f"[Self-RAG] 模型决定：需要检索")
            
            # Step 2: 执行检索
            documents = self.retrieve_documents(question)
            result["documents_retrieved"] = documents
            
            if documents:
                # Step 3: 模型评估文档相关性
                for doc in documents:
                    if self.evaluate_relevance(question, doc):
                        result["relevant_documents"].append(doc)
                        print(f"[Self-RAG] 文档相关: {doc[:50]}...")
                    else:
                        print(f"[Self-RAG] 文档不相关: {doc[:50]}...")
            
            # Step 4: 生成答案
            result["answer"] = self.generate_answer(question, documents)
        else:
            print(f"[Self-RAG] 模型决定：不需要检索，直接回答")
            prompt = self.format_prompt(question)
            output = self.model.generate([prompt], self.sampling_params)[0]
            result["answer"] = self.clean_output(output.outputs[0].text)
        
        return result


# 使用示例
if __name__ == "__main__":
    selfrag = SelfRAGSystem()
    
    # 简单问题（可能不需要检索）
    result1 = selfrag.run("1 + 1 等于多少？")
    print(f"答案: {result1['answer']}")
    print(f"是否检索: {result1['retrieval_needed']}")
    
    # 需要外部知识的问题
    result2 = selfrag.run("2024年诺贝尔物理学奖得主是谁？")
    print(f"答案: {result2['answer']}")
    print(f"是否检索: {result2['retrieval_needed']}")

五、CRAG 详解

5.1 核心原理

CRAG（Corrective RAG，纠正式 RAG）的核心是「在工作流层面增加纠错机制」，不需要专门训练模型：

代码语言：javascript
复制
┌─────────────────────────────────────────────────────────────────┐
│                       CRAG 工作流程                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  用户问题 → 检索文档 → 评估文档质量                               │
│                           │                                     │
│         ┌─────────────────┼─────────────────┐                   │
│         ↓                 ↓                 ↓                   │
│     [Correct]         [Ambiguous]       [Incorrect]             │
│     文档正确            文档模糊           文档错误               │
│         ↓                 ↓                 ↓                   │
│     直接使用          知识精炼后使用     丢弃，网络搜索            │
│         ↓                 ↓                 ↓                   │
│         └─────────────────┼─────────────────┘                   │
│                           ↓                                     │
│                      生成最终回答                                │
└─────────────────────────────────────────────────────────────────┘

「三个核心组件：」

组件

作用

「Retrieval Evaluator」

评估检索文档的质量（Correct/Ambiguous/Incorrect）

「Knowledge Refinement」

对模糊文档进行知识提炼，去除无关信息

「Web Search Fallback」

当本地检索失败时，使用网络搜索兜底

「与 Self-RAG 的核心区别：」

代码语言：javascript
复制
┌─────────────────────────────────────────────────────────────────┐
│                         CRAG                                     │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐                   │
│  │ 你定义   │ →  │ 你定义   │ →  │ 你定义   │                   │
│  │ 检索逻辑 │    │ 评估Prompt│   │ 分支逻辑 │                   │
│  └──────────┘    └──────────┘    └──────────┘                   │
│  LLM 只是执行你定义的评估 prompt，流程由你控制                    │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                       Self-RAG                                   │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │              专门训练的模型                                │   │
│  │  内置: 是否检索? → 文档相关吗? → 回答有依据吗?            │   │
│  │        [Retrieval]   [Relevant]   [Fully Supported]       │   │
│  └──────────────────────────────────────────────────────────┘   │
│  模型本身会输出判断令牌，你只需解析这些令牌                       │
└─────────────────────────────────────────────────────────────────┘

5.2 环境配置

代码语言：javascript
复制
# 创建环境
conda create -n crag python=3.11
conda activate crag

# 安装依赖
pip install langgraph langchain langchain-openai langchain-community
pip install chromadb tiktoken tavily-python
pip install python-dotenv

# 可选：使用本地模型
pip install langchain-ollama

「环境变量配置（.env 文件）：」

代码语言：javascript
复制
OPENAI_API_KEY=your_openai_api_key
TAVILY_API_KEY=your_tavily_api_key  # 用于网络搜索

5.3 完整代码实现（LangGraph）

代码语言：javascript
复制
"""
CRAG (Corrective RAG) 完整实现
基于 LangGraph 的工作流编排
"""

import os
from typing import List, Literal
from typing_extensions import TypedDict
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END

load_dotenv()


# ============================================================
# 1. 定义状态
# ============================================================

class GraphState(TypedDict):
    """工作流状态定义"""
    question: str                    # 用户问题
    documents: List[Document]        # 检索到的文档
    generation: str                  # 生成的回答
    web_search_needed: bool          # 是否需要网络搜索
    relevance_scores: List[str]      # 文档相关性评分


# ============================================================
# 2. 初始化组件
# ============================================================

# LLM（可替换为本地模型）
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 向量存储
embeddings = OpenAIEmbeddings()

# 网络搜索工具
web_search_tool = TavilySearchResults(max_results=3)


# ============================================================
# 3. 定义评估器和生成器
# ============================================================

# 文档相关性评估 Prompt
relevance_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个文档相关性评估专家。

评估给定的文档是否与用户问题相关。
只回答 'yes' 或 'no'。

- 'yes': 文档包含与问题相关的关键信息
- 'no': 文档与问题无关或信息不足
"""),
    ("human", """问题: {question}

文档内容:
{document}

该文档是否与问题相关？"""),
])

relevance_chain = relevance_prompt | llm | StrOutputParser()


# 知识精炼 Prompt
refinement_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个知识提炼专家。

从给定的文档中提取与用户问题最相关的关键信息。
去除无关内容，只保留核心知识点。
如果文档完全无关，返回 "无相关信息"。
"""),
    ("human", """问题: {question}

文档内容:
{document}

请提炼关键信息："""),
])

refinement_chain = refinement_prompt | llm | StrOutputParser()


# 生成回答 Prompt
generation_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个知识问答助手。

基于提供的上下文信息回答用户问题。
如果上下文信息不足，请诚实说明。
回答要准确、简洁、有条理。
"""),
    ("human", """上下文信息:
{context}

问题: {question}

请回答："""),
])

generation_chain = generation_prompt | llm | StrOutputParser()


# ============================================================
# 4. 定义节点函数
# ============================================================

def retrieve(state: GraphState) -> GraphState:
    """
    检索节点：从向量库检索相关文档
    """
    print("---[节点] RETRIEVE: 执行检索---")
    question = state["question"]
    
    try:
        vectorstore = Chroma(
            persist_directory="./chroma_db",
            embedding_function=embeddings
        )
        documents = vectorstore.similarity_search(question, k=4)
        print(f"  检索到 {len(documents)} 个文档")
    except Exception as e:
        print(f"  检索失败: {e}")
        documents = []
    
    return {
        **state,
        "documents": documents,
        "relevance_scores": []
    }


def grade_documents(state: GraphState) -> GraphState:
    """
    评估节点：评估每个文档的相关性
    这是 CRAG 的核心 - Retrieval Evaluator
    """
    print("---[节点] GRADE: 评估文档相关性---")
    question = state["question"]
    documents = state["documents"]
    
    relevant_docs = []
    relevance_scores = []
    
    for i, doc in enumerate(documents):
        score = relevance_chain.invoke({
            "question": question,
            "document": doc.page_content
        }).strip().lower()
        
        relevance_scores.append(score)
        
        if score == "yes":
            relevant_docs.append(doc)
            print(f"  ✓ 文档 {i+1} 相关")
        else:
            print(f"  ✗ 文档 {i+1} 不相关")
    
    # 判断是否需要网络搜索（没有相关文档时触发）
    web_search_needed = len(relevant_docs) == 0
    
    if web_search_needed:
        print("  → 没有相关文档，需要网络搜索")
    else:
        print(f"  → 找到 {len(relevant_docs)} 个相关文档")
    
    return {
        **state,
        "documents": relevant_docs,
        "relevance_scores": relevance_scores,
        "web_search_needed": web_search_needed
    }


def refine_knowledge(state: GraphState) -> GraphState:
    """
    知识精炼节点：提炼文档中的关键信息
    这是 CRAG 的 Knowledge Refinement 组件
    """
    print("---[节点] REFINE: 精炼知识---")
    question = state["question"]
    documents = state["documents"]
    
    refined_docs = []
    for i, doc in enumerate(documents):
        refined_content = refinement_chain.invoke({
            "question": question,
            "document": doc.page_content
        })
        
        if refined_content != "无相关信息":
            refined_docs.append(Document(page_content=refined_content))
            print(f"  文档 {i+1} 精炼完成")
        else:
            print(f"  文档 {i+1} 无有效信息，跳过")
    
    return {
        **state,
        "documents": refined_docs
    }


def web_search(state: GraphState) -> GraphState:
    """
    网络搜索节点：当本地检索失败时使用
    这是 CRAG 的 Fallback 机制
    """
    print("---[节点] WEB SEARCH: 执行网络搜索---")
    question = state["question"]
    
    try:
        search_results = web_search_tool.invoke({"query": question})
        web_docs = [
            Document(page_content=result["content"])
            for result in search_results
        ]
        print(f"  网络搜索返回 {len(web_docs)} 个结果")
    except Exception as e:
        print(f"  网络搜索失败: {e}")
        web_docs = []
    
    return {
        **state,
        "documents": web_docs
    }


def generate(state: GraphState) -> GraphState:
    """
    生成节点：基于文档生成最终回答
    """
    print("---[节点] GENERATE: 生成回答---")
    question = state["question"]
    documents = state["documents"]
    
    # 合并文档内容
    if documents:
        context = "\n\n".join([doc.page_content for doc in documents])
    else:
        context = "没有找到相关信息。"
    
    # 生成回答
    generation = generation_chain.invoke({
        "question": question,
        "context": context
    })
    
    return {
        **state,
        "generation": generation
    }


# ============================================================
# 5. 定义条件边
# ============================================================

def decide_to_search(state: GraphState) -> Literal["web_search", "refine"]:
    """
    决策函数：根据文档评估结果决定下一步
    - 如果没有相关文档 → 网络搜索
    - 如果有相关文档 → 知识精炼
    """
    if state.get("web_search_needed", False):
        print("---[决策] 转向网络搜索---")
        return"web_search"
    else:
        print("---[决策] 转向知识精炼---")
        return"refine"


# ============================================================
# 6. 构建工作流图
# ============================================================

def build_crag_graph():
    """构建 CRAG 工作流图"""
    workflow = StateGraph(GraphState)
    
    # 添加节点
    workflow.add_node("retrieve", retrieve)
    workflow.add_node("grade_documents", grade_documents)
    workflow.add_node("refine", refine_knowledge)
    workflow.add_node("web_search", web_search)
    workflow.add_node("generate", generate)
    
    # 定义边
    workflow.add_edge(START, "retrieve")
    workflow.add_edge("retrieve", "grade_documents")
    
    # 条件边：根据评估结果决定下一步
    workflow.add_conditional_edges(
        "grade_documents",
        decide_to_search,
        {
            "web_search": "web_search",
            "refine": "refine"
        }
    )
    
    workflow.add_edge("refine", "generate")
    workflow.add_edge("web_search", "generate")
    workflow.add_edge("generate", END)
    
    return workflow.compile()


# ============================================================
# 7. 辅助函数：创建向量库
# ============================================================

def create_vectorstore(texts: List[str] = None, urls: List[str] = None):
    """
    创建或更新向量库
    
    Args:
        texts: 文本列表
        urls: URL 列表（会自动抓取内容）
    """
    from langchain_community.document_loaders import WebBaseLoader
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    
    documents = []
    
    # 从 URL 加载
    if urls:
        for url in urls:
            try:
                loader = WebBaseLoader(url)
                documents.extend(loader.load())
                print(f"已加载: {url}")
            except Exception as e:
                print(f"加载失败 {url}: {e}")
    
    # 从文本加载
    if texts:
        for text in texts:
            documents.append(Document(page_content=text))
    
    ifnot documents:
        print("没有文档可加载")
        returnNone
    
    # 分块
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50
    )
    splits = text_splitter.split_documents(documents)
    
    # 创建向量库
    vectorstore = Chroma.from_documents(
        documents=splits,
        embedding=embeddings,
        persist_directory="./chroma_db"
    )
    
    print(f"向量库创建完成，包含 {len(splits)} 个文档块")
    return vectorstore


# ============================================================
# 8. 主函数
# ============================================================

def main():
    """CRAG 使用示例"""
    
    # 构建工作流
    app = build_crag_graph()
    
    # 可视化工作流（可选）
    try:
        from IPython.display import Image, display
        display(Image(app.get_graph().draw_mermaid_png()))
    except:
        pass
    
    # 运行查询
    print("\n" + "="*60)
    print("CRAG 系统启动")
    print("="*60 + "\n")
    
    result = app.invoke({
        "question": "什么是量子计算？它有哪些应用？",
        "documents": [],
        "generation": "",
        "web_search_needed": False,
        "relevance_scores": []
    })
    
    print("\n" + "="*60)
    print("最终回答:")
    print("="*60)
    print(result["generation"])


if __name__ == "__main__":
    # 首次运行时，可以先创建向量库
    # create_vectorstore(urls=["https://example.com/doc1", "https://example.com/doc2"])
    
    main()
使用本地模型的 CRAG
代码语言：javascript
复制
from langchain_ollama import ChatOllama

# 替换 LLM 为本地模型
llm = ChatOllama(
    model="qwen2.5:7b",  # 或 llama3.1:8b, deepseek-r1:8b 等
    temperature=0
)

# 其余代码保持不变

六、RAGFlow 平台

6.1 平台定位

RAGFlow 是一个「支持静态和动态两种模式的 RAG 平台」，不是纯粹的动态 RAG：

模式

说明

「Chat 助手」

静态 RAG，单次检索 + 生成

「Agent 工作流」

动态 RAG，通过 Graph 编排实现多步骤迭代

6.2 Agent 工作流机制

RAGFlow 的动态能力通过 「Graph 工作流编排」 实现：

代码语言：javascript
复制
用户输入 → Graph 工作流引擎 → 节点1(判断) → 节点2(检索) → 节点3(评估) 
                                    ↓ 不满意
                              节点4(改写查询) → 重新检索 → ...
                                    ↓ 满意
                              节点5(生成回答)

「关键点：」

  • 动态逻辑需要你在 UI 中配置或通过 DSL 定义
  • 不是系统自动决定，而是你定义规则，系统执行

6.3 SDK 使用

安装
代码语言：javascript
复制
pip install ragflow-sdk
基础用法
代码语言：javascript
复制
from ragflow_sdk import RAGFlow

# 1. 初始化客户端
rag = RAGFlow(
    api_key="<YOUR_API_KEY>",
    base_url="http://<YOUR_HOST>:9380"
)

# 2. 创建数据集
dataset = rag.create_dataset(
    name="my_knowledge_base",
    embedding_model="BAAI/bge-large-zh-v1.5@BAAI",
    chunk_method="naive"
)

# 3. 上传文档
dataset.upload_documents([
    {"display_name": "doc.pdf", "blob": open("doc.pdf", "rb").read()}
])

# 4. 解析文档
dataset.async_parse_documents(["doc_id"])

# 5. 创建对话助手
assistant = rag.create_chat(
    name="my_assistant",
    dataset_ids=[dataset.id],
    llm={"model_name": "gpt-3.5-turbo"},
    prompt={"similarity_threshold": 0.2}
)

# 6. 对话
session = assistant.create_session()
for msg in session.ask("你的问题", stream=True):
    print(msg.content, end="")
Agent 模式
代码语言：javascript
复制
# 创建 Agent（需要先配置工作流 DSL）
agent = rag.create_agent(
    title="智能助手",
    description="支持多轮检索的助手",
    dsl={
        "canvas": {
            "nodes": [...],  # 节点定义
            "edges": [...]   # 边定义
        }
    }
)

# 列出已有 Agent
agents = rag.list_agents()

七、技术对比与选型建议

7.1 静态 vs 动态 RAG

维度

静态 RAG

动态 RAG

「检索次数」

1 次

多次（按需）

「决策能力」

无

LLM 自主决策或规则驱动

「适用场景」

简单问答、FAQ

复杂推理、多跳问题

「延迟」

低

较高

「实现难度」

简单

复杂

「成本」

低（单次 LLM 调用）

高（多次 LLM 调用）

7.2 Self-RAG vs CRAG

维度

Self-RAG

CRAG (LangGraph)

「实现方式」

专门微调的模型

工作流编排 + 普通 LLM

「"是否检索"判断」

模型内部自动决定

你写 prompt 让 LLM 判断

「模型要求」

必须用 selfrag 模型

任意 LLM（OpenAI、本地等）

「GPU 需求」

必须（16GB+）

可选（用 API 则不需要）

「安装复杂度」

高（vLLM + 大模型）

低（pip install）

「可定制性」

低（逻辑固化在模型中）

高（可自定义任何逻辑）

「推理延迟」

较低（单次推理含判断）

较高（多次 LLM 调用）

「适合场景」

对事实性要求极高、研究

快速原型、业务迭代

7.3 选型建议

代码语言：javascript
复制
┌─────────────────────────────────────────────────────────────────┐
│                        选型决策树                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  问题复杂度如何？                                                │
│       │                                                         │
│       ├── 简单（单跳问答）→ 静态 RAG + Reranker                  │
│       │                                                         │
│       └── 复杂（多跳推理）→ 动态 RAG                             │
│                │                                                │
│                ├── 有 GPU 资源？                                 │
│                │       │                                        │
│                │       ├── 是 → Self-RAG（效果最好）             │
│                │       │                                        │
│                │       └── 否 → CRAG + API（灵活易用）           │
│                │                                                │
│                └── 需要高度定制？                                 │
│                        │                                        │
│                        ├── 是 → CRAG / LangGraph                │
│                        │                                        │
│                        └── 否 → RAGFlow Agent                   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

「具体建议：」

场景

推荐方案

快速原型验证

CRAG (LangGraph) + OpenAI API

生产环境简单问答

静态 RAG + Reranker + 混合检索

生产环境复杂问答

CRAG + 本地模型 或 RAGFlow Agent

研究/对事实性要求极高

Self-RAG

企业级知识库平台

RAGFlow / Dify / FastGPT

八、参考资源

论文

论文

链接

Self-RAG: Learning to Retrieve, Generate and Critique

arXiv:2310.11511

Corrective Retrieval Augmented Generation

arXiv:2401.15884

Adaptive-RAG

arXiv:2403.14403

开源项目

项目

链接

说明

Self-RAG

GitHub

官方实现

LangGraph

GitHub

工作流编排框架

RAGFlow

GitHub

RAG 平台

LlamaIndex

GitHub

RAG 框架

LangChain

GitHub

LLM 应用框架

模型资源

模型

HuggingFace

Self-RAG Llama2-7B

selfrag/selfrag_llama2_7b

Self-RAG Llama2-13B

selfrag/selfrag_llama2_13b

BGE Reranker

BAAI/bge-reranker-base

本文参与 腾讯云自媒体同步曝光计划，分享自微信公众号。
原始发表：2026-01-17，如有侵权请联系 cloudcommunity@tencent.com 删除
原理
工作流
模型
搜索
网络

本文分享自 有文化的技术人 微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ，欢迎热爱写作的你一起参与！

原理
工作流
模型
搜索
网络
评论
登录后参与评论
0 条评论
热度
最新
登录 后参与评论
推荐阅读
目录
  • 目录
  • 一、RAG 技术概述
    • 为什么需要 RAG？
    • RAG 基础流程
  • 二、静态 RAG
    • 2.1 核心原理
    • 2.2 优化技术
    • 2.3 主流实践方案
      • 开源框架
      • 向量数据库
      • 企业级产品
    • 2.4 代码示例
      • 基础静态 RAG（LangChain）
      • 带 Reranker 的静态 RAG
  • 三、动态 RAG
    • 3.1 核心原理
    • 3.2 主流实现方案
  • 四、Self-RAG 详解
    • 4.1 核心原理
    • 4.2 反思令牌机制
    • 4.3 环境配置
    • 4.4 代码实现
      • 基础使用
      • 完整 Self-RAG 系统
  • 五、CRAG 详解
    • 5.1 核心原理
    • 5.2 环境配置
    • 5.3 完整代码实现（LangGraph）
      • 使用本地模型的 CRAG
  • 六、RAGFlow 平台
    • 6.1 平台定位
    • 6.2 Agent 工作流机制
    • 6.3 SDK 使用
      • 安装
      • 基础用法
      • Agent 模式
  • 七、技术对比与选型建议
    • 7.1 静态 vs 动态 RAG
    • 7.2 Self-RAG vs CRAG
    • 7.3 选型建议
  • 八、参考资源
    • 论文
    • 开源项目
    • 模型资源
领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 粤公网安备44030502008569号

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

Copyright © 2013 - 2026 Tencent Cloud.

All Rights Reserved. 腾讯云 版权所有

登录 后参与评论