走读 LangChain：Agent Loop 如何连接模型与工具

上一篇文章《走读 DeepSeek Harness：Agent Loop 如何驱动一次完整执行》分析了一种显式 Agent Loop：ReactLoopAgent 用 while 推进 Turn 和 Step，Session 事件日志负责保存执行事实。

继续阅读 Deep Agents 时，却很难找到类似的主循环。create_deep_agent() 做完模型、工具和状态配置后，直接调用 LangChain 的 create_agent()，返回一个 LangGraph CompiledStateGraph。这意味着 Deep Agents 没有重新实现 Agent Loop，真正的循环位于 LangChain。

本文以项目锁定的 LangChain 1.3.14 为准，沿着 langchain/agents/factory.py 回答几个问题：

1. create_agent() 如何把 Agent Loop 编译成图？
2. model 节点如何产生下一步决策？
3. 两条条件边如何在 model、tools 和 END 之间路由？
4. 多个工具调用如何并发执行，结果又如何回到下一次模型请求？

一、Agent Loop 不是 while，而是一张状态图

create_agent() 的 docstring 已经给出了核心语义：模型收到消息并返回 AIMessage；如果消息中包含 tool_calls，图就执行工具，将结果作为 ToolMessage 写回消息列表，然后再次调用模型，直到响应不再包含工具调用。

忽略可选扩展节点后，图可以简化为：

START
  -> model
     -> no tool calls: END
     -> pending tool calls: tools
          -> direct result: END
          -> normal result: model

对应的伪代码仍然是熟悉的模型—工具循环：

while True:
    ai_message = call_model(messages)
    messages.append(ai_message)

    if not ai_message.tool_calls:
        break

    tool_messages = execute_tools(ai_message.tool_calls)
    messages.extend(tool_messages)

但源码并没有执行这段 while。它先创建 StateGraph，添加 model 和 tools 节点，再用 add_conditional_edges() 描述循环：

graph.add_node("model", model_node)
graph.add_node("tools", tool_node)

graph.add_conditional_edges("model", model_to_tools)
graph.add_conditional_edges("tools", tools_to_model)

最后编译并返回 CompiledStateGraph。因此，调用方执行的是：

agent.invoke(...)
agent.stream(...)

循环的控制权在 LangGraph，而不是一个长期存活的 AgentLoop 对象中。

二、model 节点如何完成一次 Step

每次进入 model 节点，LangChain 都会从当前 state 构造 ModelRequest：

request = ModelRequest(
    model=model,
    tools=default_tools,
    system_message=system_message,
    response_format=initial_response_format,
    messages=state["messages"],
    state=state,
    runtime=runtime,
)

这里最重要的是 messages 和 tools：前者携带前面所有决策与观察，后者告诉模型本轮可以采取哪些行动。

真正调用模型前，_get_bound_model() 会根据当前请求绑定工具：

model_with_tools = request.model.bind_tools(final_tools)

随后 _execute_model_sync() 把 system message 放在历史消息前面并调用模型：

messages = request.messages
if request.system_message:
    messages = [request.system_message, *messages]

output = model_with_tools.invoke(messages)

模型可能直接给出最终回答：

AIMessage(content="The default port is 8080")

也可能请求工具：

AIMessage(
    content="",
    tool_calls=[
        {
            "id": "call_1",
            "name": "read_file",
            "args": {"path": "config.py"},
        }
    ],
)

model 节点只负责产生并记录决策，不负责执行工具。响应写入 state 后，条件边才决定本次 Step 是结束整个 run，还是转入 tools。

三、model_to_tools 如何决定下一跳

模型侧的路由函数是 _make_model_to_tools_edge()。它首先从 state["messages"] 中向前查找最近一条 AIMessage，并收集它之后出现的 ToolMessage。

核心判断可以压缩为：

last_ai_message, tool_messages = fetch_messages(state["messages"])

if last_ai_message is None:
    return END

if not last_ai_message.tool_calls:
    return END

pending_tool_calls = [
    call
    for call in last_ai_message.tool_calls
    if call["id"] not in completed_tool_call_ids
]

if pending_tool_calls:
    return [Send("tools", [call]) for call in pending_tool_calls]

这里检查的是 pending tool calls，而不是简单判断 tool_calls 是否为空。路由函数会用下面两个字段匹配调用和结果：

AIMessage.tool_calls[].id
ToolMessage.tool_call_id

例如 state 中已经有：

AIMessage(tool_calls=[call_1, call_2])
ToolMessage(tool_call_id=call_1)

那么 call_1 已经完成，只有 call_2 会被送入 tools。这让 Agent 从 checkpoint 恢复时能够识别已有结果，避免把已经执行过的副作用工具再运行一次。

