☰
agency-agents:构建可观测、可控的自主智能体框架实践
2026/10/10 10:49:01 网站建设 项目流程

1. 一开始为什么要做 "agency-agents"

这几年"智能体"这个概念被炒得很热,各种框架层出不穷。可我自己把主流方案都搭了一遍之后,体会就一个字:虚。表面上都能聊、能调工具、能"多步推理",你真拿一个有压力的真实业务过去,它就原形毕露了——要么在大模型幻觉里打转,要么把一个简单任务拆成一堆互相矛盾的子步骤,要么压根不知道什么时候该停手。

我的需求其实很朴素:做一个真正能自主推进任务的系统,而不是"prompt + 循环 + 工具调用"的三件套。项目标题里的 agency 不是"代理机构"这个意思,而是指智能体的"能动性/自主决策能力"。所以这个项目叫 agency-agents,想做的就是把这种自主性落成一个可工程化、可观测、可控制的框架。

1.1 那些年我用"链式编排"的憋屈

在动工之前,我习惯性地先盘了一遍已有方案的痛点。最常见的问题是,大家把"智能体"理解成"在一个循环里反复调用大模型,直到它说做完"。听上去很聪明,实际跑起来你会看到它反复做同一件事,或者一次任务里调了十几次模型接口,结果答案还不如一次性直接问来得好。

第二个痛点是编排粒度太粗。很多框架把"步骤"定义成一个大函数,开发者只需要塞进去一个"角色设定",剩下全靠模型自由发挥。业务一旦复杂,这个"自由发挥"就会变成灾难。你没法在某个关键节点打断它、修正它,也没法事后复盘"它当时为什么选了这条路径"。

还有一个非常现实的问题:多智能体协作基本靠互相甩 prompt。A 智能体把一大段结果堆给 B 智能体,B 再堆给 C,链越长,信息损耗和 token 消耗越离谱。更别提多个智能体同时操作外部资源时,连最基本的并发互斥都没有。

1.2 "自主性"不是靠多写几个 prompt 就能实现的

我在设计 agency-agents 时给自己定了几条原则,后来发现它们几乎成了整个项目的地基:

第一,自主性必须建立在显式任务图之上。智能体不是从第一个字开始自由翱翔,而是先根据用户目标做任务分解,再把子任务组织成一张有依赖关系的图。每个节点都知道自己的输入、输出、可用工具和完成条件。这样它既能自主决定下一步,又能被人类在任意节点插话。

第二,每个决策都要留痕。大模型本身是黑盒,如果系统不主动记录它选了哪个工具、为什么选、输入输出是什么、中间有没有重试,那整个系统就是不可信的。我把"决策日志"当成一等公民,而不是事后从 API 调用记录里猜。

第三,一切可以被量化治理。预算上限、最大步数、最大分解深度、工具白名单、超时时间,这些不能只是配置文件里的摆设,而要在运行时被强制执行。自主不等于失控,我们要的是"在给定边界内的自主"。

这三条原则听上去简单,把它们同时落到一个可运行的代码架构里,花了我相当长的时间。后面几章我把关键设计一个个讲清楚。

2. 核心抽象:任务图、决策点与可回溯日志

agency-agents 的第一版其实并不复杂,我甚至没有用任何现成的 Agent 框架,核心就是三个数据结构:任务图(TaskGraph)、决策节点(DecisionNode)和工具路由(ToolRoute)。整个运行过程可以概括为:用户目标进来,规划器生成任务图,执行器沿着图推进,每个节点上智能体也许要选择工具、也许要生成子任务,每走一步都写日志。

2.1 从线性链条到 DAG:让 Agent 自己决定下一步

我见过很多"流程编排"框架其实只是一根直线:步骤一、步骤二、步骤三,最多加个 if 分支。真实业务里,任务之间往往是"依赖关系",而不是"先后关系"。比如一个"生成竞品分析报告"的任务,它包含三个子任务:抓取竞品页面、清洗数据、写分析结论。写分析结论依赖清洗后的数据,但抓取和清洗之间并不是严格的先后关系——你可以直接抓取原始数据,也可以先从已有数据库里取。

