企业官网建设流程全解析

这些年“Agents”这个火热，各种框架层出不穷，但每次看到“agency-agents”这种命名，我还是会忍不住多琢磨几秒。它不像“agent-demo”那么直白，也不是某个模型厂商的官方仓库名，而更像是一个描述“一群智能体如何组织起来干活”的集合概念。我刚接触这个方向的时候，也以为它只是一个简单的学习项目，直到自己动手搭了一遍，才发现“agent”本身不是难点，“agency”也就是多个智能体之间的协作、权责分配、消息流转才是真正烧脑的部分。

这篇博文就围绕我在一个模拟项目X里，如何把一个“一堆智能体”的混沌想法，逐步落地成一个可运行、可扩展的“agency-agents”协作框架。我会把设计思路、架构选择、踩过的坑、还有关键的代码骨架都写出来，尽量做到既能让刚入门的朋友看懂，也能让已经在做多智能体应用的同学找到点共鸣。

1. 破题：从“agent”到“agency-agents”，到底是多了什么

1.1 拆解标题里的三层含义

先说结论，我认为“agency-agents”这个组合词，至少表达了三层意思：

• 第一层：它是一个智能体的集合。单个agent能处理的问题有限，比如让它写一段代码、做一次检索、整理一份摘要，这些都算是单点能力。而“agency-agents”天然暗示着“一批”智能体同时存在，彼此之间有关系、有分工。
• 第二层：它强调“代理机构”式的组织感。在真实世界里，一家agency有策划、有执行、有审核、有对接客户的岗位。映射到系统里，就是不同的agent扮演不同职能角色，而不是一堆同样的agent各干各的。这种角色化设计，是多智能体系统和“并行调用多个模型”最本质的区别。
• 第三层：它指代“可以协作的框架”。既然是复数，又带有组织含义，那实现上就必须有通信机制、任务分发机制、状态同步机制。哪怕只是一个很简化的实现，也得有这些骨架，否则顶多算一个“多进程脚本”。

我当时拿到“agency-agents”这个题目时，第一反应是：需求方到底想要一个能用的产品原型，还是一个用来验证多智能体协作概念的技术Demo？后来通过和某导师沟通，确认了定位是后者——一个可扩展的、代码清晰的参考框架，目的是验证“多角色智能体协作解决复杂任务”的可行性。

1.2 为什么单Agent模式不能满足需求

很多人刚接触大模型应用时，会习惯性地把任务丢给一个超长的Prompt，指望一个Agent全流程搞定。我也走过这条路，实际结果通常是：

• Prompt写得越长，模型越容易丢失早期指令。尤其是超过上下文窗口的临界点之后，前面的“角色设定”和“步骤要求”会被后续内容稀释，甚至是遗忘。
• 任务一旦出现异常分支，单个Agent只能靠着“if-else”式的自我推理去兜底，可它往往既要做决策，又要做执行，还得自查错误，脑袋里塞了太多事，效果自然下降。
• 单Agent模式下，所有的工具调用、中间结果、上下文记忆都耦合在一起，出了问题很难定位，是模型推理错了，还是工具返回异常，还是上下文被污染？排查成本极高。

而“agency-agents”模式就好比把一家公司的流程搬到系统里：策划Agent只负责拆解目标和制定计划，执行Agent只负责调用工具和生成内容，审核Agent只负责挑错。各司其职，上下文被隔离，问题域被切分。实测下来，整体成功率比单Agent硬扛要稳定得多。

2. 方案选型：为什么我选择“规划-执行-审核”三级协作

2.1 常见的多Agent协作模式对比

多智能体协作的实现方式其实有很多种，我之前大概梳理过三类主流模式，各有各的适用场景：

这次的项目X采用的是“编排模式”的变体，叫“规划-执行-审核”。选择它不是因为它最先进，而是因为它在可控性和效果之间最平衡。项目要求是用尽量简单的方式验证多智能体的价值，那就必须让责任边界足够清晰，每一个环节出的问题都能被单独观测和修正。

2.2 固定流程编排的取舍

在项目初期，我也考虑过要不要上更“智能”的动态编排，也就是让模型自己决定下一步该调用哪个Agent。后来果断放弃了，原因有三点，都是实操层面的：

