企业官网建设流程全解析

做多智能体项目时，有一个很微妙的阶段：单个 Agent 已经能跑通，但一旦让它们协同工作，各种问题就来了——上下文互相污染、任务反复推翻重做、对话绕圈子，完全不像一个团队，倒像几个新手在抢同一台电脑。这个名为 agency-agents 的项目，本质上就是针对这类问题做的一套多智能体协作框架。它的核心思路不是去发明更聪明的 Agent，而是把"个体智能"的问题转化为"组织协调"的问题：像搭一个机构一样去搭 Agent 系统，每个 Agent 有明确的岗位职责、汇报关系和协作流程，靠结构和制度来弥补单个模型的不确定性。这篇文章会从设计思路拆解到代码实现，再到调试过程中踩过的坑，完整记录这套框架是怎么搭起来的，以及它适合解决什么样的问题。

如果你现在刚准备入手多智能体开发，或者已经在做但总觉得协作效果不稳定，这篇内容应该能帮你省掉不少弯路。

1. 整体设计：为什么要把 Agent 组织成"机构"而不是"聚会"

1.1 单智能体到多智能体：问题性质的变化

单个 Agent 的本质是一个"输入到输出"的高智能映射：你给它一个任务，它借助大语言模型的推理能力，拆解步骤、调用工具、产出结果。这种情况下，质量瓶颈主要在模型能力和提示词设计上，属于"个体能力"问题。

但一旦有了多个 Agent，情况就不同了。Agent 与 Agent 之间的交互会产生组合爆炸：消息怎么传递、先后顺序怎么定、意见冲突时听谁的、失败时谁来兜底，这些都不再是提示词能简单解决的。我见过不少项目，把几个 Agent 丢在一起，结果就是它们反复确认同一件事，把 Token 白白烧掉，最后给出的答案还不如单 Agent 直接跑一遍。这不是模型不够聪明，而是组织方式出了问题。多智能体系统最大的风险不是单个 Agent 变笨了，而是整个系统的混乱度呈指数上升。

1.2 "机构化"思路：用制度对抗不确定性

agency-agents 这个名字，英文里 agency 既有"代理"的意思，也有"机构"的意思。这个项目刻意取了双关：一方面它是一个代理（Agent）的集合，另一方面它尝试把真实的组织管理经验移植到智能体系统里。

真实机构是怎么保证一群能力参差不齐的员工产出稳定结果的？靠流程、分工、协作规范。不是每个人都要全能，但每个人都要清楚自己的职责边界、汇报对象和交付标准。这套逻辑放到 Agent 系统里是成立的，而且很有效。

具体来说，这套框架的核心是三层结构：

• 顶层是调度层，相当于项目经理，负责任务接收、拆解和分发。
• 中间是执行层，由多个专职 Agent 组成，每个 Agent 只负责一种类型的任务。
• 底层是工具层，封装了模型调用、数据读写、外部 API 等能力，对上层透明。

调度层不直接干活，但掌控全局；执行层不思考全局，但深耕局部。这种分工方式有两个直接好处：一是每个 Agent 的提示词可以做得非常聚焦，不需要塞入一大堆上下文来维持"全局观"，降低了提示词复杂度；二是排查问题时定位非常快，某个环节出错，直接找对应的执行 Agent，而不是在一条超长的提示词里层层推理。

1.3 什么场景适合这种架构

不是所有多 Agent 需求都要上这种框架。我自己的判断标准很简单：如果任务链条短、步骤固定、依赖关系简单，用传统的工作流引擎加单个模型调用就够了，上多 Agent 反而徒增开销。

agency-agents 适合的场景通常是这几个特征：

• 任务可以明显拆分为多个专业模块，例如"信息收集 - 分析 - 文案生成"三段式。
• 各模块之间需要传递中间结果，而不是一次性给出完整需求。
• 任务执行过程中可能出现不确定性，需要反复修订，例如生成的内容需要经过质量检查和返工。
• 有并行处理需求，多个独立模块可以同时推进。

