把一个 Agent Demo 跑通,大概四十行代码就够了;把它放进生产环境连续跑三个月不出事,是完全另一个量级的事情。我第一次把这类项目推上线时,本地测试全部通过,灰度第二天就被业务方叫停——模型在某个边界场景下反复调用同一个查询工具,一轮对话烧掉了平时两百倍的工具调用量。从那次之后我才真正理解:开源 Agent 框架的价值不在于帮你写出一个会思考的循环,而在于它把工具调用、上下文管理、状态流转这些脏活抽成了可替换的零件。这篇内容聊的就是这件事——一套头部互联网公司开源的 Agent 框架,从它的运行原理讲到生产环境落地时需要补的那些坑。适合已经能跑通 Demo、正准备往真实业务里塞的开发者,也适合正在评估"要不要自己造轮子"的技术负责人。下面所有代码里的framework都是代称,替换成你实际使用的包名即可,思路是通用的。
1. 先把边界划清楚:框架替你做了什么,不做什么
1.1 从"四十行 Demo"到"线上三个月"的断崖
很多人对 Agent 的第一印象来自演示视频:给个目标,模型自己规划、自己调工具、自己纠正,最后吐出结果。这套流程在受控环境里确实优雅,但真实业务场景会同时施加三重压力——输入是不可控的自然语言,工具有副作用不能重试,用户对延迟的容忍度以秒计。演示环境里这三条都不存在,所以 Demo 和上线之间的落差不是"优化一下",而是"补一整套工程设施"。
我用过一个很朴素的判断标准:如果这个 Agent 出错了,你能不能在一分钟内定位到是哪一步、哪个工具、哪段上下文导致的。Demo 阶段没人问这个问题,生产阶段这是第一个被问的问题。开源框架通常在前半段做得很好(循环、工具注册、消息拼装),后半段留白很多(追踪、评测、限流、成本核算)。认清这个分界线,后面所有的投入才有方向。
1.2 一次完整的 Agent 循环里到底发生了什么
抛开营销话术,主流 Agent 框架的内核就是一个带状态的 while 循环,每一轮大致经过五个阶段:
- 上下文组装:把系统提示、历史消息、记忆片段、可用工具的描述拼成一个请求体。
- 模型推理:模型输出要么是自然语言答案(结束),要么是一个结构化的工具调用意图(继续)。
- 意图解析与校验:框架把模型输出解析成工具名 + 参数,这一步是解析失败率最高的地方。
- 工具执行:真正跑代码、查数据库、调外部接口,拿到结果。
- 结果回填:把工具返回值塞回消息列表,进入下一轮。
理解这五步的意义在于:每一轮都在重新消耗一次完整的上下文。假设系统提示加工具描述占 2000 token,历史对话累积到 6000 token,那么一个需要 8 轮才能完成的任务,实际输入 token 消耗接近 6.4 万。这个数字直接决定了你的成本和延迟上限,也是后面所有裁剪策略的出发点。
1.3 哪些事框架不管,必须你自己补
我把生产落地必须自建的部分整理成了一张表,你可以对照检查自己的项目缺哪块:
| 能力 | 框架是否自带 | 缺失后的典型症状 |
|---|---|---|
| 循环控制与最大轮次 | 通常自带 | 无限循环、费用失控 |
| 工具注册与描述 | 自带 | —— |
| 参数结构校验 | 部分自带 | 模型传错类型直接抛异常 |
| 幂等与重试策略 | 基本不带 | 重复下单、重复扣款 |
| 全链路追踪 | 不带或很弱 | 出问题只能靠日志猜 |
| 离线评测集 | 不带 | 改一句提示词全靠肉眼比 |
| 成本与延迟监控 | 不带 | 月底账单才知道超支 |
| 权限分级与人工闸门 | 不带 | 高危操作被模型直接执行 |
这张表里最容易被忽略的是最后一行。我见过一个内部工具,Agent 被授予了"删除测试数据"的能力,本意是清理环境,结果模型在一次语义误判后把一批还没验证的数据删了。技术上没有 bug,流程上有。凡是不可逆的写操作,都必须过一道人工确认闸门,这条不是建议,是底线。
2. 最小可运行骨架:先让它"可调试",再让它"能跑"
2.1 依赖与运行时版本的地基工程
这类框架的迭代速度通常快得离谱,主版本之间经常有破坏性变更,而且它依赖的模型 SDK 也在同步变动。我踩过最冤的一次坑是:本地环境跑得好好的,容器里一启动就报参数不识别,排查了两小时才发现是镜像里的 SDK 版本比本地低了一个小版本,tool_choice字段被静默忽略了。
地基工程的做法很朴素但必须做:锁定精确版本而不是范围版本,把框架、模型 SDK、HTTP 客户端三个包全部钉死,并且在 CI 里加一条"依赖一致性检查"。Python 项目里我通常直接用一份requirements.lock.txt配合哈希校验:
framework==x.y.z \ --hash=sha256:... model-sdk==a.b.c \ --hash=sha256:...另外两个容易翻车的点:一是异步运行时,大部分框架的工具执行是并发的,如果你的工具里有同步阻塞的数据库调用,记得包一层线程池,否则并发量一上来整个 Agent 就卡在事件循环上;二是超时层级,框架层、工具层、HTTP 层各自有超时,必须保证外层超时大于内层之和,不然会出现"框架已经放弃但工具还在跑"的幽灵任务。
2.2 四个构件搭出第一个可跑的骨架
不要一上来就设计复杂的多 Agent 协作。先把单 Agent 的四个构件搭好:系统提示、工具集、运行时配置、输出解析。下面是一段可以直接改巴改巴用的骨架:
from framework import AgentRuntime, Tool, tool_registry @tool_registry.register( name="query_order_status", description="根据订单号查询订单当前状态。仅用于查询,不修改任何数据。" ) def query_order_status(order_id: str) -> dict: # 参数在进入函数体前已被框架按 schema 校验 row = db.query_one("select status, updated_at from orders where id = %s", order_id) if not row: return {"found": False, "hint": "订单号不存在,请向用户确认是否为完整订单号"} return {"found": True, "status": row.status, "updated_at": str(row.updated_at)} runtime = AgentRuntime( tools=[query_order_status], system_prompt=SYSTEM_PROMPT, max_turns=6, # 硬性刹车,先给保守值 per_turn_timeout=20, # 单轮上限,秒 total_timeout=90, # 整任务上限,秒 )注意query_order_status的返回值:found=False时我额外给了一个hint字段。这不是装饰,是给模型的自救线索。如果只返回一个空对象,模型大概率会换个参数再试一次,然后又一次,直到把轮次耗光;给它一句明确的下一步建议,绝大多数情况下它一次就纠正过来了。
2.3 从第一天就埋下追踪,别等出事再补
我的建议是:第一个版本就要有 trace,哪怕打印到控制台。Agent 的调试体验和传统程序差异极大——你不能打断点,因为"执行路径"是模型生成的,每次都可能不同。唯一的抓手就是完整记录每一轮的输入输出。
最小的追踪记录应该包含这些字段:trace_id、轮次序号、本轮的完整消息列表(或哈希)、模型的原始输出、解析后的工具名与参数、工具耗时、工具返回摘要、本轮 token 消耗。这些信息用一行 JSON 打到日志里就够了,不用一上来就上正式的链路系统:
runtime.on_turn_end(lambda evt: log.info("agent_turn", extra={ "trace_id": evt.trace_id, "turn": evt.turn_index, "tool": evt.tool_name, "args": redact(evt.tool_args), # 脱敏后再落盘 "latency_ms": evt.tool_latency_ms, "tokens_in": evt.usage.input, "tokens_out": evt.usage.output, }))redact那一步千万别省。工具参数里经常夹着手机号、身份证号、内部路径这类信息,一旦原样落盘,后面做数据治理会非常痛苦。
3. 工具层设计:落地成败的八成在这里
3.1 工具描述写得糙,后面全是补丁
模型选工具的唯一依据就是你写的description。我见过太多项目的工具描述是"查询订单"三个字,然后抱怨模型老是调错。这就像给新同事一份只有函数名没有注释的代码,还指望他一次写对。
我总结了一套工具描述的四段式写法,实测调用准确率提升非常明显:
- 做什么:一句话说清功能,动词开头。
- 什么时候用:给出触发条件的自然语言描述,这是最关键的。
- 什么时候不用:显式列出反例,避免和相似工具混淆。
- 返回什么:字段含义和可能的空值情况。
举个对比例子。模糊版:根据用户ID查询会员等级。具体版:查询指定用户的会员等级与积分余额。当用户询问自己的等级、积分、可享受的折扣时使用;当用户询问订单物流时不要使用本工具。返回 level(等级名称)、points(积分整数,可能为 0)、expire_at(等级到期时间,可能为空)。后者长了两倍,但换来的是准确率和可解释性。
还有个技巧:同类工具命名要有明显区分度。get_user_info和get_user_detail这种命名,人和模型都会懵,不如改成get_user_profile和get_user_permissions。
3.2 参数校验和幂等:把"模型猜错"的成本压到最低
模型生成参数出错是常态而非异常,你的工具层必须假设"每一个参数都可能是错的"。三层防护我一般这样铺:
第一层是schema 约束。类型、枚举、正则、长度上限全部写进工具的入参定义里,让框架在调用前就挡掉一批。枚举尤其有用——把"订单状态"限定成几个固定值,比让模型自由发挥强太多。
第二层是业务校验。schema 过了不代表业务上合理,比如订单号格式对但不存在、日期范围反了、金额为负。这些检查要在工具函数体最前面做,并且返回结构化的错误而不是抛异常,因为异常通常会中断整个循环,而结构化错误能让模型自己纠正。
第三层是幂等保护。凡是有副作用的工具,都要带一个幂等键。我常用的是"会话ID + 工具名 + 参数哈希"作为键,写入前先查一下这个键是否已经执行过:
def create_ticket(session_id: str, title: str, content: str) -> dict: idem_key = hashlib.sha256(f"{session_id}|create_ticket|{title}|{content}".encode()).hexdigest()[:32] if cache.exists(f"idem:{idem_key}"): return {"ok": True, "ticket_id": cache.get(f"idem:{idem_key}"), "note": "重复调用,已返回上次结果"} ticket_id = ticket_service.create(title=title, content=content) cache.setex(f"idem:{idem_key}", 3600, ticket_id) return {"ok": True, "ticket_id": ticket_id}这段代码救过我至少两次。模型在网络抖动导致工具超时后,会倾向于"再试一次",没有幂等保护就是两张工单。
3.3 高危操作必须有闸门,不能靠提示词约束
在系统提示里写"不要执行危险操作"是没有任何保障力的。正确的做法是把工具按风险分级,在框架层做硬拦截:
| 风险级别 | 典型工具 | 处理策略 |
|---|---|---|
| L0 只读 | 查询、检索、计算 | 直接执行 |
| L1 低危写 | 创建草稿、追加备注 | 执行 + 记录审计日志 |
| L2 中危写 | 修改状态、发送通知 | 执行前二次确认参数 |
| L3 高危 | 删除、退款、批量变更 | 强制人工确认,Agent 只生成待办 |
L3 的实现方式是:工具不真正执行,而是返回一个"已生成待确认操作单"的结果,附带操作单号,由人在管理后台确认后再走正常业务链路。这样 Agent 依然能完成"帮我把这批数据清理掉"的语义理解,但最终动作的扳机握在人手里。
4. 上下文工程:既要记得住,又不能让 token 爆掉
4.1 短期上下文的三层裁剪
长任务跑下来,消息列表会膨胀得很快。我的裁剪策略分三层,从轻到重依次触发:
第一层,工具返回结果压缩。工具返回给模型的应该是一份"摘要"而不是原始响应。查数据库返回 200 行明细?先在工具里聚合,只给模型"共 200 条,状态分布如下……"。这一步往往能砍掉 70% 的上下文体积,收益最大且实现最简单。
第二层,历史消息折叠。超过 N 轮之后,把早期的"工具调用 + 工具结果"成对折叠成一句自然语言总结,例如"已查询订单 A123,状态为已发货"。注意要成对折叠,单独删掉工具调用而留下结果,会让消息结构不合法,很多接口会直接报错。
第三层,滑动窗口 + 锚点保留。保留系统提示、最初的用户目标、最近 K 轮原文,中间部分全部压缩成摘要。这里的"锚点"很关键——最初的任务目标如果被裁掉了,模型跑到后面会忘记自己到底要干什么,这是长任务跑偏最常见的原因。
4.2 长期记忆的写入时机比存储方案更重要
很多人一上来就纠结用什么向量库,其实更难的是什么时候写、写什么、什么时候忘。我的经验是:
写入时机只选三个——任务成功结束时写"结论",用户显式纠正时写"偏好",工具返回稳定事实时写"事实"。任务失败时不写,因为失败的路径往往包含错误假设,写进去会污染后续检索。
写入内容要带元信息:来源会话、时间戳、置信度、有效期。没有有效期的记忆是负债,比如"用户当前会员等级是黄金",三个月后大概率已经过期,检索出来反而误导。
遗忘策略上,我倾向于时间衰减 + 命中率淘汰双轨:超过有效期的直接归档不参与检索;长期零命中的记忆降权。这件事一定要做成自动化的,靠人工清理一定会烂掉。
4.3 给长任务留一个可恢复的状态点
超过三十秒的任务,用户很可能中途关页面、网络断掉、或者干脆去干别的了。如果没有状态持久化,一切从头再来,体验极差且成本翻倍。
我的做法是在每一轮结束后,把trace_id、已完成步骤列表、当前中间产物、剩余待办序列化存下来。恢复时不是简单地把历史消息塞回去,而是重新构造一个精简的上下文:系统提示 + 原始目标 + 已完成步骤摘要 + 当前待办。这样恢复成本几乎恒定,不会因为任务跑得久而变贵。
5. 上线前的三道关:追踪、评测、成本
5.1 追踪要能回答"为什么",而不只是"发生了什么"
前面说的日志只能回答"调了什么工具",但排查时真正需要的是"模型为什么选了这个工具"。所以追踪里必须保留当轮模型看到的完整输入——至少是工具列表和关键上下文的哈希,理想情况下是完整的请求体(脱敏后)。
我一般会在追踪系统里做三个视图:单次会话的时间轴(每轮的输入摘要、输出、耗时、token)、工具维度的成功率与错误分布、失败会话的聚类(把失败原因相似的会话归到一起)。第三个视图价值最高,因为 Agent 的失败模式高度重复,往往修三个问题就能覆盖八成故障。
还有一点:采样策略要按结果分流。正常的会话按 5% 采样就够了,但凡是触发了最大轮次、工具报错、超时、用户负反馈的会话,一律 100% 保留。这些才是真正有价值的样本。
5.2 评测集不是奢侈品,是改提示词的前提
没有评测集的时候,改一句提示词靠什么判断好坏?靠感觉。这是最危险的工作方式,因为你永远不知道这次改动修好了 A 场景的同时是不是弄坏了 B 场景。
我的评测集是这样攒起来的:上线第一周,把所有失败会话人工标注一遍,每条标注"期望的工具序列"和"期望的最终答案要点"。攒到五十条左右就能跑起来了。评测指标别看太多,三个足够:
- 任务完成率:最终答案是否满足标注要点。
- 工具序列准确率:调用的工具和顺序是否与期望一致(允许合理变体)。
- 平均轮次:同样任务用的轮次是否下降,这是效率指标。
回归门槛设成"完成率不低于基线 2 个百分点、平均轮次不增加 20% 以上",超过就拦住不让发。这套机制搭起来大约需要两天,但后面每次迭代都能省下大量的反复。
5.3 成本与延迟的取舍要提前算清楚
Agent 的成本结构和普通接口完全不同,它是轮次 × 上下文长度的乘积关系。我做过一个粗略的测算:同样一个客服场景,单轮问答平均 800 token,跑到 Agent 里变成平均 4.2 轮、每轮 3500 token,实际消耗是原来的 18 倍。如果不提前算这笔账,上线后账单会很难看。
常见的几个优化手段和它们的代价:
| 手段 | 省什么 | 代价 |
|---|---|---|
| 工具结果压缩 | 输入 token | 可能丢失细节,需验证 |
| 简单意图走直连不进口 Agent | 整体轮次 | 需要维护一条旁路逻辑 |
| 小模型做工具选择、大模型做生成 | 单轮成本 | 多一次调用,延迟增加 |
| 并行工具调用 | 延迟 | 需要处理结果合并与冲突 |
| 上下文缓存(前缀复用) | 输入 token | 要求系统提示稳定不变 |
表格里最后一条最容易被忽略:系统提示和工具描述如果每轮都变,缓存全部失效。我见过有项目把当前时间戳精确到秒塞进系统提示,直接把缓存命中率打到零。改成只放日期,成本立刻降了一截。
6. 上线后最常见的三类翻车与排查链路
6.1 死循环调用同一个工具:从现象到根因
现象:某个会话的工具调用次数异常高,token 消耗飙升,最终触发最大轮次限制返回兜底话术。
排查链路:第一步看这个工具的返回内容,大概率是返回了空结果或者错误结构;第二步看模型在收到这个返回后说了什么,如果它在自言自语"没有找到,我再试一次",说明返回结果没有给出有效的下一步指引;第三步看参数是否每轮都一样,如果一样,说明模型陷在了同一个假设里。
根因通常有三个:工具返回空结果时缺少hint字段(前面提过);工具描述里没有说明"查询不到时应该怎么办";最大轮次设置过高,给了它太多试错机会。
修复方案:在工具的空结果分支里补上明确指引;在系统提示里加一条"同一工具连续调用两次结果相同则必须停止并告知用户";把最大轮次从 10 降到 6。第三条听起来粗暴,但实测有效,因为大部分正常任务 4 轮内就能完成。
6.2 长任务被上下文截断:一个隐蔽的失败
现象:短任务一切正常,稍微复杂一点的任务在第 5 轮之后开始答非所问,或者突然重复之前已经做过的事。
排查链路:先打出每轮的实际输入 token 数,看是否贴近模型上限;再看裁剪逻辑是在哪一轮触发的;重点检查裁剪后最初的任务目标是否还在上下文里。
根因:滑动窗口把最早的用户消息裁掉了。这是最隐蔽的一类 bug,因为程序没有任何报错,日志看上去也正常,只是模型"忘了"。
修复方案:把系统提示、首条用户消息、当前待办列表标记为不可裁剪的锚点;裁剪时优先折叠工具往返消息;在裁剪发生后打一条显式日志,方便后续核对。
6.3 输出格式漂移:下游解析崩溃
现象:Agent 返回的最终答案,前 90% 的请求都能被下游 JSON 解析,偶尔几条报解析失败。
排查链路:把失败样本的原始输出拉出来对比,通常能看到模型在答案前后加了"好的,根据查询结果……"这类前缀,或者把 JSON 包在了 Markdown 代码块里。
根因:格式约束只写在提示词里,没有做强制。模型在上下文变长、任务变复杂时,遵循格式指令的稳定性会下降。
修复方案:三层加固。一是如果模型接口支持结构化输出参数,直接开启,这是最硬的约束;二是在解析前做一次容错提取,先用正则捞出第一个完整的 JSON 对象再解析;三是解析失败时不要直接抛给用户,而是把"格式错误,请只输出 JSON"作为一条新的系统消息重试一次,通常一次就能修好。
顺带说一个我在实践中形成的习惯:所有 Agent 的最终输出都过一遍 schema 校验,校验失败就走兜底文案。宁可给用户一句"抱歉我没能整理出结果",也不要让下游服务收到一个畸形的数据结构然后雪崩。
这套东西从头搭到尾,我一个人大概花了两周多,其中一半时间在补工具层的健壮性和评测集。回头看,真正让我少走弯路的判断只有一个:把 Agent 当成一个"会犯错的新同事"来设计系统,而不是当成一个"更聪明的接口"。接口错了会报错,新同事错了会继续往下干,所以你得给他边界、给他反馈、给他留后路,还得有人在关键节点上把一下关。