从零构建 LLM Agent（二）：手写 StateGraph，拆开 ReAct 循环

系列说明

本系列围绕「从用到懂到造」展开，记录基于自部署 LLM 构建 Agent 的实践。本文为第 2 篇——懂：拆开第 1 篇的黑盒 create_react_agent，用 StateGraph 四原语显式重建等价的 ReAct 图。

篇主题阶段
1用 create_react_agent 跑通 ReAct Agent用
2（本文）手写 StateGraph，拆解 ReAct 循环的状态机实现懂
3脱离预置图，自实现 agent loop造
4引入长期记忆：向量检索与 RAG造
5多 Agent 协同（上）：手写白盒最小心智模型造
6多 Agent 协同（中）：LangGraph 灰盒重写，逐行对照造
7多 Agent 协同（下）：CrewAI vs AutoGen 框架对比造
8生产化：可观测、容错与部署工程

运行环境：Python 3.11.15 · langgraph 1.2.9 · langchain-openai 1.3.5 · langchain-core 1.4.9
模型：Qwen3-235B-A22B（自部署，OpenAI 兼容接口）。本篇复用第 1 篇的 config.py 与 tools.py。

1. 问题：循环在 LangGraph 里如何表达

第 1 篇末尾给出了 ReAct 的最小循环伪代码：

while True:
    resp = llm(messages, tools=tools)
    if not resp.tool_calls:
        break
    messages += execute(resp.tool_calls)
    messages += resp
return resp.content

create_react_agent 把它封装成了黑盒。本文要做的是：不依赖预置封装，用 StateGraph 的四个原语把这个 while 翻译成一张状态图，并验证其与预置实现行为等价。

LangGraph 不提供命令式循环，循环由图的拓扑表达：一个节点经条件边指向自身可达的另一节点，且该节点有回流边回到前者，即构成循环。 具体到 ReAct：agent 节点经条件边分流到 tools 或 END，tools 经固定边回流 agent，循环由此成立。

2. 四原语

StateGraph 的全部构造能力可归结为四个原语，对应本文代码的四个部分：

原语作用ReAct 图中的对应
State节点间流转的数据结构messages 列表
Node接收 state、返回 state 增量的函数agent_node / tools_node
Edge固定连线START->agent、tools->agent
条件边依据 state 选择下一节点agent->tools 或 agent->END

2.1 State：消息即状态

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

ReAct 的状态只有一个字段 messages，因为推理的全部上下文就是消息历史。关键在 Annotated[..., add_messages]：add_messages 是一个 reducer，定义节点返回值如何并入状态。默认行为是覆盖，加了 reducer 后变为追加。这意味着每个节点只需返回本轮新增的消息，框架负责合并，节点之间无需感知完整状态如何拼装。

这是“节点只管增量”模型的基础，也是后续多节点协作能保持简洁的原因。

2.2 Node：增量函数

agent_node 调 LLM 产出 AIMessage（可能含 tool_calls）：

def agent_node(state: AgentState) -> dict:
    messages = state["messages"]
    if not any(isinstance(m, SystemMessage) for m in messages):
        messages = [SystemMessage(content=SYSTEM_PROMPT)] + list(messages)
    response: AIMessage = llm_with_tools.invoke(messages)
    return {"messages": [response]}   # 只返回增量

tools_node 执行工具并产出 ToolMessage。这里手写而非用预置 ToolNode，目的是暴露 ToolMessage 的构造过程：

def tools_node(state: AgentState) -> dict:
    last_msg: AIMessage = state["messages"][-1]
    results: list[ToolMessage] = []
    for tc in last_msg.tool_calls:
        tool_fn = TOOL_MAP[tc["name"]]
        try:
            output = tool_fn.invoke(tc["args"])
        except Exception as e:
            output = f"工具执行出错: {e}"
        results.append(ToolMessage(
            content=str(output),
            tool_call_id=tc["id"],
            name=tc["name"],
        ))
    return {"messages": results}

