从零构建 LLM Agent（三）：剥到白盒，纯 Python 实现 Agent Loop

系列说明

本系列围绕「从用到懂到造」展开。本文为第 3 篇——造：脱离 StateGraph，以纯 Python 实现 agent loop，看清拿掉框架抽象后 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 兼容接口）。本文复用前两篇的 config.py 与 tools.py。

1. 三阶段是一条“去抽象”的线

阶段	实现	抽象层级	可见的结构
1	create_react_agent	黑盒	只见输入输出
2	StateGraph	灰盒	节点 + 边 + 条件边
3（本文）	纯 Python loop	白盒	for 循环 + messages 列表

前两篇逐步拆开了黑盒：第 2 篇把 create_react_agent 还原成 StateGraph 的节点与条件边。本文再进一步——连 StateGraph 也不用，把那张图手动展开成一个普通循环。

目标不是“重新发明轮子”，而是回答一个问题：框架层层封装之后，内核到底是什么？ 看清这个内核，再回看 StateGraph 与 create_react_agent 的抽象会清晰得多。

2. 内核：一个循环加一个消息列表

第 1 篇给出过 ReAct 的最小循环伪代码。它就是 agent 的内核：

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

把这份伪代码落成可运行代码（复用前两篇的 llm_with_tools、TOOL_MAP），只需十几行：

def run_agent(user_input: str, max_steps: int = 10) -> list[BaseMessage]:
    # State：一个普通列表（第 2 篇的 State 在这里退化成 list）
    messages: list[BaseMessage] = [
        SystemMessage(content=SYSTEM_PROMPT),
        HumanMessage(content=user_input),
    ]

    # for 代替 while：range 自带步数上限，即 recursion_limit 的等价物
    for step in range(max_steps):
        # 【agent 节点】调 LLM，决定回答 or 调工具
        resp: AIMessage = llm_with_tools.invoke(messages)
        messages.append(resp)

        # 【条件边】没有 tool_calls 就结束（等价于导向 END）
        if not resp.tool_calls:
            return messages

        # 【tools 节点】执行工具调用，结果作为 ToolMessage 回填
        for tc in resp.tool_calls:
            try:
                output = TOOL_MAP[tc["name"]].invoke(tc["args"])
            except Exception as e:
                output = f"工具执行出错: {e}"
            messages.append(ToolMessage(
                content=str(output),
                tool_call_id=tc["id"],
                name=tc["name"],
            ))
        # 循环回到顶部：带着工具结果再次调 LLM（等价于 tools -> agent 回流边）

    print(f"⚠️ 达到最大步数 {max_steps}，强制结束")
    return messages

没有 State、没有 Node、没有条件边——只有 for、if、append。

3. 与第 2 篇 StateGraph 概念的逐项映射

这份代码与第 2 篇的 StateGraph 实现是逐概念对应的。把两张实现并排放，每个框架抽象都能找到原生 Python 的等价物：

第 2 篇 StateGraph 概念	本文的等价物	代码位置
State（messages + add_messages reducer）	一个 list，手动 append	messages = [...]
agent_node	llm_with_tools.invoke(messages)	循环体首行
tools_node	for tc: TOOL_MAP[name].invoke(args)	内层 for
条件边（tool_calls 非空?）	if not resp.tool_calls: return	中部 if
recursion_limit（默认 25）	for step in range(max_steps)	循环头
app.invoke({"messages": ...})	run_agent(user_input)	函数调用

关键区别只在状态的合并方式：

• StateGraph 用 reducer：节点只返回增量 {"messages": [resp]}，框架靠 add_messages 自动追加；
• 纯 loop 没有 reducer：直接 messages.append(resp)，手动维护。

reducer 抽象的代价是隐去了“状态如何拼装”。手动 append 把这一步显式化，于是能看清“状态就是消息列表的不断增长”。

4. 等价性验证

复刻是否正确，用前两篇的同一问题验证。

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