所以我用了一个有向无环图(DAG)来表示任务。每个节点有两种行为模式:

  • 原子任务(Atomic Task):直接调用某个工具完成,比如"调用某搜索接口获取关键词 Top 100"。
  • 复合任务(Composite Task):继续向下分解出子图,由规划器递归处理。

规划器在生成图的时候不搞"什么都让大模型自由发挥"的那一套,而是先做一次目标分析,把用户目标拆成几个语义块,然后做依赖推导,决定哪些块必须先做、哪些可以并行,最后做资源绑定,给每个块分配可用的工具和预期产出。

提示:DAG 不是越复杂越好。如果任务本身只有两个步骤,就老老实实走线性流程,别硬拆成五六个节点。后续我加了一个"最大分解深度"参数,默认是三层,最多五层,超过就触发"询问用户"的降级动作,防止规划器把简单任务切碎。

2.2 决策点记录:每一个"为什么"都要能在日志里找到答案

这是我认为整个项目里最值钱的设计。大多数框架把模型 API 的 raw response 存下来就算完事,但那些内容噪声太大,复盘时根本翻不动。我在每个节点内部增加了三个固定环节:

  1. 意图分析(Intent): 这一步到底要做什么?期望的产出格式是什么?
  2. 工具选择(Tool Selection): 当前可选工具列表是什么?模型选了哪个?候选得分是多少?
  3. 结果评估(Evaluation): 输出是否符合 schema 校验?是否触发重试?重试原因是什么?

每一次执行都会生成一条结构化日志,类似:

{ "node_id": "report_gen_001", "intent": "生成最终分析报告,需包含数据摘要与异常解释", "selected_tool": "report_writer", "candidate_tools": ["report_writer", "code_interpreter", "search_engine"], "input_summary": "清洗后数据表,共 38 行", "output_summary": "报告已生成,长度 1200 字", "retries": 1, "retry_reason": "输出缺少 '异常解释' 段落,schema 校验失败", "cost_usd": 0.023, "timestamp": "2025-06-19T08:23:19Z" }

这条日志不仅用于事后审计,它还有一个实时用途:当重试次数超过阈值时,执行器会跳过当前决策,把问题升级给人类处理。这样你就不会看到智能体在同一个节点上纠结十几次、烧掉一堆 token 的奇观。

2.3 工具路由不是函数调用,而是候选列表筛选

另一个容易出问题的点是工具调用。很多框架喜欢做"function calling",直接把模型的输出映射到 Python 函数。这听上去很顺,但模型经常瞎编参数,或者在一个不需要工具的上下文里强行调用工具。我在设计工具路由时没有走"直接映射"这条路,而是先让模型从候选列表里选,选完再做参数补全和校验:

  • 第一步,模型返回一个工具的标识符(比如工具 ID),而不是一大段 JSON 参数。
  • 第二步,系统从工具注册表里取出该工具的 JSON Schema,让模型基于这个 Schema 生成参数。
  • 第三步,参数必须通过 pydantic 校验,不过就再给一次机会,再不过就询问用户。

这个设计的额外收益是:模型不需要记住每个工具的完整参数格式,它只需要知道"这个工具能干什么、大概需要哪几类信息"。参数生成在被验证过的 Schema 约束下进行,输出质量会稳很多,幻觉参数的概率明显下降。

3. 多 Agent 协作:消息队列、资源租约与记忆隔离

单 Agent 的问题解决了,多 Agent 才是真正的硬骨头。agency-agents 里支持同时跑多个智能体,比如一个负责数据采集、一个负责数据分析、一个负责写作。如果让它们各跑各的、最后拼起来,你会发现"协作"两个字根本不成立——因为它们之间没有共同语言。

3.1 为什么不能让 Agent 之间直接互相传大段文本

