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 存下来就算完事但那些内容噪声太大复盘时根本翻不动。我在每个节点内部增加了三个固定环节意图分析Intent 这一步到底要做什么期望的产出格式是什么工具选择Tool Selection 当前可选工具列表是什么模型选了哪个候选得分是多少结果评估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_resultspayload_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 搭建之前我先把这个目标拆成了任务图数据采集节点 抓取三家竞品的公开页以及两个行业信息来源输出原始 HTML 或 RSS 条目。结构化节点 从原始数据里抽取时间、标题、摘要、链接四要素存为news_items表。分析节点 基于news_items生成 DIFF 对比重点标记新增关键词价格变动新上架功能。写作节点 基于分析结果生成日报文本要求附上数据来源链接。交付节点 调用邮件接口发送给指定收件人并在内部看板上创建一条任务记录。这里没有用线性步骤因为采集节点和结构化节点可以部分并行而且写作节点依赖的是分析结果而非原始数据这个依赖关系只有 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(graphtask_graph, logexecution_log) class Planner: def plan(self, objective: str, context: dict) - TaskGraph: subtasks self._decompose(objective, context) dependencies self._resolve_dependencies(subtasks) return TaskGraph(subtaskssubtasks, dependenciesdependencies) 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 次然后挂掉。这个体验很糟糕尤其对于无人值守的任务。我设计了四级降级路径重试如果是临时错误网络抖动、API 超时等待后重试上限三次。换工具如果当前工具持续失败尝试用同一意图下的其他工具替代。比如搜索工具挂了就切换到站点抓取工具 提取摘要的组合方案。重规划如果工具都不可用说明任务目标本身有问题规划器基于已获得的部分数据重新分解任务缩小目标范围。询问用户以上都失败暂停任务生成一份带上下文的问询请求等待人类决定是跳过、修改目标还是终止。这套降级路径救了我很多次。尤其是在外部接口不稳定的时候任务不会直接挂掉而是自动切换数据集继续完成最后生成的报告里会明确标注某数据源未更新使用了备选方案。读者至少能知道数据里哪部分可信。5.3 逃逸检测自循环、重复旧模式、工具调幻觉自主系统最容易被吐槽的问题就是跑飞了。我在执行器里内置了三个逃逸检测器循环检测如果连续五个节点的输出高度相似用向量相似度判断判定为自循环中断并触发重规划。动作熵检测如果一个任务在十个以上节点之间反复切换但整体目标没有任何进展说明规划可能失控触发询问用户。参数幻觉检测工具调用时如果参数与当前上下文完全不相关比如分析金融报告时突然调用天气查询工具直接拒绝并记录。这三个检测器都是纯代码实现的没有依赖额外的模型调用几乎零成本。它们的意义在于把智能体没问题从一种信念变成一种可验证的断言。每次任务结束时我会跑一份系统自检报告列出所有节点是否越过护栏方便复盘。6. 关于什么时候不要用 Agent 方案的一点唠叨做完这个项目之后我最大的收获不是代码而是一个反过来的认知不是所有任务都适合做智能体。如果你手上是一个流程相对固定、步骤不超过五个、输入输出模式很明确的业务直接写函数编排就是最优解——便宜、稳定、好调试。智能体方案适合的是那些目标明确但路径不确定的事比如开放性的调研、跨数据源的综合分析、需要根据实时进展调整方案的任务。我内部已经形成了一张表用来快速做技术选型判断任务特征建议方案理由单一固定流程步骤小于 5传统代码流程便宜、可控、无需调试智能体部分节点需要模型理解上下文局部小模型节点只把语义理解交给模型其余保持代码确定性路径不固定且需要动态决策智能体方案任务图动态生成的价值大多数据源 多角色协作智能体 消息队列松耦合单点失败不影响全局预算极其敏感毫厘必争别用智能体的 token 成本开销远大于固定流程写到这里agency-agents 已经稳定跑了快两个月我团队里的日报生成行业调研数据异常复盘都迁移到了它上面。说句实话它没有多炫酷核心代码甚至有点朴素——一个任务图、一套决策日志、三个护栏检测器、一个消息队列。但正是这些朴素的东西让自主性从一句口号变成了可以审计、可以干预、可以控制的工程事实。最后分享一个经验如果你也在做类似的事先别急着写代码。把你手头最熟悉的一个业务拿下来拆成任务图跑通一遍再考虑通用化。直接搭一个大而全的平台大概率会卡在你根本不知道哪些抽象是必要的这个阶段。从真实的业务长出框架才是最有生命力的路径。 agency-agents：构建可观测、可控的自主智能体框架实践
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 存下来就算完事但那些内容噪声太大复盘时根本翻不动。我在每个节点内部增加了三个固定环节意图分析Intent 这一步到底要做什么期望的产出格式是什么工具选择Tool Selection 当前可选工具列表是什么模型选了哪个候选得分是多少结果评估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_resultspayload_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 搭建之前我先把这个目标拆成了任务图数据采集节点 抓取三家竞品的公开页以及两个行业信息来源输出原始 HTML 或 RSS 条目。结构化节点 从原始数据里抽取时间、标题、摘要、链接四要素存为news_items表。分析节点 基于news_items生成 DIFF 对比重点标记新增关键词价格变动新上架功能。写作节点 基于分析结果生成日报文本要求附上数据来源链接。交付节点 调用邮件接口发送给指定收件人并在内部看板上创建一条任务记录。这里没有用线性步骤因为采集节点和结构化节点可以部分并行而且写作节点依赖的是分析结果而非原始数据这个依赖关系只有 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(graphtask_graph, logexecution_log) class Planner: def plan(self, objective: str, context: dict) - TaskGraph: subtasks self._decompose(objective, context) dependencies self._resolve_dependencies(subtasks) return TaskGraph(subtaskssubtasks, dependenciesdependencies) 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 次然后挂掉。这个体验很糟糕尤其对于无人值守的任务。我设计了四级降级路径重试如果是临时错误网络抖动、API 超时等待后重试上限三次。换工具如果当前工具持续失败尝试用同一意图下的其他工具替代。比如搜索工具挂了就切换到站点抓取工具 提取摘要的组合方案。重规划如果工具都不可用说明任务目标本身有问题规划器基于已获得的部分数据重新分解任务缩小目标范围。询问用户以上都失败暂停任务生成一份带上下文的问询请求等待人类决定是跳过、修改目标还是终止。这套降级路径救了我很多次。尤其是在外部接口不稳定的时候任务不会直接挂掉而是自动切换数据集继续完成最后生成的报告里会明确标注某数据源未更新使用了备选方案。读者至少能知道数据里哪部分可信。5.3 逃逸检测自循环、重复旧模式、工具调幻觉自主系统最容易被吐槽的问题就是跑飞了。我在执行器里内置了三个逃逸检测器循环检测如果连续五个节点的输出高度相似用向量相似度判断判定为自循环中断并触发重规划。动作熵检测如果一个任务在十个以上节点之间反复切换但整体目标没有任何进展说明规划可能失控触发询问用户。参数幻觉检测工具调用时如果参数与当前上下文完全不相关比如分析金融报告时突然调用天气查询工具直接拒绝并记录。这三个检测器都是纯代码实现的没有依赖额外的模型调用几乎零成本。它们的意义在于把智能体没问题从一种信念变成一种可验证的断言。每次任务结束时我会跑一份系统自检报告列出所有节点是否越过护栏方便复盘。6. 关于什么时候不要用 Agent 方案的一点唠叨做完这个项目之后我最大的收获不是代码而是一个反过来的认知不是所有任务都适合做智能体。如果你手上是一个流程相对固定、步骤不超过五个、输入输出模式很明确的业务直接写函数编排就是最优解——便宜、稳定、好调试。智能体方案适合的是那些目标明确但路径不确定的事比如开放性的调研、跨数据源的综合分析、需要根据实时进展调整方案的任务。我内部已经形成了一张表用来快速做技术选型判断任务特征建议方案理由单一固定流程步骤小于 5传统代码流程便宜、可控、无需调试智能体部分节点需要模型理解上下文局部小模型节点只把语义理解交给模型其余保持代码确定性路径不固定且需要动态决策智能体方案任务图动态生成的价值大多数据源 多角色协作智能体 消息队列松耦合单点失败不影响全局预算极其敏感毫厘必争别用智能体的 token 成本开销远大于固定流程写到这里agency-agents 已经稳定跑了快两个月我团队里的日报生成行业调研数据异常复盘都迁移到了它上面。说句实话它没有多炫酷核心代码甚至有点朴素——一个任务图、一套决策日志、三个护栏检测器、一个消息队列。但正是这些朴素的东西让自主性从一句口号变成了可以审计、可以干预、可以控制的工程事实。最后分享一个经验如果你也在做类似的事先别急着写代码。把你手头最熟悉的一个业务拿下来拆成任务图跑通一遍再考虑通用化。直接搭一个大而全的平台大概率会卡在你根本不知道哪些抽象是必要的这个阶段。从真实的业务长出框架才是最有生命力的路径。