第1轮: get_current_time({})
第2轮: get_stock_price({'symbol': '腾讯'})   -> 385.6 港元
第3轮: calculate({'expression': '385.6 * 100'})  -> 38560.0
终答: 市值约为 38,560 港元

与第 1、2 篇的 trace 一致：相同的串行工具链，calculate 入参 385.6 同样取自上一步 get_stock_price 返回。三套实现在同一输入下产生相同的工具调用序列与终答，机制等价。

由此可下结论：create_react_agent、StateGraph、纯 Python loop 三者层层封装、内核相同。剥到最底，agent 就是“一个循环加一个消息列表”。

5. 一个意外发现：行为等价 ≠ 输出一致

验证过程中出现了一个超出预期的现象。同一问题，把 temperature 从 0 调到 0.7，trace 变了：

[3] get_current_time({})
[3] get_stock_price({'symbol': '腾讯'})
[3] calculate({'expression': '300 * 100'})    ← 一次性并行调了三个工具，且 calculate 猜了个 300
[6] calculate 返回 300 * 100 = 30000           ← 错误的中间结果
[7] 终答: 385.6 × 100 = 38,560 港元            ← 但终答正确

三个值得记录的观察：

1. 策略随机：temperature=0.7 时模型选择了并行调用（一轮发三个工具），而 temperature=0 时稳定选择串行。同一问题，每次运行策略可能不同。
2. 并行 + 数据依赖 = 出错：calculate 依赖 get_stock_price 的结果，但并行调用时 385.6 尚未返回，模型只能凭空猜了 300，中间结果错误。
3. 自我纠错：终答仍然正确——模型在汇总阶段识别出 30000 与 385.6 的矛盾，选择相信真实查到的 385.6 自行重算。

为确认这是 LLM 采样随机性而非 loop 实现错误，做对照实验：temperature=0 下连跑两次同问题。

第1次: time → stock(腾讯) → calculate(385.6 * 100)
第2次: time → stock(腾讯) → calculate(100 * 385.6)   ← 仅乘数顺序不同，语义一致

两次 trace 结构完全一致，且均为正确的串行链。差异源自采样随机性，与 loop 实现无关。

这一发现修正了对“等价性”的理解：

三套实现的等价是“机制等价”——状态流转、终止条件、工具执行方式相同。但 agent 的具体行为（串行还是并行、参数填什么）由 LLM 决定，带采样随机性。temperature=0 时三者给出一致 trace；temperature>0 时策略会漂移。等价不等于输出 bitwise 一致。

6. 小结

• 内核：剥掉所有框架抽象，agent 是“一个循环加一个消息列表”。这是 ReAct 的最小实现。
• 逐概念映射：StateGraph 的 State/Node/条件边/recursion_limit，分别对应 list/invoke/if break/range。reducer 抽象隐藏的“状态拼装”在纯 loop 中被显式化。
• 机制等价：三套实现（create_react_agent / StateGraph / 纯 loop）层层封装、内核相同，temperature=0 时行为一致。
• 可靠性边界：框架只保证机制正确，不保证决策正确。正确性最终压在 LLM 判断上，而 LLM 决策带采样随机性，可能误判工具依赖（如并行调用有依赖的工具）。

“用-懂-造”三阶段至此闭环：从调用预置封装，到拆开状态图，再到裸写出内核循环。agent 的机制层面已无黑盒。

7. 下一阶段

机制看透之后，真正的问题才刚开始暴露。第 5 节的“并行猜参”显示：框架保证了循环正确运转，却无法保证 agent 的决策正确。 一个会为了效率并行调用、却误判数据依赖的 agent，在玩具场景能靠自纠错兜住，在生产场景（发消息、下单、改库）则不可接受。

由此引出后续两个方向：

• 可控性（第 5 篇）：强制串行、限制并行、在关键工具调用前插入人工审核（human-in-the-loop），让 agent 的决策可被外部约束。
• 可靠性（第 6 篇）：temperature 取值策略、工具依赖的显式声明、中间步错误的检测与兜底、执行步数的监控。

