作为LangChain智能体的Agent采用一种被称为ReAct循环的执行流程（如下图所示），这是一种结合了“推理”（Reasoning）与“行动”（Acting）的交互模式，旨在让Agent能像人类一样通过逻辑思考来解决复杂问题。但是从底层的执行方式来看，Agent本质上是一个Pregel对象(我的系列文章“拆解LangChain执行引擎”对Pregel具有系统的介绍)，是一个通过节点和通道构建的Actor模型，那么两者是如何结合在一起的？我希望通过这个系列文章回答这个问题。除此之外，这个序列还将深入介绍Agent状态读写、结构化输出、工具和中间的设计和实现。
1. 从create_agent工厂函数说起
LangChain提供了不同的编程模式创建Agent，其中最简单的莫过于直接使用create_agent函数，所以我们不妨先来看看该函数的定义。如下面的代码片段所示，这个工厂函数创建的Agent体现为一个CompiledStateGraph对象，它本质上是一个Pregel对象，所以此函数提供的很多参数旨在初始化这个Pregel对象。
def create_agent(
model: str | BaseChatModel,
tools: Sequence[BaseTool | Callable[..., Any] | dict[str, Any]] | None = None,
*,
system_prompt: str | SystemMessage | None = None,
middleware: Sequence[AgentMiddleware[StateT_co, ContextT]] = (),
response_format: ResponseFormat[ResponseT] | type[ResponseT] | dict[str, Any] | None = None,
state_schema: type[AgentState[ResponseT]] | None = None,
context_schema: type[ContextT] | None = None,
checkpointer: Checkpointer | None = None,
store: BaseStore | None = None,
interrupt_before: list[str] | None = None,
interrupt_after: list[str] | None = None,
debug: bool = False,
name: str | None = None,
cache: BaseCache[Any] | None = None,
) –> CompiledStateGraph[AgentState[ResponseT], ContextT, _InputAgentState, _OutputAgentState[ResponseT]]
我们有必要逐个介绍一下每个参数：
- model：智能体的“大脑”。可以传模型名称字符串（需环境配置）或已实例化的 BaseChatModel（如 ChatOpenAI）。
- tools：赋予Agent的“工具箱”。支持 BaseTool对象、普通Python函数（会自动转为工具）或描述工具JSON Schema的字典。
- system_prompt: 定义Agent的角色、行为准则和背景知识。支持纯字符串或预定义的SystemMessage对象。
- response_format: 强制结构化输出。可以传入Pydantic模型、JSON Schema 或 ResponseFormat对象，确保Agent返回的数据符合特定格式（常用于数据提取或接口调用）。
- state_schema: 定义Agent的内部状态结构（继承自 AgentState）。它决定了Agent运行过程中能“记住”哪些信息（如历史对话、工具执行中间结果）。
- context_schema: 定义外部上下文结构。与状态不同，上下文通常是运行时传入的只读配置或全局信息（如 user_id, session_id）。
- middleware: 插件系统。允许在 Agent 运行前后插入自定义逻辑（如日志记录、输入过滤、耗时统计）。
- checkpointer: 持久化层。用于保存图的状态快照，实现断点续传、多轮对话记忆以及“时间旅行”（回溯到某一步骤）。
- store: 长效存储。用于在不同线程或用户之间共享信息（如存储用户的长期偏好，不随单次对话结束而消失）。
- cache: 缓存机制。减少模型重复调用的开销，提高响应速度并节省 Token。
- interrupt_before/interrupt_after: 人机交互点。在进入或退出指定节点前挂起执行。这通常用于“人在回路（Human-in-the-loop）”场景，需用户审批后再继续。
- debug: 开启后会打印每一轮迭代的详细状态变化和节点执行信息。
- name: 为生成的 Agent 图命名，方便在多 Agent 协同系统（Multi-agent system）中进行追踪和引用。
一个Agent由最为核心的Model和Tool组成，所以这三个对象交互的视角来看待Agent基于ReAct循环的执行流程是最直接的。
2. 以Agent、Mode和Tool三者交互视角看待ReAct循环
ReAct是一个循环往复的过程，直到模型得出最终答案或达到最大尝试次数。其核心逻辑可拆解为以下四个步骤：
- Thought（推理）：LLM 接收到用户问题及当前上下文，分析当前所处的阶段，并决定下一步需要采取什么行动。
- Action（行动）：模型根据推理结果，发出指令调用特定的工具。这包括从工具库中提取参数并向外部环境发送请求。
- Observation（观察）：LangChain 接收工具返回的真实数据（如网页搜索结果、数据库查询值或代码运行输出），并将其反馈给模型。
- Loop（循环）：模型根据新的“观察”结果再次进行“推理”。如果信息足够，则输出最终答案；如果信息仍不足，则继续开启下一轮 Thought -> Action -> Observation。
我们可以从一个简单的实例来体验一下LangChain的ReAct循环。这是一个经典的利用Agent获取天气信息的例子，如代码片段所示，我们定义了一个get_weather函数作为Agent的工具，该函数根据指定的城市（通过city参数表示）得到以文本表示的天气信息。函数的docstring给出了基本的描述，Agent将会提取此信息作为工具描述。我们创建一个Agent，并初始化了通过model、tools和system_prompt字段表示的模型、注册的工具列表和系统提示符。
from langchain.agents import create_agent
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
load_dotenv()
def get_weather(city: str) –> str:
"""Get weather for a given city."""
return f"It's always sunny in {city}!"
agent = create_agent(
model= ChatOpenAI(model="gpt-5.2-chat"),
tools=[get_weather],
system_prompt="You are a helpful assistant",
)
message = HumanMessage(content="What is the weather in Suzhou?")
for step in agent.stream(
input= {"messages": [message]},
stream_mode="values"
):
step["messages"][–1].pretty_print()
具体的模型是基于“gpt-5.2-chat”创建的ChatOpenAI，相关的设置（比如api-key和base-url等）定义在.env文件通过dotenv.load_dotenv写入环境变量。我们直接将函数get_weather作为工具添加到注册的列表中。通过create_agent函数创建的Agent默认以表示状态的AgentState作为输入，这是一个TypedDict，其核心成员“messages”用于存储整个Agent执行过程中生成的消息。所以我们创建了这样的字典作为调用Agent的输入，并在其messages列表中添加了一个用于查询天气的HumanMessage对象。
我们调用Agent的stream方法，并将流模式设置为“values”，执行流程中的每个步骤（对于Pregel的Superstep）结束之后会以流的形式输出所有的状态，这里我们将每个步骤维护的消息列表中的最后一条消息的内容打印出来。
================================ Human Message =================================
What is the weather in Suzhou?
================================== Ai Message ==================================
Tool Calls:
get_weather (call_aCcssiCPOFyJa6dfu2dSltqD)
Call ID: call_aCcssiCPOFyJa6dfu2dSltqD
Args:
city: Suzhou
================================= Tool Message =================================
Name: get_weather
It's always sunny in Suzhou!
================================== Ai Message ==================================
Right now, the weather in **Suzhou** is reported as **sunny** ☀️.
If you’d like, I can also help with things like the temperature, forecast for the next few days, or weather in a specific Suzhou (there’s one in Jiangsu, China, and another in Anhui).
这个例子涉及的ReAct循环体现在如上的输出中：
- 首先我们以HumanMessage的形式向模型提供查询，模型经过推理确定这是“天气查询”，并且自身绑定的工具列表具有一个与之匹配的工具get_weather，所以模型认为完成当前任务需要先调用get_weather工具。模型不会自行调用工具（服务端工具除外），但是它需要将工具调用意图封装成一个TooCall对象，该对象承载的信息包括：包括工具的名称（get_weather）、标识工具调用的ID（call_aCcssiCPOFyJa6dfu2dSltqD），参数列表（city: Suzhou）。这样一个ToolCall对象被封装成AIMessage返回给Agent。
- Agent接收到AIMessage之后，从中提取出ToolCall提供的参数对目标工具发起调用，并将结果（“It’s always sunny in Suzhou!”）封装成ToolMessage再次发送给模型。
- 模型经过推理确定能够对查询做出最终的答复了，于是组织文本并组织文本并以AIMessage的形式予以回复。
message = HumanMessage(content="What is the weather in Suzhou?")
result = agent.invoke(input= {"messages": [message]})
for message in result["messages"]:
message.pretty_print()
如果不使用流式输出，我们也可用按照如上方式调用Agent的invoke方法，并从作为结果的字典中利用“messages”作为Key得到整个“交谈”过程中生成的消息。上面这段程序也可以得到类似的输出:
================================ Human Message =================================
What is the weather in Suzhou?
================================== Ai Message ==================================
Tool Calls:
get_weather (call_CIUcZcYEGEbI0IJlDs19RhTd)
Call ID: call_CIUcZcYEGEbI0IJlDs19RhTd
Args:
city: Suzhou
================================= Tool Message =================================
Name: get_weather
It's always sunny in Suzhou!
================================== Ai Message ==================================
The weather in Suzhou is sunny ☀️
3. 以Pregel的视角看到Agent和ReAct循环
既然Agent是一个Pregel对象，这是一个由节点和通道构建的Actor模型，我们不免会好奇：它由哪些节点和通道构成呢？每个通道采用何种类型？通道和节点之间的订阅和读写关系是什么样子？我们为此编写了如下的演示程序，将上面创建的Agent作为一个Pregel对象，将其所有的通道（及其类型）、节点（包括名称、输入和驱动Channel）以及Pregel对象自身的输入/输出通道，以及驱动通道与对应节点的映射关系打印出来。
from langchain.agents import create_agent
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from PIL import Image as PILImage
import io
load_dotenv()
def get_weather(city: str) –> str:
"""Get weather for a given city."""
return f"It's always sunny in {city}!"
agent = create_agent(
model= ChatOpenAI(model="gpt-5.2-chat"),
tools=[get_weather],
system_prompt="You are a helpful assistant",
)
print("channels:")
for (name, chan) in agent.channels.items():
print(f"\\t[{chan.__class__.__name__}]{name}")
print("nodes:")
for (name,node) in agent.nodes.items():
print(f"\\tname: {name}")
print(f"\\tchannels: {node.channels}")
print(f"\\ttriggers: {node.triggers}\\n")
print(f"input_channels:{agent.input_channels}")
print(f"output_channels：{agent.output_channels}")
print("trigger_to_nodes")
for (name, nodes) in agent.trigger_to_nodes.items():
print(f"\\t{name}: {nodes}")
payload = agent.get_graph(xray=True).draw_mermaid_png()
PILImage.open(io.BytesIO( payload)).show()
从如下的输出可以看出，通过create_agent函数创建的Agent有且只有三个节点（前提是没有注册中间件），分别是作为启动节点的节点“start”、封装模型和工具的节点model和节点tools。也就是说，注册的所有工具都封装在同一个节点tools中并由它管理并选择执行。
channels:
[BinaryOperatorAggregate]messages
[EphemeralValue]jump_to
[LastValue]structured_response
[EphemeralValue]__start__
[Topic]__pregel_tasks
[EphemeralValue]branch:to:model
[EphemeralValue]branch:to:tools
nodes:
name: __start__
channels: __start__
triggers: ['__start__']
name: model
channels: ['messages', 'jump_to', 'structured_response']
triggers: ['branch:to:model']
name: tools
channels: ['messages', 'jump_to', 'structured_response']
triggers: ['branch:to:tools']
input_channels:__start__
output_channels：['messages', 'structured_response']
trigger_to_nodes
__start__: ['__start__']
branch:to:model: ['model']
branch:to:tools: ['tools']
再来看看Agent的通道列表。三个节点具有对应的触发通道，它们的名称分别是“start”、“branch:to:model”和“branch:to:tools”，对应的类型均为EphemeralValue。针对Agent的调用实际上就是通过写入“start”驱动Agent从“start”节点开始执行。model节点完成模型调用后如果需要调用工具，就写入“branch:to:tools”；工具调用之后通过写入“branch:to:model”将工具执行的结果反馈给模型并驱动模型节点执行。
model和tools节点具有相同的输入通道，分别是承载消息列表的messages、用于跳转的jump_to和表示结构化输出的structured_response。messages通道的类型为BinaryOperatorAggregate，其operator用于执行列表添加操作，这意味着针对此通道的更新总是在做“消息追加”操作，这样会将所有的消息历史保存下来，上面演示的实例也证明了这一点。
工具或者中间件可以通过写入jump_to通道实现“跳转”。该通道支持三个选项“model”、“tools”和“end”，前两个选项实现针对model和tools节点的跳转，“end”用于终止整个处理流程。结构化输出最终写入通道structured_response。演示程序的最后两行代码会将Agent的结构以可视化的形式展现出来，具体呈现的结构如下所示。
如果我们站在Pregel的视角审视ReAct循环将会更加接近真实的执行流程：
- 当我们调用Agent对象时，指定的HumanMessage会写入messages通道中;
- 作为入口节点，model节点将其读取出来之后与绑定的系统消息（通过system_prompt参数指定的系统提示词创建）和工具列表一并发送给语言模型。对于我们的例子来说，就是调用OpenAI API，具体的请求如下所示。
{
"messages": [
{
"content": "You are a helpful assistant",
"role": "system"
},
{
"content": "What is the weather in Suzhou?",
"role": "user"
}
],
"model": "gpt-5.2-chat",
"stream": false,
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather for a given city.",
"parameters": {
"properties": {
"city": {
"type": "string"
}
},
"required": [
"city"
],
"type": "object"
}
}
}
]
}
- LLM经过推理，决定需要调用工具，并将针对工具调用的描述置于响应中。从如下所示的响应内容可以看出，针对工具get_weather的调用被置于响应的"choices">"tool_calls"节点(包括调用工具的名称，标识工具调用的ID和参数列表）。model节点将接收到的响应内容封装成AIMessage，并写入messages通道。通过分析内容确定下一步需要调用工具，于是通过写入“branch:to:tools”通道驱动执行tools节点。
{
"choices": [
{
"content_filter_results": {},
"finish_reason": "tool_calls",
"index": 0,
"logprobs": null,
"message": {
"annotations": [],
"content": null,
"refusal": null,
"role": "assistant",
"tool_calls": [
{
"function": {
"arguments": "{\\"city\\":\\"Suzhou\\"}",
"name": "get_weather"
},
"id": "call_cK9LgQ76xAalnjxUH1q945CM",
"type": "function"
}
]
}
}
],
"created": 1772717196,
"id": "chatcmpl-DG30m9xH7c6ujOxbuMIJ2cid9mU30",
"model": "gpt-5.2-chat-2025-12-11",
"object": "chat.completion",
"prompt_filter_results": […],
"system_fingerprint": null,
"usage": {…}
}
- 当tools节点开始执行的时候，它从messages通道中读取最后一条AIMessage，在将工具调用部分的内容读取出来后，对涉及的工具实施调用（并发调用）。对于每个成功完成的工具调用，tools节点都会根据指定的结果创建一个ToolMessage并写入messages通道。待所有工具调用完成之后，它通过写入“branch:to:model”通道驱动执行model节点。
- model节点和之前一样，从messages通道提取写入的三条消息，并附加上一条系统消息和工具列表，生成如下的请求第二次对作为LLM的OpenAI API发起调用。
{
"messages": [
{
"content": "You are a helpful assistant",
"role": "system"
},
{
"content": "What is the weather in Suzhou?",
"role": "user"
},
{
"content": null,
"role": "assistant",
"tool_calls": [
{
"type": "function",
"id": "call_cK9LgQ76xAalnjxUH1q945CM",
"function": {
"name": "get_weather",
"arguments": "{\\"city\\": \\"Suzhou\\"}"
}
}
]
},
{
"content": "It's always sunny in Suzhou!",
"role": "tool",
"tool_call_id": "call_cK9LgQ76xAalnjxUH1q945CM"
}
],
"model": "gpt-5.2-chat",
"stream": false,
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather for a given city.",
"parameters": {
"properties": {
"city": {
"type": "string"
}
},
"required": [
"city"
],
"type": "object"
}
}
}
]
}
- 经过LLM的分析，此时能够对用于的查询做出答复，于是它充分利用现有的语料组织文本对请求予以响应。model节点依然将响应内容封装成AIMessage写入messages通道。由于messages作为Agent的两个输出通道之一，所以我们能从调用结果中的messages成员提取完整的消息历史。
{
"choices": [
{
"content_filter_results": {…},
"finish_reason": "stop",
"index": 0,
"logprobs": null,
"message": {
"annotations": [],
"content": "The weather in **Suzhou** is currently **sunny** ☀️.\\n\\nIf you’d like more details—such as temperature, humidity, orr a multi-day forecast—just let me know!",
"refusal": null,
"role": "assistant"
}
}
],
"created": 1772719213,
"id": "chatcmpl-DG3XJbOOkM3usPXR56MEVrlRrthMO",
"model": "gpt-5.2-chat-2025-12-11",
"object": "chat.completion",
"prompt_filter_results": […],
"system_fingerprint": null,
"usage": {…}
}
