LangGraph中的ReAct：从手动循环到ToolNode与StateGraph的实现

一个基本的工具调用循环只需一次遍历：模型请求使用工具，获取结果后输出答案。而对于那些需要链式调用的任务，即第二次调用依赖于第一次调用的结果时，就需要一个能够思考、行动、查看结果后再再次行动的智能体。这种循环结构就是ReAct模式。下面展示了其概念、手写的ReAct循环，以及使用LangGraph重新构建的相同智能体，其中包含了工具节点、工具消息、条件边和状态图。

LangGraph为此提供的组件包括：

• 工具节点：解析工具调用，执行工具操作并返回结果。
• 工具消息：将工具的输出传回大语言模型。
• tools_condition：一种路由函数，用于检测工具调用并触发条件边。
• StateGraph：负责运行整个工作流程。

状态：用于存储代理动态数据的键值对字典。

节点：用于更新状态的 Python 函数，工具节点即为此类函数之一。

边：节点之间的转换关系，条件边可实现分支和循环结构。

设置

将敏感信息保存在项目根目录下的 .env 文件中：

OPENAI_API_KEY="<Your OpenAI API Key>"
TAVILY_API_KEY=<Your TAVILY API Key>

通过 Pydantic 的 BaseSettings 类来加载它们：

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

model_config = SettingsConfigDict(env_file=".env", extra="ignore")

项目结构：

project/
├── react_loop/
│   └── weather_assistant.py # implement a vanilla react loop
│
├── react_langgraph     # implement ReAct by LangGraph workflow
│   ├── nodes.py        # graph nodes. Defined as Python functions
│   ├── graph.py        # graph workflow
│   ├── consts.py       # define node names as constants
│   └── state.py        # typed dictionary to store graph states
|
├── tools
│   ├── tool_call.py  # implement simple tool-calling loop
|   └── weather.py    # get_weather LangChain tool using open source APIs
|
|──    # implement tool-calling loop using LangChain tool
├── config.py         # pydantic BaseConfig
├── utils.py          # utility method to print messages
├── .env              # environment variable definition
├── .gitignore
└── requirements.txt  # package requirements

ReAct的含义

ReAct代表“推理加行动”。执行多步骤任务的智能体会不断进行推理、采取行动、观察情况，然后再重复推理，直至任务完成。传统流程仅允许执行一轮工具调用，而这正是ReAct所要弥补的缺陷。

一次运行包含六个步骤：

1. 用户发送消息。
2. 大语言模型读取请求，确定所需信息，并决定是否以及调用哪些工具。
3. 智能体运行一个或多个工具，例如搜索网络、查询API或发送邮件。
4. 工具返回结果并更新智能体的当前状态。
5. 大语言模型评估新获取的信息，要么给出答案，要么要求再次调用工具。
6. 当无需进一步操作时，最终答案会被输出。

该循环之所以强大，是因为智能体能够根据刚刚学到的信息调整自己的计划，因此每个决策都可以依赖于之前的结果。

简单的 ReAct 循环

这个版本位于 react_loop/weather_assistant.py 中，与单次遍历循环类似，但有两点不同：

async def chat(self, message: str):
    # Add user message
    self.messages.append(HumanMessage(content=message))

    # loop to handle serial tool calls
    while True:

        # Get AI response with potential tool calls
        response = await self.llm_with_tools.ainvoke(self.messages)
        self.messages.append(response)

        # Check if there is any tools to call
        if not response.tool_calls:
            # no tools to call
            break
            # Execute each tool call

        # process tool calls asynchronously
        results = await asyncio.gather(*[self.tools[tc["name"]].ainvoke(tc) for tc in response.tool_calls])
        self.messages.extend(results)

首先，工具调用通过 asyncio.gather 异步执行，因此多个调用可以并行运行。其次，工具执行步骤位于 while 循环中，不断循环进行推理与操作，直到模型的响应中不再包含工具调用。

LangGraph 中的同一智能体

