今年上半年,我在做一个内部Agent项目时,同事抛过来一句话:“你那个Demo跑得挺欢,但能不能扛住100个真实用户同时用?”我当时愣了一下。Demo能跑和能上生产,中间隔着的从来不是模型效果,而是一整套没人愿意明说的工程化支撑。于是我把手头一个代号叫PI的Agent框架彻底拆开,重新搭了一层生产级Harness。这篇就当复盘记录,适合正在做Agent开发、特别是准备把Agent从Notebook搬到生产环境的朋友。我会尽量讲清楚为什么拆、拆完看到什么、以及哪些地方值得你直接抄作业。
1. 我为什么要把Agent框架拆开去重写Harness
1.1 很多Agent项目,不是死在模型能力上,而是死在“没有Harness”上
先说一个容易被忽略的事实:Agent框架和Harness是两回事。框架解决的是“Agent怎么组织决策”,比如模型该怎么被调用、工具该怎么接、上下文该怎么传;Harness解决的是“Agent怎么在真实环境里安全地跑起来”,比如请求怎么追踪、状态怎么恢复、工具调用怎么限流、失败怎么降级。
我见过太多团队把LangChain或者自研框架一接,调通两个工具,就觉得Agent已经做完了。结果一压测,问题全冒出来:模型偶尔返回一段非JSON,工具执行没有幂等导致重复下单,进程一重启整个任务状态归零。这些问题都不是模型能力能解决的,是Harness层该背的锅。
测试领域早就有“test harness”的说法,意思是给被测对象搭一套执行环境、桩件和断言工具。Agent领域的Harness把这个思想搬了过来,但多了一层:它不只是测Agent,还要让Agent在生产里被“约束着运行”。就像赛马必须配马具,不然再快的马也跑不出赛道,还会伤到骑手。
1.2 PI的定位,以及我给自己定的四条改造目标
PI是我对内部一个Agent框架的代号,网上也有叫PI的个人智能体产品,但这不是重点。重点是任何Agent框架拆到底,都逃不过五块:模型接入、工具层、记忆层、调度内核、评估手段。我当时对PI做的第一件事,就是把这五块在代码里明确划出边界,而不是混在一坨类里。
围绕这个拆分,我给新Harness定了四条铁律,后续所有开发决策都以它们为准:
| 铁律 | 含义 | 落点 |
|---|---|---|
| 可追踪 | 每次决策都能回放 | 端到端trace_id,从HTTP请求到LLM调用全程串起来 |
| 可重入 | 任务随时可以恢复 | 状态持久化,崩溃后从断点继续而不是从头开始 |
| 可降级 | 任何外部依赖挂了都不拖垮主流程 | 模型、工具、记忆库全部做熔断与降级 |
| 可评估 | 每次改动有据可依 | 离线回放评测 + 在线观测双轨 |
当时有人问我,这四条是不是有点重了。我的回答是:只要你的Agent要面对真实用户,这四条一条都省不掉。你以为省的是工作量,实际省掉的是可靠性。
2. 拆开Agent框架后看到的五个职责层
2.1 调度内核:把“链式调用”升级为“状态机循环”
很多Agent框架最原始的形态是一条链:第一步调用模型,第二步调用工具,第三步再调模型收尾。链式结构写起来很爽,但稍微复杂点的任务就卡住了。比如一个查询任务,Agent要先查用户权限,再决定检索哪几个数据源,中途还可能要找人确认参数,最后才能汇总——这种路径必须靠状态机来表达。
我给PI改调度内核时,核心思路是:把一次任务拆成一组带状态的步骤,每个步骤有id、类型、依赖关系和回调。每一步执行完都写回状态库,Harness的主循环去查“现在该执行哪个步骤”,而不是模型一路推理到底。模型在循环里只做两件事:根据当前状态选择一个动作,或者结束任务。这样所有决策都变成可观测的节点,也为后面的可重入打下了基础。
2.2 技能层:工具注册、参数校验与权限边界
Agent调用的工具,本质上是暴露给模型的函数。但暴露给模型的东西,不能和暴露给内部研发的函数一样。生产环境里,工具要过一道注册表:名称、描述、参数schema、权限矩阵、超时阈值、调用配额,一个都不能少。
我在这块踩过一个很典型的坑:早期图省事,直接把内部Python函数塞给模型,结果模型可以根据描述猜出一些本来不打算给它的函数。后来所有工具都改成显式注册,并且加了参数校验层,模型传进来的参数先过schema校验,不合法就直接返回错误,而不是扔进真实函数里跑。权限矩阵也放到这一层:同一个工具,不同用户能调用的数据范围完全不同。这块做扎实之后,后面再加新工具就变成了配置工作,而不是每次都要写胶水代码。
2.3 记忆层:短期上下文和长期经验要分开存
记忆是Agent框架最容易做成一锅粥的地方。很多人以为记忆就是“把对话历史都塞给模型”,但短期上下文和长期经验是两回事。短期上下文是当前任务窗口内的对话、工具结果和中间状态,讲究的是快、准、不丢信息;长期经验是跨任务沉淀下来的方法论,比如“用户问发票问题时,先查清公司抬头再调开票接口”,讲究的是可检索、不膨胀。
我把PI的记忆层拆成两套存储:短期上下文用带窗口的缓冲,超阈值就做摘要压缩;长期经验用结构化记录,按任务类型和标签索引。两条线都有生命周期管理:写入、合并、过期、清理。没有清理机制的记忆库,跑三个月就会被无效信息塞满,到时候模型检索到的全是噪音。
2.4 模型层:多供应商路由与故障降级
生产环境最忌讳的事,就是把身家性命押在一家模型供应商上。模型层要做路由,按任务类型、成本和延迟打分,选一个合适的模型来执行当前步骤。比如简单分类任务走便宜快的小模型,复杂推理任务才动用强模型。
路由之外必须有降级链。当时社区讨论比较多的一些Harness实践,包括基于DeepSeek这类底座做的开发套件,有一点是共通的:主模型超时或返回异常时,要自动切到备选模型,而不是让用户干等。模型返回的JSON解析也要在这一层做容错,LLM偶尔会输出多余文字,解析器要能剥离后用严格模式再解析一次。这些坑不在Harness层挡掉,就会一层层漏到业务代码里。
2.5 评估层:Harness里的“体检中心”
很多Agent项目跑起来之后,没有任何手段判断“这次改动是变好了还是变坏了”。评估层就是干这个的。我把评估拆成离线和在线两条线:离线评估用历史任务记录回放,自动跑一遍,对比关键指标;在线观测用真实trace数据,看成功率、耗时、工具误用率、无响应率。
安全评测也要挂在评估层。比如指令注入防护,用户输入里藏了“忽略之前所有指令”这类话,模型反应是否安全;再比如工具误用,模型没有权限却试图调用受限工具,这类行为必须被Harnes察觉并记录。没有评估层的Harness就像不开仪表盘的飞机,你觉得在飞,其实早偏航了。
3. 开发生产级Harness的关键工程点:六问六答
这一章我直接用问答形式写,因为这些问题是我被同事和同行问得最多的,也是我自己在真实环境里持续吃的苦头。
| 工程问题 | 生产要求 | 对应模块 |
|---|---|---|
| 出问题能追踪吗 | 端到端trace | 调度内核 |
| 工具会重复执行吗 | 幂等与超时 | 技能层 |
| 中断后能从哪接着跑 | 可重入状态 | 状态存储 |
| 模型抽风怎么办 | 熔断与降级 | 模型层 |
| Agent能访问什么 | 沙箱隔离 | 工具执行环境 |
| 记忆会污染吗 | 原子写入 | 记忆层 |
3.1 出问题能追踪吗:端到端trace
生产环境下的Agent排查,最怕的是只有结果没有过程。模型为什么调了那个工具?工具返回了什么?下一步为什么走的分支?这些问题没有trace根本答不上来。我给每条请求分配一个trace_id,从网关开始一直透传到每次LLM调用、工具执行和记忆读写,所有日志都带这个id。
实践上有个小技巧:trace日志不要只记录“做了什么”,还要记录“被什么驱动着做的”,也就是把模型当时的决策输入摘要一起打进去。这样回放的时候,你能看到模型是基于哪几条上下文做的决定,而不是对着孤零零的一行“调用工具search_invoice”瞎猜。
3.2 工具会重复执行吗:幂等与超时
Agent调用工具不像人敲命令,失败之后它会自己重试。但如果工具本身没有幂等性,重试就是事故。比如发送邮件、扣减积分、创建订单这一类操作,同一请求执行两次和一次,后果完全不同。我在工具注册表里强制要求:凡是写操作,必须支持幂等键。调用方生成一个幂等键,服务端记住这个键的处理结果,重复请求直接返回原结果。
超时也一样,一定要在Harness层硬性规定。模型调用工具,3秒没返回就按失败处理,不能无限等下去。否则一个下游服务假死,整条Agent链路全部卡住。真实环境里我甚至见过一个工具把整个任务拖了40分钟,就是因为没有超时控制。
3.3 中断后能从哪接着跑:可重入状态
生产Agent一定要假设自己会崩溃。进程被杀、机房断网、发布重启,这些不是小概率事件。我把任务的每一步状态写进持久化存储,字段包括pending、running、done、failed四种。Harness重启后,扫描所有处于pending或者running状态的任务,恢复到断点,路由进主循环继续跑。
这里有个容易忽略的细节:一个步骤已经处于running但实际没执行完,恢复时不能盲目重跑。我的做法是配合工具层的幂等键,恢复后先检查该步骤对应的外部调用是否已经生效,再决定是继续等待还是重试。没有这层检查,可重入反而会变成重复执行的帮凶。
3.4 模型抽风怎么办:熔断与降级
模型供应商再稳定,也会有连续失败的时候。我给模型层加了一个简单的熔断器:连续失败N次,就打开熔断开关,直接把流量切到备选模型,同时启动半开探测,过一段时间放一点量回来试探。配置上,主备模型的切换策略要提前写清楚,不要等到线上炸了再开会讨论。
降级也包括“向用户说明现状”这一环。模型不可用时,Harness要能返回一个可读的错误信息,而不是让前端收到一个500。用户的体验可以打折,但不能被糊弄。该说“服务暂时不可用请稍后再试”的时候,就老老实实说。
3.5 Agent能访问什么:沙箱和权限收敛
语音上大家都在说Agent,但Agent能访问什么资源,很多人没有认真收敛过。我把所有工具执行都放到受限环境里:文件系统只读虚拟目录,网络访问走白名单代理,CPU和内存设置配额。代码执行类工具还会额外开子进程或者容器,跑完直接销毁。
这部分的取舍是:灵活性和安全性的平衡。早期我为了让Agent“能力更强”,把沙箱放开了一些,结果它在一个测试环境里调用了一个不该调的接口。还好有审计日志,事后一眼就查出来。从那以后我的原则就一句话:默认最小权限,每次放开都要有理由,所有动作都要留痕。
3.6 记忆会污染吗:写入原子性与版本控制
长期记忆一旦写脏,Agent会在之后很长一段时间内持续犯错,这种错误还特别隐蔽。我给记忆层加了版本号,每次写入都基于版本做乐观锁,冲突时采取新旧合并或弃旧留新。并行跑的任务同时往同一个记忆主题写内容时,这条设计就特别关键。
记忆的污染还有一个来源:Agent自己编造的经验。模型在总结长期经验时可能会脑补一些没有发生过的事。解决方案是让经验沉淀必须走校验链路:要么由另一个模型交叉验证,要么定期人工抽检。好记性不如烂笔头,但烂笔头写错内容,比没有笔头还可怕。
4. 借鉴PI控制思想设计记忆与反馈回路
控制工程里有一堆热词,“电压电流双闭环PI控制”“电流环PI参数整定”“转速外环P与PI对比”等等。我刚开始觉得这些词跟Agent八竿子打不着,后来把PI的记忆和反馈回路拆开一看,发现Agent本质上就是一个数字控制器,只不过控制对象变成了“决策质量”。
4.1 比例项:当前任务的即时决策
PI控制里的P项,输出和当前偏差成正比,偏差越大动作越大。对应到Agent里,就是当前这一步基于眼前上下文的即时决策。工具该不该调、问题该不该追问,这些反应要快、要直接。
但P增益不能调太大。增益太大,Agent就会过于敏感,工具返回一个轻微的异常就立刻发起重试,造成震荡;增益太小,该行动时又犹豫不决,宁可自己编答案也不去查工具。实际整定时,我会用“工具触发阈值”来控制这个增益:除非上下文明确包含需要查询的意图,否则先按常识回答;一旦触发,就直接执行而不是反复向用户确认。
4.2 积分项:长期经验沉淀
I项负责消除稳态误差。没有I项的PID控制器,系统会一直存在一个残差;没有长期记忆的Agent,会反复犯同一个错误,每次都在同一个坑里摔得一模一样。这就是“稳态误差”在Agent世界的表现。
把失败经验沉淀进长期记忆,就是I项在累积误差之后给出补偿。比如Agent在一次任务里发现“这个数据源在下午四点后拉取经常超时”,这条经验被写入长期记忆,之后同一时段的任务会直接选择备用数据源。但积分项会饱和,对应到记忆就是经验过度累积。所以要有遗忘机制和学习率:旧经验按时间衰减,新经验逐渐加权,不然记忆库会被过期结论占据。
4.3 微分项:防错前置
D项在PID里预测误差的变化趋势,起到阻尼作用。Agent里可以做成“历史失败匹配”的预检:在每一步工具调用前,拿当前上下文去匹配相似的历史失败样例,一旦命中高风险模式,就先切换策略而不是硬试。
这套东西成本不高,一个带相似度检索的小模块就能跑起来。它的价值在于,你不需要等错误发生再收拾,而是让系统自己有“预感”。当然,这不是必须项,如果团队资源紧张,可以先做P和I,D项等有数据积累之后再上。
顺带说一个双闭环的类比:内环电流环相当于短期上下文和工具反馈回路,要求响应快、动态好;外环电压环相当于长期记忆和目标对齐,要求稳态准、不漂移。设计记忆系统时,我会提醒自己:短期记忆的窗口大小和摘要压缩阈值,就是在整定“内环参数”;长期记忆的经验沉淀条件和清理策略,就是在整定“外环参数”。这比引用一堆时髦架构词好理解得多,也更好向团队解释。
5. 从原型到上线的实测踩坑记录
5.1 记忆框架选型:热门的未必适合
“Agent记忆框架以及选型”这个话题在社区里很热,我也被带着看过向量库、图数据库、混合检索一堆方案。最后我选了最简单的那一个:核心记忆用结构化存储,相似度检索用一个嵌入式向量索引就够了。
选型的逻辑就三条:第一,你的记忆是单条事实还是关系网络?单条事实为主就别上图数据库。第二,写入频率高不高?写入频繁就别把向量库当主存储,容易拖垮链路。第三,团队对这套组件的运维能力够不够?不够就不碰。技术栈这东西,适合永远比时髦重要。
5.2 工具调用死循环:Agent“二进宫”的问题
上线后遇到的最诡异问题,是Agent在同一个失败工具上反复折腾,每次生成不同的参数再试,直到把上下文塞满。排查下来发现根因有两层:第一层,工具返回的异常信息太长太杂,模型误以为换一个参数就能成功;第二层,调度循环没有限制单任务的最大工具调用轮数。
修复方案是在Harness层加了一道硬约束:单任务最多执行15次工具调用,超过直接进入人工处理分支;工具异常信息做规范化截断,只保留错误码和一句可读描述;同一工具同一参数失败两次后,禁止同路径重试,强制退回上一个决策分支。这套组合拳上线后,这类问题基本绝迹。
5.3 Harness与Agent职责越界的代价
早期我习惯把业务规则直接写进prompt里,比如“如果用户是内部员工,走A通道;否则走B通道”。一开始确实灵活,但后来发现只要业务规则一变,就要重新调prompt甚至微调模型,费时费力且效果不可控。
后来我把边界清理干净:一切可枚举的确定性规则,全部下沉到Harness层用代码实现;只有无法枚举的语义判断才交给Agent模型。二者的区分标准很简单——规则能不能写成if-else?能,就别让模型决策。这条原则让系统的可维护性上升了一个台阶。
5.4 评测集污染:别拿标准答案训练Harness
我在早期评估Harness效果时,用了一套精心整理的标准数据,调参调到分数很好看。结果换了一批真实用户指令,效果直接打回原形。原因是评测集被污染了:我在调试过程中反复用它,模型参数和Harness配置都向着这套数据过拟合了。
现在的做法是:评测集按月轮换,持续加入真实用户指令的脱敏样本,单独留一批“新指令集”做泛化测试,不进调参循环。评测脚本记录的不只是总得分,还有失败模式标签,比如“工具误用”“指令注入失防”“记忆检索无关”,这样迭代Harness时才能针对性改进。
6. 一个可运行的Harness最小骨架
6.1 核心对象设计
下面给一个能跑的最小骨架,代码以Python为主,重点是结构而不是功能完备。核心就五个对象:Harness负责主循环;StateStore负责状态持久化;ToolRegistry负责工具注册与执行;ModelRouter负责模型路由;MemoryStore负责短期上下文和长期经验读写。
from dataclasses import dataclass, field from typing import Optional import uuid import time @dataclass class ToolCall: name: str arguments: dict idempotency_key: str = field(default_factory=lambda: uuid.uuid4().hex) status: str = "pending" # pending/running/done/failed @dataclass class TaskState: task_id: str = field(default_factory=lambda: uuid.uuid4().hex) steps: list = field(default_factory=list) current_step: int = 0 status: str = "running" # running/done/failed trace_id: str = field(default_factory=lambda: uuid.uuid4().hex)6.2 主循环逻辑:带状态、带幂等、带超时保护
主循环的逻辑并不复杂:读取当前状态,决定下一步动作;动作如果是调用工具,就带上幂等键执行;执行后把结果写回状态库。模型路由和记忆读写在这里只留了接口示意。
class Harness: def __init__(self, store, tools, model_router, memory): self.store = store self.tools = tools self.model_router = model_router self.memory = memory self.max_tool_calls = 15 def run(self, task_state: TaskState, user_request: str): tool_call_count = 0 while task_state.status == "running": context = self.memory.get_context(task_state.task_id) decision = self.model_router.decide(context, user_request) if decision.action == "finish": task_state.status = "done" self.store.save(task_state) break if decision.action == "call_tool": tool_call_count += 1 if tool_call_count > self.max_tool_calls: task_state.status = "failed" task_state.steps.append( {"type": "error", "detail": "max_tool_calls_exceeded"} ) self.store.save(task_state) break result = self.tools.execute( name=decision.tool_name, arguments=decision.arguments, timeout_seconds=5 ) self.memory.append_tool_result( task_state.task_id, decision.tool_name, result ) task_state.steps.append( {"type": "tool_call", "name": decision.tool_name, "result": result} ) self.store.save(task_state) return task_state实际使用的时候,StateStore会落到Redis或PostgreSQL,Tools会带上权限和幂等控制,ModelRouter会有真实的降级链。这套骨架的价值在于:它把主循环的形状定下来了,你往里面填东西不会跑偏。
6.3 最小骨架的扩展顺序
如果你照着这个骨架去做自己的Harness,我个人建议按这个顺序扩展:先挂状态存储到真实数据库,让任务断了能恢复;再加观测,把trace_id对应的日志结构化收集起来;然后接人审队列,让风险操作有一个人工确认的入口;最后才去优化工具数量、记忆策略这些锦上添花的东西。
这套结构从PI框架拆出来到现在,已经在我们内部扛过了几轮真实流量。我自己最大的体会是:Agent框架决定了一个项目的想象力上限,而Harness决定了一个项目的生存下限。很多人一上来就追模型、追框架,但我建议你先把你手头那个框架拆开看看,哪怕它再简陋,也要知道哪几层是必须由你自己扛起来的。这些层,就是你的生产级Harness。