• 动态编排意味着每一个Agent返回的格式都可能是半结构化的，解析成本高，错误率高。模型生成的JSON字段经常缺胳膊少腿，处理这些“格式垃圾”的时间比写业务逻辑还长。
• 协商模式的token消耗非常大。一次任务来回二十轮对话，成本翻好几倍，而项目X只是验证概念，不需要无限探索。
• 动态编排的调试体验极其痛苦。今天能跑通的链路，明天换一个模型版本可能就走不通了，完全不可控。

所以最后采用了一个看似“笨”但非常可靠的方法：固定流程 + 角色隔离 + 节点级重试。也就是流程的骨架是固定的，但每个节点内部允许根据输出质量自动触发一次重试或修复。这种方式既保留了多Agent协作的优势，又把实现复杂度控制在了合理范围内。

3. 核心架构：消息总线、任务状态机和角色定义

3.1 消息总线：Agent之间怎么说话

在“agency-agents”框架里，Agent之间不能直接互相调用方法，否则就变成强耦合了。我参考了消息队列的思路，实现了一个轻量级的内存消息总线。所有的Agent实例只和总线打交道，不关心消息是谁发的，也不关心最终谁会消费。

消息总线的数据结构非常简单，但非常关键。我定义了这样几个字段：

@dataclass class AgentMessage: message_id: str sender: str receiver: str msg_type: str # task_request / task_result / review_report / error payload: dict timestamp: float retry_count: int = 0

别小看这个结构。“receiver”字段决定了消息是点对点发送还是广播；“msg_type”字段则方便消费者快速路由。之前我见过很多项目把所有信息都塞进payload，然后用一个大的type字段区分，结果后面根本没法维护。这样拆开之后，每个Agent收到消息时只需要匹配自己关心的msg_type，其他一律忽略，逻辑非常清爽。

3.2 任务状态机：一个任务从生到死的完整轨迹

多智能体系统中最容易“失控”的地方就是任务状态。A任务被B、C、D三个Agent同时处理时，谁先完成，谁在等待，谁报错了，这些必须一目了然。所以我在框架里建了一个任务状态机，流转路径是：

pending -> queued -> dispatching -> processing -> reviewing -> completed \-> failed -> retry

状态机的实现并不复杂，核心是围绕任务ID维护状态字典，并且每次状态变化时都会触发一次回调，把当前状态推送给相关的Agent。有一次我在测试中发现，执行Agent明明已经往总线上发了task_result，但规划Agent那边迟迟没有反应。排查了好一会儿才发现，是任务状态还挂在“processing”上，没有正确更新为“reviewing”。当时就在状态机里加了一个强制校验：只有上游状态合法，才允许切换到下一个状态。从那以后，这类问题基本绝迹了。

3.3 角色定义：三个核心Agent的职责边界

项目X里我一共定义了三种角色，正好对应前面提到的三级协作：

• 规划Agent（Planner）：负责任务的拆解、步骤编排、以及最终结果的汇总。它不直接调用底层工具，只做“脑力劳动”。
• 执行Agent（Executor）：负责任务的具体执行，包括调用搜索工具、文档解析工具、代码生成工具等。它会读取规划Agent下发的子任务，执行并生成结果。
• 审核Agent（Reviewer）：负责对执行结果进行评估和纠错。如果发现质量问题，打回重做；如果通过，则上报给规划Agent进行汇总。

这三者的关系有点像现实项目里的甲方、干活的和监理。规划者不干活，只管方向；干活的不决策，只管实现；监理不讨好任何人，只负责挑毛病。角色边界一旦清晰，系统行为的可预期性就会大幅提升。

4. 实操过程：从零搭建一个最小可用的agency-agents框架

4.1 环境准备与依赖选型

先说环境，我的建议是不要在这上面花太多时间纠结版本，选一套自己最熟的组合就行。项目X用的是Python 3.10版本，模型接口走OpenAI兼容的SDK，消息总线从零手写，没有引入Celery这类重型任务队列。工具调用方面，直接把几个常见的工具封装成了函数注册表，方便后续扩展。

依赖只需要三个包：pydantic用于定义数据结构，openai用于调用模型接口，python-dotenv用于管理环境变量。如果只跑Demo，其实连pydantic都可以不用，但为了后续维护方便，加上它会舒服很多。