当模型没有产生工具调用时，条件边直接走向 END。这就是 Agent Loop 最常见的终止条件。

四、ToolNode 如何执行一批工具调用

create_agent() 会把普通 Python callable 和 BaseTool 收集起来，构造 LangGraph 的 ToolNode：

tool_node = ToolNode(tools=available_tools)

模型一次响应可以包含多个工具调用。model_to_tools 不会把整批调用作为一个串行任务交给 ToolNode，而是为每个待执行调用返回一个 Send：

return [
    Send("tools", [tool_call])
    for tool_call in pending_tool_calls
]

假设模型同时请求：

AIMessage
  -> read_file("a.py")
  -> read_file("b.py")
  -> grep("create_agent")

图会产生三个发送任务：

Send(tools, read_file_a)
Send(tools, read_file_b)
Send(tools, grep_create_agent)

LangGraph 可以并行调度它们。每个工具完成后产生与调用 ID 对应的 ToolMessage，结果随后通过 state reducer 合并进消息历史。

这里的并发边界很明确：同一条 AIMessage 中的多个 tool call 可以被扇出；下一次模型调用要等这些结果进入 state 后，才能基于完整观察继续决策。

五、tools_to_model 为什么通常返回模型

工具执行完成后，第二条条件边 _make_tools_to_model_edge() 决定是退出还是继续：

if all_executed_tools_return_direct:
    return END

if structured_output_tool_executed:
    return END

return model

普通工具默认回到模型，因为工具结果只是 Observation，还不是面向用户的最终答案：

tools
  -> append ToolMessage
  -> model reads ToolMessage
  -> model chooses next action

如果工具设置了 return_direct=True，工具结果本身就可以成为 Agent 输出，不需要再付出一次模型调用成本。结构化输出工具成功执行后也可以直接结束，因为目标结果已经写入 structured_response。

两条条件边合在一起，才构成完整循环：

model_to_tools
  -> END or tools

tools_to_model
  -> END or model

前者解释“模型是否还要行动”，后者解释“工具结果是否还需要模型处理”。

六、messages 如何连接相邻 Step

LangChain 的默认 AgentState 很小，Agent Loop 主要依赖三个字段：

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    jump_to: JumpTo | None
    structured_response: ResponseT

其中 messages 使用 add_messages reducer。节点不需要复制和返回整段历史，只需返回本次新增的消息，reducer 会把它们合并进 state。

一次完整执行可能形成下面的序列：

HumanMessage("Read config.py and report the default port")
AIMessage(tool_calls=[read_file("config.py")])
ToolMessage(tool_call_id="call_1", content="DEFAULT_PORT = 8080")
AIMessage("The default port is 8080")

这个消息序列同时承担两种职责：

• 它是下一次模型调用的上下文；
• 它是条件边判断工具是否待执行、循环是否该结束的状态依据。

因此，messages 不只是聊天记录，而是 Agent Loop 的控制状态。模型输出什么、哪些工具已经返回，都通过消息结构反映到下一次路由中。

七、一次完整执行与终止条件

假设用户要求读取配置并回答默认端口，一次 run 会这样推进：

1. START -> model
2. model returns read_file tool call
3. model_to_tools -> tools
4. tools appends ToolMessage
5. tools_to_model -> model
6. model returns final answer
7. model_to_tools -> END

这里包含两次模型调用，但仍然是一个用户 Turn。第一次模型调用结束，不代表任务结束；只有条件边走向 END，整个 Agent run 才结束。

结合源码，主要终止条件包括：

• 最近的 AIMessage 没有 tool_calls；
• 没有找到可以继续处理的 AIMessage；
• 本批客户端工具全部设置了 return_direct=True；
• 结构化输出已经生成；
• 外部取消执行，或模型、工具抛出不可恢复异常；
• LangGraph 达到 recursion limit。

这也说明 Agent Loop 不等于“不断调用模型直到模型说完成”。真正决定生命周期的是模型消息、工具结果、结构化状态和图运行时共同形成的终止协议。

八、总结

与 DeepSeek Harness 的显式 Driver -> Turn -> Step 相比，LangChain 把同一套执行语义声明成 LangGraph 状态图：

LangChain Agent Loop 可以压缩成一句话：

model 把当前消息状态变成下一步决策，tools 把决策变成真实结果，两条条件边根据 AIMessage 和 ToolMessage 在二者之间循环，直到路由进入 END。

参考资料：

• LangChain create_agent() 源码
• LangChain AgentState 源码
• Deep Agents create_deep_agent() 源码