很多团队把 Agent 从 Demo 推向生产环境时,都会卡在同一个坎上:模型能力好选,Agent 编排却很难做。Prompt 写得再漂亮,一接上工具调用,上下文混乱、循环失控、失败恢复困难、线上问题难排查这些问题就会集体爆发,而这些问题最集中的地方,就是我们常说的 Agent Runtime。
所谓 Agent Runtime,不是某个服务框架,而是承载 Agent 执行循环、上下文管理、工具调用、状态持久化、失败恢复这些通用能力的基础设施。它是 Agent 的"操作系统":模型负责思考,Runtime 负责让思考落地。这篇文章想从底层原理聊到工程落地,讲清楚一个真正好用的 Agent Runtime 应该怎么拆、怎么搭、怎么避坑。无论你是在做单 Agent 应用,还是已经在设计多 Agent 协作系统,这份内容都能给你一个可参考的框架。
- 先看清"编排"二字:Runtime 要兜住的不只是流程
1.1 编排的本质是管理不确定性,而不是画流程图
我见过不少团队一开始做 Agent 编排,第一反应是画一张漂亮的流程图:先调用 A 工具,再调用 B 工具,如果失败就调 C。这种思路在传统后端里没问题,但在 Agent 场景下会立刻翻车。
原因很简单:传统流程是确定的,每一步的输入输出都可预期;而 Agent 的每一步决策都由 LLM 现场生成,同样的用户输入,模型今天可能选择工具 A,明天可能选择工具 B,甚至可能连续三次都在瞎试。Runtime 要处理的不是"流程怎么走",而是"模型在下一次决策时手上有哪些信息、能用哪些工具、超出预算怎么办"。
我习惯把编排拆成五个维度:策略(谁来决策、按什么节奏决策)、状态(上下文和历史怎么保存)、能力(哪些工具对模型暴露)、失败(出错后如何恢复)、观测(每一步发生了什么可不可以回放)。真正好用的 Runtime,核心就是把这五个维度全部收拢成基础设施,让上层业务不用反复实现。
1.2 Runtime 和业务代码的边界怎么划
很多项目做到后面,Runtime 和业务逻辑会糊成一团。一个常见迹象是:Runtime 的代码里开始出现"如果用户是会员就调用这个工具"之类的判断。一旦出现,说明边界已经失守。
我推荐的划分方式很粗暴:
Runtime 负责 | 业务层负责 上下文窗口管理、历史裁剪与摘要 | Prompt 的措辞与角色设定 工具注册、参数校验、超时与熔断 | 工具内部的业务实现 执行主循环与终止条件控制 | 业务规则、审批流、结果呈现 状态持久化、断点恢复 | 数据权限与业务授权逻辑 通用事件记录与可观测性 | 面向用户的通知与话术
这样划分的底气来自一个原则:Runtime 解决的是"模型会失误"这件事,业务层解决的是"业务要求必须满足"这件事。两者混在一起,最直接的后果就是 Runtime 无法复用,换一个业务场景就得重写一套循环,测试也没法做。
1.3 为什么先做好单 Agent,再谈多 Agent
"Agent 编排"四个字很容易让人直接奔着多 Agent 去,仿佛角色越多越高级。我自己的实践结论恰恰相反:多 Agent 的稳定性上限取决于单 Agent 的 Runtime 下限。
单 Agent Runtime 至少要看住四件事:一个 Agent 实例的消息循环是否可靠、工具调用失败能否自愈、状态能否持久化恢复、日志能否定位到具体某一步。多 Agent 协作本质上是在这四件事之上再做消息路由、任务分发和结果汇总。如果单 Agent 自身循环都经常死锁,多 Agent 只会把错误放大成一个更乱的故事。
所以我后面讲的每个模块,都是先围绕"一个 Agent 实例"展开。把这个底座做扎实,再往上叠加多 Agent 编排,就是水到渠成的事。
- 运行时主循环:上下文、工具调用与终止条件的设计
2.1 三种主循环结构:从消息循环到计划执行
Runtime 的心脏是一个主循环。我梳理过三种主流结构,各自适合不同场景:
循环结构 | 基本思路 | 优点 | 主要风险 消息循环 | 把 LLM 调用和工具执行交替进行,直到模型不再请求工具 | 实现最简单、逻辑最直接 | 步骤感弱,复杂任务容易迷失 推理行动循环(ReAct) | 模型先输出思考过程,再执行工具,观察结果后继续 | 适合需要推理的任务,过程可解释 | token 消耗大,思考过程可能绕圈子 计划后执行(Plan-then-Execute) | 模型先生成完整计划,再逐项执行计划 | 适合长流程任务,整体路径清晰 | 计划一旦偏离现实,后续步骤容易失效
我建议大多数团队从消息循环起步,而不是一上来就上复杂的图编排。消息循环长这样:
async def run_agent(request_id, messages, max_steps=10): for step in range(max_steps): response = await llm.chat(messages, tools=tool_schemas) if not response.tool_calls: return response.text for call in response.tool_calls: result = await execute_tool(call) messages.append(tool_message(call.id, serialize(result))) raise StepLimitExceeded(request_id)这段逻辑简单到让人怀疑:这也能叫 Runtime?能。它把最核心的不确定性——"模型下一轮到底还要不要调用工具"——用循环兜住了。真正复杂的是循环内部每个环节的可靠性,而不是循环本身。
2.2 上下文管理的组合拳:裁剪、摘要与动态槽位
上下文窗口是有限资源,尤其工具调用结果经常很长。一条 SQL 查询可能返回 5000 行,直接把原始结果塞进 messages,几轮下来 token 就爆了。我的做法是三层组合拳。
第一层是滑动窗口裁剪:只保留最近 N 条消息。这个 N 要按场景调,我建议从 20 条起步。第二层是摘要压缩:把较早的消息用一次额外的 LLM 调用压缩成一段摘要,放进系统提示词。第三层是动态槽位:把用户目标、已确认约束、当前任务阶段这些"关键信息"单独维护,每轮刷新,不依赖历史消息推导。
一个简化的上下文组装函数:
def build_context(static_prompt, summary, slots, messages, max_messages=20): recent = messages[-max_messages:] slot_text = ";".join(f"{k}:{v}" for k, v in slots.items()) system_content = f"{static_prompt}\n历史摘要:{summary}\n关键信息:{slot_text}" return [{"role": "system", "content": system_content}] + recent这里有个容易踩的坑:裁剪不是简单丢消息。如果裁掉了一条工具调用结果,而模型后续还要参考它,就会凭空"失忆"。所以裁剪触发时,最好把被裁掉的工具结果做一次摘要归入历史摘要,并明确告诉模型"早期的细节可能被压缩了,必要时请通过工具重新获取"。
2.3 工具协议设计:一次失败的调用代价有多大
工具调用是 Agent 对外部世界的触手,协议设计直接决定 Runtime 的容错能力。一个合格的工具定义至少包含:名称、描述、参数 JSON Schema、超时时间、结果大小上限、是否幂等。描述写得好不好,直接决定模型能不能正确选工具,这比大多数人的直觉更重要。
参数校验必须放在 Runtime 而不是工具内部。模型返回的 tool_call 里的 arguments 只是一个字符串,必须先做 JSON 解析,再用 JSON Schema 校验。我的经验是:模型填错参数太常见了,要么缺字段,要么枚举值不在范围内,要么干脆返回的不是合法 JSON。Runtime 要在执行前把错误接住,而不是把一串解析异常抛给工具。
工具返回结果也要统一信封,不要裸返回字符串:
{"ok": true, "data": {...}, "truncated": false} {"ok": false, "error": {"code": "tool_timeout", "message": "执行超时,请简化查询条件"}}统一信封的价值在失败时体现得最明显:工具执行失败后,Runtime 可以把错误信息作为 tool 消息回灌给模型,让模型自己修正。但错误信息不能太长,我控制在 300 字符以内,只给错误码和一句人话提示,不给完整 traceback。否则模型会被细节带偏,还可能把内部异常信息当成"指令"来执行——这里的水很深,后面说安全边界的时候再展开。
- 编排引擎的三个硬骨头:状态机、失败恢复与并发隔离
3.1 用显式状态机替代散落的 if-else
Agent 执行过程中,"状态"不只是消息列表,还包括当前处于哪个阶段、已经尝试了多少次、还剩下多少预算。很多 Runtime 代码把这些信息散落在循环里的局部变量和 if-else 里,导致两个问题:没有办法从崩溃现场恢复;没有办法回答"这个 Agent 现在到底进行到哪一步了"。
我的做法是引入显式状态机。哪怕只有六个状态,也比没有好:
class AgentStage(Enum): RECEIVED = 1 # 收到用户请求 PLANNING = 2 # 模型规划下一步 EXECUTING = 3 # 执行工具调用 VERIFYING = 4 # 校验执行结果 DONE = 5 # 正常结束 FAILED = 6 # 达到终止条件,失败退出状态转移规则用一张表写清楚:RECEIVED 收到消息后进入 PLANNING;PLANNING 输出 tool_call 则进入 EXECUTING,输出最终答案则进入 DONE;EXECUTING 拿到工具结果回到 PLANNING;任何阶段如果步数超限、token 超限或连续失败达到阈值,进入 FAILED。
显式状态机带来的直接好处有三个:可测试(每个转移都能写单测)、可恢复(持久化当前状态后,可以从任意节点重启)、可观测(状态字段直接进日志,排查问题不用猜)。
3.2 三层失败恢复:模型调用、工具执行与业务流程
真正好用的 Runtime 不会把失败恢复压在同一层处理。我分成三层,每层只管自己的事。
第一层是模型调用层。LLM 接口超时、限流、返回非法 JSON,这些问题在这一层解决。超时用指数退避重试,连续三次失败直接把这个模型调用标记为失败。最容易被忽略的是:模型返回的 JSON 解析失败时,不要直接中断,把"返回内容无法解析,请重新生成工具调用参数"作为一条系统反馈消息放回去,让模型自己纠正。
第二层是工具执行层。工具超时、抛异常、返回超大结果,都在这层处理。核心原则是把失败变成一个标准化的 tool 消息返回给模型,允许模型换参数或换工具重试。但要有兜底:同一个工具连续失败三次,就不再让它无限试了,转交人工或直接终止。
async def safe_execute_tool(name, args, max_attempts=3): for attempt in range(max_attempts): try: result = await call_tool(name, args) return normalize_success(result) except ToolTimeout: args = suggest_simplified_args(args) except ToolRejected as e: return normalize_error("tool_rejected", str(e)[:300]) return normalize_error("tool_failed", "工具连续失败,已中止")第三层是业务流程层。这一层做的是 checkpoint 和降级。每隔一定步骤把消息列表和状态机阶段持久化一次,进程挂了可以从最近 checkpoint 恢复。如果是多 Agent 流程,还可以在这里做降级:复杂编排失败后,退回单 Agent 模式,用"集中决策"把任务硬啃下来,至少不让用户面对一次白屏。
3.3 并发控制:状态隔离、限流与一致性问题
Agent 一旦上线,面对的一定是高并发。我见过最典型的并发事故是:把消息列表放在 Agent 类的实例属性里,每个请求 new 一个 Agent,看着没问题,但异步代码里如果共享了某个 session 连接或全局缓存,还是会互相污染。
第一条铁律:一个 request_id 对应一个独立的状态容器,所有消息、槽位、状态机的数据都不能放全局变量。状态容器可以放内存、Redis 或数据库,但读写必须原子。第二条铁律:LLM 调用和工具调用都要限流。LLM 用令牌桶限流,工具按每个工具的独立配额限流。
sem = asyncio.Semaphore(8) async def call_llm_with_limit(messages, tools=None): async with sem: return await llm.chat(messages, tools=tools)第三条经验谈的是"先写后读"的一致性:如果状态存储在 Redis 里,工具执行完把结果写回 Redis,紧接着模型又要读取最新状态,这时候异步框架的调度时机很微妙。解决办法很简单:同一请求的所有状态读写都串行化——要么都走同一个任务队列,要么用 per-request 的锁,不要依赖"写完了后面总能看到"这种直觉。
- 工程落地最隐蔽的三个坑:可观测性、安全边界与测试策略
4.1 可观测性:不只是日志,而是可回放的事件流
Agent 应用的排错难度远超普通接口,因为同一个用户问题可能触发 10 次模型调用、5 次工具调用。如果只在每个模块里打几行日志,出了线上问题根本拼不出完整画面。
我建议从第一天就把 Runtime 设计成"事件流"模式:每一步都产生一个结构化事件,统一写入日志或消息队列。事件至少包括:LLM 调用、工具调用、工具结果、状态转移、重试、上下文裁剪、错误。统一格式长这样:
{ "run_id": "8f3a...", "seq": 12, "ts": "2025-01-01T10:00:00.123Z", "event": "tool_call", "payload": { "tool": "query_order", "args": {"order_id": "A10086"}, "duration_ms": 230 } }有了 run_id 和 seq,排查问题时可以把一个会话的事件全部捞出来,按顺序回放,等于看了一遍 Agent 的完整思考和执行过程。成本方面我的经验是:全量记录 LLM 响应可能很贵,可以做一个采样开关,默认对生产环境按 10% 采样记录完整 messages,其余只记录 token 数和耗时。
4.2 安全边界:别让工具成为后门
Agent 的安全问题比普通后端更棘手,因为工具调用参数是由模型生成的,而模型可能被 prompt 注入。工具结果里如果混入了恶意指令,模型还很可能照做。
我的做法是三层防线。第一层:每个工具声明所需权限,Runtime 在执行前检查当前会话是否具有该权限,而不是让工具内部自己判断。第二层:所有工具参数在进入业务逻辑前,必须重新经过 Runtime 的校验和过滤,比如隐藏的敏感字段、危险路径、越权 ID 都不能放行。第三层:工具返回的数据要脱敏,给模型看到必要字段即可,不要把所有数据库字段原样返回。
还有一个细节很多人会忽略:工具返回的内容只是"数据",不应该被当作系统指令。最稳妥的写法是,把工具返回放在 user 消息或独立的 tool 消息里,绝对不要拼接到 system prompt 里。这样即便工具结果里被注入了恶意指令,也不会获得系统级优先级。
4.3 替换式测试:把 LLM 从单元测试里拿掉
直接用真实 LLM 做回归测试,既贵又不稳定,模型一升级测试结果就变红。我强烈建议做"替换式测试":把 LLM Provider 替换成录制回放的 Mock,让 Runtime 测试完全确定性。
实现思路不复杂。先把真实运行中的 LLM 请求和响应录制为一份 cassette,类似接口测试里的 fixture。然后在测试环境里让 Runtime 走相同的代码路径,但 LLM Provider 返回录制的响应。这样每次运行的结果都是确定的,可以准确断言:Runtime 是否发起了预期的工具调用、状态机是否正确转移、最终输出是否符合预期。
class ReplayProvider: def __init__(self, cassette): self.cassette = cassette self.idx = 0 async def chat(self, messages, tools=None): item = self.cassette[self.idx] self.idx += 1 return item["response"]除了回放正常流程,对抗样本也要准备:模型返回非法 JSON、连续请求同一个工具、工具返回超大结果、LLM 接口超时、突然的并发高峰。这些用例的价值不是测模型,而是测 Runtime 的护栏是否真的在干活。我自己的经验是:这批对抗用例救过的线上事故,比业务功能用例多一倍。
- 从零搭一个最小 Agent Runtime:模块划分与落地顺序
5.1 八个模块,职责一次说清
如果从零开始搭,我建议按八个模块来划分,避免所有逻辑塞进一个大循环里。
模块 | 职责 | 关键点 LLMProvider | 统一封装模型调用 | 支持重试、超时、流式 MessageStore | 消息与状态的持久化 | 按 run_id 隔离 ToolRegistry | 工具注册与 Schema 管理 | 集中维护工具元数据 ContextBuilder | 上下文组装与裁剪 | 动态槽位、摘要、滑动窗口 AgentLoop | 主循环控制 | 终止条件、步数限制 StateManager | 状态机转移与校验 | 可测试、可恢复 Checkpointer | 快照保存与恢复 | 定期持久化阶段 EventBus | 事件发布与记录 | 所有模块统一上报
这八个模块合起来就是一个最小的 Runtime。模块之间只通过接口调用,不直接共享内部状态。
5.2 我推荐的落地顺序与验收标准
我见过不少人一上来就想把所有模块都做完,结果两个月后还在内部讨论抽象设计。我的建议是分四步走,每步都有明确验收标准。
第一步:跑通消息循环,注册三个以内的工具,不做持久化。验收标准:模型能连续调用两个工具后给出最终答案,中途某个工具报错可以被模型自我纠正。
第二步:接入 StateManager 和 Checkpointer。验收标准:模拟进程被杀后,能从最近 checkpoint 恢复执行,用户不需要重新描述需求。
第三步:接入 EventBus,做结构化事件流。验收标准:通过 run_id 可以完整回放一次会话的模型调用、工具调用和状态转移。
第四步:做安全校验和替换式测试。验收标准:越权工具调用被拒绝,回放测试和对抗用例全部通过,可以在 CI 里稳定运行。
这套顺序的核心逻辑是:先让"能用"跑通,再让"可维护"跑通,最后让"敢上生产"跑通。
5.3 哪些事现阶段千万别做
最后聊几个我踩过跟头的"别做"清单。
别急着做多 Agent。单 Agent 连状态恢复都做不到稳,多 Agent 只会把问题放大到更难看。别在 Runtime 里堆业务规则。一旦你在主循环里写了"如果订单金额大于 X 就跳过工具调用"这种逻辑,Runtime 就失去了通用性,换一个场景全部重写。也别追求全自动编排。真正好用的 Runtime 会给人留下干预通道:关键决策点可以让用户确认,失败多次可以让步去人工,而不是让模型无限循环。
最重要的一个体会是:Runtime 设计的最高优先级是"体面地失败",而不是"每次都成功"。模型不可能每次都对,工具不可能每次都不超时,网络不可能每次都稳定。一个能让失败变成可理解信息、而不是一场灾难的 Runtime,才是我理解的真正好用的 Agent Runtime。