4.2 定义消息模型和工具注册表

整个框架的代码量其实不大，最复杂的部分反而是消息模型的定义。我用Pydantic定义了所有Agent之间通信的结构，同时在工具层做了一层注册机制。这样执行Agent就不需要写死“能干什么”，而是运行时查注册表，判断自己有没有能力处理某个子任务。

class ToolRegistry: _tools = {} @classmethod def register(cls, name): def decorator(func): cls._tools[name] = func return func return decorator @classmethod def get(cls, name): return cls._tools.get(name) @classmethod def has(cls, name): return name in cls._tools

之后，我用@ToolRegistry.register("web_search")这样的方式，把一个个真实工具函数挂载进去。这种注册表模式的优点在于扩展性极好，想加一个工具，不需要改执行Agent的内部逻辑，只需要新写一个函数并注册即可。这个设计我在很多实际项目中反复使用，成本低，效果却非常显著。

4.3 核心循环：消息路由与任务分发

系统的核心引擎其实就是一个持续运行的消息循环。它不断从消息总线里取出消息，根据消息的receiver字段找到对应的Agent，然后调用Agent的handle方法。听起来很简单，但真正实现时有两个细节容易踩坑。

第一个细节是消息的幂等性。因为消息循环可能会因为异常而重复读取一条消息，如果处理逻辑不是幂等的，就会产生重复执行，浪费token还污染结果。我的解决方案是给每条消息带上message_id，并在已处理集合里留痕，重复消息直接丢弃。

第二个细节是超时处理。每个Agent处理消息时，如果等待模型响应时间过长，不能无限等下去。我引入了asyncio.wait_for机制，超时后会触发自定义的“AgentBusyError”异常，让规划Agent重新调度相邻节点。关键代码大致如下：

async def dispatch_message(message: AgentMessage): agent = agent_registry[message.receiver] try: result = await asyncio.wait_for( agent.handle(message), timeout=30.0 ) return result except asyncio.TimeoutError: logger.error(f"Agent {message.receiver} timeout") raise AgentBusyError(message.receiver, message)

4.4 规划Agent：如何把需求拆成可执行的子任务列表

规划Agent是整个框架里“最像人”的一个环节，它负责把一个总任务拆分成多个子任务，并为每个子任务指定执行Agent和期望输出格式。系统提示词中必须明确约束它：只输出结构化的JSON数组，不要输出任何解释性文字，否则后面解析起来会非常痛苦。

我给规划Agent定义的任务拆分数据结构如下：

class SubTask(BaseModel): task_id: str description: str agent_role: str requires_tool: list[str] output_format: str

每个字段都有它的意义。agent_role决定由哪个角色的Agent执行，requires_tool是一个工具列表，执行Agent会先检查自己注册表里有没有这些工具，如果缺工具就直接返回错误，而不是硬着头皮瞎编。实测下来，这种“工具预检”机制能挡住至少半数的无效任务。因为模型有时候会展开想象，让执行Agent去做它根本没能力做的事，与其等它失败，不如一开始就拒绝。

4.5 执行Agent：工具调用与结果标准化

执行Agent的工作相对机械但极为重要：接收子任务，解析需要调用的工具，执行工具，然后把结果整理成标准格式返回。这里最大的坑在于工具的返回结果五花八门，有些是纯文本，有些是JSON，有些是渲染后的HTML。如果这些结果直接塞回大模型上下文，不仅浪费token，还可能干扰模型输出格式。

所以我在执行Agent里强制加了一个“标准化输出”步骤。不管工具返回什么内容，最终往外发消息时，payload里的content字段必须是规范化的文本，通常不超过1500字。超出部分用截断和摘要处理。这个过程听起来像是在“损失信息”，但实际上是为了保住消息流的有序性。毕竟执行Agent的下游是审核Agent，它需要的是结构清晰、逻辑自洽的结果，而不是一堆原始垃圾。

标准化处理的核心逻辑，其实就是调用一次轻量级模型做摘要，把工具原始返回结果压缩成“核心事实”和“关键引用”两个部分。实测下来，这种压缩能让后续所有Agent的推理准确率显著提升，因为上下文里的信噪比提高了。

