自适应 RAG

上一篇 LangGraph 构建 RAG 走通了最基础的 retrieve → generate，但它有个致命假设：每个问题都该检索、检索结果一定相关、生成结果一定对。现实往往不是这样。本篇引入 Self-RAG 的思想，让模型自己判断"要不要检索""检索的相关吗""回答有没有幻觉"，并用条件边把判断结果接到不同的分支上。

一、为什么需要"自适应"

基础 RAG 在这些场景会翻车：

1. 闲聊问题也去检索：用户说"你好"，模型还去向量库捞一堆文档，浪费 token 还可能把回答带偏。
2. 检索到不相关文档：硬塞进 prompt 让模型"强行回答"，反而制造幻觉。
3. 回答与检索文档脱节：模型脑补了文档里没有的内容，没人检查。

自适应 RAG（受 Self-RAG 论文启发）的思路是：在关键节点上加评估器，由 LLM 自己打分，再用条件边根据分数路由。这正好是 LangGraph 条件边的用武之地。

对 Java/LangGraph4j 同学：这相当于在工作流里加 if/else 分支节点，每个分支是一个独立的处理路径。

二、整体流程图

自适应 RAG 在基础 RAG 上加了三个评估环节：

flowchart TD
    S([START]) --> Q[query_analyze 查询分析]
    Q -- 需检索 --> R[retrieve 检索]
    Q -- 无需检索 --> G2[generate_direct 直接回答]
    R --> D[grade_docs 文档评分]
    D -- 有相关文档 --> G1[generate 生成]
    D -- 都不相关 --> RQ[rewrite_query 改写问题]
    RQ --> R
    G1 --> H[grade_answer 回答评分]
    H -- 合格 --> E([END])
    H -- 不合格/幻觉 --> G1
    G2 --> E

三个评估点：

评估点	判断什么	结果分流
query_analyze	这个问题需要检索吗	需检索 / 直接回答
grade_docs	检索到的文档相关吗	有相关 / 全不相关(改写后重检索)
grade_answer	回答是否基于文档(无幻觉)	合格 / 重生成

三、状态设计

需要在基础 RAG 状态上多加几个字段：原始问题、改写后的问题、文档相关性标记、回答是否合格。

from typing import TypedDict, List
from langchain_core.documents import Document

class AdaptiveRAGState(TypedDict):
    question: str               # 当前使用的问题（可能被改写）
    original_question: str      # 原始问题，便于追溯
    documents: List[Document]   # 检索到的文档
    need_retrieval: bool        # 是否需要检索
    relevant_docs: List[Document]  # 评分后筛出的相关文档
    answer: str                 # 最终回答
    retry_count: int            # 重生成次数，防止死循环

注意 retry_count 这个字段——任何带循环的图都必须有"刹车"，否则一旦评分一直不合格就会无限重生成。

四、节点实现

1. 查询分析节点

让 LLM 判断问题是否需要查资料。用一个结构化输出会更稳。

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

class QueryDecision(BaseModel):
    need_retrieval: bool = Field(description="是否需要检索知识库")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# with_structured_output 让模型直接返回结构化对象，省去解析 JSON 的麻烦
decision_llm = llm.with_structured_output(QueryDecision)

def query_analyze(state: AdaptiveRAGState) -> dict:
    """判断问题是否需要检索"""
    question = state["question"]
    prompt = (
        "判断下面这个问题是否需要查询知识库才能回答。\n"
        "事实性、知识性问题需要检索；闲聊、问候、写作类不需要。\n\n"
        f"问题：{question}"
    )
    decision = decision_llm.invoke(prompt)
    return {"need_retrieval": decision.need_retrieval}

2. 文档评分节点

对每篇检索文档打"相关/不相关"分，只保留相关的。

class DocGrade(BaseModel):
    relevant: bool = Field(description="文档是否与问题相关")

grade_llm = llm.with_structured_output(DocGrade)

def grade_docs(state: AdaptiveRAGState) -> dict:
    """逐条给文档打相关性分"""
    question = state["question"]
    docs = state["documents"]
    relevant = []
    for d in docs:
        prompt = (
            "判断下面这段资料是否与问题相关，能帮助回答问题。\n\n"
            f"问题：{question}\n\n资料：{d.page_content}"
        )
        result = grade_llm.invoke(prompt)
        if result.relevant:
            relevant.append(d)
    return {"relevant_docs": relevant}

3. 问题改写节点

当文档全不相关时，把问题改写得更"好检索"。

def rewrite_query(state: AdaptiveRAGState) -> dict:
    """改写问题，使其更利于检索"""
    question = state["question"]
    prompt = (
        "把下面问题改写得更具体、更利于向量检索，只输出改写后的问题：\n"
        f"{question}"
    )
    new_q = llm.invoke(prompt).content
    return {"question": new_q, "retry_count": state.get("retry_count", 0) + 1}

