拆解 Agent 循环的本质

写给开发者的"祛魅"指南。读完你会知道:所谓 agent 编排,99% 是一个 while 循环加几个钩子。

引子:先打破神秘感

"AI Agent"这个词被炒得太玄。各种框架——LangChain、LangGraph、AutoGen、CrewAI——把简单的事情包装成图、节点、边、状态机,让人以为这是个高深领域。

事实是:一个能干活的 agent,核心是一个 20 行的 while 循环。所有框架的价值都在循环之外(工具集成、记忆、监控),而不在"编排"本身。

我用 pi 这个开源 coding agent 的源码作参照(packages/agent/src/agent-loop.ts,全部精华就这一个文件),带你从零拆到生产级。


一、最小可用 agent:一个 while 循环

扔掉所有框架,agent 的本质是 ReAct 模式(Reason + Act):

模型思考 → 调用工具 → 看到工具结果 → 再思考 → 再调用 → ... → 不需要工具了,给出答案

用 Python 写出来,就这么点:

def agent_loop(question, tools, max_turns=10):
    messages = [{"role": "user", "content": question}]
    
    for _ in range(max_turns):
        # 1. 把整个对话历史喂给模型
        response = llm.chat(messages, tools=tools)
        messages.append(response.message)
        
        # 2. 模型没有调工具 → 它给出了最终答案,结束
        if not response.tool_calls:
            return response.message.content
        
        # 3. 模型调了工具 → 执行,把结果回灌进对话
        for call in response.tool_calls:
            result = tools[call.name](**call.arguments)
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            })
    
    raise RuntimeError("超过最大轮数,可能死循环了")

就这些。 这就是一个 agent。剩下的所有工程都是把这个骨架加固、加速、可观测。

记住这张图,它是后面所有讨论的基础:

       ┌─────────────────────────────────────┐
       │  messages = [用户问题]                │
       └────────────────┬────────────────────┘
                        │
       ┌────────────────▼────────────────────┐
  ┌───►│  ① 调 LLM(把整个 messages 送去)     │
  │    └────────────────┬────────────────────┘
  │                     │ 返回 assistant message
  │    ┌────────────────▼────────────────────┐
  │    │  ② 这条消息里有 tool_calls 吗?        │
  │    └──────┬─────────────────┬────────────┘
  │           │ 否              │ 是
  │    ┌──────▼──────┐   ┌──────▼─────────────────────────┐
  │    │  返回答案,   │   │  ③ 执行每个 tool_call            │
  │    │  结束        │   │     把结果作为 tool message      │
  │    └─────────────┘   │     append 进 messages          │
  │                       └──────┬─────────────────────────┘
  │                              │
  └──────────────────────────────┘

三个要素:对话历史(messages) 是状态,LLM 调用 是决策,工具执行 是动作。三者循环,直到模型不再调工具。


二、关键认知:为什么不是图?

很多人第一反应是"agent 应该用状态图建模"。LangGraph 的核心卖点就是这个。但你要问自己一个问题:

你的 agent 真的有"分支"吗?

99% 的 agent 没有。它的下一步永远由 LLM 决定,而不是由代码里的 if 决定。状态图的"边"是开发者预定义的转换,但 agent 的转换是模型在运行时动态选的——这根本不是图,是模型驱动的循环

状态图适合确定性的多步管道

[用户问题] → [生成SQL] → [执行SQL] → [校验结果] → [生成图表] → [写摘要]

每一步是固定的,分支有限(成功/失败/重试)。这种场景 LangGraph 很合适。

agent 循环适合开放式探索

[用户问题] → "我需要先查X" → 查 → "再看Y的关联" → 查 → "还不够,算个Z" → ...

下一步做什么完全由模型根据上一步结果决定,无法预先画出流程图。pi 的 coding agent 就是这种——它可能读文件、改代码、跑测试、再读文件,顺序完全不可预测。

判断标准很简单:你能提前画出流程图的,用状态图;画不出来的,用循环。强行用图建模 agent,你会写出一堆"if 模型说了X就转Y节点"的伪图,本质还是循环,只是更难调试。


三、从玩具到生产:五个加固要素

最小循环能跑,但上生产会死得很惨。下面是必须加的五层。

要素 1:流式响应

同步等模型把整段回答吐完才返回,体验极差。生产 agent 必须流式——每个 token 边生成边推给 UI。

def stream_assistant(messages, tools):
    partial = AssistantMessage()
    for event in llm.stream(messages, tools):
        if event.type == "text_delta":
            partial.content += event.text
            yield {"type": "text_delta", "text": event.text}  # 推给 UI
        elif event.type == "tool_call":
            partial.tool_calls.append(event.call)
        elif event.type == "done":
            yield {"type": "done", "message": partial}
            return partial

关键设计:边流式输出边累积,结束后才把完整 message 塞回 history。pi 里 streamAssistantResponse() 就是这么做的(agent-loop.ts:275)。

要素 2:工具并行执行