反过来，如果任务是"给出一段文本，翻成英文"，这种单步骤任务硬拆成多 Agent 是完全没必要的，纯属自找麻烦。

2. 核心细节：角色编排、消息协议与上下文隔离

2.1 角色怎么划分才算合理

角色划分是这套框架里最重要也最容易被忽视的一步。很多人喜欢把 Agent 划分得很细，什么"意图识别 Agent""情感分析 Agent""代码生成 Agent"，最后项目变得极其臃肿。我的经验是，角色划分的粒度应该跟任务的"专业化程度"对齐，而不是跟功能清单对齐。

一个有用的标准：如果一个子任务交给单个 Agent 直接处理时，你仍然需要大量提示词来控制它的行为，说明这个 Agent 的角色还太宽，需要继续拆分；如果一个子任务用一段简单的提示词就能稳定完成，说明它已经足够聚焦，再拆反而增加通信开销。

以这个模拟项目里的实际案例来说，一个内容生产流水线被拆成了四个角色：资料收集者、内容策划者、撰稿人和质量审查人。资料收集者只输出事实信息列表，不输出任何判断；内容策划者只输出大纲和观点，不写具体内容；撰稿人只负责把大纲扩展成完整文稿；质量审查人只输出问题和修改建议，不直接改稿。每个角色的输出边界都被严格约束，这样审查人的反馈才能准确落到撰稿人头上，而不是一团混战。

2.2 消息格式：结构化比自然语言可靠

多个 Agent 之间通信，最自然的想法是让它们直接用自然语言对话，但实际操作中发现并不可靠。自然语言本身有歧义，模型生成的自然语言更甚。同样一句话，不同模型对它的理解可能相差很远。而且自然语言消息很难做程序化校验，一个 Agent 漏传某个字段，另一个 Agent 也不会主动发现，只会硬着头皮往下跑。

因此在这个框架里，Agent 之间传递的消息被强制定义成了结构化格式。每条消息包含三个固定字段：发送者、接收者、任务类型，以及一个 JSON 格式的数据载荷。任务类型决定了接收方应该用哪个处理函数去解析，数据载荷则承载实际的业务数据。这样一来，消息本身就带有了足够的元信息，接收方不需要靠"猜"来理解对方意图，而可以像解析 API 请求一样处理消息。

结构化消息还有一个好处，就是让日志变得非常可读。调试时打开消息流水，一眼就能看出哪个 Agent 在什么时候给谁发了什么消息、数据长什么样。这在排查多智能体问题时能省下大量时间，比盯着对话记录猜要直观得多。

2.3 上下文隔离：每个 Agent 只该看到自己需要的

多智能体系统里最隐蔽的问题，是上下文污染。当多个任务在同一个对话环境中流转时，模型很容易把不同任务的信息混在一起，造成输出逻辑混乱。用户明明只问了 A 项目的数据，模型却把 B 项目的结论也带了进来，这样的系统看起来"懂了很多"，实际上非常危险，因为输出结果不可控。

agency-agents 的做法是物理隔离：每个执行 Agent 维护自己独立的上下文存储，调度层只传递必要的结构化数据，绝不把整个历史对话倒给下一个 Agent。就像真实机构里，财务部门不需要看到销售部门的全部通话记录，只需要拿到对方签好的合同金额。这样既减少了 Token 浪费，也避免了模型被无用信息干扰。

实际操作中，这个设计带来的效果非常明显。同样一个数据统计需求，上下文隔离前，模型时不时会把上一次任务的旧数据混进结果里；隔离之后，只要保证传入的数据载荷是干净的，输出就基本不会再出现张冠李戴的情况。

2.4 任务状态机：让系统知道自己走到哪一步了

多智能体系统很容易陷入一种"走一步看一步"的被动状态，尤其是使用链式调用的时候。Agent A 返回结果，Agent B 拿到后继续处理，整个过程没有一个明确的进度概念。一旦中间环节失败或者结果异常，整个链路就卡住了，你也说不清卡在哪里。