4. 生成与回答评分

from langchain_core.prompts import ChatPromptTemplate

gen_prompt = ChatPromptTemplate.from_template(
    "只根据资料回答，资料不足就说不知道。\n资料：\n{context}\n\n问题：{question}\n回答："
)

def generate(state: AdaptiveRAGState) -> dict:
    docs = state["relevant_docs"]
    context = "\n\n".join(d.page_content for d in docs)
    resp = llm.invoke(gen_prompt.invoke({"context": context, "question": state["question"]}))
    return {"answer": resp.content, "retry_count": state.get("retry_count", 0) + 1}

def generate_direct(state: AdaptiveRAGState) -> dict:
    """不需要检索时直接回答"""
    resp = llm.invoke(state["question"])
    return {"answer": resp.content}

class AnswerGrade(BaseModel):
    grounded: bool = Field(description="回答是否完全基于资料，没有幻觉")

answer_grade_llm = llm.with_structured_output(AnswerGrade)

def grade_answer(state: AdaptiveRAGState) -> dict:
    """检查回答是否基于资料（无幻觉）"""
    docs = state["relevant_docs"]
    context = "\n\n".join(d.page_content for d in docs)
    answer = state["answer"]
    prompt = (
        "判断回答是否完全基于下面资料，没有编造资料外的内容。\n\n"
        f"资料：{context}\n\n回答：{answer}"
    )
    result = answer_grade_llm.invoke(prompt)
    # 把"是否合格"塞回状态，供条件边读取
    return {"answer": answer, "_answer_ok": result.grounded}

五、条件边与路由

条件边是自适应 RAG 的核心，所有"判断"在这里转成"走向"。

def route_after_query(state: AdaptiveRAGState) -> str:
    if state["need_retrieval"]:
        return "retrieve"
    return "generate_direct"

def route_after_grade(state: AdaptiveRAGState) -> str:
    # 没有相关文档 → 改写重检索
    if not state.get("relevant_docs"):
        return "rewrite_query"
    return "generate"

def route_after_answer(state: AdaptiveRAGState) -> str:
    # 防死循环：最多重生成 2 次
    if state.get("_answer_ok", False) or state.get("retry_count", 0) >= 3:
        return END
    return "generate"

六、组装图

from langgraph.graph import StateGraph, START, END

gb = StateGraph(AdaptiveRAGState)
for name, fn in [
    ("query_analyze", query_analyze),
    ("retrieve", retrieve),       # 复用上一篇的检索节点
    ("grade_docs", grade_docs),
    ("rewrite_query", rewrite_query),
    ("generate", generate),
    ("generate_direct", generate_direct),
    ("grade_answer", grade_answer),
]:
    gb.add_node(name, fn)

gb.add_edge(START, "query_analyze")
gb.add_conditional_edges("query_analyze", route_after_query, {
    "retrieve": "retrieve", "generate_direct": "generate_direct"
})
gb.add_edge("retrieve", "grade_docs")
gb.add_conditional_edges("grade_docs", route_after_grade, {
    "rewrite_query": "rewrite_query", "generate": "generate"
})
gb.add_edge("rewrite_query", "retrieve")     # 改写后重新检索
gb.add_edge("generate", "grade_answer")
gb.add_conditional_edges("grade_answer", route_after_answer, {
    END: END, "generate": "generate"
})
gb.add_edge("generate_direct", END)

adaptive_app = gb.compile()

七、完整可运行示例

下面给一个能跑的最小例子，沿用上一篇的知识库。为节省篇幅只贴关键差异部分，检索器构造同 上一篇。

import os
from typing import TypedDict, List
from pydantic import BaseModel, Field

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.graph import StateGraph, START, END

# ---- 知识库（同上一篇，略）----
docs = [Document(page_content=t) for t in [
    "LangGraph 把流程建模成图，节点是函数，边决定执行顺序。",
    "条件边可以根据状态选择下一个节点，实现分支逻辑。",
    "Python 用缩进表示代码块，不需要大括号。",
]]
vs = InMemoryVectorStore(OpenAIEmbeddings(model="text-embedding-3-small"))
vs.add_documents(docs)
retriever = vs.as_retriever(search_kwargs={"k": 2})

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ---- 状态 ----
class S(TypedDict):
    question: str
    documents: List[Document]
    relevant_docs: List[Document]
    need_retrieval: bool
    answer: str
    retry_count: int
    _answer_ok: bool

# ---- 三个结构化输出模型 ----
class QDec(BaseModel):
    need_retrieval: bool
class DGrade(BaseModel):
    relevant: bool
class AGrade(BaseModel):
    grounded: bool

# ---- 节点 ----
def query_analyze(state):
    d = llm.with_structured_output(QDec).invoke(
        "判断是否需要查知识库(事实/知识性=需要, 闲聊=不需要)：\n" + state["question"])
    return {"need_retrieval": d.need_retrieval}