两个细节值得注意：

  1. tool_call_id 关联。ToolMessage 必须带上对应 tool_call 的 id。一次 AIMessage 可能含多个 tool_calls（并行调用），LLM 靠 id 把“这次工具结果”关联到“那次工具请求”。漏掉 id 会导致协议解析失败。
  2. 工具失败也是 Observation。工具抛异常时不中断图，而是把错误字符串包成 ToolMessage 喂回 LLM，由 LLM 决定重试、换工具或放弃。这比直接抛错更健壮，因为 LLM 有上下文判断如何处理。

2.3 条件边：终止条件所在

def should_continue(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if isinstance(last_msg, AIMessage) and last_msg.tool_calls:
        return "tools"
    return END

这是整个循环的“终止条件”。没有 while 的 break，终止表现为条件边返回 END：当某次 agent_node 产出的 AIMessage 不含 tool_calls，下一次转移直接到 END，图执行结束。create_react_agent 内部也是同样的判定。

2.4 组装

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tools_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")          # 回流边 -> 构成循环
app = graph.compile()

add_edge(START, "agent") 设定入口；add_conditional_edges 在 agent 之后按 should_continue 分流；add_edge("tools", "agent") 是回流边。三条边共同构成循环：agent -> tools -> agent -> ...，直到条件边把控制流导向 END。

编译后的拓扑（grandalf 绘制）：

        +-----------+
        | __start__ |
        +-----------+
               |
          +-------+
          | agent |
          +-------+
          /         \
+---------+         +-------+
| __end__ |         | tools |
+---------+         +-------+

tools -> agent 的回流边未在 ASCII 图中显式画出，但在 add_edge 中真实存在，正是它让拓扑闭环。

3. 等价性验证

复刻是否正确，用第 1 篇的同一问题验证，对比行为是否一致。

输入：「现在几点了？腾讯股票多少钱？持有 100 股市值多少港元？」

本文手写图 trace：

[1] user: 现在几点了？腾讯股票多少钱？持有100股市值多少港元？
[2] AIMessage tool_calls: get_current_time({})
[3] ToolMessage: 2026年07月16日 10:20:34 Thursday
[4] AIMessage tool_calls: get_stock_price({'symbol': '腾讯'})
[5] ToolMessage: 腾讯控股 当前 385.6 港元，+1.2%
[6] AIMessage tool_calls: calculate({'expression': '385.6 * 100'})
[7] ToolMessage: 385.6 * 100 = 38560.0
[8] AIMessage: 市值约为 38,560 港元。

与第 1 篇 create_react_agent 的 trace 一致：相同的 get_current_time -> get_stock_price -> calculate 串行链，calculate 入参 385.6 同样取自上一步 get_stock_price 返回。两套实现在同一输入下产生相同的工具调用序列与终答，可认为行为等价。由此可下结论：create_react_agent 即上述 StateGraph 结构 + 默认 should_continue + 预置 ToolNode 的封装，无额外机制。

4. 小结

  • 循环即拓扑：ReAct 的 while 在 LangGraph 中由“条件边 + 回流边”表达，节点拓扑闭环即循环。
  • State 即消息：add_messages reducer 使节点只需返回增量，框架负责合并，这是状态累积的机制。
  • ToolMessage 的契约：tool_call_id 关联请求与结果，是工具结果正确回传 LLM 的关键。
  • 终止即条件边导向 END：tool_calls 为空时条件边返回 END，无显式 break。
  • 工具失败不中断：异常被包成 ToolMessage 回流，由 LLM 处理。

四个原语组合出一个完整的、可解释的 ReAct 状态机。create_react_agent 不再是黑盒。

5. 遗留问题与下一阶段

至此仍有两个问题悬而未决：

  1. 终止的可靠性。终止依赖 LLM 自觉不再发起 tool_calls。若模型持续发起调用，理论上循环不终止。实践中 LangGraph 以 recursion_limit（默认 25 步）作硬性防线，超限抛错。这是工程兜底，但暴露了“纯靠模型自觉”的脆弱性。
  2. 过程不可干预。当前图一旦执行即跑完，中途无法插入人工审核、无法在关键工具调用前暂停。这对低风险工具无碍，但对有副作用的操作（发消息、下单、删数据）不可接受。

这两个问题指向同一组能力：对循环过程的控制。这正是第 5 篇（人机协同）与第 6 篇（生产化）的主题。

但在此之前，第 3 篇先做更彻底的拆解——连 StateGraph 也不用，以纯 Python 实现 agent loop。StateGraph 用 reducer、节点、条件边把循环抽象得很优雅，但优雅的代价是隐去了底层的状态搬运与转移控制。手写 loop 能把这层完全展开，回答“拿掉框架之后，ReAct 的最小实现到底是什么样”。理解了这个最小实现，再回头看 StateGraph 的抽象会清晰得多。

# -*- coding: utf-8 -*-
"""第 2 阶段：手写 StateGraph，等价复刻 create_react_agent。

目的：把上一阶段那个“黑盒”拆开。create_react_agent 内部其实是这样一张图：

    START ─▶ agent ──tool_calls 非空──▶ tools ─┐
                   │                          │
                   │                          │ 回流
              tool_calls                      │
                 为空                          │
                   ▼                          │
                  END ◀───────────────────────┘
                              （不会走这条，仅示意）

这里我们用 StateGraph 的四个原语自己搭出来：
    State  : 节点间流转的数据 -- 这里就是 messages 列表
    Node   : 接收 state、返回 state 增量的函数 -- agent 节点 + tools 节点
    Edge   : 固定连线 -- START->agent、tools->agent（回流）
    条件边 : 根据 state 决定下一站 -- agent 出来后看 tool_calls 决定去 tools 还是 END

核心认知：ReAct 的“循环”不是 while，而是「条件边 + 一条回流边」。
图自己在 agent <-> tools 之间转，直到某次 agent 不再发起 tool_calls，走条件边到 END。

运行方式：
    conda activate agent
    python agent_graph.py
"""
from typing import TypedDict, Annotated

from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage, SystemMessage, BaseMessage
from langchain_core.tools import BaseTool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

from config import LLM_BASE_URL, LLM_API_KEY, LLM_MODEL
from tools import get_weather, get_current_time, calculate, get_stock_price


# ============================================================
# 1. 定义 State：在节点间流转的数据
# ============================================================
# TypedDict 声明状态字段；Annotated[list, add_messages] 表示：
# 节点返回的 messages 会用 add_messages 这个 reducer “追加”进状态，而非覆盖。
# 这就是为什么每个节点只需返回【新增的消息】，框架会自动合并。
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]