最开始我的方案很简单:A 智能体把结果作为一段长文本塞给 B 智能体,B 看完再干活。结果遇到了两个大问题。第一是 token 消耗爆炸,一份数据表如果直接渲染成文本传过去,一次就得吃掉几千 token,多传几轮任务还没做完,账单先疯了。第二是"上下文污染",B 智能体分不清哪些是 A 的最终结论、哪些是 A 的过程困惑,容易被无关信息带偏。

后来我改成消息队列 + 结构化数据包的模式。智能体之间的通信不再传输"对话文本",而是传输一个带有明确 schema 的数据包:

  • task_id:属于哪个父任务
  • sender/receiver:谁发的、谁接收
  • payload_type:数据类型,比如table_chunk、summary_section、search_results
  • payload_ref:指向某个数据存储位置的引用,而不是直接塞大对象
  • ttl:这个包的有效时间,过期自动丢弃

B 智能体拿到的是"引用 + 元信息"而非整段文本。它需要具体数据的时候,再通过专门的数据读取工具去取。这个改动让多智能体协作从"人传话"变成了"仓库调度",信息失真和 token 浪费同时下降。

3.2 资源租约:避免两个 Agent 抢同一个文件

多智能体并发操作外部资源的时候,一定会遇到锁问题。比如数据采集智能体和数据清洗智能体可能同时访问某个缓存文件,如果没有互斥机制,会出现半写状态或者脏读。

我早期用过一个笨办法:让第二个智能体重试。"等一会儿再读也许就好了"——事实证明这个办法在大多数情况下只是把问题延后。最后我引入了一个非常简单的资源租约表,本质就是一个 SQLite 表,每次工具要访问共享资源前先申请租约:

CREATE TABLE resource_leases ( resource_key TEXT PRIMARY KEY, lease_holder TEXT NOT NULL, expires_at TIMESTAMP NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

工具访问资源的流程变成:先尝试申请租约,成功则执行操作,完成后释放;失败则等待并重试,超过等待上限就把节点标记为blocked,等待人工协调。这个表排除了死锁的可能,因为租约一定有截止时间,不会出现"等到天荒地老"的情况。

3.3 记忆要分账本:会话隔离与长期知识的取舍

做过智能体的人都知道,记忆是个双刃剑。让智能体记住上一轮对话,它能提供更好的上下文;让它记住太多历史,它就开始混淆边界、把旧任务的信息套到新任务上。我在设计记忆机制的时候用了"三本账":

  • 会话记忆(Ephemeral):只在当前 run 内有效,任务结束自动清空。存放节点之间的临时输出。
  • 项目记忆(Project):同一个项目下共享,带版本号。比如"这个任务的图结构怎么生成的,上一次失败在哪里"。
  • 长期知识(Long-term):写入独立的向量库,但每次写入必须经过一个"知识提炼"步骤,把原始信息压缩成结构化条目,并记录来源。

核心的取舍是:智能体永远不要直接读历史对话原文,只允许读"提炼后的知识条目"。这样做避免了上下文污染,也降低了 token 成本。

4. 一个完整的实测案例:自动生成竞品监测日报

理论说了一堆,还是上一个完整的案例来讲。假设我们老大要求每天上午十点自动产出一份"竞品监测日报",内容包括:三家竞品网站当天是否有更新、有哪些新变化、这些变化对我们应该意味着什么、以及三条运营建议。这个任务看起来不复杂,但前提是"每天自动跑",任何一步出问题都得知道自己怎么修复。

4.1 先把任务拆解清楚

用 agency-agents 搭建之前,我先把这个目标拆成了任务图:

  1. 数据采集节点: 抓取三家竞品的公开页,以及两个行业信息来源,输出原始 HTML 或 RSS 条目。
  2. 结构化节点: 从原始数据里抽取"时间、标题、摘要、链接"四要素,存为news_items表。
  3. 分析节点: 基于news_items生成 DIFF 对比,重点标记"新增关键词""价格变动""新上架功能"。
  4. 写作节点: 基于分析结果生成日报文本,要求附上数据来源链接。
  5. 交付节点: 调用邮件接口发送给指定收件人,并在内部看板上创建一条任务记录。

这里没有用"线性步骤",因为采集节点和结构化节点可以部分并行,而且写作节点依赖的是分析结果而非原始数据,这个依赖关系只有 DAG 才能清晰表达。规划器在运行时自动生成了这张图,我只在配置里给了它"数据源、产出物、发送对象"这些边界信息。

4.2 代码骨架:三个核心类怎么协作

架构层面只写了三个核心类,整体逻辑不算复杂:

class Coordinator: def __init__(self, planner, executor): self.planner = planner self.executor = executor def run(self, objective: str, context: dict) -> RunResult: task_graph = self.planner.plan(objective, context) execution_log = self.executor.execute(task_graph) return RunResult(graph=task_graph, log=execution_log) class Planner: def plan(self, objective: str, context: dict) -> TaskGraph: subtasks = self._decompose(objective, context) dependencies = self._resolve_dependencies(subtasks) return TaskGraph(subtasks=subtasks, dependencies=dependencies) class Executor: def execute(self, graph: TaskGraph) -> ExecutionLog: for node in graph.ordered_nodes(): if node.is_compound() and node.needs_decomposition(): node.graph = self.planner.plan(node.description, node.context) self._run_node(node) return ExecutionLog(...)

注意这里的ordered_nodes()不是简单的拓扑排序。它还要考虑"哪些节点可以并行执行、哪些节点目前处于 blocked 状态需要跳过"。执行器有一个工作线程池,并行节点会被分到不同线程,互不依赖的节点能同时跑。

4.3 实际跑起来之后的几个意外

第一个意外是规划器把"结构化节点"和"分析节点"合并成了一个,因为这两个节点之间只隔了一个字段转换,模型判断没必要分开。这个行为本身合理,但也暴露出一个问题:模型过度合并会牺牲可观测性。后来我加了规则,要求任意两个节点之间如果存在"数据格式转换"或"语义理解",就必须保持独立,不能合并。

第二个意外是采集节点反复请求同一个页面,因为前一次请求超时了。原始的设计里没有"去重缓存"的概念,导致同一个 URL 被请求了三次。后来我加了请求级缓存和幂等键,同一任务内相同参数的请求直接返回缓存结果,成本马上就下来了。

第三个意外是写作节点生成的日报里,引用了结构化节点里不存在的数据编号。这个属于典型的"模型幻觉输入错位",最后通过断言工具解决:写作节点在输出日报前,会自动检查"报告里引用的所有编号是否存在于数据源中",不过断言就重新生成,最多重试两次。实际效果很好,之后几乎没有出现过"无中生有"的引用。

5. 护栏与失败降级:让自主系统不作死

自主系统最怕的不是能力不够,而是能力够但行为不可控。我给 agency-agents 设计护栏的时候,参考的是弹幕游戏里那种"保命机制":你可以在边界内随便操作,但一旦越界,系统立刻接管。

5.1 成本上限与 Token 预算实时拦截

大模型不是无限便宜的,尤其一个长时间运行的多智能体系统,一天跑下来成本控制不好真的会吓人。我的做法是三层预算:

第一层是总预算,这个 run 最多允许消耗多少费用,超了直接停止并告知用户"任务因为预算不足被终止"。

第二层是节点预算,每个节点有自己的 token 上限。比如结构化节点预计最多输入 2000 输出 1000 token,实测超过 4 倍就判定为异常,触发重规划。

第三层是实时拦截器,每次调用大模型前,库里存有"这个节点的历史调用耗费"的计数,如果本次预计 token + 已消耗 token 会超过节点预算,拦截器会先拒绝这次调用,而不是等到账单出来再后悔。

预算这个东西,宁可设小一点频繁补充,也不要一开始设得很大指望模型节制。模型一点都不节制,它只会想着"完成任务",不会替你心疼钱。

5.2 失败降级路径:从工具摘除到询问用户

一个节点失败了怎么办?很多框架的做法是"重试 N 次然后挂掉"。这个体验很糟糕,尤其对于无人值守的任务。我设计了四级降级路径:

  1. 重试:如果是临时错误(网络抖动、API 超时),等待后重试,上限三次。
  2. 换工具:如果当前工具持续失败,尝试用同一意图下的其他工具替代。比如"搜索工具"挂了,就切换到"站点抓取工具 + 提取摘要"的组合方案。
  3. 重规划:如果工具都不可用,说明任务目标本身有问题,规划器基于已获得的部分数据重新分解任务,缩小目标范围。
  4. 询问用户:以上都失败,暂停任务,生成一份带上下文的问询请求,等待人类决定是跳过、修改目标还是终止。

这套降级路径救了我很多次。尤其是在外部接口不稳定的时候,任务不会直接挂掉,而是自动切换数据集继续完成,最后生成的报告里会明确标注"某数据源未更新,使用了备选方案"。读者至少能知道数据里哪部分可信。

5.3 逃逸检测:自循环、重复旧模式、工具调幻觉

自主系统最容易被吐槽的问题就是"跑飞了"。我在执行器里内置了三个逃逸检测器:

  • 循环检测:如果连续五个节点的输出高度相似(用向量相似度判断),判定为自循环,中断并触发重规划。
  • 动作熵检测:如果一个任务在十个以上节点之间反复切换但整体目标没有任何进展,说明规划可能失控,触发"询问用户"。
  • 参数幻觉检测:工具调用时,如果参数与当前上下文完全不相关(比如分析金融报告时突然调用"天气查询"工具),直接拒绝并记录。

这三个检测器都是纯代码实现的,没有依赖额外的模型调用,几乎零成本。它们的意义在于:把"智能体没问题"从一种信念变成一种可验证的断言。每次任务结束时,我会跑一份"系统自检报告",列出所有节点是否越过护栏,方便复盘。

6. 关于"什么时候不要用 Agent 方案"的一点唠叨

做完这个项目之后,我最大的收获不是代码,而是一个反过来的认知:不是所有任务都适合做智能体。如果你手上是一个流程相对固定、步骤不超过五个、输入输出模式很明确的业务,直接写函数编排就是最优解——便宜、稳定、好调试。智能体方案适合的是那些"目标明确但路径不确定"的事,比如开放性的调研、跨数据源的综合分析、需要根据实时进展调整方案的任务。

我内部已经形成了一张表,用来快速做技术选型判断:

任务特征建议方案理由
单一固定流程,步骤小于 5传统代码流程便宜、可控、无需调试智能体
部分节点需要模型理解上下文局部小模型节点只把"语义理解"交给模型,其余保持代码确定性
路径不固定,且需要动态决策智能体方案任务图动态生成的价值大
多数据源 + 多角色协作智能体 + 消息队列松耦合,单点失败不影响全局
预算极其敏感,毫厘必争别用智能体的 token 成本开销远大于固定流程

写到这里,agency-agents 已经稳定跑了快两个月,我团队里的"日报生成""行业调研""数据异常复盘"都迁移到了它上面。说句实话,它没有多炫酷,核心代码甚至有点朴素——一个任务图、一套决策日志、三个护栏检测器、一个消息队列。但正是这些朴素的东西,让"自主性"从一句口号变成了可以审计、可以干预、可以控制的工程事实。

最后分享一个经验:如果你也在做类似的事,先别急着写代码。把你手头最熟悉的一个业务拿下来,拆成任务图,跑通一遍,再考虑通用化。直接搭一个大而全的平台,大概率会卡在"你根本不知道哪些抽象是必要的"这个阶段。从真实的业务长出框架,才是最有生命力的路径。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询