给 Agent 装上护栏——harness 层设计
给 Agent 装上护栏——harness 层设计
这是「Agent 工程化」系列的第九篇。前面八篇把 Agent 的"能力"讲完了:会动手、记性好、懂业务、有经验、还会拆任务。但能力越强越让人害怕——它跑飞了怎么办?这篇讲护栏(harness):预算、白名单、重试、断点恢复,让"能干"的 Agent 也"听话"。
跑飞现场:上线第一天,钱花光了还在跑
Agent 上线第一天的真实场景:
用户:帮我查一下这个行业的最新动态 Agent:(调用搜索工具,第 1 次)搜到 10 条结果 Agent:(第 2 次)嗯,再搜下另一个关键词…… Agent:(第 30 次)…… Agent:(第 200 次)还没完,再搜搜……用户早走了,Agent 还在自己循环:搜索工具被调了几百次,token 费用哗哗涨,但没人能叫停它。
为什么会这样?回看第二篇那个核心循环:
while任务未完成:# ← 问题在这llm 输出{思考,行动}执行工具循环的条件是"任务未完成"——谁来判断完成?LLM 自己。而 LLM 可能陷入"再搜一次也许有结果"的死循环。循环是油门,没有刹车。
护栏要做的,就是给这个循环装上刹车。专业点说:在 Agent 循环外面包一层 harness(缰绳),它不管 Agent 怎么想,只管"不许越过这些线"。
护栏的整体结构:循环外的一层壳
harness 不管"怎么完成任务",只管四件事:预算、白名单、重试、断点。逐个拆。
核心一:三层预算——先定义"多少钱算完"
跑飞的本质是没有成本上限。解法是给循环套三层预算,任何一层超了都强制收尾:
| 层级 | 管什么 | 例子 |
|---|---|---|
| 每次调用 | 单次工具调用的开销上限 | 单次返回最多 2000 token |
| 每轮循环 | 单轮思考+行动的 token | 单轮最多 1500 token |
| 整个会话 | 总调用次数 / 总 token / 总费用 | 最多 20 次工具调用,总费用 0.5 元 |
classBudget:def__init__(self):self.max_calls=20# 整个会话最多调 20 次工具self.max_cost=0.5# 最多花 0.5 元self.calls=0self.cost=0.0defcheck(self)->bool:"""循环每轮开始时检查,超了就刹车"""returnself.calls<self.max_callsandself.cost<self.max_cost循环每轮都问一句"预算还够吗":
budget=Budget()whilenottask_doneandbudget.check():# 刹车在这里...# Agent 循环本体budget.calls+=1budget.cost+=round_cost(...)else:ifnottask_done:finish_gracefully("已达本次会话上限,先到这里")# 优雅收尾优雅收尾是关键:预算耗尽不是硬中断,而是告诉用户"已达上限,需要的话可以继续"。
核心二:工具白名单——只允许它碰该碰的
Agent 能力越强,越要给它的工具划定范围:
| 方式 | 说明 | 风险 |
|---|---|---|
| 黑名单 | 默认都可用,禁止某些 | 漏一个就出事 |
| 白名单 | 默认都不可用,只放行名单内的 | 最安全,写操作天然受限 |
生产里两个原则:
- 白名单 + 按场景下发:查资料任务只下发搜索工具,订票任务才下发下单工具——每个 Agent 看到的工具是"这个任务该有的"
- 写操作强制人工确认:下单、付款、删数据这类工具,执行前必须过人工确认闸(第三篇讲过两道闸,harness 在这里落地)
WHITELIST={"research_task":["web_search","read_url"],# 研究任务:只给查询工具"booking_task":["search_flight","book_ticket"],# 订票任务:才给下单工具}defallowed_tools(task_type:str)->list:return[tfortinALL_TOOLSift.nameinWHITELIST[task_type]]白名单 = 不给模型做坏事的机会,比事后追责便宜得多。
核心三:重试——什么时候重来,什么时候别重来
工具调用会失败,但失败要分类——有的值得重试,有的重试等于烧钱:
| 失败类型 | 例子 | 重试? |
|---|---|---|
| 网络抖动 | 连接超时、服务 503 | 重试(退避重试) |
| 限流 | 429 请求过多 | 重试(等几秒) |
| 参数错误 | 4xx、必填参数缺失 | 不重试(重试也一样错) |
| 逻辑错误 | 业务校验不通过 | 不重试(改参数才行) |
importtimedefcall_with_retry(fn,*args,max_retries=3):foriinrange(max_retries):try:returnfn(*args)exceptNetworkErrorase:# 5xx/429:退避重试time.sleep(2**i)# 1s, 2s, 4s 退避exceptParameterErrorase:# 4xx:参数错了,重试没用returnf"参数错误:{e}"# 把错误反馈给模型让它改return"多次重试仍失败"一条判断口诀:网络层的问题重试,业务层的问题反馈给模型改。把错误信息以工具结果形式喂回模型(第三篇讲过"反馈修正"),让模型自己改参数——这比重试高效得多。
核心四:断点恢复——聊到一半崩了,能接着聊
生产环境进程可能随时重启(发布、OOM、宕机)。用户聊到一半,会话就没了,这是最伤体验的事。
断点恢复 = 会话状态持久化:
defsave_checkpoint(session_id,messages,budget):db.save(session_id,{"messages":messages,"budget":budget})defrestore(session_id):state=db.load(session_id)returnstate["messages"],state["budget"]# 重建上下文继续跑注意:持久化的不只是 messages,还有预算剩余和已完成步骤——恢复后不能"白跑一遍"(那会重复扣费、重复下单)。
踩坑:Agent 死循环,怎么都停不下来
最经典的护栏翻车场景:
场景:工具返回了一个错误,Agent 觉得"我没调好,再调一次",结果同一个错误循环重试了几十次——步数没超限,但轮次全部浪费在"重试同一个注定失败的调用"上。
根因:步数上限只防"无限循环",防不了"低效循环"。
解法:加一层连续失败检测——连续 N 次(比如 3 次)工具调用都失败或结果相同,就认定卡住了:
CONSECUTIVE_FAIL_LIMIT=3defshould_stop(history)->bool:fails=0forhinhistory[-CONSECUTIVE_FAIL_LIMIT:]:ifh.is_errororh.result==history[-1].result:fails+=1returnfails>=CONSECUTIVE_FAIL_LIMIT卡住后的收尾方式也有讲究,不是沉默硬停,而是"强制提示一次 + 标记未解决":
助手:抱歉,我连续尝试了几次都未能完成(标记为未解决), 可能是接口暂时不可用。建议稍后再试或联系人工客服。把"未解决"状态记录下来——这是第 11 篇评测的数据来源:哪些问题 Agent 搞不定,比哪些问题它搞定了,更有价值。
小结
- 护栏 = 循环外的壳,不管"怎么干",只管"不许越线"
- 三层预算:每次 / 每轮 / 整个会话,超了就优雅收尾
- 工具白名单:按场景下发,写操作人工确认,不给做坏事的机会
- 重试分两类:网络层问题重试(退避),业务层问题反馈给模型改
- 断点恢复:messages + 预算 + 已完成步骤都持久化,崩溃了接着聊
- 死循环:连续失败检测 + 强制提示一次 + 标记"未解决"
下篇预告
护栏装好了,Agent 终于敢放心上线了。但上线只是开始——线上 Agent 出了岔子,你怎么知道?它调用了几次工具、花了多少钱、哪一步答错了,这些必须有记录。
下一篇讲生产部署与可观测:日志、trace、指标、告警,让线上 Agent 的一举一动都在掌握之中。
本系列路线(从 0 到 1):Agent 是什么 → 手写最小 ReAct → Function Calling 与工具设计 → 上下文管理 → Memory 记忆系统 → RAG 知识库 → Skill 自学习 → 编排模式与多 Agent → 给 Agent 装护栏 → 生产部署与可观测 → 评测与回归