传统 RAG 面临的困境

现行 RAG 系统普遍存在一个未被重视的缺陷：不论问题性质如何，统一执行检索操作。

当用户询问"RAG 系统如何做效果评估"时——执行检索。 当用户询问"1 加 1 等于几"时——同样执行检索。 当用户请求"帮我写一个计算最大公约数的函数"时——还是执行检索。

上述后两个问题完全依赖模型内部知识即可解答，强制检索不仅消耗计算资源，还可能将无关文档混入上下文，干扰大语言模型的最终判断。

2023 年，Asai 等人提出的 Self-RAG 框架通过引入"反思机制"有效破解了这一难题：模型在文本生成期间会输出特殊反思 token，自主判断是否需要检索、检索内容是否相关、最终答案是否可验证。

Self-RAG 的四种反思 token

论文中训练了一个能够输出四类特殊 token 的模型：

Token	语义含义	取值范围
[Retrieve]	判断是否需要检索？	yes / no / continue
[IsRel]	检索到的文档与问题是否相关？	relevant / irrelevant
[IsSup]	生成内容是否有文档依据？	fully supported / partially supported / no support
[IsUse]	回答对用户是否有价值？	1~5 分

这四种 token 贯穿完整生成流程，使模型在各阶段做出自适应抉择，而非机械地"始终检索、始终采用"。

工程落地层面，无需训练包含特殊 token 的专用模型——采用常规 LLM 配合 Prompt 模拟这四项判断逻辑，即可获得良好效果。

基于 LangGraph 构建 Self-RAG 流程

处理流程图

用户输入
    ↓
[router] 判断是否需要检索？
    ├─ yes → [search] 向量库查询 top-4
    │            ↓
    │         [relevance_filter] 逐篇评估相关度，排除无关文档
    │            ↓
    │         [context_generate] 基于过滤后文档生成回答
    │            ↓
    └─ no  → [direct_generate] 直接依据内部知识生成
                 ↓
             [citation_check] 回答是否有文档支撑？
                 ↓
              输出最终结果

状态定义

LangGraph 的核心在于 State——节点间传递的状态容器：

class SelfRAGState(TypedDict):
    query: str
    should_search: str           # "yes" | "no"
    raw_documents: list[Document]
    filtered_documents: list[Document]
    response: str
    citation_status: str         # "supported" | "unsupported"
    total_tokens: int
    execution_trace: list[str]   # 记录执行轨迹

核心节点实现

节点一：检索路由判断（router）

RETRIEVE_ROUTING_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "你是一个 RAG 系统的智能路由模块。请判断下方问题是否需要从外部知识库检索内容。\n\n"
     "必须检索：涉及具体技术参数、产品选型、实现细节等事实性信息\n"
     "无需检索：基础常识、数学运算、逻辑推导、日常寒暄等\n\n"
     "仅输出 yes 或 no，无需附加说明。"),
    ("human", "问题：{query}"),
])

节点二：相关性过滤（relevance_filter）

RELEVANCE_FILTER_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "请判断给定文档片段是否与问题相关。\n"
     "相关：文档内容能帮助回答问题\n"
     "无关：文档内容与问题无直接关联\n\n"
     "输出 relevant 或 irrelevant。"),
    ("human", "问题：{query}\n\n文档：{document_content}"),
])

def relevance_filter_node(state: SelfRAGState) -> SelfRAGState:
    query = state["query"]
    docs = state["raw_documents"]
    
    filtered = []
    for doc in docs:
        result = relevance_model.invoke(
            RELEVANCE_FILTER_PROMPT.format(
                query=query,
                document_content=doc.page_content[:500]
            )
        )
        if "relevant" in result.content.lower():
            filtered.append(doc)
    
    return {"filtered_documents": filtered}

节点三：支撑性验证（citation_check）

CITATION_CHECK_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "请验证生成的回答是否有充分的文档依据。\n"
     "fully supported：回答内容完全可从文档中找到依据\n"
     "partially supported：部分内容有依据，部分为模型自行补充\n"
     "no support：回答内容与检索文档无关\n\n"
     "输出上述三种判断之一。"),
    ("human", "问题：{query}\n\n回答：{response}\n\n参考文档：{documents}"),
])

def citation_check_node(state: SelfRAGState) -> SelfRAGState:
    query = state["query"]
    response = state["response"]
    docs = state.get("filtered_documents", [])
    
    doc_texts = "\n\n".join([d.page_content[:300] for d in docs])
    
    result = citation_model.invoke(
        CITATION_CHECK_PROMPT.format(
            query=query,
            response=response,
            documents=doc_texts
        )
    )
    
    verdict = "supported"
    if "no support" in result.content.lower():
        verdict = "unsupported"
    
    return {"citation_status": verdict}

工作流组装

workflow = StateGraph(SelfRAGState)

workflow.add_node("router", routing_node)
workflow.add_node("search", search_node)
workflow.add_node("relevance_filter", relevance_filter_node)
workflow.add_node("context_generate", context_generate_node)
workflow.add_node("direct_generate", direct_generate_node)
workflow.add_node("citation_check", citation_check_node)

workflow.set_entry_point("router")

workflow.add_conditional_edges(
    "router",
    lambda x: "search" if x["should_search"] == "yes" else "direct_generate"
)

workflow.add_edge("search", "relevance_filter")
workflow.add_edge("relevance_filter", "context_generate")
workflow.add_edge("context_generate", "citation_check")
workflow.add_edge("direct_generate", "citation_check")
workflow.add_edge("citation_check", END)

app = workflow.compile()

实际效果与适用场景

引入 Self-RAG 后，系统能够智能区分"需要外部知识"与"可内部解答"的问题类型。对于技术文档查询、产品参数确认等场景正常触发检索流程；而对于基础计算、简单问答等则跳过检索步骤，直接基于模型知识生成回答。

该方案在减少不必要计算开销的同时，也降低了无关文档干扰模型判断的风险，使 RAG 系统的响应质量更加稳定可靠。