引言

  在上篇文章中，我们搭建了Agentic RAG系统的基础设施，包括文档的加载、处理和检索工具的创建。现在，我们的系统就像一个拥有了一个强大“搜索引擎”的初学者，但它还不知道如何思考和行动。本文将和大家一起完成这个智能代理的“大脑”构建。我们将定义代理的决策节点，包括：何时使用工具、如何判断检索结果是否有效、以及如何优化失败的查询。最终，我们将把这些节点连接成一个流程图，让代理能够自主地执行“规划-检索-评估-再规划”的循环，并成功回答复杂问题。

一、让模型（LLM）学会使用工具

  在上一篇文章的最后，我们创建了一个名为“百度百科”的检索工具。现在，我们需要让大语言模型（LLM）学会使用这个工具。我们将使用阿里云的通义千问模型qwen-plus作为我们代理的“大脑”。generate_query_or_respond函数是这个“大脑”的第一个决策点：它是直接回答用户，还是需要调用检索工具来获取更多信息？我们通过给模型“绑定”工具来实现这一点。当模型认为需要更多信息时，它会输出一个“工具调用”指令，而不是直接输出文本回答。

from langgraph.graph import MessagesState
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import convert_to_messages
import os

# 初始化大语言模型
response_model = ChatTongyi(
    model="qwen-plus",
    temperature=0,
    api_key=os.getenv("DASHESCOPE_API_KEY"))

def generate_query_or_respond(state: MessagesState):
    # 将检索工具绑定到模型，模型可以决定是否调用它
    response = response_model.bind_tools([retriever_tool]).invoke(state["messages"])
    return {"messages": [response]}

  接下来，是系统的核心评估环节。代理在使用工具后，需要判断检索到的文档是否真的与用户问题相关。如果相关，则进入“生成答案”阶段；如果不相关，则说明检索策略可能失败，需要“重写问题”再次尝试。我们定义一个grade_documents节点来完成这个评估工作。这里我们使用了一个带有结构化输出的模型GradeDocuments，并要求它对给定的问题和上下文给出“yes”（相关）或“no”（不相关）的评分。

from pydantic import BaseModel, Field
from typing import Literal

# 定义评分模型的结构
class GradeDocuments(BaseModel):
    binary_score: str = Field(
        description="Relevance score: 'yes' if relevant, or 'no' if not relevant"
    )

GRADE_PROMPT = (
    "You are a grader assessing relevance of a retrieved document to a user question. \n "
    "Here is the retrieved document: \n\n {context} \n\n"
    "Here is the user question: {question} \n"
    "If the document contains keyword(s) or semantic meaning related to the user question, grade it as relevant. \n"
    "Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question."
)

def grade_documents(state: MessagesState,) -> Literal["generate_answer", "rewrite_question"]:
    # 获取原始问题和检索到的上下文
    question = state["messages"][0].content
    context = state["messages"][-1].content

    # 格式化提示词并调用模型进行评分
    prompt = GRADE_PROMPT.format(question=question, context=context)
    response = (
        response_model.with_structured_output(GradeDocuments).invoke(
            [{"role": "user", "content": prompt}]
        )
    )
    score = response.binary_score

    # 根据评分决定下一步去哪个节点
    if score == "yes":
        return "generate_answer"
    else:
        return "rewrite_question"

  通过提示模板引导模型理解用户问题的深层意图，并使用大模型生成改进后的查询语句。函数从状态中提取原始问题，调用大模型返回优化后的问题，最终以消息形式输出，供后续节点重新处理。如果评估结果是不相关，我们就需要启动“问题重写”节点rewrite_question。

REWRITE_PROMPT = (
    "Look at the input and try to reason about the underlying semantic intent / meaning.\n"
    "Here is the initial question:"
    "\n ------- \n"
    "{question}"
    "\n ------- \n"
    "Formulate an improved question:"
)