4.6 审核Agent：质量关卡如何设计

审核Agent的设计是整个系统里最容易被低估的环节。很多人觉得审核Agent就是拿另一个模型生成一句“看起来不错”，那就大错特错了。真正的审核Agent必须能提出具体的修改建议，并且判断结果是否达到“放行”标准。

我设计的审核维度有四个：

• 完整性：结果是否覆盖了任务描述中的所有要求。
• 准确性：事实、数据、引用是否可靠，是否有一本正经地胡说八道。
• 格式合规性：输出结构是否符合output_format的要求。
• 安全与尺度：是否包含越界、敏感或低质量的内容。

每次审核完成后，审核Agent必须返回pass或fix结果。如果返回fix，必须附带具体的原因说明和修改建议。规划Agent在收到fix结果后，会决定是让原执行Agent带着修改建议重做一次，还是换一个执行Agent重做一次。这个“重做策略”在我们的测试中新能解决大约30%的错误案例。

5. 验证场景与能力边界：跑通了，但也看到了天花板

5.1 一个完整的任务流转示例

为了验证框架可用性，项目X里构造了一个“跨领域信息整理”的任务，具体内容是：调研某行业近半年的发展趋势，整理出关键事件列表，并针对其中一个事件进行深度分析。这类任务在单Agent模式下往往表现得很一般，因为既要检索、又要梳理、还要做分析，工作量和上下文复杂度都摆在那里。

任务下发后，系统经历了一条完整的链路：

• 规划Agent拆解出四个子任务：趋势概览、关键事件提取、特定事件深度分析、最终汇总。
• 执行Agent依次处理四个子任务，调用搜索和文本解析工具，每种结果都经过标准化压缩。
• 审核Agent对每个子任务结果做了详细检查，其中“特定事件分析”被判定为不达标，原因是缺少了来源引用且结论过于情绪化。
• 规划Agent根据修改建议，让执行Agent重新跑了那个子任务，补充了引用，调整了口径，二次审核通过。
• 规划Agent汇总所有结果，输出最终报告。

整个流程大约耗时三分钟，调用模型次数在8次左右。如果让单Agent干这件事，我和同行比较过，大部分情况下会在一千多个token之后开始遗漏早期要求，内容变得飘忽不定。而多Agent模式虽然花掉的调用次数更多，但每一段结果基本都在预期轨道上。

5.2 能力边界：多Agent也不是银弹

在实际使用中，我对这套框架的边界有了清晰认识，写在下面，方便大家参考：

• 规划Agent的任务拆分能力决定了上限。如果规划Agent本身就理解错了需求，后面所有Agent再努力也没用。所以对需求模糊、表达不清的任务，系统表现会显著下降。
• 工具能力决定了执行Agent的物理上限。我之前试过不注册工具，让执行Agent纯靠模型“编造”结果，审核Agent居然还能给出“看起来合理”的评价。这说明审核链路再强，也挡不住事实数据源的缺失。
• 消息丢失和状态不一致的问题在高并发时会放大。单任务系统里，内存消息总线足够用；但一旦同时跑几十个任务，消息顺序、状态同步都开始变得脆弱。我后来加入了简单的按任务ID分桶的队列，才缓解了这个问题。
• 成本控制是绕不开的话题。多Agent协作一定会比单Agent调用更费token。每个Agent各带一套上下文、对话历史，累计起来非常惊人。项目X里我强制要求每条消息在进入模型之前必须做一次轻量压缩，效果显著，但也会引入信息损耗。

6. 踩坑记录与排错技巧：那些文档里不会写的坑

6.1 模型返回JSON格式不稳定怎么治

这应该是做Agent应用的人都会遇到的第一个头疼问题。用模型生成JSON，结果经常多一个换行、少一个引号，甚至带着md代码块标记一并返回。我的处理方式分两步走。第一步，在后处理时允许用正则容忍最外层被Markdown代码块包裹的情况。第二步，无论模型接口的响应格式写的多标准，都必须在解析层做重试兜底。

