跳到主要内容

第 1 章:agent 循环如何编译成状态图

本章讲整个项目的主线:你调 create_agent,它在内存里画了一张什么样的图,以及数据在图里怎么转、什么时候停。

1.1 先建立直觉:循环 = 图​

最朴素的 agent 循环长这样(示意,非源码):

# 示意,非源码:agentic loop 的本质
messages = [user_msg]
while True:
    ai = model.invoke(messages)        # 1. 问模型
    messages.append(ai)
    if not ai.tool_calls:              # 2. 模型不调工具了 → 收工
        break
    for call in ai.tool_calls:         # 3. 执行工具
        result = run_tool(call)
        messages.append(result)        # 4. 结果塞回去,继续 while

LangChain 不用 while 写这个。它把每一步变成图里的一个节点,把"该往哪走"变成节点之间的边:

         START
           │
           ▼
     ┌──────────┐   有 tool_calls    ┌──────────┐
     │  model   │ ─────────────────▶ │  tools   │
     │ 问大模型 │                    │ 执行工具 │
     └──────────┘ ◀───────────────── └──────────┘
           │        执行完绕回
           │ 没有 tool_calls
           ▼
          END

为什么要绕这一圈? 因为一旦它是"图"而不是"循环",LangGraph 运行时就能在节点之间存档(checkpointer)、暂停(interrupt)、流式吐 token——这些用裸 while 都得自己造。这是 LangChain v1 把 agent 建在 LangGraph 上的根本原因(libs/langchain_v1/README.md:25:"LangChain agents are built on top of LangGraph in order to provide durable execution, streaming, human-in-the-loop, persistence")。

1.2 入口:create_agent 做的第一批事​

create_agent 的签名一眼能看出它收什么(factory.py:808-827):

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,
    ...
) -> CompiledStateGraph[...]:

开头几步都是"归一化输入"(factory.py:962-990):

  • model 是字符串就先 init_chat_model(model) 变成真模型对象(factory.py:963-964);
  • system_prompt 统一包成 SystemMessage(factory.py:966-972);
  • response_format 包进结构化输出策略(详见第 4 章)。

关键点:create_agent 最终返回的是 StateGraph(...).compile(...) 的产物——一个 CompiledStateGraph(factory.py:825-827、1159-1166)。它是图,不是函数。

1.3 图里的状态(state)是什么​

图节点之间传递的不是裸消息,而是一个 AgentState(middleware/types.py:347-352):

class AgentState(TypedDict, Generic[ResponseT]):
    messages: Required[Annotated[list[AnyMessage], add_messages]]
    jump_to: NotRequired[Annotated[JumpTo | None, EphemeralValue, PrivateStateAttr]]
    structured_response: NotRequired[Annotated[ResponseT, OmitFromInput]]

三个字段各有讲究:

  • messages 带 add_messages reducer——节点返回的新消息是追加进列表,不是覆盖。这就是"对话历史会增长"的实现。
  • jump_to 带 EphemeralValue——一次性的"跳转指令",中间件用它说"直接去 model / tools / end",用完即消(下面 1.6 讲)。
  • structured_response 是结构化输出的落点。

1.4 装配节点​

图至少有两个核心节点:

graph.add_node("model", RunnableCallable(model_node, amodel_node, trace=False))  # factory.py:1502
if tool_node is not None:
    graph.add_node("tools", tool_node)                                          # factory.py:1505-1506

注意 model 节点用 RunnableCallable(sync, async) 同时持有同步和异步两个实现——这样 graph.invoke 和 graph.ainvoke 都能跑,不用你写两遍。

model_node 本身很薄(factory.py:1433-1451):组一个 ModelRequest,如果没有 wrap_model_call 中间件就直接执行,否则交给"洋葱"包装器:

def model_node(state, runtime) -> list[Command[Any]]:
    request = ModelRequest(model=model, tools=default_tools, ...)
    if wrap_model_call_handler is None:
        model_response = _execute_model_sync(request)
        return _build_commands(model_response)
    result = wrap_model_call_handler(request, _execute_model_sync)
    return _build_commands(result.model_response, result.commands)