def rewrite_question(state: MessagesState):
    question = state["messages"][0].content
    prompt = REWRITE_PROMPT.format(question=question)
    response = response_model.invoke([{"role": "user", "content": prompt}])
    # 返回一个包含重写后问题的消息，这个新问题将被重新送入generate_query_or_respond节点
    return {"messages": [{"role": "user", "content": response.content}]}

  其中节点rewrite_question的任务是理解用户原始问题的深层意图，并生成一个更精确、更适合检索的新问题。

  接着定义一个生成答案的节点：使用提示模板，要求模型基于检索到的上下文回答问题，若不知道则直接说明，如果评估结果是相关，则进入“生成答案”节点generate_answer。这个节点的任务是将原始问题和检索到的上下文作为参考，生成最终答案。

GENERATE_PROMPT = (
    "You are an assistant for question-answering tasks. "
    "Use the following pieces of retrieved context to answer the question. "
    "If you don't know the answer, just say that you don't know. "
    "Use three sentences maximum and keep the answer concise.\n"
    "Question: {question} \n"
    "Context: {context}"
)

def generate_answer(state: MessagesState):
    question = state["messages"][0].content
    context = state["messages"][-1].content
    prompt = GENERATE_PROMPT.format(question=question, context=context)
    response = response_model.invoke([{"role": "user", "content": prompt}])
    return {"messages": [response]}

  现在，所有“零部件”都准备就绪。我们需要使用 LangGraph 将它们连接成一个完整的流程图。流程图定义了信息如何在这些节点之间流动。使用 LangGraph 构建一个检索增强生成（RAG）工作流。流程从“生成查询或回答”节点开始，通过条件判断决定是直接结束还是进入“检索”节点。检索后由自定义评分函数评估文档相关性，若相关则进入“生成答案”节点后结束，若不相关则进入“重写问题”节点，再返回起点重新处理。

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition

workflow = StateGraph(MessagesState)

# 添加所有节点
workflow.add_node("generate_query_or_respond", generate_query_or_respond)
workflow.add_node("retrieve", ToolNode([retriever_tool]))
workflow.add_node("rewrite_question", rewrite_question)
workflow.add_node("generate_answer", generate_answer)

# 设置起点
workflow.add_edge(START, "generate_query_or_respond")

# 第一个决策点：模型决定是调用工具还是直接回答
workflow.add_conditional_edges(
    "generate_query_or_respond",
    tools_condition, # 内置的条件判断函数
    {
        "tools": "retrieve", # 如果调用工具，去retrieve节点
        END: END, # 如果直接回答，结束流程
    },
)

# 第二个决策点：评估检索结果是否相关
workflow.add_conditional_edges(
    "retrieve",
    grade_documents, # 我们自定义的评分函数
)

workflow.add_edge("generate_answer", END) # 生成答案后结束
workflow.add_edge("rewrite_question", "generate_query_or_respond") # 重写问题后，重新开始“生成查询或回答”的流程

# 编译成可运行的应用
graph = workflow.compile()

  最后，我们终于可以运行这个完整的Agentic RAG系统了。让我们问一个需要实时信息的问题：“国足刚刚赢比赛的比分是多少？”看看我们的智能代理如何工作。

for chunk in graph.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "国足刚刚赢比赛的比分是多少?",
            }
        ]
    }
):
    for node, update in chunk.items():
        print("Update from node", node)
        update["messages"][-1].pretty_print()

  其中generate_query_or_respond 节点判断需要检索，并生成了一个查询“中国国家足球队 最近比赛结果”。retrieve 节点执行检索，返回了包含“国足2比0库拉索”等新闻的片段。grade_documents 节点评估这些结果，发现与问题相关，决定进入generate_answer。generate_answer 节点根据问题“国足刚刚赢比赛的比分是多少?”和检索到的信息，生成了“国足刚刚赢的比赛是2比0战胜库拉索。”的最终答案。