前三篇的 agent 每次对话都从零开始，没有跨会话的记忆。引入向量检索（复用本系列待用的 embedding 服务）后，agent 才能记住用户说过的话、检索私有知识。这是从“能推理”到“有记忆”的跨越。

# -*- coding: utf-8 -*-
"""第 3 阶段：脱离 StateGraph，纯 Python 手写 agent loop。

把第 2 阶段被 StateGraph 封装起来的那个循环，手动展开。
没有 State / Node / 条件边——就是一个 for 循环 + 一个 messages 列表。

目的：看清拿掉框架之后，ReAct 的最小实现到底长什么样。
你会看到：agent 的核心，其实就这么十几行。

与第 2 阶段 StateGraph 概念的逐项对照：
    StateGraph 概念             <==>   这里的等价物
    -----------------------------------------------------------
    State (messages + reducer)  <==>   一个 list，手动 append
    agent 节点                  <==>   llm_with_tools.invoke(messages)
    tools 节点                  <==>   for tc: TOOL_MAP[...].invoke(...)
    条件边 (tool_calls 非空?)   <==>   if not resp.tool_calls: break
    recursion_limit (默认 25)   <==>   for step in range(max_steps)
    app.invoke(...)             <==>   run_agent(user_input)

运行：conda activate agent && python agent_loop.py
"""
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    AIMessage, HumanMessage, ToolMessage, SystemMessage, BaseMessage,
)

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

# ---- 准备 LLM + 工具（和前两阶段完全一致）----
llm = ChatOpenAI(
    model=LLM_MODEL, base_url=LLM_BASE_URL, api_key=LLM_API_KEY, temperature=0.7,
)
TOOLS = [get_weather, get_current_time, calculate, get_stock_price]
llm_with_tools = llm.bind_tools(TOOLS)          # 决策侧：schema 下发给模型
TOOL_MAP = {t.name: t for t in TOOLS}           # 执行侧：名字 -> 函数 查找表

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


def run_agent(user_input: str, max_steps: int = 10) -> list[BaseMessage]:
    """纯 Python 的 ReAct loop。返回完整 messages 历史。"""
    # 【State】状态就是一个普通列表（第 2 阶段的 State 在这里退化成 list）
    messages: list[BaseMessage] = [
        SystemMessage(content=SYSTEM_PROMPT),
        HumanMessage(content=user_input),
    ]

    # 用 for 代替 while：range 自带步数上限，就是 recursion_limit 的等价物
    for step in range(max_steps):
        # 【agent 节点】调 LLM，决定回答 or 调工具
        resp: AIMessage = llm_with_tools.invoke(messages)
        messages.append(resp)

        # 【条件边】没有 tool_calls 就结束（等价于导向 END，循环终止）
        if not resp.tool_calls:
            return messages

        # 【tools 节点】执行所有工具调用，结果作为 ToolMessage 回填进 messages
        for tc in resp.tool_calls:
            try:
                output = TOOL_MAP[tc["name"]].invoke(tc["args"])
            except Exception as e:
                output = f"工具执行出错: {e}"
            messages.append(ToolMessage(
                content=str(output),
                tool_call_id=tc["id"],
                name=tc["name"],
            ))
        # 循环回到顶部：带着工具结果再次调 LLM（等价于 tools -> agent 回流边）

    # 步数耗尽仍未结束（等价于 recursion_limit 超限）
    print(f"⚠️ 达到最大步数 {max_steps}，强制结束")
    return messages


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():
    print("=" * 50)
    print("  纯 Python Agent Loop 已就绪（输入 quit 退出）")
    print("=" * 50)
    while True:
        user = input("\n你: ").strip()
        if user.lower() in ("quit", "exit", "q"):
            print("再见！")
            break
        if not user:
            continue
        messages = run_agent(user)
        print_trace(messages)


if __name__ == "__main__":
    main()