def retrieve(state):
    return {"documents": retriever.invoke(state["question"])}

def grade_docs(state):
    rel = []
    for d in state["documents"]:
        g = llm.with_structured_output(DGrade).invoke(
            f"资料是否与问题相关：\n问题：{state['question']}\n资料：{d.page_content}")
        if g.relevant:
            rel.append(d)
    return {"relevant_docs": rel, "retry_count": state.get("retry_count", 0)}

def rewrite_query(state):
    new = llm.invoke("改写得更利于检索，只输新问题：\n" + state["question"]).content
    return {"question": new, "retry_count": state.get("retry_count", 0) + 1}

def generate(state):
    ctx = "\n\n".join(d.page_content for d in state["relevant_docs"])
    a = llm.invoke(f"只根据资料回答，不足就说不知道。\n资料：{ctx}\n问题：{state['question']}")
    return {"answer": a.content, "retry_count": state.get("retry_count", 0) + 1}

def generate_direct(state):
    return {"answer": llm.invoke(state["question"]).content}

def grade_answer(state):
    ctx = "\n\n".join(d.page_content for d in state["relevant_docs"])
    g = llm.with_structured_output(AGrade).invoke(
        f"回答是否基于资料无幻觉：\n资料：{ctx}\n回答：{state['answer']}")
    return {"_answer_ok": g.grounded}

# ---- 路由 ----
def r1(state): return "retrieve" if state["need_retrieval"] else "generate_direct"
def r2(state): return "rewrite_query" if not state.get("relevant_docs") else "generate"
def r3(state): return END if (state.get("_answer_ok") or state.get("retry_count",0)>=3) else "generate"

# ---- 建图 ----
gb = StateGraph(S)
for n, f in [("query_analyze",query_analyze),("retrieve",retrieve),("grade_docs",grade_docs),
             ("rewrite_query",rewrite_query),("generate",generate),
             ("generate_direct",generate_direct),("grade_answer",grade_answer)]:
    gb.add_node(n, f)
gb.add_edge(START, "query_analyze")
gb.add_conditional_edges("query_analyze", r1, {"retrieve":"retrieve","generate_direct":"generate_direct"})
gb.add_edge("retrieve", "grade_docs")
gb.add_conditional_edges("grade_docs", r2, {"rewrite_query":"rewrite_query","generate":"generate"})
gb.add_edge("rewrite_query", "retrieve")
gb.add_edge("generate", "grade_answer")
gb.add_conditional_edges("grade_answer", r3, {END:END,"generate":"generate"})
gb.add_edge("generate_direct", END)
app = gb.compile()

if __name__ == "__main__":
    for q in ["条件边是干嘛的？", "你好呀"]:
        r = app.invoke({"question": q, "retry_count": 0})
        print(q, "=>", r["answer"], "| 检索：", r.get("need_retrieval"))

第二个问题"你好呀"会被 query_analyze 判为不需要检索，直接走 generate_direct，省掉一轮检索和打分。

八、常见踩坑

1. 评分 prompt 设计太松或太严

• 太松：所有文档都判"相关"，等于没筛。
• 太严：相关文档也被判不相关，陷入"改写→检索→全不相关→再改写"的死循环。

调试建议：先固定一组问答对，把评分节点的判断打印出来人工核对，再调 prompt 措辞。结构化输出（with_structured_output）比让模型回 "yes/no" 字符串稳得多。

2. 循环过多导致 token 失控

rewrite → retrieve → grade → rewrite 和 generate → grade → generate 都是潜在死循环。任何带循环的图都必须有 retry_count 上限，本例用 retry_count >= 3 强制退出。生产环境还要叠加 token 总量上限和单次请求超时。

3. 条件边返回了未注册的节点名

add_conditional_edges 的第二个参数返回的字符串必须和 add_node 的名字一致，且要在第三个参数的 path_map 里列出。写错一个字母图就编译不过。建议把 path_map 显式写出来（如上例），别让 LangGraph 自己猜。

4. 状态字段污染

_answer_ok 这种"内部信号"字段一旦写进状态，下游节点都能看到。如果不想让它进最终输出或干扰别的判断，命名上加下划线前缀做约定，并在出口节点显式清理。本系列 实战项目 里会更系统地处理状态卫生。

九、小结

• 自适应 RAG = 基础 RAG + 三个评估节点（查询分析、文档评分、回答评分）+ 条件边路由。
• 三个评估点分别决定：要不要检索、要不要重检索、要不要重生成。
• 结构化输出（with_structured_output）是让 LLM"打分"最稳的方式。
• 带循环的图必须有 retry_count 刹车，否则会无限重试烧钱。

下一篇 自纠正 RAG (CRAG) 会引入更精细的"三档评分"和 web 搜索兜底，进一步处理检索质量不行的情况。