这两年做 Agent 应用的人越来越多,但大多数团队卡住的不是模型本身,而是让 Agent 稳定地跑起来。模型会自己决定调用哪个工具、观察结果、调整计划,再决定下一步,这一整套“感知—决策—行动”的循环就是 Agent 编排;而承载这套循环、状态、上下文和容错机制的底座,就是 Agent Runtime。这篇文章把我自己在设计和落地 Runtime 过程中的底层思考、工程取舍、踩坑记录完整写出来,适合正在做智能客服、自动化运维、数据分析助手这类项目的工程师,也适合那些被“Agent 老是掉链子”折磨得头疼的团队。
1. 先搞清楚:Agent 编排到底在编排什么
1.1 Agent 与传统 LLM 应用的分水岭
传统 LLM 应用其实不复杂。用户问一句,模型答一句,顶多中间穿插几次固定的 API 调用,比如查个天气、检索一下知识库,然后根据结果生成回答。这种模式本质上还是“请求—响应”,应用逻辑是写死的:先调用什么、再调用什么,全部由代码决定。
Agent 应用就不一样了,决策权被交给了模型。模型拿到用户任务后,自己判断要先做什么:是查数据库,还是调用搜索,还是直接回答。每执行完一步,它看到结果后再决定下一步。这个过程可能经历 3 次、5 次甚至 10 次工具调用,直到模型认为任务已经完成,才输出最终答案。
所以编排这个词,核心不是“把步骤串起来”,而是回答四个问题:循环由谁驱动、何时停止、状态放在哪里、步骤之间如何衔接。这四个问题只要有任何一个没想清楚,Agent 就会表现出典型的“失控感”:反复调用同一个工具、上下文爆炸、跑着跑着彻底跑偏、甚至陷入死循环。
我见过不少团队直接写一段 while 循环,让模型不断调用工具,直到它自己说“完成”。这种原型实现确实最快,但它完全没有边界:没有状态管理,没有错误恢复,没有预算控制,也没有可观测性。Demo 能跑,一到生产就露馅。真正的产物其实是一套能承载循环、状态、容错和恢复机制的 Runtime,而不是几行 Python 代码。
1.2 Runtime 到底要承担哪些职责
如果拿操作系统做类比,“业务应用”是 Agent 的 Prompt 和工具逻辑,Agent Runtime 就是内核。它负责在最底层把模型调用、工具执行、状态流转、异常处理统一接管,让上层的业务代码只需要专注“任务怎么做”,不用操心“循环怎么转”。
一套合格的 Runtime 至少要承担六件事:
- 调度与循环控制:维护 Agent 的推理循环,决定模型输出之后是继续调用工具、直接结束,还是进入人工兜底。
- 状态保存与恢复:把当前对话上下文、已执行的步骤、中间结果完整保存,出现故障时可以续跑。
- 工具协议与执行:统一处理工具的参数校验、超时、重试、异常返回,让模型获得高质量的结果反馈。
- 上下文与记忆管理:控制输入给模型的消息数量,压缩、摘要、裁剪,保证不超出窗口限制。
- 可观测性与审计:记录每一步的输入输出、耗时、Token 消耗,事后能复盘 Agent 为什么这么干。
- 安全与预算控制:限制工具权限、熔断异常、控制成本和步数,防止 Agent 跑飞到不可收拾。
这六个职责里,前三个解决“能不能跑起来”的问题,后三个解决“能不能跑得久、跑得稳”的问题。很多自研 Runtime 只解决了前三件事,后三件等到线上出了事故再补,实际上代价会高得多。
2. 底层原理:编排引擎的四个核心零件
2.1 状态机:Runtime 的心脏
Agent 的执行过程天然可以建模成一个有限状态机。这不是理论上的洁癖,而是工程上的必需。状态机让运行时可以明确知道“当前在哪个阶段,下一步能做什么”,而不是让循环逻辑散落在业务代码的各个角落里。
最基础的状态集合只需要五个:
- IDLE(空闲):接到新任务时的初始状态。
- RUNNING(推理中):正在调用模型等待决策。
- TOOL_CALLING(工具调用中):已收到模型要调用工具的指令,正在执行工具。
- WAITING_RESULT(等待结果中):工具执行完成或超时,准备把结果回传模型。
- DONE / ERROR(完成 / 异常终止):任务输出最终结果,或进入人工兜底。
状态迁移的事件就是模型输出和工具执行结果。写成伪代码大概是这个意思:
class StateMachine: def __init__(self): self.state = "IDLE" def on_model_responded(self, response): if self.state != "RUNNING": raise InvalidTransitionError() if response.has_tool_calls: self.state = "TOOL_CALLING" else: self.state = "DONE" def on_tool_completed(self, result): if self.state != "TOOL_CALLING": raise InvalidTransitionError() self.state = "WAITING_RESULT" def on_result_observed(self, message): self.state = "RUNNING"为什么一定要用状态机而不是随便一个 while 循环?因为真实场景中,一次 Agent 任务可能持续几十秒甚至几分钟。期间可能发生网络超时、服务重启、用户中途催促,甚至另一个操作员要介入干预。只有状态被显式建模,系统才能做到“随时暂停、随时恢复、随时审计”。
另一个好处是打点。每次状态切换都是一个天然的埋点位置:进入 RUNNING 时记录模型请求,进入 TOOL_CALLING 时记录工具参数,进入 DONE 时记录最终答案。这些日志就是后续做回放测试和故障排查的基础。
2.2 上下文管理:给模型看什么,不看什么
模型输入窗口是有限的。你不可能把整个任务过程中的所有工具返回都塞给模型,让它每次都完整“重看”一遍。上下文管理的本质,是决定哪些信息必须让模型看到,哪些信息可以用摘要代替,哪些信息干脆不出现。
如果不做任何管理,Agent 跑 4 到 5 次工具调用,输入 token 就很容易翻几倍。尤其是工具返回了金融行情表、数据库查询结果、日志片段这些“含金量低但长度大”的内容,窗口被占满后,模型会用得很吃力,响应质量明显下降,成本也直线上涨。
我常用的上下文管理策略是三层结构:
- 新鲜层:最近两轮到三轮的完整消息,包括工具返回原文,因为模型当下做决策最依赖这几步。
- 压缩层:更早的步骤不保留完整内容,而是由运行时或模型生成一段摘要,比如“已完成用户身份校验,确认订单存在,正在查询物流状态”。
- 记忆层:从整个会话中提取出来的关键事实,比如用户的偏好、问题的前提条件、已经确认过的参数,存在独立的记忆存储中,每次重放进消息列表。
这三层配合起来,模型看到的是一个“该知道的全知道、不该知道的绝不多占地方”的上下文窗口。我遇到过一些团队想省事,把所有历史一股脑丢进去,结果每轮请求成本高得吓人,最后我只能建议他们先做一次消息中间层的裁剪和摘要,成本能降一半不止。
另外还有一个容易被忽视的问题:工具返回之后,模型可能自己产生下一步的中间思考。这些“模型思考”要不要继续放进上下文?如果只是做即时决策,尽量只保留必要的工具结果和上一轮决策,不要保留模型的长篇推理链,否则很快又会溢出。
2.3 工具调用协议:Agent 与世界的连接器
工具是 Agent 唯一能触达外部世界的手段,所以 Runtime 对工具协议的设计直接决定了 Agent 的成功率。工具调用不是简单“执行一个函数”,它涉及参数校验、异常处理、结果标准化和超时兜底四个环节。
这里最需要注意的是“结果标准化”。模型决策依赖工具返回,但这个返回不是给用户看的,是给模型看的。所以工具的返回内容应该结构化、语义化,而不是直接把原始数据一股脑返回。一个设计良好的工具返回格式大致长这样:
{ "status": "ok", "summary": "查询到 3 条符合条件的数据,最新一条在 3 分钟前更新", "data": "此处是模型进一步决策所需的必要字段" }status 直接告诉模型这一步成功还是失败;summary 用几句话概括结果全貌;data 才放具体内容。如果 status 是 error,还要带上 error_type 和 possible_cause,模型才能判断下一步怎么办,而不是看到一个 Python 异常堆栈后不知所措。
这里引入一个“失败反馈闭环”的原则:工具调用失败,不要只是把错误信息回传模型,然后让模型重新试一次。更有效的做法是把错误标准化成一个“可行动描述”,比如“库存查询失败,原因:接口超时,已完成一次自动重试,请考虑改用另一个仓库查询接口”。模型看到的是一个可推理的初始状态,而不是一个需要猜的异常。
另外要给每个工具的返回值做大小限制。任何工具的输出都不能无限制地塞进上下文,超出指定阈值(我一般限制在 2000 到 3000 个字符)就截断,并显式提示模型“数据量过大,已截断,请根据摘要信息继续”。这样既保护上下文,也让模型知道它看到的不是完整数据,必要时可以去查询更细的接口。
2.4 记忆与长期状态:让 Agent 记住该记住的
上下文管理管的是“当前窗口里有哪些信息”,记忆管的是“让 Agent 跨会话、跨轮次记住关键事实”。这两件事经常被混为一谈,但它们是不同层次的问题。
简单来说,短期记忆就是当前任务进行中的状态和中间结果,可以理解为工作记忆,这部分通常直接存在会话上下文里,不需要额外存储。长期记忆则是用户偏好、历史结论、已经确认的约束条件,这些信息不该每次都让模型重新推导,而是通过检索注入到上下文中去。
我在工程实践里会把长期记忆分成两种形态:
- 显式记忆:结构化的键值对或者配置,比如“用户所在城市=上海”“订单号=2025061101”,在上下文中直接以事实列表形式注入。
- 语义记忆:一段带向量索引的文本记录,比如用户曾经提到过某个偏好,下次任务开始时通过向量召回相关段落。
很多 Runtime 把所有历史对话都塞进向量库,然后每次任务开始前做一次召回,这当然可以,效果也尚可。但我认为语义记忆适合“经验沉淀”,不适合“关键状态”。关键状态应该做到确定性存储,而不是靠相似度召回,否则一旦召回失败,Agent 就表现得很“健忘”。
这块设计核心原则是:运行时只在真正需要的时候才做记忆检索,不要在每个循环步骤都查一遍记忆库。检索本身有耗时、有成本,而且召回结果会分散模型的注意力。代价更低的做法是开场注入一次,任务中途除非明确需要,否则不再动记忆库。
3. 工程落地:怎么设计一套能上生产的 Agent Runtime
3.1 先定编排模式:链式、图式、还是自主循环
很多人在设计 Runtime 之前先纠结技术选型,但恰恰忘了最重要的事——想清楚你的业务需要哪种编排模式。市场上常见的编排框架基本可以归为三类,它们的控制权、可预测性和灵活性完全不同。
| 编排模式 | 控制方式 | 适用场景 | 主要风险 |
|---|---|---|---|
| 链式编排 | 代码写死步骤顺序,每一步固定下一步 | 流程高度稳定的内部系统,比如订单信息补全 | 灵活性差,无法应对业务分支 |
| 图式编排 | 节点和边组成 DAG,条件路由决定下一步 | 大部分有明确流程边界但存在分支的面向业务 Agent | 需要投入更多设计成本,且流程无法覆盖所有真实情况 |
| 自主循环 | 模型完全自由决策下一步动作 | 探索性任务、开放性问题,比如竞品调研、方案分析 | 不可控,成本高,容易死循环或跑偏 |
我给团队的建议是:业务确定性越强,越要收敛到图式编排;探索性越强的任务,才开放自主循环。换句话说,能用图解决的问题不要轻易放开成自主循环。生产环境里,可控性往往比灵活性更重要。
图式编排看起来复杂,但它带来的好处是决定性的:可预测。你可以在图的每个节点上做拦截、加超时、做审计。用户问“Agent 现在为什么卡住了”,你能直接告诉他当前走到哪个节点了,下一步有哪些分支。
自主循环也有它的位置。比如你让 Agent 自己写方案、查资料、做对比,这个任务本身就没有固定流程,不放开模型就发挥不了价值。但就算是自主循环,我也建议用图式编排中的“沙箱节点”去包裹:明确这个阶段模型可以自由行动,但行动范围、可用工具、最大步数都提前限制住。
3.2 五个工程能力,缺一个都难上线
状态持久化、可观测性、流控与预算、安全隔离、回放测试,这五个能力是我认为一套 Runtime 能不能上生产的分水岭。任何一个缺失,都会在线上以事故的形式教训你。
第一,状态持久化。Agent 的每一个步骤都应该能落地到存储里,不只是最终结果,而是中间每一步的消息、工具调用、状态转移。这样做有三个直接好处:一是任务中断后可以从最近的稳定点恢复;二是每一步都有审计记录;三是事后可以拿真实数据做分析和优化。
落库的粒度我一般建议以“Step”为单位。每个 Step 记录任务 ID、步骤序号、输入消息、模型输出、工具调用及结果、耗时、Token 消耗。存储本身选关系型数据库就够了,一条任务几十步,数据量并不大。
第二,可观测性。这里的重点不是指标监控,而是把每一步推理的来龙去脉串起来。要做到每一条日志都能回答五个问题:当前是哪个任务、正在执行哪一个步骤、模型看到什么内容、它做了什么决策、工具返回了什么结果。把这些问题串成一条 Trace,Agent 的行为就会变得完全透明。
第三,流控与预算。Agent 的最大特点是不确定性。一个任务可能 5 步收工,也可能 20 步还在原地打转。没有预算控制,一次线上失控就可以烧掉一大笔钱。所以 Runtime 必须内置硬性限制:最大步数、单次工具超时、整体任务超时、Token 总预算、成本上限。超过上限,一律强制终止并转人工。
第四,安全隔离。给 Agent 的工具权限要遵循最小化原则。工具注册表里能调用哪些接口、每个接口最多能操作多少数据,都要提前定好。敏感操作还必须二次确认,比如删除数据、发消息、转账,不能放给模型自主决策。可以加一个“人工审批”节点:模型决定要执行敏感操作时,Runtime 先暂停,等操作员确认后才放行。
第五,回放测试。后面单独开一节说,这是保证 Runtime 持续演进而不会倒退的关键能力。
3.3 一个轻量 Runtime 的最小可运行实现
讲了这么多原则,还是给一个能直接跑起来的骨架最简单。这里我用 Python 写一个最简 Runtime,不依赖任何外部框架,核心逻辑大概 100 行左右,足够演示状态机、工具调用、结果标准化和兜底限制是怎么组合在一起的。
import json import time from dataclasses import dataclass, field @dataclass class ToolResult: status: str # ok / error / truncated summary: str data: object = None class ToolRegistry: def __init__(self): self._tools = {} def register(self, name, description, schema, handler): self._tools[name] = { "description": description, "schema": schema, "handler": handler, } def get_schemas(self): return [ { "type": "function", "function": { "name": name, "description": meta["description"], "parameters": meta["schema"], }, } for name, meta in self._tools.items() ] def invoke(self, name, arguments, timeout=30): if name not in self._tools: return ToolResult("error", f"工具 {name} 不存在") handler = self._tools[name]["handler"] try: result = handler(**arguments) return ToolResult("ok", result["summary"], result.get("data")) except Exception as e: return ToolResult("error", f"工具调用失败: {str(e)}") class SimpleAgentRuntime: def __init__(self, model_fn, registry, system_prompt, max_steps=10, step_timeout=30, output_limit=2000): self.model_fn = model_fn self.registry = registry self.system_prompt = system_prompt self.max_steps = max_steps self.step_timeout = step_timeout self.output_limit = output_limit def run(self, user_input): messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": user_input}, ] for step in range(self.max_steps): response = self.model_fn( messages, tools=self.registry.get_schemas(), timeout=self.step_timeout, ) if not response.tool_calls: return response.content messages.append({"role": "assistant", "content": response.content, "tool_calls": response.tool_calls}) for call in response.tool_calls: result = self.registry.invoke(call["name"], json.loads(call["arguments"])) content = json.dumps({"status": result.status, "summary": result.summary}) # 对超长数据做截断保护 if result.data is not None: data_str = json.dumps(result.data) if len(data_str) > self.output_limit: content += " data_truncated=true summary=data超过限制,请根据summary继续" else: content += " " + data_str messages.append({"role": "tool", "tool_call_id": call["id"], "content": content}) raise RuntimeError("任务超过最大步数限制,与模型协商终止")这套骨架有几个关键设计点。第一,每个工具调用都会被 ToolRegistry 包裹,统一处理异常,不会让裸异常扩散到模型决策层。第二,工具输出通过 output_limit 强制截断,避免上下文膨胀。第三,max_steps 是硬边界,超过就终止,不指望模型自己“变聪明”。
实际的工程版 Runtime 会在每步落数据库、记录 Trace、加预算熔断,但核心骨架就是上面这个循环。把状态机、持久化、观测点往这个骨架上加,会非常顺。如果一开始就引入很重的框架,你会发现很多控制点都被框架藏起来了,反而不好做精细化治理。
4. 实测总结:常见故障与排查方法
4.1 模型反复调用同一个工具,怎么办
这是 Agent 应用里最常见的故障,没有之一。现象是模型对一个工具连续调用三五次,内容都差不多,然后输出一个勉强能看的答案。排查看上去是模型“笨”,实际上是工具返回让模型无法进入下一步。
最常见的原因有两个。第一,工具返回里没有明确 success / failure / need_more_info 这样的判断信号。模型看到一大堆业务数据,不知道查询到底成没成功,于是只能再查一次碰运气。第二,工具参数描述写得模糊,模型每次传的参数都差一点,导致结果永远不符合预期。
解决办法分两头。一头是工具返回必须标准化,第一条就告诉模型“这一步完成了,结论是什么”,而不是丢一堆原始数据。另一头是给工具描述加上“fail 时的建议”,比如“查询失败且已重试一次后,请直接尝试备用查询工具”。此外,Runtime 最好做一个检测,如果在连续 N 步内模型提交了相同名称和相同参数的调用,就自动打断,并补充一条系统消息提示“你已经在重复上一个动作,请换一种策略”。
4.2 上下文突然膨胀,Token 成本失控
这个问题通常出现在工具返回数据较大的场景。数据库查询返回几百行、搜索接口返回大段网页正文、代码执行工具返回完整 Log——每一条都会让下一轮模型的输入暴涨。
我见过一个极端案例,某团队做数据分析 Agent,模型每查一次数据库,就把完整结果塞进上下文,结果一个 8 步的任务,最后一步输入了将近 40 万 token。后来改成两步:数据库查询结果只回传前 500 字符 + 一个统计摘要,具体明细留在运行时内部,模型需要明细时再调一个专门的取数工具。成本立刻降了七成,完成率反而更稳定了。
写代码的时候要注意,截断数据不是只截掉尾部,而是要保留模型决策最需要的信息。一个列表查询,保留“总数、页码、关键字段分布”,远比保留前 500 条原始记录更有决策价值。如果这个摘要运行时做不出来,可以让模型先跑一个轻量总结工具,再决定下一步。
4.3 Agent 陷入死循环,怎么兜底
死循环的表现有很多种:模型反复猜测同一个问题、反复调用邮件工具发送同一封内容、或者一直在几个工具之间反复横跳。这类问题不解决,Agent 就谈不上可用。
最基础的兜底就是 max_steps。我给团队定的默认值是 10 步,超过直接终止。但这个兜底只是止损,不是治疗。真正要做的,是把循环“拆开”来看:模型为什么认为还需要下一步?它的判断依据是不是有问题?
一种典型的循环原因是“模型试图自我纠错但始终离正确答案差一步”。比如让它搜索一个答案,它搜了三次,每次都觉得少了点啥,又不知道少什么。这时候纠错方向是让工具返回更明确的“完整性判断”,比如搜索工具在返回结果末尾附加一句“该结果已经覆盖问题的主要维度”,模型看到这句就不再继续搜了。
另一种实用手段是连续动作检测。Runtime 记录最近几步的工具调用序列,如果发现模式完全重复,比如“A 工具 -> B 工具 -> A 工具”连续出现两轮,就主动中断,并提示模型“检测到你的动作模式没有进展,请重新解释当前状态,然后换一个策略”。这相当于在循环里插入一次计划修正,多数模型都能被拉回来。
4.4 会话中断与恢复:长任务的最后一道防线
Agent 任务有可能跑很久。如果一次任务执行到第 20 步,中间某个工具超时导致整个进程崩溃,没有状态持久化的 Runtime 就会让用户重新开始,这是不可接受的。
我的做法是事件溯源:每完成一步,就把这一步骤的所有信息打包成一个事件写入存储。事件内容包括步骤序号、消息列表上下文、工具调用记录、状态机状态、完成时间。如果进程崩溃,重启后扫描未完成任务列表,找到最后一条已提交事件,从对应步骤重新调用模型继续。
这个恢复方案有个细节:恢复时要把最后一条工具结果重新放进消息列表,否则模型会看到“assistant 发起了工具调用”,但看不到工具结果,容易再次发出重复的工具调用。恢复逻辑里一定要保证“发起调用”和“调用结果”成对出现,不能只恢复一半。
生产环境我还习惯在恢复前加一个检查:断点距离现在已经超过 X 分钟,中间数据可能已经过期,那就先请求模型重新评估一次上下文,再决定是否继续。这一步能避免很多因为中间数据状态变化导致的二次错误。
5. 如何评估一套 Runtime 好不好用
5.1 先把指标定下来,再谈优化
很多团队评估 Agent 只盯一个指标:最终成功率。我觉得这远远不够。成功率只能说明终点,说明不了过程。一个任务要是多花了三倍步数才成功,那它只是“勉强成功”,成本、延迟、用户体感都很差。
我建议至少跟踪五类指标:
- 任务完成率:最终产出用户可接受结果的比例。
- 平均步数与平均耗时:衡量 Agent 是否够“直给”。
- 单任务平均 Token 成本:直接关联预算。
- 工具调用失败率与重试率:反映工具协议设计质量。
- 人工介入率:多少任务需要中途转人工,转人工的环节集中在哪。
这五个指标一起看,才能定位问题。比如完成率高但平均步数很高,说明 Agent 能完成但不会收敛,问题大概率出在工具返回的决策信号上。完成率不高且介入率高,说明当前任务的工具链和流程设计可能有缺口,不是模型换个参数就能解决的。
Runtime 的优化不是让模型“变聪明”,而是让模型“更容易走对路”。工具结果的标准化、上下文摘要的质量、状态机的恢复策略,这些环节每优化一块,模型的表现就会跟着上一个台阶。
5.2 用回放测试给改动上保险
Runtime 是一个持续迭代的工程产物,不会写完就定型。你会改工具返回格式、改上下文裁剪策略、改超时逻辑,每一次改动都可能让原本能跑通的任务突然跑歪。这时就需要回放测试,保证改动是“正向优化”而不是“引入回归”。
回放测试的原理不复杂:从线上记录的真实任务中抽一批样本,把每一步的模型输入、模型输出、工具返回都录制下来。然后在新版本 Runtime 上重放这些样本,对比每一步的行为是否和旧版本一致。由于模型有随机性,重放测试的关键在于把模型输出也固定下来:用录制好的模型响应来代替真实模型调用,只测 Runtime 的逻辑变化。
工具返回也要做固定化处理。真实工具可能返回动态数据,回放时不能用真实数据,而是用录制下来的固定返回。这样测试的因变量只有一个:Runtime 本身的逻辑。如果新版重放结果和旧版完全一致,说明改动没有影响已有任务;如果行为变了,就要逐个分析是不是预期内的变化。
我在团队里把回放测试做成了一套自动化流程,每次改动后跑一遍全量回放,大概覆盖几百条真实任务样本。这套机制让我们对 Runtime 的改动一直很有底气,敢于持续重构,而不是每次改完都提心吊胆。
在我自己的实践里,Agent Runtime 最难的不是某一个点,而是把状态机、上下文管理、工具协议、预算控制这些能力咬合成一个整体。任何一个环节掉链子,Agent 都会以各种难以预料的姿势失败。但只要这套底座扎实了,上层加业务功能就会非常顺手。最后分享一个小技巧:给 Runtime 加一个“调试面板”,把每一步的 Token 数、步数、工具结果摘要实时显示出来,你第一天上线就能发现一堆优化机会。