# ============================================================
# 2. 准备 LLM + 工具（和上一阶段一致）
# ============================================================
llm = ChatOpenAI(
    model=LLM_MODEL,
    base_url=LLM_BASE_URL,
    api_key=LLM_API_KEY,
    temperature=0.7,
)

TOOLS: list[BaseTool] = [get_weather, get_current_time, calculate, get_stock_price]
# bind_tools 把工具 schema 下发给 LLM，使其能返回 tool_calls
llm_with_tools = llm.bind_tools(TOOLS)

print("llm_with_tools",llm_with_tools)
SYSTEM_PROMPT = (
    "你是一个能干的个人助手，可以查天气、查时间、做计算、查股票。"
    "回答用户问题时，请优先调用合适的工具获取信息或完成计算，不要凭空编造。"
    "需要组合多个工具才能完成的任务，大胆分步组合使用。回答用中文。"
)

# 工具名 -> 工具对象 的映射，供 tools 节点查表执行
TOOL_MAP = {t.name: t for t in TOOLS}
print("TOOL_MAP", TOOL_MAP)

# ============================================================
# 3. 定义节点
# ============================================================
def agent_node(state: AgentState) -> dict:
    """agent 节点：调 LLM，产出 AIMessage（可能含 tool_calls）。"""
    # 首轮注入 system prompt（放在消息列表最前）
    messages = state["messages"]
    if not any(isinstance(m, SystemMessage) for m in messages):
        messages = [SystemMessage(content=SYSTEM_PROMPT)] + list(messages)
    # 调用 LLM：它看着全部历史消息，决定回答 or 调工具
    print("message",messages)
    response: AIMessage = llm_with_tools.invoke(messages)
    print("agent_node response", response)
    # 只返回新增消息，reducer 会自动 append
    return {"messages": [response]}