为了避免这种局面，这套框架给每个任务引入了一个简单的状态机：待处理、执行中、已完成、已失败、需修订。调度层维护一张任务表，任务每流转一步，状态就更新一次。这样做直接带来的好处是：任何时刻你都能准确知道系统忙在哪、卡在哪，而不是面对一条长长的日志大海捞针。

状态机还让"任务重试"和"结果修订"变得规范化。某个 Agent 输出不符合要求时，不需要把整个链路推翻重来，只需要把对应任务的状态改成"需修订"，重新放回调度队列，让对应 Agent 针对修订意见再做一轮即可。这比整个流程从头跑一遍要高效得多，尤其是在任务链较长的情况下。

3. 实操过程：从零搭一个最小可用的多 Agent 协作系统

3.1 环境准备与框架选型

这个项目的语言选型是 Python，主要是生态成熟，模型调用库多，而且团队更熟悉。如果是从头搭建类似项目的话，依赖很简单：一个支持工具调用的模型 SDK，加上 Python 内置的 asyncio 用来做并发调度就够了。项目初期不需要引入重量级框架，避免被框架约束了思路，等跑通后再考虑抽象成通用库。

代码结构也刻意保持简单，核心目录只有三个：

• agents/：放置各个执行 Agent 的实现，每个 Agent 一个文件。
• core/：消息协议、状态机、上下文存储这些基础设施。
• scheduler/：调度器，负责任务的接收与分发。

下面是我在这个模拟项目中实际使用的调度器核心代码骨架，去掉了业务细节，保留了关键逻辑。

import asyncio import json from typing import Dict, Any class TaskState: PENDING = "pending" RUNNING = "running" DONE = "done" FAILED = "failed" NEEDS_REVISION = "needs_revision" class Message: def __init__(self, sender: str, receiver: str, task_type: str, payload: Dict[str, Any]): self.sender = sender self.receiver = receiver self.task_type = task_type self.payload = payload def to_dict(self) -> Dict[str, Any]: return { "sender": self.sender, "receiver": self.receiver, "task_type": self.task_type, "payload": self.payload, } class Scheduler: def __init__(self): self.agents = {} self.task_table = {} def register_agent(self, name: str, agent): self.agents[name] = agent async def dispatch(self, task_id: str, message: Message) -> Any: self.task_table[task_id] = TaskState.RUNNING agent = self.agents.get(message.receiver) if not agent: self.task_table[task_id] = TaskState.FAILED raise ValueError(f"Agent {message.receiver} not found") try: result = await agent.process(message) self.task_table[task_id] = TaskState.DONE return result except Exception as e: self.task_table[task_id] = TaskState.FAILED raise e

调度器本身不执行业务逻辑，它只做三件事：找到消息接收者对应的 Agent、调用它、记录任务状态。这个设计保持了调度层的轻量，也让新增 Agent 变得非常简单——只需要实现process方法并注册到调度器里。

3.2 Agent 基类与执行器实现

所有执行 Agent 都继承同一个基类。基类负责消息解析、模型调用、错误包装这些公共逻辑，子类只需要实现核心的业务处理函数。