模型经常一次返回多个 tool_call(比如同时读三个文件)。串行执行浪费时间。

# 串行:3个工具各1秒 = 3秒
for call in tool_calls:
    execute(call)

# 并行:3个工具各1秒 = 1秒
import asyncio
results = await asyncio.gather(*[execute(c) for c in tool_calls])

注意:有些工具必须串行(比如写文件后再读)。pi 的做法是给工具标 executionMode: "sequential",任一工具标了就降级整批为串行(agent-loop.ts:384)。生产 agent 必须支持这种混合模式。

要素 3:事件流(替代回调)

UI 怎么知道 agent 在干嘛?新手会写一堆回调。更好的做法是把所有状态变化建模成事件,agent 只管发事件,谁爱听谁听。

# agent 内部
def execute_tool_call(call):
    emit({"type": "tool_start", "name": call.name, "args": call.arguments})
    result = tools[call.name](**call.arguments)
    emit({"type": "tool_end", "name": call.name, "result": result})
    return result

# UI 那边
for event in agent.run(question):
    if event["type"] == "tool_start":
        spinner.show(f"正在执行 {event['name']}...")
    elif event["type"] == "text_delta":
        print(event["text"], end="", flush=True)

pi 定义了 9 种事件(types.ts:408):agent_start/endturn_start/endmessage_start/update/endtool_execution_start/update/end。整个 TUI 就是这 9 种事件的消费者。事件流让 agent 内核和 UI 完全解耦——同样的内核可以接 CLI、Web、IDE 插件。

要素 4:两个消息队列(steering vs follow-up)

这是 pi 最精彩的设计,新手 agent 都没有。

痛点:agent 跑了 30 秒,跑了 5 轮工具调用。这时用户想插话"顺便也查下上个月的数据"。怎么塞进去?

错误做法:等 agent 跑完再处理。用户体验差。

正确做法:维护两个队列,在不同时机注入。

turn 1: [LLM] → tool → tool_result
turn 2: [LLM] → tool → tool_result
                              │
                              │  ← steering 队列在这里 drain
                              │     注入到下一轮 LLM 调用前
turn 3: [LLM] → tool → tool_result
turn 4: [LLM] → (无 tool) → 答案   ← agent 本来要停了
                              │
                              │  ← follow-up 队列在这里 drain
                              │     agent 继续干活
turn 5: [LLM] → ...
  • steering:agent 还在干,用户的话要在当前任务没结束时插进去影响方向。
  • follow-up:agent 干完了,用户的追加任务让它重新启动
class Agent:
    def __init__(self):
        self.steering_queue = []
        self.follow_up_queue = []
    
    def run(self, question):
        messages = [{"role": "user", "content": question}]
        while True:
            # steering 注入点:当前 turn 结束后、下次调 LLM 前
            if self.steering_queue:
                messages.extend(self.steering_queue)
                self.steering_queue.clear()
            
            response = llm.chat(messages, tools=tools)
            messages.append(response.message)
            
            if not response.tool_calls:
                # agent 想停了 → 检查 follow-up
                if self.follow_up_queue:
                    messages.extend(self.follow_up_queue)
                    self.follow_up_queue.clear()
                    continue
                return response.message.content
            
            for call in response.tool_calls:
                result = execute(call)
                messages.append(tool_message(call.id, result))

steering/follow-up 的区别,是区分"严肃 agent"和"玩具脚本"的分水岭。

要素 5:钩子点(hook points)

生产 agent 需要:工具调用前拦截(权限)、调用后改写结果(脱敏)、每轮后换模型(省钱)、超 token 时停(防失控)。这些不能写死在循环里,否则没法复用。

解法是用函数注入。循环骨架不变,行为靠传入的钩子定制:

def agent_loop(question, config):
    messages = [{"role": "user", "content": question}]
    for _ in range(config.max_turns):
        response = llm.chat(messages, config.model, tools=config.tools)
        messages.append(response.message)
        
        if not response.tool_calls:
            return response.message.content
        
        for call in response.tool_calls:
            # 钩子1: 执行前拦截(权限检查、参数校验)
            if config.before_tool_call:
                decision = config.before_tool_call(call)
                if decision.block:
                    messages.append(tool_error(call.id, decision.reason))
                    continue
            
            result = execute(call)
            
            # 钩子2: 执行后改写(脱敏、截断、加注释)
            if config.after_tool_call:
                result = config.after_tool_call(call, result)
            
            messages.append(tool_message(call.id, result))
        
        # 钩子3: 每轮后决策(换模型、停止、压缩历史)
        if config.after_turn:
            update = config.after_turn(messages)
            if update.stop: break
            if update.new_model: config.model = update.new_model

pi 把这套钩子定义在 AgentLoopConfig(types.ts:135),共 8 个钩子点。整个 coding agent 的权限、脱敏、成本控制、上下文压缩,全靠这几个钩子实现,循环本身一行没改。 这就是好的抽象——核心循环极简,所有变化点都是可注入的函数。


四、钩子点全景图

把上面五层叠起来,一个生产级 agent 循环长这样:

agent_loop():
┌──────────────────────────────────────────────────────────┐
│  while True:                                              │
│                                                           │
│    ┌─ [steering 队列 drain] ◄── 用户工作中插话            │
│    │                                                      │
│    │  ┌───────────────────────────────────────────────┐  │
│    │  │  stream_assistant(messages, model, tools)     │  │
│    │  │    │ 流式事件 → emit 事件给 UI                  │  │
│    │  │    └─► assistant message (含 N 个 tool_calls)  │  │
│    │  └───────────────────────────────────────────────┘  │
│    │                                                      │
│    ├─ if 无 tool_calls:                                   │
│    │    ┌─ [follow-up 队列 drain] ◄── 用户追加任务        │
│    │    │  有 → continue                                  │
│    │    └─ 无 → return 答案                               │
│    │                                                      │
│    ├─ for each tool_call (并行 or 串行):                  │
│    │    ├─ [before_tool_call 钩子] ◄─ 权限/校验/拦截      │
│    │    ├─ execute(args) → 流式 emit 进度                 │
│    │    ├─ [after_tool_call 钩子] ◄─ 脱敏/截断/改写       │
│    │    └─ append tool_result 到 messages                 │
│    │                                                      │
│    ├─ [prepare_next_turn 钩子] ◄─ 换模型/换上下文         │
│    ├─ [should_stop 钩子] ◄─ 超 token/成本/手动停止        │
│    └─ continue                                            │
└──────────────────────────────────────────────────────────┘

核心循环代码不到 100 行。所有 [] 标注的都是可注入的扩展点。这就是 pi 的 agent-loop.ts 实际的样子——我建议你打开那个文件通读一遍,它是这个模式最干净的实现。


五、什么时候不该用 agent 循环

诚实的工程师要会说不。agent 循环不是银弹。

不该用的场景

  1. 单步问答。"上个月销售额多少" → 一条 SQL + 一个数字。直接调 LLM 生成 SQL,执行,返回。不需要循环,不需要 agent。这是大部分企业数据分析的真实需求。
  2. 确定性管道。ETL、报表生成、固定流程审批。状态图或普通函数式管道更合适,agent 的不确定性反而是 bug 源。
  3. 高准确率要求。财务、合规。agent 循环每多一轮,不确定性乘一次。能一轮搞定就别两轮。

该用的场景

  1. 探索式分析。"找出去年所有异常报销并分析原因"——模型需要先查列表,再看明细,可能要交叉对比,路径不可预知。
  2. 多文件代码改动。pi 的本职工作。
  3. 需要工具组合的任务。任务分解方式取决于中间结果。

判断标准:如果你能在写代码前画出完整流程图,就别用 agent。如果流程图要靠模型在运行时画,才用 agent。


六、选型建议:写自己的,还是用框架?

我的建议可能反直觉:生产 agent,先自己写循环骨架

理由:

  1. 核心循环就 100 行。你一天能写完,并且完全理解。框架的"agent"抽象通常是黑盒,出问题调试地狱。
  2. 框架的价值在循环之外:工具生态、向量检索、记忆管理、监控。这些你可以按需挑组件,不必整个框架上身。
  3. 你需要的钩子点因业务而异。pi 的 8 个钩子是为 coding agent 设计的,你的业务可能需要完全不同的钩子(比如"每次工具调用后写审计日志到 PostgreSQL")。自己写循环,钩子点你自己定。

何时考虑框架

  • 团队没人懂 LLM,需要快速起步 → 用框架的 cookbook
  • 任务确实是确定性多步管道 → LangGraph
  • 需要多 agent 协作(这个场景其实比想象中少)→ AutoGen/CrewAI

但即便用框架,也请先花半天手写一个 100 行的循环。理解了本质,你才知道框架在哪帮你、在哪害你。


七、延伸阅读

如果你想看工业级的 agent 循环实现,按这个顺序读 pi 的源码(每个文件都不长):

  1. packages/agent/src/agent-loop.ts — 核心循环,必读,全部精华
  2. packages/agent/src/types.ts — 看钩子点和事件类型怎么定义
  3. packages/agent/src/agent.ts — 看怎么把无状态循环包成有状态的 Agent
  4. packages/ai/src/api-registry.ts — 看多 provider 怎么用注册制抽象

读完这四个文件,你对"agent 编排"的理解会超过 90% 喊概念的人。


结语

agent 不神秘。它是一个 while 循环,状态是消息历史,决策靠 LLM,动作靠工具,扩展靠钩子。所有花哨的框架名词——节点、边、图、状态机、supervisor、planner——剥到底,要么是这个循环的包装,要么是可被这个循环替代的过度设计。

把循环吃透。剩下的工程问题(语义层、权限、监控、成本)才是你真正该花时间的。


文中代码为示意,省略了错误处理、并发控制等生产细节。完整生产实现参考 pi 项目 packages/agent/src/agent-loop.ts

©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

相关阅读更多精彩内容

友情链接更多精彩内容