import re, json def extract_json(text): text = text.strip() code_block_match = re.search(r"```(?:json)?\s*([\s\S]*?)```", text) if code_block_match: text = code_block_match.group(1).strip() # 尝试去掉首尾多余字符 start = text.find("{") end = text.rfind("}") if start != -1 and end != -1: return json.loads(text[start:end + 1]) raise ValueError("JSON parse failed")

这个方法算不上完美，但能解决80%的格式问题。剩下20%的情况，我在框架里做了重试机制，如果解析失败，就把原始响应丢回给模型，提示它“上一次输出格式不符合要求，请重新输出JSON”。这里有个技巧：重试提示里最好附带上上次输出中的具体错误原因，而不是笼统地说“格式错误”，这样才能给模型足够的信息去自我更正。

6.2 上下文污染：多Agent之间信息泄露如何隔离

在设计初期，我犯过一个错误：把多个Agent的对话历史放在同一个session里管理，结果导致模型的角色设定互相混淆。比如执行Agent在回答里突然带上了审核Agent的口吻，一看就是系统提示词被冲撞了。

解决办法是每个Agent完全独立的对话session，不共享任何上下文。所有跨Agent的信息传递都走消息总线，用结构化的payload承载。刚开始我担心这样会显得很“隔”，但实测发现，只有彻底隔离，每个Agent才能稳定扮演自己的角色。共享上下文这个想法看起来很高效，实际是灾难。

6.3 调试工具：如何追踪一条消息的完整生命周期

多Agent系统调试时的痛苦源于“黑盒”。模型输出不可控，消息流转又交错复杂，出了问题很难定位。我的做法是在消息总线入口处加了一个完整的日志记录器，每条消息从入队到被消费的每一步都记录时间戳和状态。调试时直接搜索message_id，就能看到这消息的完整生命轨迹。

这个日志记录器看起来很简单，但在排查问题时帮了大忙。有一次系统运行到一半突然挂起，我通过日志发现是消息总线的消费队列里卡了一条receiver为“executor”的消息，但是对应的执行Agent实例因为之前的异常已经退出了，没有人消费它，而且没有超时清理机制。加了“孤儿消息扫描”后，定期清理超过10分钟未消费的消息，这个问题就解决了。

6.4 重试风暴与熔断机制

多Agent系统还有一个隐蔽问题，就是“重试风暴”。当一个执行Agent连续报错时，规划Agent可能机械地反复给它下发相同任务，导致错误无限放大。我当时在审核Agent里加了一个阈值计数：同一个子任务被重试超过3次，就会触发告警，并直接标记为失败，让规划Agent放弃该节点。

这个方法源于现实运维里的熔断思想。在流程类系统里，无限重试远比一次失败更危险，因为每次重试都在消耗真金白银的模型调用和宝贵的时间。设置重试上限，其实也是在倒逼规划Agent注意任务拆分的合理性，而不是盲目地把“锅”甩给下游。

7. 后续可以横向扩展的方向

项目X跑通之后，我在原有框架的基础上又做了几个方向的延伸尝试，其中有些已经验证可用，有些仍在打磨中。

• 引入记忆模块：目前每个Agent的任务上下文都是即时生成的，任务结束后就清空了。如果要让系统具备持续学习能力，就得给每个Agent配一个长期记忆存储，把历史任务的经验教训沉淀下来。
• RAG增强工具链：执行Agent目前依赖的外部工具比较基础，如果接入一个成熟的知识库检索组件，让它能访问领域内的文档库，执行能力会有一个质的提升。
• 可视化编排界面：现在整个流程是代码配置出来的，不够直观。后续如果做成配置面板，用拖拽方式定义Agent角色和消息路由，就能让非技术背景的人也能用上这套框架。
• 动态流程控制：虽然现在固定流程很稳，但某些业务场景确实需要动态分支，我目前在实验用审核Agent的结果自动引导下一步调用路径，实现一种“半固定流程”的模式。

就我个人的体会而言，做“agency-agents”这类项目，最大的收获不是代码本身，而是通过亲手搭建，真正体会到“智能体协作”和“单点调API”之间的鸿沟。角色边界、消息协议、状态管理、失败重试，这些东西看起来都是工程上的老生常谈，但一旦放到Agent这个新物种身上，每一个老问题都会换个马甲重新出现。希望这篇分享能帮到正在折腾同样方向的朋友，少走几步弯路。