手动循环虽然可行，但需要用户自行管理推理-操作周期、消息列表、工具字典以及结果收集。LangGraph 则负责处理这些环节：ToolNode 负责解析调用并执行工具，StateGraph 用于在图结构中存储消息，而 tools_condition 边则替代了手动编写的 while 循环及其工具调用检查逻辑。

状态。 一个带类型的字典用于存储消息列表（位于 react_langgraph/state.py 中）：

from typing import TypedDict, Annotated
from langgraph.graph.message import AnyMessage, add_messages

class ReactState(TypedDict):
    """
    Define the state schema for our hotel assistant.

    Attributes:
        messages: list of messages
    """

    messages: Annotated[list[AnyMessage], add_messages]

节点通过它读取和更新消息，而 add_messages reducer 会追加节点返回的任何内容。

节点。 在 react_langgraph/nodes.py 中，初始化过程会创建一个 LangChain 工具列表、一段告知模型如何使用这些工具的系统提示，以及一个与该工具列表绑定的 OpenAI 模型：

tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY), get_weather]

SYSTEM_PROMPT = f"""You are an assistant for question-answering tasks with access to multiple tools.
    use these tools to retrieve 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.

    Tool Use Instructions:

    - Use tools thoughtfully—only when they add value to the response
    - Chain tools when needed (e.g., search for info, then use that to search
      for relevant hotels)
      - Be conversational and helpful
    """

system_message = SystemMessage(content=SYSTEM_PROMPT)

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)


async def run_agent_reasoning(state: ReactState) -> dict:
    """
    run agent reasoning node
    :param state: ReactState, manage graph state
    :return: dict containing new message
    """

    print("----AGENT REASONING------")
    messages = [system_message] + state["messages"]

    if isinstance(messages[-1], ToolMessage):
        print_message(messages[-1])

    response = await llm_with_tools.ainvoke(messages)
    print_message(response)

    return {"messages": [response]}


tool_node = ToolNode(tools)

最终会生成两个节点。推理节点负责构建消息列表，调用大语言模型并返回其回复；而动作节点则是一个ToolNode，它负责解析工具调用、执行相应工具并返回工具生成的消息。

图结构。在react_langgraph/graph.py中，StateGraph负责注册这些节点，并通过边将它们相互连接：

workflow = StateGraph(ReactState)

workflow.add_node(AGENT_REASON, run_agent_reasoning)
workflow.add_node(ACT, tool_node)

workflow.add_edge(START, AGENT_REASON)
workflow.add_edge(ACT, AGENT_REASON)

workflow.add_conditional_edges(
    AGENT_REASON,
    tools_condition,
    {"tools": ACT, END: END}
)tion is invoked first

执行过程遵循四条规则。流程图从推理节点开始。如果大语言模型的回复中包含工具调用，tools_condition会将控制流转至执行节点。工具运行完成后，控制权总会返回到推理阶段。如果回复中没有工具调用，流程图即结束并返回最终答案。这实际上就是ReAct循环，只不过无需手动循环或管理消息。

编译并运行。编译过程会生成智能体，同时还可以将流程图导出为PNG格式的图片：

react_agent = workflow.compile()

react_agent.get_graph().draw_mermaid_png(output_file_path="graphs.png")

question = """
           tell me the first hotel you find in Boston if
            its temperature is lower than 30 degree."""


async def main():
    print("hello ReAct agent by Langraph")
    message = HumanMessage(content=question)
    res = await react_agent.ainvoke({"messages": [message]})


if __name__ == "__main__":
    asyncio.run(main())

通过用户消息调用它，生成的消息列表会显示推理节点与工具节点的交替使用情况，从而构成一个可运行的多步骤流程。

当前进展

有三个LangGraph功能负责实现这一功能：ToolNode用于查找并运行工具，tools_condition用于循环控制，而add_messages则负责自动处理消息。在理解并构建了ReAct之后，下一步就是将其重新应用到LangGraph RAG流程中，使其具备完全的智能代理能力。