本文是 Agent 工程里的经典架构 系列的一篇。文中的业务场景和代码为讲解设计,产品能力以所链接的官方资料为准。

一个团队只有一个 Agent 时,知道它调用了哪些工具、用了多少预算并不难。十几个团队分别接入模型、数据库和外部接口以后,问题就变成了:哪些 Agent 还在运行,谁给了它权限,某次费用增长来自哪里,以及出现故障时能否快速停止影响。

按官方介绍,Agent/works是 Thoughtworks 面向企业 Agent 的治理方案,关注行为控制、可观测性和成本管理。本文借这个入口讨论治理能力的工程用意,不把官网宣传直接当成效果验证,也不假设存在未公开的产品 API。

我会先区分执行和治理。执行层负责调用模型与工具,治理层定义哪些身份可以做哪些事、消耗多少预算、留下哪些记录。两者可以部署在不同位置,但所有关键调用都应经过约束点。只有一个展示运行数量的仪表盘,还不能算真正控制了系统行为。

机制图:在执行入口统一管理策略、用量和可追踪记录。

预算是一个很具体的例子。调用开始前需要预留额度,结束后根据实际用量结算。如果等所有请求完成后才统计,多路并发可能同时认为余额充足,最后一起超支。下面用线程锁演示同一进程内的原子预留;额度是抽象单位,不对应任何厂商价格。

from threading import Lock


class Budget:
    def __init__(self, limit):
        self.limit, self.spent = limit, 0
        self.pending, self.closed = {}, set()
        self.lock = Lock()

    def reserve(self, key, ceiling):
        with self.lock:
            if ceiling <= 0 or key in self.pending or key in self.closed:
                return False
            available = self.limit - self.spent - sum(self.pending.values())
            if ceiling > available:
                return False
            self.pending[key] = ceiling
            return True

    def settle(self, key, actual):
        with self.lock:
            ceiling = self.pending[key]
            if not 0 <= actual <= ceiling:
                raise ValueError("实际用量超出约定上限,需异常清算")
            self.spent += actual
            del self.pending[key]
            self.closed.add(key)


budget = Budget(20)
assert budget.reserve("call-1", 15)
assert not budget.reserve("call-2", 10)
budget.settle("call-1", 8)
assert budget.reserve("call-2", 10)
assert not budget.reserve("call-1", 1)
assert budget.spent + sum(budget.pending.values()) <= budget.limit
print("并发额度先预留,再结算,操作编号不能复用")

完整示例可下载,使用 Python 3.9 或以上版本运行:python3 example.py。只依赖标准库,不调用真实业务服务。

示例要求实际用量不超过预留上限。这项前提需要调用适配器通过输入估算、输出上限等方式落实;如果不能确定成本上界,就只能提供软预算和超限处理,不能宣称严格防止超支。跨进程部署还需要共享存储中的原子操作,单机锁无法保护其他 Worker。

调用结果未知时也不能立即释放预留额度。模型服务可能已经完成推理,只是响应丢失,费用仍然发生。需要查询、对账或超时清算策略。重复回调则不能重复结算,因此预留记录需要稳定标识和状态。这个问题与幂等、结果未知和持久化,实际上属于同一组工程约束。

权限管理同样应该有身份与范围。一个负责查询合同的 Agent 不必拥有所有客户数据的导出权限;临时任务结束后,临时授权也应失效。工具注册只是知道“有什么能力”,真正治理还要检查“谁在什么条件下可以使用”。策略变化需要审计,避免管理员一次修改影响所有业务却没有记录。

可观测性要围绕任务组织。任务标识串起模型调用、工具执行、审批和业务结果,才能解释一次费用增长是否带来了更多成功任务。只收集 token 数会遗漏重复失败、长时间等待和人工补救;只看接口延迟又可能看不到整件事一直没有结束。

我也会给治理系统设计故障策略。策略服务不可用时,哪些只读请求可以有限降级,哪些写入必须暂停,需要提前决定。治理服务本身可能成为共享依赖,应该有容量、缓存一致性和可用性设计。随意缓存授权虽然提高了性能,也可能延迟权限撤销生效。

选择现成平台时,可以用同一组用例验证:跨租户调用是否被拒绝,预算是否能承受并发竞争,旧授权撤销多久生效,任务记录能否导出,模型或框架替换后哪些能力仍然有效。验证应以实际部署方案为准,不能仅凭产品功能表判断集成成本。

对小规模团队,统一的工具网关、身份管理、费用记录和任务追踪,可能已经覆盖当前需要。随着团队和调用链增长,再评估集中治理平台的收益。Agent 的数量不是成熟度指标;能够解释它们做了什么、限制它们能做什么,并把成本和业务结果联系起来,才更接近可以持续运营的系统。

上一篇 · 系列目录