class BaseAgent: def __init__(self, name: str, model_client): self.name = name self.model_client = model_client self.context_history = [] async def process(self, message: Message) -> Any: raise NotImplementedError async def call_model(self, system_prompt: str, user_content: str) -> str: self.context_history.append({"role": "user", "content": user_content}) response = await self.model_client.chat( system=system_prompt, messages=self.context_history[-8:], # 控制上下文长度 ) self.context_history.append({"role": "assistant", "content": response}) return response

调用模型时只保留最近八条消息，这是个非常粗暴但有效的上下文截断策略。它能防止上下文无限膨胀，同时因为前面做了上下文物理隔离，这个截断基本不影响结果质量。如果你想优化模型对长上下文的理解，可以换成更精细的摘要机制，但在这个项目规模下，截断策略已经够用了。

以资料收集 Agent 为例，它的process方法是这样实现的：

class ResearchAgent(BaseAgent): async def process(self, message: Message) -> Any: query = message.payload.get("query") system_prompt = ( "你是资料收集员。请围绕用户给出的问题，检索并整理相关信息。" "只输出事实性信息列表，每条信息包含来源、时间、关键内容。" "不要输出任何观点、结论和建议。" "输出格式为JSON数组，每个元素包含 source, time, summary 三个字段。" ) raw = await self.call_model(system_prompt, query) return self._parse_json(raw)

可以看到，这个 Agent 的提示词非常聚焦，约束了输出范围、输出格式，还严格限定了"不要输出观点"。限制越清晰，模型的输出越稳定，后续的策划 Agent 拿到数据时也就越容易处理。

3.3 多轮协作：任务拆分与结果汇总示例

接下来用一个具体的场景把各模块串起来。假设任务是"撰写一篇关于本地社区健身需求的分析报告"。调度层收到这个任务后，会把它拆成三个子任务：

• 资料收集：查找社区健身相关的公开数据、政策文件和案例。
• 内容策划：根据收集到的资料提炼核心观点，形成报告大纲。
• 撰稿执行：按照大纲撰写完整报告。

调度层拆完后，依次向三个 Agent 派发消息。如果前一步结果不合格，比如策划 Agent 发现收集到的数据不足以支撑观点，它会给调度层返回一条"需修订"的消息，附带缺什么信息的具体说明。调度层把对应任务状态更新为NEEDS_REVISION，再把修订请求发回资料收集 Agent，让它补查数据，然后重新触发策划任务。

下面这段代码模拟了调度器如何编排这两轮协作：

import asyncio async def run_pipeline(scheduler: Scheduler): # 任务一：资料收集 research_msg = Message( sender="scheduler", receiver="research_agent", task_type="collect", payload={"query": "本地社区健身需求的相关信息", "depth": "medium"}, ) research_result = await scheduler.dispatch("task_collect", research_msg) # 任务二：内容策划，依赖任务一结果 plan_msg = Message( sender="scheduler", receiver="planning_agent", task_type="make_plan", payload={"data": research_result, "target": "社区健身需求分析报告"}, ) plan_result = await scheduler.dispatch("task_plan", plan_msg) # 如果策划结果要求补料，则回到任务一 if plan_result.get("needs_more_data"): supplement_msg = Message( sender="scheduler", receiver="research_agent", task_type="collect", payload={"query": plan_result["missing_query"], "depth": "high"}, ) extra_data = await scheduler.dispatch("task_collect_extra", supplement_msg) plan_msg.payload["data"].append(extra_data) plan_result = await scheduler.dispatch("task_plan_revise", plan_msg) # 任务三：撰稿 write_msg = Message( sender="scheduler", receiver="writer_agent", task_type="write_report", payload={"outline": plan_result, "style": "professional"}, ) final_report = await scheduler.dispatch("task_write", write_msg) return final_report

这个流程最值得注意的地方是任务 ID 的写法。task_collect和task_collect_extra虽然都是让同一个 Agent 做资料收集，但它们是两个独立任务，状态互不影响。这样即使补料环节失败，系统也能明确知道失败的是补料这一步，而不是把整个流程标记为失败。

3.4 质量审查环节：把"返工"变成正规流程

多智能体系统输出质量不稳定是常态，如果只管生成不管审查，最终结果往往达不到可用标准。因此这套框架里增加了一个独立的审查 Agent，专门做质量把关，不参与内容生成。

审查 Agent 的输入是撰稿人产出的完整报告加上一份质量检查清单。它不会直接给出"修改后的版本"，而是输出一个结构化的审查意见表，逐条列出问题类型、所属章节、修改建议。供稿人拿到这张表之后，只针对列出的问题做定向修改，不重新展开全文。

class ReviewAgent(BaseAgent): async def process(self, message: Message) -> Any: report_content = message.payload.get("report") checklist = message.payload.get("checklist") system_prompt = ( "你是质量审查员。检查报告是否符合以下要求：完整性、数据一致性、逻辑连贯性、格式规范。" "不要修改原文。输出JSON格式的问题列表，" "每个问题包含 issue_type, section, severity, suggestion 四个字段。" "如果所有检查项均通过，输出空列表。" ) raw = await self.call_model(system_prompt, json.dumps({"report": report_content, "checklist": checklist})) return self._parse_json(raw)

审查环节的价值不只是提高质量，它还给整个流程引入了"退出条件"。没有审查时，多 Agent 协作的产出是好是坏全凭感觉，有了审查，系统就对"什么时候算做完"有了明确的判断依据。审查通过就结束，不通过就进入返工循环，直到通过或达到最大修订次数上限。

这里要提一个实操细节：必须设置最大修订次数。如果不加限制，系统在极端情况下会陷入无限循环：审查员始终不满意，撰稿人始终在改，Token 持续燃烧。我在调试时设的是最多三轮修订，超过就直接报"质量未达标"，并把最后一次结果和审查意见一并返回给上层，至少保证有产物出来，还能说清楚为什么没达标。

4. 常见问题与排查技巧实录

4.1 上下文污染导致的"串味"问题

最常遇到的现象：报告里出现了跟本次任务完全无关的内容，比如上次任务的旧数据、旧结论，甚至旧措辞。问题根源是上下文的物理隔离没做好。排查时先检查调度层传参，看是不是把历史消息或者旧的结果数据混在了 payload 里传给了 Agent；再检查 Agent 内部的 context_history 是否被多个任务共用了。

我在调试时踩过的一个坑，是贪图省事让同一个 Agent 实例处理多个任务，它的context_history一直在累积。虽然每次只取最近几条消息，但模型看到了之前的任务痕迹，输出就会"穿越"。解决办法很简单：每次任务开始前重置context_history，或者确保每个任务使用独立的 Agent 实例。我自己最终选择了后者，因为多个任务共享实例虽然省点内存，但心智负担太重，随时要提防消息串场。

4.2 循环调用与死锁：系统为什么会绕圈子

多智能体系统一旦角色之间可以"互相反馈"，就很容易出现循环。A 让 B 改，B 改完让 C 审，C 审完不通过又回到 A，然后 A 又发给 B，如此循环。这种循环在任务逻辑不严格时特别容易出现。

排查循环问题的第一步就是看消息流水。如果发现同一对收发方的消息反复出现，且内容高度相似，基本可以断定是循环了。根因通常是某个 Agent 的职责边界不清晰，比如审查 Agent 既提出修改意见又要求补充内容，而补充内容的动作会触发另一个 Agent 的连锁反应，反馈路径就闭合了。

解决方式是在调度层加上"环路检测"：记录每条消息的哈希值，如果同一任务类型在短时间内重复出现超过阈值，直接终止并提示人工介入。这是兜底方案，更根本的解法还是在角色设计时把反馈路径限定为"单向"——在项目里，审查 Agent 的意见只会发给撰稿人，不会发给调度层，调度层只负责在超限时截断流程。

4.3 Token 消耗失控：成本怎么控制

多智能体系统的 Token 消耗通常比想象中高得多，主要原因不是单个调用贵，而是调用次数多了。一个任务走完"收集-策划-撰写-审查-修订"五个环节，可能产生几十次模型调用，每次调用还带着越来越长的上下文。

控制 Token 有几个有效手段。第一，严格执行上下文截断，不要把所有历史消息都传给模型。第二，payload 里只塞"必要数据"，不要图省事把中间结果全打包传下去。第三，对简单子任务使用更小的模型，比如格式转换、字段抽取这种任务，完全不需要顶配模型来处理。这套框架在设计时就把模型名作为 Agent 的可配置参数，这样不同 Agent 可以用不同规格的模型，成本能省下来一大块。

我实际观察过一个案例：同一个任务，不做任何优化时消耗约 12 万 Token，做了上下文截断和精简 payload 后降到 4 万出头，效果是立竿见影的。所以如果觉得成本高，先别急着怀疑模型贵，大概率是通信和上下文的浪费。

4.4 调试利器：消息流水与任务状态表

多智能体系统最可怕的不是出错，而是出错后不知道哪一步错了。传统单 Agent 系统只要看日志和输入输出就能定位，多智能体系统则必须依赖持久化的消息记录。

我在这个项目里加了一个简单的消息日志模块：所有 Message 在发送和接收时都会追加到一张数据表里。字段包括时间戳、任务 ID、发送者、接收者、任务类型、消息摘要、状态。排查问题时，只需要按任务 ID 过滤，就能完整还原一个任务的执行时间线。这是整个项目里性价比最高的一项功能，强烈建议任何做多 Agent 项目的人第一时间加进来。

任务状态表同样重要。调度器维护的任务表会记录每个任务从PENDING到DONE或FAILED的流转过程。如果发现任务一直停在RUNNING，基本就是 Agent 调用挂住了，需要检查模型调用是否超时；如果停在NEEDS_REVISION没后续动作，则是调度器的反馈路径没走通，需要检查修订任务是否被正确重新入队。

来看一个实际的排查场景。某次运行中，报告任务在"策划"节点反复失败。查看任务状态表发现task_plan一直是FAILED，再看消息日志，发现策划 Agent 收到了资料收集结果，但 payload 里的数据格式不是 JSON 数组，而是一个字符串。进一步追查发现，资料收集 Agent 在模型输出非标准 JSON 时用了简单的字符串兜底，导致下游解析失败。修复方式是在解析函数里增加容错逻辑，遇到非标准格式时尝试提取代码块中的 JSON 片段。整个排查过程不到十分钟，全靠消息日志把问题范围一步步缩窄。

4.5 常见问题速查表

这张表是我在实际调试中整理出来的高频问题。多智能体系统的常见故障其实就集中在几个点上，跑通一次流程后，把容易出错的环节提前用规范和兜底逻辑卡住，系统稳定性会显著提升。

5. 扩展思考：这套框架还能往哪些方向走

agency-agents 目前的实现还比较轻量，但它留下的几个设计接口让我看到不少扩展空间。第一个方向是为 Agent 增加"工具调用"能力。当前框架中的 Agent 只能调用模型，如果接入真实的数据查询、文件读写、API 请求等工具，整个系统就能从"分析讨论"走向"动手执行"。实现上只需要在 BaseAgent 里增加一个工具注册表，Agent 可以通过结构化消息申请调用工具，然后由工具层执行并返回结果。

第二个方向是引入人工介入的审批节点。有些场景下，AI 系统产出的内容需要经过人工确认才能进入下一步，比如发送对外邮件、发布正式文档。可以在调度器上加一个"人工审核"状态，流经该节点的任务会先挂起，等人工确认后再继续。这种模式特别适合对准确性要求高的业务场景。

第三个方向是评估体系的建设。现在系统的"做得好不好"只靠审查 Agent 的主观判断，这种判断本身也是模型生成的，不保证稳定。长远来看，应该沉淀一套带客观指标的效果评估体系，例如报告内容的命中率、格式规范率、返工率等指标，用统计数据分析系统质量，而不是听模型自我感觉良好。

第四个方向是复用和编排更复杂的组织形态。机构里不只有老板和员工，还有部门、项目组、专家团，对应的系统设计也可以往这个方向靠。比如同样是资料收集，可以并行多个收集 Agent 再合并结果；同样是审查，可以让两个不同视角的审查 Agent 交叉验证。这些高级形态都从"角色可替换、调度可编程"这两个基础能力上生长出来，所以基础框架的设计是否留足了余地，非常关键。

我在实际探索中发现，系统复杂度上升到一定阶段后，最重要的反而不再是模型能力，而是整个系统的可预期性和可观测性。一个固定输出格式的小模型 Agent，有时候比一个聪明但不可控的大模型 Agent 更让你省心。多智能体的本质不是堆砌模型，而是用架构把能力组织成系统，想清楚这一点，后续的扩展都会顺畅很多。