def tools_node(state: AgentState) -> dict:
    """tools 节点：执行上一条 AIMessage 里的 tool_calls，产出 ToolMessage。

    这里手写而非用预置 ToolNode，是为了看清楚 ToolMessage 是怎么来的：
    遍历 tool_calls -> 查表找到函数 -> 执行 -> 用结果构造 ToolMessage。
    ToolMessage 的 tool_call_id 必须与对应 tool_call 的 id 对上，LLM 才能把
    “这次工具结果”正确关联到“那次工具请求”。
    """
    last_msg: AIMessage = state["messages"][-1]
    results: list[ToolMessage] = []
    for tc in last_msg.tool_calls:
        tool_fn = TOOL_MAP[tc["name"]]
        try:
            output = tool_fn.invoke(tc["args"])
            print(f"tools_node {tc['name']}({tc['args']}) -> {output}")
        except Exception as e:
            # 工具失败也是一种 Observation，喂回 LLM 让它自己处理（如换工具/重试）
            output = f"工具执行出错: {e}"
        results.append(ToolMessage(
            content=str(output),
            tool_call_id=tc["id"],
            name=tc["name"],
        ))
    return {"messages": results}


# ============================================================
# 4. 定义条件边的路由函数
# ============================================================
def should_continue(state: AgentState) -> str:
    """agent 节点出来后的分流：有 tool_calls 去 tools，否则结束。

    返回值是“下一个节点的名字”或 END。这就是循环的“终止条件”所在：
    没有 while，全靠这条条件边决定继续转还是停下。
    """
    last_msg = state["messages"][-1]
    if isinstance(last_msg, AIMessage) and last_msg.tool_calls:
        return "tools"
    return END


# ============================================================
# 5. 组装图
# ============================================================
def build_graph():
    graph = StateGraph(AgentState)

    # 5.1 加节点
    graph.add_node("agent", agent_node)
    graph.add_node("tools", tools_node)

    # 5.2 加边
    graph.add_edge(START, "agent")                       # 入口 -> agent
    graph.add_conditional_edges(                         # agent -> tools 或 END（条件边）
        "agent",
        should_continue,
        # path_map 仅用于可视化，列出所有可能去向
        {"tools": "tools", END: END},
    )
    graph.add_edge("tools", "agent")                     # tools -> agent（回流，构成循环）

    # 5.3 编译成可执行对象
    return graph.compile()


def print_graph(app):
    """打印图结构（ASCII），直观看清节点与边。"""
    print("\n-- 图结构 --")
    try:
        print(app.get_graph().draw_ascii())
    except Exception:
        # 某些环境缺_asciiart依赖时退化为文字版
        print("START -> agent -> [tools | END]; tools -> agent")


def print_trace(messages):
    """打印 ReAct 循环每一步（与上一阶段一致的输出格式，方便对比）。"""
    print("\n-- Agent 推理过程 --")
    for i, m in enumerate(messages, 1):
        if isinstance(m, SystemMessage):
            continue
        if isinstance(m, HumanMessage):
            print(f"[{i}] 👤 用户: {m.content}")
        elif isinstance(m, AIMessage):
            if m.tool_calls:
                for tc in m.tool_calls:
                    print(f"[{i}] 🤖 AI 调用工具: {tc['name']}({tc['args']})")
            else:
                print(f"[{i}] 🤖 AI 回答: {m.content}")
        elif isinstance(m, ToolMessage):
            print(f"[{i}] 🔧 工具返回: {m.content}")
    print("-- 推理结束 --")


def main():
    app = build_graph()
    print_graph(app)
    print("=" * 50)
    print("  手写 StateGraph Agent 已就绪（输入 quit 退出）")
    print("=" * 50)

    while True:
        user = input("\n你: ").strip()
        if user.lower() in ("quit", "exit", "q"):
            print("再见！")
            break
        if not user:
            continue
        result = app.invoke({"messages": [{"role": "user", "content": user}]})
        print_trace(result["messages"])


if __name__ == "__main__":
    main()
# 人工智能
Logo
DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

cover

我用 Qwen3.8-Max 做了一个采购合同风险审查工具：从能分析，到能定位、复核和交付

avatar DeepSeek技术社区
cover

我用 Qwen3.8-MAX 搭了一个教研纪要 Agent，一次集体备课几秒出七段、分歧还不被和稀泥

avatar DeepSeek技术社区
cover

OpenCode Go 与 Command Code 的性价比分析 2026.08.31更新

avatar DeepSeek技术社区