真正调模型的是 _execute_model_sync(factory.py:1406-1431):拿到绑好工具的模型 → 加上 system 消息 → model_.invoke(messages) → 处理输出。短短一行 output = model_.invoke(messages) 就是整个 agent 唯一真正"问大模型"的地方(factory.py:1419)。

1.5 装配边:循环的进退靠纯函数​

图能"转圈"全靠两条条件边。条件边 = 一个看 state、返回"下一个节点名"的纯函数。

model → tools 边(_make_model_to_tools_edge,factory.py:1840-1891),决定模型说完话后去哪。它的判定顺序很清晰:

顺序条件去向
1state 里有 jump_to按中间件指定跳
2没有 AIMessage(历史被清空)END
3模型没调任何工具END(经典退出条件)
4有待执行的工具调用Send("tools", ...) 逐个分发
5已有结构化响应END

第 3 条"没有 tool_calls 就结束"就是 agent 循环的经典停止条件,源码注释直白写着 # this is the classic exit condition for an agent loop(factory.py:1866-1867)。

tools → model 边(_make_tools_to_model_edge,factory.py:1921-1953),决定工具执行完后回不回模型:

  • 若所有执行的工具都标了 return_direct=True → 直接 END(工具结果就是最终答案,不再问模型,factory.py:1940-1943);
  • 若执行了结构化输出工具 → END;
  • 否则 → 绕回 model,让模型看着工具结果继续想(factory.py:1949-1952)。

这两条边一拼,就是 1.1 那张"转圈"图。return_direct 是个实用的小逃生口:某些工具(比如"直接返回检索片段")不需要模型再润色,标上它就能省一次模型调用。

1.6 jump_to:中间件的"瞬移"指令​

普通节点只能走默认边,但中间件常常想"提前结束"或"跳回模型"。机制是:中间件返回 {"jump_to": "end"},而 _add_middleware_edge 在装配时,如果该钩子声明了 can_jump_to,就给它接一条条件边 jump_edge(factory.py:1978-1999):

def jump_edge(state):
    return _resolve_jump(state.get("jump_to"), ...) or default_destination

can_jump_to 来自钩子上的 @hook_config(can_jump_to=["end"]) 装饰(middleware/types.py:894-897)。只有声明过能跳的钩子,图里才会给它接出额外的分支边——没声明的就只有一条直边,省得图里全是用不到的分叉。这是"图结构按需生成"的典型。

1.7 巧妙之处​

  • 退出条件是数据驱动的纯函数,不是控制流。 把"循环要不要继续"写成 model_to_tools(state) -> 节点名,意味着它可被 LangGraph 在任意 checkpoint 处重放、可被流式观测——比 while/break 可观测得多(factory.py:_make_model_to_tools_edge)。
  • Send("tools", [tool_call]) 逐调用分发。 多个工具调用不是串行 for 循环,而是 fan-out 成多个 Send,可并行执行;且注释指出工具节点运行时才从 channel 拉 state,避免把整份 state 塞进每个 Send(原来 O(N²) 的写放大,factory.py:1877-1882)。
  • sync/async 双实现塞进一个节点。 RunnableCallable(model_node, amodel_node) 让一张图同时支持 .invoke 和 .ainvoke(factory.py:1502)。

1.8 边界与局限​

  • 真正"跑图"的运行时(状态合并、持久化、interrupt 暂停/恢复)在 LangGraph 仓库,不在本 clone 里;本章只讲到"图怎么被画出来"。
  • wrap_model_call 里返回的 Command 目前不支持 goto——会抛错并提示改用 jump_to + before_model/after_model(factory.py:218-221)。

1.9 代码地图​

主题文件路径关键符号
工厂入口与签名libs/langchain_v1/langchain/agents/factory.pycreate_agent
建图、加节点libs/langchain_v1/langchain/agents/factory.pyStateGraph / graph.add_node
模型节点libs/langchain_v1/langchain/agents/factory.pymodel_node / _execute_model_sync
model→tools 边libs/langchain_v1/langchain/agents/factory.py_make_model_to_tools_edge
tools→model 边libs/langchain_v1/langchain/agents/factory.py_make_tools_to_model_edge
中间件跳转边libs/langchain_v1/langchain/agents/factory.py_add_middleware_edge / _resolve_jump
state 定义libs/langchain_v1/langchain/agents/middleware/types.pyAgentState