别把 Agent 的运行记录当成授权证据:用 SHACKLE 给工具调用加可复现的熔断门
Agent 的事故并不总是从“模型答错”开始。更常见的路径是:某次工具调用得到 401、超时或格式错误;Agent 把失败信息塞回上下文后再试一次;随后继续调用同一个工具、重复提交同一个请求,或者在没有人确认的情况下恢复一条此前已经被拒绝的动作。日志会显示它“做过检查”,但日志存在不等于下一次调用已经获准。
SHACKLE 是一个 Python 运行时治理层,PyPI 包名为 pyshackle。它把每次待执行的工具调用放到一个明确的决策面:ALLOW、DENY 或 HITL(暂停并等待人工处理)。项目同时发布了 SP/1.0 一致性规范、参考 decide(config, state, call) 实现和公开测试向量。对开发团队而言,值得关注的不是又多一个“安全提示词”,而是它试图把“这次为什么放行”变成可运行、可复查的程序结果。
先把问题拆开:历史可见,不代表当前可执行
很多 Agent 框架都会保存对话、工具结果和 checkpoint。它们对恢复任务很有帮助,但恢复机制有一个危险的默认假设:既然某个动作在历史里出现过,重放它似乎就是合理的。
SHACKLE 的规范刻意把这两个概念分离:一条历史记录只能说明动作曾被观察到;真正执行前,运行时仍要根据当前配置、执行状态和调用内容重新给出 verdict。它的核心判断模型写为 Valid(τ) ⇔ Required(τ) ⊆ Supported(τ):一个状态转换所要求的能力,必须全部落在系统明确支持的能力集合里。这个表述的工程价值在于,重试、恢复、人工修改后的继续执行都不再是“上下文里看起来合理”的隐式分支,而是可测试的状态转换。
因此,SHACKLE 不适合作为权限系统的替代品。数据库 RBAC、云 IAM、支付审批仍应在目标系统中执行;它更像运行时的第二道门,负责在调用抵达下游前发现重复、超预算、超时与需要人类决断的转换。
四类熔断条件,分别解决什么
项目提供的参考运行时围绕四种条件建立断路器:
| 条件 | 关注的失败模式 | 处理意义 |
| — | — | — |
| 重复工具调用 | 同一工具与同一输入反复出现,或输入已带错误信号 | 防止把 401、500、timeout 当成可无限重试的问题 |
| 预算超限 | 累积的 token 成本超过上限 | 让一次会话的成本边界显式化 |
| 执行超时 | 工作流超过设定墙钟时间 | 阻止卡死的线程或迟迟不返回的下游依赖 |
| 调用次数上限 | 工具调用总量超过阈值 | 限制“不断换参数试试”的级联行为 |
这里有一个很重要的边界:成本是在调用后的 usage 信息中累计读取的。因此预算熔断能阻止后续循环,却未必能撤销刚刚越过上限的那一次请求。不要把它描述成每个 API 请求都能在服务端原子拦截的账单防火墙;更稳妥的做法是把单次请求上限交给模型网关或供应商配额,同时由运行时断路器控制会话级后果。
从现有 CrewAI 工作流接入
SHACKLE 的 README 展示了用 Guard 装饰器保护既有运行入口的形式。下面的预算、重复调用阈值与超时时间均是示例配置,团队应按任务风险和模型成本调整:
from shackle import Guard
from crewai import Crew, Agent, Task
researcher = Agent(
role="researcher",
goal="只整理已验证的资料",
backstory="遇到权限或网络错误时停止并上报",
)
task = Task(
description="调研一个公开技术主题,并输出来源列表",
expected_output="带来源链接的简要说明",
agent=researcher,
)
crew = Crew(agents=[researcher], tasks=[task])
@Guard(budget=0.25, max_repeat_calls=3, timeout_seconds=180)
def run_research():
return crew.kickoff()
result = run_research()
print(result)
示例的关键不在装饰器只有一行,而在于应把它放在真正会触发模型与工具调用的边界上。如果只装饰一个预处理函数,后续 Agent loop 仍能绕过检查。接入前先用故意失败的测试工具制造“相同输入连续失败”的场景,确认断路器确实包住了实际执行路径。
触发后,项目的交互式控制台提供 Resume/Reset、Skip 与 Abort 三种处理方向。生产环境要谨慎看待这个交互模式:无头 worker、定时任务或 HTTP 请求线程通常没有人可以在终端上及时选择。此类场景可把 HITL 结果映射为队列中的待审批任务、失败状态或工单,而不是让进程无限等待。README 也将其定位为本地开发、调试、CLI Agent 和受监督工作流更合适;没有人工响应通道的 API 服务,应采用能自动处理的框架原生策略。
用测试向量,而不是口头承诺验证规则
只写“支持 HITL”还不够。SHACKLE 随项目提供一致性 fixtures,包含决策核心与人工处理状态转换的测试案例。克隆源码并安装开发依赖后,可以先运行项目给出的测试命令:
git clone https://github.com/Fame510/SHACKLE.git cd SHACKLE pip install -e . pytest tests/test_conformance.py
把这一步放进自己的 CI 时,建议额外写三类贴近业务的集成测试。第一类是重复失败:固定让下游返回 401,断言第三次或达到阈值后不再发出请求。第二类是恢复路径:先拒绝高风险动作,再验证重放同一动作不会因为历史中存在记录而自动执行。第三类是人工修改:审批人修改参数后,断言修改后的调用会被重新判断,而不是沿用旧 verdict。
公开 fixtures 能证明实现遵循它定义的合约,但不能替你证明 IAM、数据库事务或业务审批本身正确。因此测试报告应同时记录:被保护的调用边界是什么、哪些工具绕过了 Python 进程、以及 HITL 被谁在多长时间内处理。
分阶段上线:先观测,再阻断
在已有生产 Agent 上直接把所有 DENY 变成硬失败,往往会把历史上被隐藏的依赖问题一次性暴露出来。更低风险的路径是分三步推进。第一周只记录 verdict、触发条件和调用指纹,不改变执行结果;重点统计重复调用真正来自模型重试、SDK 自动重试,还是任务编排器的恢复逻辑。第二步只对明显无副作用的查询工具启用拒绝和超时,观察任务完成率、人工介入量及误拦截案例。最后才把写数据库、发消息、支付或部署等动作接入 HITL 与更严格的预算上限。
同时要为每一次拒绝保留最小但足够的证据:策略版本、会话 ID、工具名、参数摘要、触发的阈值和最终处置。参数摘要应避免保存密钥、完整用户内容或原始凭据。这样排查时能够回答“为什么被拦”,又不会把治理日志本身变成新的敏感数据仓库。规则变更也应走代码评审和回归测试;否则一次为了降低误报而放宽的阈值,可能悄悄重新打开无限重试的通道。
何时值得引入,何时不必强上
如果团队正在本地调试 CrewAI、AutoGen、LangGraph 一类多工具工作流,且经常遇到错误重试、token 失控或需要人工接管,运行时熔断层能将隐性的失败循环变成显式事件。它也适合在接入新工具前作为“先停住再看”的缓冲层。
反过来,若工作流只有受严格 IAM 控制的单次 API 调用,或生产系统完全无人值守却没有异步审批通道,直接启用交互式断路器可能只会增加阻塞。此时先定义失败返回、重试预算、死信队列和告警,再决定是否把 SHACKLE 放在开发与验证阶段。
可靠的 Agent 治理不是让模型承诺更谨慎,而是把每一次可能造成副作用的转换放进可检查的边界。运行记录当然要保留,但下一次调用是否执行,应由当下的策略、状态和可复现测试共同决定。