写给开发者的"祛魅"指南。读完你会知道:所谓 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/end、turn_start/end、message_start/update/end、tool_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 循环不是银弹。
不该用的场景:
- 单步问答。"上个月销售额多少" → 一条 SQL + 一个数字。直接调 LLM 生成 SQL,执行,返回。不需要循环,不需要 agent。这是大部分企业数据分析的真实需求。
- 确定性管道。ETL、报表生成、固定流程审批。状态图或普通函数式管道更合适,agent 的不确定性反而是 bug 源。
- 高准确率要求。财务、合规。agent 循环每多一轮,不确定性乘一次。能一轮搞定就别两轮。
该用的场景:
- 探索式分析。"找出去年所有异常报销并分析原因"——模型需要先查列表,再看明细,可能要交叉对比,路径不可预知。
- 多文件代码改动。pi 的本职工作。
- 需要工具组合的任务。任务分解方式取决于中间结果。
判断标准:如果你能在写代码前画出完整流程图,就别用 agent。如果流程图要靠模型在运行时画,才用 agent。
六、选型建议:写自己的,还是用框架?
我的建议可能反直觉:生产 agent,先自己写循环骨架。
理由:
- 核心循环就 100 行。你一天能写完,并且完全理解。框架的"agent"抽象通常是黑盒,出问题调试地狱。
- 框架的价值在循环之外:工具生态、向量检索、记忆管理、监控。这些你可以按需挑组件,不必整个框架上身。
- 你需要的钩子点因业务而异。pi 的 8 个钩子是为 coding agent 设计的,你的业务可能需要完全不同的钩子(比如"每次工具调用后写审计日志到 PostgreSQL")。自己写循环,钩子点你自己定。
何时考虑框架:
- 团队没人懂 LLM,需要快速起步 → 用框架的 cookbook
- 任务确实是确定性多步管道 → LangGraph
- 需要多 agent 协作(这个场景其实比想象中少)→ AutoGen/CrewAI
但即便用框架,也请先花半天手写一个 100 行的循环。理解了本质,你才知道框架在哪帮你、在哪害你。
七、延伸阅读
如果你想看工业级的 agent 循环实现,按这个顺序读 pi 的源码(每个文件都不长):
-
packages/agent/src/agent-loop.ts— 核心循环,必读,全部精华 -
packages/agent/src/types.ts— 看钩子点和事件类型怎么定义 -
packages/agent/src/agent.ts— 看怎么把无状态循环包成有状态的Agent类 -
packages/ai/src/api-registry.ts— 看多 provider 怎么用注册制抽象
读完这四个文件,你对"agent 编排"的理解会超过 90% 喊概念的人。
结语
agent 不神秘。它是一个 while 循环,状态是消息历史,决策靠 LLM,动作靠工具,扩展靠钩子。所有花哨的框架名词——节点、边、图、状态机、supervisor、planner——剥到底,要么是这个循环的包装,要么是可被这个循环替代的过度设计。
把循环吃透。剩下的工程问题(语义层、权限、监控、成本)才是你真正该花时间的。
文中代码为示意,省略了错误处理、并发控制等生产细节。完整生产实现参考 pi 项目 packages/agent/src/agent-loop.ts。