首页 / 文章 / 在LangGraph中构建ReAct研究智能体:大脑、手部与路由器。

在LangGraph中构建ReAct研究智能体:大脑、手部与路由器。

学习如何将ReAct的推理-行动-观察循环实现为LangGraph子图,包括强制反思、迭代预算以及并行散收集研究机制。

4678 词

单个大语言模型调用无法研究它完全不了解的问题。一个实用的研究代理必须进行搜索、阅读返回的结果、判断还缺少什么,然后再次搜索,直到拥有足够的信息来给出答案。ReAct模式为这种行为提供了明确的结构,而LangGraph则允许你用简洁的显式图来表示它,而非一堆复杂的while循环。完成本教程后,你将理解一个可运行的ReAct研究子图中的每个节点,知道它可能出错的地方,并掌握一系列升级方法,使其能够在生产环境中安全运行。

其设计借鉴了名为deep-research-agent的开源项目中的研究员组件。从高层次上看,一次运行流程如下:

  1. 用户提出一个研究问题。
  2. 作为“大脑”的大语言模型会对该问题进行推理,并选择要调用的工具。
  • 工具执行器,即“手”,负责运行该工具并返回观察结果。
  • 路由器会检查大脑发出的最新消息,判断是否还有更多工具调用请求。
  • 如果有,则控制权返回到第2步。
  • 如果没有,就会将积累的所有信息整合为最终答案。
  • ReAct模式究竟是什么

    ReAct代表“推理与行动”。这一概念源自研究论文《ReAct:在语言模型中协同实现推理与行动》(首次发布于2022年,并在2023年的ICLR会议上展示)。该研究的核心发现是:当语言模型交替执行两种操作——思考下一步该做什么,以及使用工具获取真实信息时,表现会更佳。单独执行其中任何一种操作都会导致问题:仅进行推理的模型会凭空编造事实,因为没有任何依据;仅执行行动的模型则会机械地调用工具,而不去理解其返回的结果。

    Google Cloud的架构指南将这种模式描述为对自然语言指令的循环处理,直到满足退出条件为止。在实际应用中,它可分解为三个重复的阶段:

    1. 思考。模型会查看迄今为止收集到的所有信息,判断该请求是否已得到解答,或是下一步该做什么。
    2. 行动。根据这一判断,它要么调用工具获取更多数据,要么撰写最终答案,从而结束循环。
    3. 观察。工具的输出结果会返回并保存在对话中。由于之前的观察记录仍然可见,模型可以在此基础上继续处理,而无需重复搜索或遗忘上下文。

    这正符合经验丰富的工程师处理陌生问题时的方式:先查找相关资料,思考问题,再查找更多信息,最后才得出结论。Anthropic的工具使用文档从API层面描述了相同的机制:模型会返回一个工具使用请求,你的应用程序执行该请求并返回结果,随后这一过程不断重复。无论底层的模型是Claude、GPT还是Gemini,这个循环都是相同的。

    这一点至关重要:ReAct是一种模式,而非库的功能。LangGraph恰好提供了通过StateGraph来简洁表达它的方法,但你也可以使用任何模型和编排代码实现同样的循环逻辑。deep-research-agent项目将其封装为LangGraph的子图,从而实现了可组合性:更大的系统可以将其作为一个整体来调用。如果在深入编写代码之前需要概念上的回顾,可参阅我们关于AI智能体如何将推理与现实世界行动相结合的概述。

    三个组件及其共享的状态

    “研究者循环”由三个小型函数构成,每个函数负责一项任务。大脑模块(llm_call)会读取当前的对话内容,然后以纯文本或工具调用请求的形式作出回应。操作模块(tool_node)会执行所有被请求的工具调用,并返回其结果。路由模块(should_continue)则负责判断是否需要进入下一轮循环。将这些职责分开,意味着可以分别对每个模块进行单元测试,独立更换模型或工具,同时也能通过查看这三个简短的函数来理解整个循环的运作方式。

    定义研究者状态

    在 LangGraph 图中流动的所有数据都存储在状态对象中,该对象被定义为 TypedDict。研究者状态包含五个字段:researcher_messages 用于保存当前的对话内容,它被封装在带有 add_messages 约简器的 Annotated 中,该约简器指示 LangGraph 将新消息合并到现有列表中而非直接覆盖它。tool_call_iterations 用于统计循环次数,research_topic 记录智能体正在研究的内容,compressed_research 用于存储最终总结结果,而 raw_notes 则通过使用 operator.add 作为约简器来收集笔记,这样不同节点返回的列表就能被拼接在一起。

    # Define the state that flows through the entire ReAct loop
    from typing import Annotated, Sequence, List, TypedDict
    from langgraph.graph.message import add_messages
    from langchain_core.messages import BaseMessage
    import operator
    
    class ResearcherState(TypedDict):
        # The message history accumulates as the loop runs
        researcher_messages: Annotated[Sequence[BaseMessage], add_messages]
        # Tracks how many tool call iterations have occurred
        tool_call_iterations: int
        # The topic this researcher is investigating
        research_topic: str
        # The final compressed output after the loop ends
        compressed_research: str
        # Raw notes collected during research
        raw_notes: Annotated[List[str], operator.add]
    

    正是这些 reducer 让循环得以运行。每当“大脑”发出指令或“手”返回结果时,节点只会返回新的消息,而 reducer 会将其追加。如果没有 add_messages,每个节点都会替换历史记录,模型就会丢失在之前迭代中所学到的一切。

    该项目还定义了更严格的输出结构。它控制着当父图调用子图时哪些字段会被传递出去。

    # Output schema controls what the parent graph sees
    class ResearcherOutputState(TypedDict):
        compressed_research: str
        raw_notes: Annotated[List[str], operator.add]
        researcher_messages: Annotated[Sequence[BaseMessage], add_messages]
    

    将内部状态与输出状态分开是 LangGraph 的良好习惯。像 tool_call_iterations 这样的计数信息仅在循环内部有用,因此父图永远无法看到它。父图接收的是经过压缩的研究结果、原始笔记以及消息,这样就能保持各图之间的接口简洁且目的明确。

    为何选择 TypedDict 而非普通字典?因为它能明确记录在图中传递的数据内容,还能让类型检查工具发现拼写错误的键。add_messages 减算子在此基础上增加了功能:它会添加新消息,而当传入的消息包含列表中已有消息的 ID 时,会替换该消息而非重复添加,从而保持顺序和唯一性的一致性。

    核心部分:llm_call

    核心部分是进行推理的地方。它会根据系统消息以及完整的消息历史构建提示语,然后询问模型下一步该做什么。需要注意的是,源代码将这段代码标记为 JavaScript,但实际上它是 Python 代码。

    # The "brain" of the researcher: analyzes current state and decides next action
    from langchain_core.messages import SystemMessage
    
    def llm_call(state: ResearcherState):
        # Invoke the LLM with the system prompt and full conversation history
        return {
            "researcher_messages": [
                model_with_tools.invoke(
                    [SystemMessage(content=research_agent_prompt.format(date=get_today_str()))]
                    + state["researcher_messages"]
                )
            ]
        }
    

    这里会发生三件事。首先,根据research_agent_prompt创建一个SystemMessage,并通过get_today_str()注入当天的日期,以便模型判断其信息的时效性。其次,该系统消息会被添加到state[“researcher_messages”]中的所有内容之前,这样模型就能始终看到完整的上下文。最后,model_with_tools.invoke()会将这些内容发送给配备了工具的模型。回复要么是表示研究已完成的纯文本,要么是一个或多个表示需要更多信息的工具调用请求。该函数会将回复以列表形式返回到researcher_messages中,而reducer则将其追加进去。

    系统提示词会在每次调用时重新生成,而不会被存储在状态中。这样既能保持历史记录的整洁,也能确保即使在多次迭代之后,指令始终处于优先位置。

    model_with_tools是在设置过程中创建的。该项目没有导入诸如ChatOpenAI这样的第三方类,而是使用了LangChain的init_chat_model()函数,该函数接受以提供者名称为前缀的模型字符串。

    # Initialize the model using LangChain's provider-agnostic helper
    from langchain.chat_models import init_chat_model
    
    # The project uses different models for different tasks
    model = init_chat_model(model="openai:gpt-4o")
    
    # Bind the research tools so the model knows what actions are available
    model_with_tools = model.bind_tools([tavily_search, think_tool])
    

    提供商前缀的优势在于:从 "openai:gpt-4o" 更改为 Anthropic 模型(格式为 "anthropic:claude-sonnet-4-20250514")仅需修改配置,无需更改导入方式。请查看您所使用提供商的当前模型列表,因为这些标识符会随时间变化。.bind_tools() 用于向模型通报可用的操作。目前绑定了两种工具:通过 Tavily API 进行网络搜索的 tavily_search,以及下文将详细介绍的反射工具 think_tool

    执行工具:tool_node

    “手”负责接收来自“大脑”最新消息中的工具调用并执行相应操作。尽管标注为 JavaScript,但这段代码实际上仍是 Python 编写的。

    # The "hands" of the researcher: executes all tool calls from the brain
    from langchain_core.messages import ToolMessage
    
    def tool_node(state: ResearcherState):
        # Get the tool calls from the last message (the brain's output)
        tool_calls = state["researcher_messages"][-1].tool_calls
        observations = []
    
        # Execute each tool call and collect raw results
        for tool_call in tool_calls:
            tool = tools_by_name[tool_call["name"]]
            observations.append(tool.invoke(tool_call["args"]))
    
        # Convert raw results into properly formatted ToolMessage objects
        tool_outputs = [
            ToolMessage(
                content=str(observation),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            )
            for observation, tool_call in zip(observations, tool_calls)
        ]
    
        return {"researcher_messages": tool_outputs}
    

    该函数从最新的消息中读取tool_calls,在tools_by_name字典中按名称查找每个被请求的工具,然后使用模型提供的参数来调用这些工具。确保正确性的关键在于:每个原始结果都会被封装在包含三个字段的ToolMessage对象中。content字段存储字符串形式的处理结果,name字段记录生成该结果的工具名称,而tool_call_id字段则将结果与触发它的具体请求关联起来。模型API要求必须提供这个标识符;无法与任何请求匹配的工具结果会被拒绝,而没有对应结果的请求则会使对话处于无效状态。

    tools_by_name 这一映射在项目笔记本中被使用,但从未被定义。你需要自行构建它,例如设置为 {"tavily_search": tavily_search, "think_tool": think_tool},或者通过工具列表的字典推导式来确保名称始终与已绑定的内容保持一致。

    如果“大脑”要求 tavily_search 查找“最新的人工智能研究”,“双手”就会执行该查询并返回类似这样的消息:

    ToolMessage(content="Search results for 'latest AI research': ...", name="tavily_search", tool_call_id="call_abc123")
    

    由于在普通的 for 循环中工具调用是顺序执行的,因此包含多次搜索的那一轮所需时间相当于所有搜索时间之和。这对于演示来说是可以接受的;后续我们会探讨如何让这个节点更加稳健。

    路由器:should_continue

    路由器是功能最简单的组件,负责控制整个循环。它会查看最后一条消息并选择下一个节点。

    # The "router": determines whether to loop again or finish
    from typing import Literal
    
    def should_continue(state: ResearcherState) -> Literal["tool_node", "compress_research"]:
        # Check the last message in the conversation
        messages = state["researcher_messages"]
        last_message = messages[-1]
    
        # If the brain requested tool calls, continue the loop
        if last_message.tool_calls:
            return "tool_node"
    
        # If no tool calls, the brain is done researching
        return "compress_research"
    

    如果大脑发出的最新消息包含tool_calls,路由器会返回"tool_node",循环继续进行。如果大脑仅生成文本,则路由器会返回"compress_research",从而跳出循环进入总结阶段。这一决策完全基于模型的输出;路由器本身并不判断该研究是否足够好。

    Literal[“tool_node”, “compress_research”] 返回的注解会告知 LangGraph 可能的目标有哪些。LangGraph 利用这些信息来了解路由器的分支结构(例如在绘制图表时或未提供明确映射时),因此它的作用远不止于文档说明。不过,这并不妨碍该函数在运行时返回其他字符串;一旦选择某个分支,这些字符串就会以错误的形式体现出来。

    为何要先进入压缩步骤而非直接结束?因为这个子图是为监管代理设计的。监管代理需要简洁、结构化的答案,而非冗长的搜索记录、分析过程及工具数据内容。在子图内部进行压缩可以减少监管代理自身的上下文负担。

    使用 StateGraph 连接循环

    在三个功能都准备就绪后,下一步就是将它们连接起来。无需手动编写控制流程,只需声明节点和边,LangGraph便会运行该图结构。执行过程从“大脑”开始,经过“路由器”,要么传递到“手部”(之后总会返回“大脑”),要么通过compress_research退出。

    图的组装与编译

    尽管标注为JavaScript,但这段代码实际上也是Python编写的。

    # Build the ReAct loop as a LangGraph StateGraph
    from langgraph.graph import StateGraph, START, END
    
    # Initialize the graph with both input state and output schema
    agent_builder = StateGraph(ResearcherState, output_schema=ResearcherOutputState)
    
    # Add the three nodes to the graph
    agent_builder.add_node("llm_call", llm_call)                       # The brain
    agent_builder.add_node("tool_node", tool_node)                     # The hands
    agent_builder.add_node("compress_research", compress_research)     # The exit point
    
    # Wire the entry point: execution starts at the brain
    agent_builder.add_edge(START, "llm_call")
    
    # Wire the router: after the brain thinks, decide what to do next
    agent_builder.add_conditional_edges(
        "llm_call",
        should_continue,
        {
            "tool_node": "tool_node",
            "compress_research": "compress_research",
        },
    )
    
    # Wire the loop: after the hands act, always go back to the brain
    agent_builder.add_edge("tool_node", "llm_call")
    
    # Wire the exit: after compression, end the graph
    agent_builder.add_edge("compress_research", END)
    
    # Compile the graph into a runnable agent
    researcher_agent = agent_builder.compile()
    

    从上到下阅读:

    1. StateGraph(ResearcherState, output_schema=ResearcherOutputState)用于创建包含完整内部状态以及父图所能看到的受限输出模式的图结构。
    2. 注册了三个节点:llm_calltool_nodecompress_research
  • add_edge(START, "llm_call")将大脑设为入口点。
  • add_conditional_edges将路由器连接到大脑的输出端,同时通过字典将每个可能的返回值映射到对应的节点。
  • tool_node回到llm_call的固定边构成了循环回路。
  • add_edge("compress_research", END)在摘要生成完成后终止图结构。
  • .compile()会将这些定义转换为可运行的对象。它被命名为researcher_agent而非简单的agent,因为在整个项目中它是由监督节点调用的子图。

    传递给add_conditional_edges的映射关系值得重视。其键值"tool_node""compress_research"必须与should_continue返回的值完全一致,且其值必须是真实的节点名称。编译时会检查这些映射目标是否存在,因此若节点名称有误,会在程序运行初期就失败,而不会等到中途。相反,如果路由器返回的值为映射表中不存在的,则只有在对应分支被执行时才会失败,因此需在两个分支上都对路由器进行测试。即便如此,这种方式仍比手动编写的等效while循环更容易验证,因为在后者中,错误的分支只会导致异常行为。

    可视化循环结构

    编译后的图结构呈现为如下流程:

    START
      │
      ▼
    llm_call (Brain reasons about the query)
      │
      ├── has tool_calls? ──► tool_node (Hands execute tools)
      │                            │
      │                            └──► llm_call (Back to brain)
      │
      └── no tool_calls? ──► compress_research (Summarize and exit)
    

    这就是经典的ReAct循环。只要模型持续请求工具,大脑和双手就可以交替运作多次。每次迭代都会向状态中添加观测数据,因此每一次新的决策都能基于比上一次更多的信息来做出。

    添加检查点功能

    目前的示例是一个独立的、运行在内存中的智能体。对于需要长时间运行的任务,需要在编译时添加检查点机制,以便在每一步之后保存图的结构状态。

    # Production: add checkpointing for fault tolerance
    from langgraph.checkpoint.memory import MemorySaver
    
    checkpointer = MemorySaver()
    agent = agent_builder.compile(checkpointer=checkpointer)
    

    MemorySaver会将检查点保存在进程内存中,这非常适合开发和测试,但重启后就会消失。对于生产环境,LangGraph提供了诸如PostgresSaverSqliteSaver这样的持久化检查点工具,它们能够让被中断的运行从上次保存的步骤继续执行,并记录下每一次状态变化。一个实际需要注意的细节是:一旦图结构使用了检查点工具,每次invoke调用都需要在其配置中包含线程标识符(例如{"configurable": {"thread_id": "..."}}),这样LangGraph才能知道应该加载和更新哪个已保存的对话内容。

    增强循环性能:反射、预算限制与并行处理

    一个单纯的ReAct循环存在两种很快就会显现的故障模式:其一,它可能无限循环而无法收敛,不断消耗令牌和API额度进行搜索;其二,它可能在未真正分析搜索结果的情况下快速完成搜索,从而产生肤浅的答案。该项目针对这两种问题进行了改进,并通过三项新增功能进一步扩展了该循环的功能。

    使用think_tool强制反思

    最有趣的新功能是一种完全不执行任何外部操作的工具,它的唯一作用就是让模型停下来以书面形式进行思考。

    # A tool that forces the agent to pause and reflect
    from langchain_core.tools import tool
    
    @tool(parse_docstring=True)
    def think_tool(reflection: str) -> str:
        """Tool for strategic reflection on research progress and decision-making.
    
        Use this tool after each search to analyze results and plan next steps
        systematically. This creates a deliberate pause in the research workflow
        for quality decision-making.
    
        Args:
            reflection: Your detailed reflection on research progress, findings,
                        gaps, and next steps.
    
        Returns:
            Confirmation that reflection was recorded for decision-making.
        """
        return f"Reflection recorded: {reflection}"
    

    think_tool接收一个reflection字符串,并在前面加上确认标识后返回。冗长的文档字符串并非装饰性内容:当设置parse_docstring=True时,LangChain会从文档字符串中提取工具描述和参数说明,模型在选择工具时会读取这些文本。其作用源于系统提示词,该提示词要求智能体在每次搜索后调用think_tool,从而迫使模型说明它刚刚学到了什么、还缺少什么以及下一步打算做什么。

    如果没有这一步,智能体往往会持续进行搜索而不会在过程中整合任何信息。Anthropic的工程团队也有过相关观察:能够自我审查并修正输出结果的智能体更为可靠,因为它们能在错误恶化之前发现并及时纠正,当偏离目标时也能重新调整方向。think_tool会在每次迭代中加入一个简单的自我审查环节。由于仅需消耗反思内容本身的令牌量以及额外的一次往返处理时间,成本很低,同时还能让模型始终聚焦于目标。

    假设智能体在搜索后发现有三种RAG索引方法,但仍然缺乏用于比较的基准数据。该工具会直接用确认前缀将这一反思内容原样反馈回去:

    Reflection recorded: The search results show three approaches to RAG indexing. I still need to find benchmarks comparing them.
    

    该字符串被存储为ToolMessage,因此在下一次迭代时,系统会读取它自己的计划。为何要使用工具而非直接要求模型“逐步思考”?因为工具调用是追踪日志中清晰可见的独立事件,会以可预测的格式存入消息历史记录,而且提示词可以在循环的特定节点要求使用它。

    系统提示中的预算控制

    第二个添加项则限制了智能体的搜索次数:

    Budget rules embedded in the system prompt:
    
    - Simple queries: 2 to 3 search calls maximum
    - Complex queries: up to 5 search calls maximum
    - Always stop after 5 calls if sources are not found
    

    这些限制存在于系统提示中,而非代码里。模型会被告知针对简单或复杂查询应进行多少次搜索以及何时该停止。这是一种务实的选择:无需计数器或额外的图结构逻辑,只需给出模型能够遵循的指令即可。

    这种权衡确实是存在的。Google Cloud的指南指出,与单次查询相比,迭代式方法会增加延迟,且搜索结果在很大程度上取决于模型质量。搜索预算通过限制可进行的迭代次数来直接控制这种延迟。

    不过,基于提示的约束较为宽松。模型可能会误判任务的复杂性,或者直接忽略指令。当前系统已有tool_call_iterations字段,因此很容易设置一个由路由器强制执行的硬性上限。一种更为稳健的方案是同时使用这两种方法:提示词用于规范正常行为,而代码则确保存在上限。我们在关于LLM工具使用中的受限智能体循环的TypeScript实现一文中也探讨了同样的理念。

    带监督者的分散收集机制

    第三种扩展方式将整个ReAct循环视为一个可重用的工作节点。监督代理会将一个复杂问题拆分成多个子问题,然后为每个子问题并行启动一个独立的研究者子图。

    Supervisor receives: "Compare the economic impact of AI on healthcare vs. education"
    
    Supervisor creates two parallel research tasks:
    ├── ReAct Agent 1: Research AI impact on healthcare
    └── ReAct Agent 2: Research AI impact on education
    
    Both agents run their ReAct loops independently.
    Results are gathered and synthesized by the supervisor.
    

    这就是分散处理模式:将任务分配给独立的工作者,再收集并合并他们的结果。每个研究者都有自己独立的状态、工具和资源限制,因此某个子问题的长时间搜索历史不会影响其他子问题的处理过程。监督代理仅能看到经过压缩后的输出结果。

    监督代理本身也是一个小型图结构。它并非通过固定的for循环来处理各个子问题,而是利用LangGraph的Command返回类型,让节点能够在一步之内更新状态并指定下一个处理节点:

    Supervisor sub-graph nodes:
    ├── supervisor          (LLM decides what to do next)
    ├── supervisor_tools    (executes supervisor-level tools like ConductResearch)
    ├── red_team            (attacks draft logic to find flaws)
    └── context_pruner      (clears raw notes to manage context size)
    
    The supervisor_tools node uses Command to route dynamically:
      - If research is needed → spawns researcher sub-graphs via ConductResearch tool
      - If critique is needed → routes to red_team node
      - If context is bloated → routes to context_pruner node
      - If research is complete → routes to END
    

    对监督者而言,研究者就像一个黑箱。它调用ConductResearch工具,该工具再启动已编译好的研究者子图,由后者独立执行完整的ReAct循环。在其周围还有其他专用节点:一个red_team节点用于攻击草案的推理过程以找出弱点,还有一个context_pruner节点在上下文规模过大时清除原始笔记。

    使用Command而非静态边为监督者提供了运行时的灵活性。根据当前状态,它可以在每一步之后选择启动更多研究者、发送草案以供评审、修剪上下文或结束流程。不过这样的代价是路由逻辑被嵌入到节点代码中,因此仅从边定义无法清晰看出图的结构;此时良好的日志记录与追踪功能就显得更为重要。

    追踪完整运行过程

    要查看各组件如何协同工作,可以给智能体一个中等复杂度的问题。源代码将此调用标记为纯文本;实际上它是 Python 代码。

    # Run the agent with a research question
    result = agent.invoke({
        "researcher_messages": [
            HumanMessage(content="What are the main approaches to reducing hallucination in RAG systems?")
        ]
    })
    

    如果自行运行此代码,需要做两处小修改。之前编译的图模型名为 researcher_agent,请使用该名称(或上一节中保存的 agent 模型)。若使用已保存的版本,则需按照上述说明在配置中传入 thread_id

    示例执行轨迹如下所示,它属于日志文件而非 Python 代码。

    --- Iteration 1 ---
    [Brain] Reasoning: I need to search for approaches to reducing RAG hallucination.
    [Brain] Tool call: search_tool(query="reducing hallucination in RAG systems approaches")
    [Hands] Executing search_tool...
    [Hands] Results: Found 5 relevant articles about RAG hallucination reduction.
    [Router] Last message has tool_calls? No (think_tool was called)
    
    --- Iteration 2 ---
    [Brain] Tool call: think_tool(reflection="The search results mention three main
    approaches: better chunking strategies, re-ranking retrieved documents, and
    adding citation verification. I should search for specific implementations.")
    [Hands] Executing think_tool...
    [Router] tool_calls present -> route to tool_node
    
    --- Iteration 3 ---
    [Brain] Tool call: search_tool(query="citation verification RAG pipeline implementation")
    [Hands] Executing search_tool...
    [Router] tool_calls present -> route to tool_node
    
    --- Iteration 4 ---
    [Brain] Tool call: think_tool(reflection="I now have solid coverage of the three
    main approaches with implementation details. I have enough information to
    provide a comprehensive answer.")
    [Hands] Executing think_tool...
    [Router] tool_calls present -> route to tool_node
    
    --- Iteration 5 ---
    [Brain] No tool calls. Generating final response.
    [Router] No tool_calls -> route to compress_research
    [Compress] Summarizing all findings into structured output.
    

    轨迹显示的内容:

    1. 智能体进行了两次搜索和两次反思调用,均在提示词规定的范围内。
    2. 每次反思都会总结已学到的内容并准备下一次搜索,这正是强制反思规则所期望产生的效果。
  • 当代理判断已获取足够信息后,它会自动停止:先进行四轮工具调用,随后给出最终文本回复。
  • 每当存在工具调用时,路由器会将执行任务发送给tool_node;而当没有工具调用时,则将其发送给compress_research
  • 最终输出的是压缩后的摘要,而非原始对话内容。
  • 应将此跟踪记录视为概要,而非字面意义上的程序输出。其中将搜索工具命名为search_tool,但实际上使用的工具是tavily_search;此外,在第一次迭代中路由器显示没有进行任何工具调用,尽管实际上已提出了搜索请求,按理说应该将任务路由到tool_node。需要记住的是整体流程:交替进行搜索与思考,直到模型以纯文本形式给出答案。

    将此循环应用于实际应用

    研究子图是坚实的基础,但通过五项升级后,它在实际应用中的可靠性大幅提升:

    1. 在代码中强制执行预算限制。 每次迭代时增加 tool_call_iterations 的值,一旦达到上限,无论模型有何要求,都让 should_continue 转向 compress_research。这便是对提示词软限制的保障机制。
    2. 向用户实时反馈进度。 LangGraph 的 .astream_events() 可以为节点转换和工具调用发出事件,因此界面可以在循环运行时显示“正在搜索…”或“正在分析结果…”之类的消息,而无需使用旋转加载图标。
  • 从工具错误中恢复。目前,若出现网络超时或工具内部的速率限制错误,会导致任务运行失败。应将每次调用封装在try/except结构中,并返回描述该错误的ToolMessage。这样,系统就能识别到失败情况,并可以尝试使用不同的参数重新执行或改变处理方式。Anthropic的文档建议以这种方式将工具错误反馈给模型。
  • 保持长时间研究任务的连续性。对于需要多次迭代的任务,可使用PostgresSaver,这样即使任务被中断,也能从停止处继续执行,且所有的推理步骤和工具调用记录都可供查看。
  • 添加人工干预的断点。LangGraph可以在中断点暂停并等待批准。对于高风险研究,每隔N次迭代就暂停一次,向人工展示已发现的结果,再由他们决定是让智能体继续、重新定向还是停止。
  • 核心要点

    • ReAct是一个包含思考、行动和观察的循环;它与具体框架无关,而LangGraph只是将这一循环明确化并使其可被查看。
    • 三个专用节点(大脑、手、路由器)再加上由归约器支持的状态,就足以构建出一个可运行的研究智能体。
    • 始终要将每个工具的运行结果与其tool_call_id关联起来,并将内部状态与子图公开的内容分开。
    • 一个无实际功能的反射工具是一种成本低廉且易于观察的方式,可用于强制在不同搜索结果之间进行综合。
    • 提示词预算可以影响智能体的行为,但只有代码层面的限制才能确保其最终终止。
  • 一旦循环被编译为子图,监督者就可以将其并行展开处理,将每位研究人员视为一个黑盒。
  • 相关阅读

  • 路由、扇出、ReAct、评估与批准:五种LangGraph工作流模式 — 了解LangGraph中的五种智能工作流模式,涵盖路由器、ReAct循环、评估器以及人工审批等功能,并阐述每种模式在实际应用中所需的约束条件。
  • 深入了解LangGraph的InMemorySaver:检查点、写入数据与二进制数据的存储方式 — 详解LangGraph的InMemorySaver中的存储结构、写入数据及二进制数据管理机制,追踪单个小型图计算如何转化为三个相互关联的检查点。
  • 专业代理、关键词路由器与中断机制:一款LangGraph辅导工具 — 设计一个拥有确定性路由器、可在任务交接间保持状态同步的两专业代理型LangGraph助手,以及基于中断机制和命令功能构建的人工检查点系统。
  • 从单节点聊天机器人到LangGraph中的MCP支持型代理 — 逐步构建LangGraph应用:包括状态与还原器、边连接、工具循环、带检查点的线程、三种流式处理模式,以及通过MCP提供的工具功能。