企业官网建设流程全解析

“agency-agents”这个词，我第一次真正重视起来是在一个跨部门协作的项目里。当时我们接到需求：要搭建一套内容审校系统，把一篇长文拆给不同角色去处理——有人查事实、有人挑逻辑、有人补案例、还有人做最终风格统一。一开始试的是“超级单体Agent”，上下文几千行，单次回答又长又乱，效果非常不稳定。后来切换到“agency-agents”思路——不是让一个智能体做完所有事，而是像搭建一个虚拟内容公司那样，让多个角色化、职责互斥的Agent在同一套机制里有次序地协作。效果立刻不一样了。这篇文章就来拆解这类“智能体机构”的设计核心、工程实现和实战避坑，适合正在做多Agent系统、或者准备从单Agent往复杂协作分布式架构迁移的开发者参考。

1. agency-agents的核心设计思路拆解

1.1 先搞清楚：它解决的到底是什么问题

很多人在刚接触“agency-agents”时，第一反应是“多Agent嘛，我跑几个Agent并行调用不就行了”。实际远不止于此。单Agent的能力边界在三个地方会被卡死：上下文窗口不够用、单一思维链路容易漏事实、以及没有“校对/制衡”机制导致错误会一路滚下去。

我见过最典型的失败案例是让一个Agent同时做资料检索、观点提炼、段落改写和数据校验。它既要理解业务背景，又要输出严谨结论，还得时刻记得约束条件。结果就是上下文越堆越长，前面的关键指令到后面已经被稀释掉，模型开始“自由发挥”。agency-agents的核心思路是把“全能者”切换成“专业团队”：每个Agent只做一类事，职责边界清晰，任务通过结构化消息传递。

这种设计解决的问题非常直接：

• 上下文隔离：每个Agent只处理自己领域的输入输出，不背负全流程信息
• 错误抑制：下游Agent可以检测上游输出问题，形成制衡
• 并行与顺序可编排：可以按依赖关系安排任务，哪些环节串行、哪些并行可以自由控制
• 可追踪可回放：每步都有独立结果，定位出错环节很容易

1.2 关键词拆解：Agency不是“群聊”，是“机构”

英文里“agency”容易让人误解为“代理”的复数，其实不对。它更接近“代理机构”。这意味着你设计的不是一堆Agent乱聊天，而是它们遵守相同的协作协议、有明确汇报路径、有任务工单流转。

我在设计系统时一直用一套“虚拟公司”类比：老板Agent（Planner）负责拆任务；执行岗Agent（Worker）负责具体活儿；审核岗Agent（Critic）负责挑错；还有一个调度器（Dispatcher）维护一个任务队列，决定下一个任务分给谁。这个类比在向业务方解释方案时特别管用，因为他们不需要理解模型原理，只需要理解“有人想方案、有人干活、有人质检”。

关键的组成要素有四个：

1.3 为什么即使有了大模型，仍需要这套结构

直接给一个超强提示词行不行？行，但不可持续。我实测过一个场景：把任务拆成四步交给同一个Agent去做，前三步结果都正常，到第四步需要引用前文数据时，它开始胡编。后来我换成三个不同的Agent实例协同，每个只做一步且都拿到上一步的结构化输出，错误率降低了将近一半。

原因在于：模型的注意力机制决定了长上下文的利用率是递减的。而Agency这种结构，本质上是在用系统设计弥补模型短板——不让模型“记太多”，只让它在有限范围内做到专注和可靠。

2. 角色职责划分与编排策略

2.1 最小可行角色集怎么定

刚开始搞agency-agents的人特别容易犯一个毛病：角色设得过多，甚至一个Agent专门负责“生成标题”，另一个专门负责“生成副标题”。这会造成大量的信息传递损耗和接口开销。

我的经验是，最小可行系统先配四个角色：

• Planner：理解用户目标，输出一份任务分解方案
• Executor：按方案执行单一子任务，输出结果片段
• Reviewer：审核执行结果是否符合要求，输出通过/修改建议
• Aggregator：把所有通过的结果组合成最终产出

这四个角色基本覆盖了“拆解—执行—验收—汇总”的完整闭环。如果任务领域比较特殊，再按需增加辅助角色，比如RAG场景里的Retriever、代码场景里的Runner。

2.2 编排模式：顺序、并行、还是协商

编排方式决定了整个系统的吞吐量和稳定性。我常用的三种模式：

• 顺序管线（Pipeline）：适合有强依赖的任务，比如“先生成提纲，再写正文，再校勘”
• 扇出汇聚（Fan-out/Fan-in）：适合子任务相互独立的方式，比如“多路并行搜集素材，最后统一汇总”
• 协商评审（Critic loop）：适合对质量要求极高的场景，比如“执行Agent产出，评审Agent打回，循环直到通过”

我见过有人把协商评审搞成无上限循环，结果系统在半夜跑了几百轮仍没有结果。解决办法是设置最大评审轮次，比如三到五次，超出后自动降级采用当前版本并打上“待人工复核”标记。

2.3 如何选择合理的编排方案

选择编排模式前，先画出任务依赖图。凡是存在上下游数据依赖的位置，必须用顺序；没有依赖的，全部设计成并行。这里有一个实操技巧：先用文字写下“每步需要什么输入、产出什么输出”，判断依赖关系时画一个简单的二维矩阵（行是输入，列是输出），非对角线有值则说明有关联，要串行排队。实际上后端实现时再使用dag库或图数据库来承担依赖表达。

3. 工程实现：一套可落地的agency-agents最小系统

3.1 环境与基础组件准备

这一步主要是选择语言和基础框架。我现在常用的技术栈是Python + 异步事件循环 + 消息队列（本地开发用内存队列即可，生产可以替换成Redis Stream或RabbitMQ）+ LangGraph或自研状态机。如果你是从零开始，优先推荐自研一个轻量调度内核，因为框架层抽象太高时，查问题反而困难。

需要的外部依赖就只有大模型API封装，业务逻辑用Pydantic定义消息体。消息结构非常关键，错误的大多数来源于消息字段定义不清，导致下游Agent不知道从哪里拿数据、拿到的数据是什么类型。

3.2 消息协议和角色基类设计

我在这里分享一个核心代码骨架，已经过简化但结构完整。首先定义消息类型：

from datetime import datetime from typing import Optional, Any, List from enum import Enum from pydantic import BaseModel, Field class TaskStatus(str, Enum): PENDING = "pending" RUNNING = "running" REVIEWING = "reviewing" DONE = "done" FAILED = "failed" class TaskMessage(BaseModel): task_id: str = Field(..., description="全局唯一任务标识") task_type: str = Field(..., description="任务类型，对应Agent角色职责") payload: dict = Field(default_factory=dict, description="实际任务内容") meta: dict = Field(default_factory=dict, description="元数据：优先级、超时时间") status: TaskStatus = TaskStatus.PENDING created_at: datetime = Field(default_factory=datetime.now) updated_at: datetime = Field(default_factory=datetime.now) class WorkResult(BaseModel): task_id: str status: TaskStatus output: Any = None error: Optional[str] = None review_comment: Optional[str] = None

然后定义一个Agent基类，所有具体Agent只要继承并实现run方法即可：

from abc import ABC, abstractmethod class BaseAgent(ABC): name: str = "base" def __init__(self, model_client, system_prompt: str): self.model_client = model_client self.system_prompt = system_prompt @abstractmethod async def run(self, message: TaskMessage) -> WorkResult: """处理传入的任务消息，返回结构化结果""" pass async def call_model(self, prompt: str, **kwargs) -> str: """封装对大模型接口的调用，支持超时与重试""" response = await self.model_client.completion( messages=[ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": prompt}, ], **kwargs, ) return response.choices[0].message.content

这样设计的初衷很简单：把Agent变成“输入输出都是结构化消息”的纯函数。任何一个Agent都可以在单独进程中测试，不需要和其他Agent耦合推理。

3.3 Planner、Executor、Reviewer、Aggregator 四个角色的实现示例

Planner的system prompt要写得很具体，要求它输出JSON格式的任务分解方案，并且明确指定每个子任务的类型和数据依赖。

PLANNER_PROMPT = """ 你是一个项目拆解专家。请根据用户提供的目标，生成一份可执行的子任务计划。 要求： 1. 输出JSON数组，数组每个元素包含：task_type（搜索/写作/审校/汇总）、description、dependencies（依赖哪些任务ID）。 2. 每个子任务必须边界清晰，不得重复描述同一件事。 3. 子任务数量控制在3-8个，不要拆得过于碎片化。 4. 返回示例： [{"task_id": "t1", "task_type": "search", "description": "查找2024年行业数据", "dependencies": []}] 只返回JSON，不返回其他文字。 """

Executor采用通用实现，根据task_type内部分发到不同处理函数：

class ExecutorAgent(BaseAgent): name = "executor" def __init__(self, model_client): system_prompt = "你是具体任务的执行者。只根据输入完成任务，不要跳出职责。按要求输出内容。" super().__init__(model_client, system_prompt) async def run(self, message: TaskMessage) -> WorkResult: task_type = message.payload.get("task_type") if task_type == "search": # 此处可调用内部检索服务 content = f"模拟检索结果：{message.payload['description']}" elif task_type == "writing": content = await self.call_model(f"请撰写一段内容。要求：{message.payload['description']}") else: content = await self.call_model(f"完成任务：{message.payload['description']}") return WorkResult(task_id=message.task_id, status=TaskStatus.DONE, output=content)

Reviewer的职责是发现问题并给出可修改建议。这里给Agent设置了明确的否决/通过机制，在prompt中写明输出格式：

REVIEWER_PROMPT = """ 你是质量审核员。请对执行结果进行审核。 如果通过，回复：{"pass": true, "comment": "通过"}。 如果不通过，回复：{"pass": false, "comment": "具体修改建议，要求可执行"}。 审核维度：事实准确性、逻辑一致性、格式合规性、与目标的相关性。 只返回JSON。 """

Aggregator的最后一步组装则简单一些，把所有通过审校的子结果按既定模板拼接。

3.4 调度内核和任务编排

调度器是整个系统的核心，我常将其设计成一个异步循环。伪代码如下：

import asyncio from collections import deque class AgencyScheduler: def __init__(self, agents: dict[str, BaseAgent]): self.queue = deque() self.agents = agents self.results = {} async def submit(self, message: TaskMessage): self.queue.append(message) async def run_until_empty(self, max_review_rounds=3): while self.queue: message = self.queue.popleft() agent_name = message.payload.get("assigned_agent") agent = self.agents[agent_name] result = await agent.run(message) self.results[message.task_id] = result # 如果审校不通过，并且没有超出最大轮次，则重新投递到执行Agent if result.status == TaskStatus.REVIEWING and result.review_comment: if result.meta.get("round", 0) < max_review_rounds: new_task = TaskMessage( task_id=message.task_id, task_type=message.task_type, payload=message.payload, meta={...}, ) self.queue.append(new_task) # 如果通过，则投递到Aggregator # ...

真实线上运行的时候，我会额外加上：任务超时控制、Agent异常重试、每个任务的生命周期日志。用状态机来防止消息重复进入同一位置。

3.5 错误处理与回退策略

系统跑起来后，最怕的不是模型答错，而是接口超时或返回格式解析失败。我的策略是所有非预期格式的输出统一定级为“可重试错误”，最多重试两次；两次后标记为FAILED，并由聚合器统一收集到“待人工处理批次”。

Reviewer否决后重新投递的任务，在meta中带round计数字段，防止无限循环；超过最大轮次的直接用当前版本加“quality warning”字段，而不是阻塞全流程。这里要特别注意：在prompt中强调“如果结果已经比上一版好，允许通过”，避免系统为了改而改。

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

4.1 任务死循环：最典型的“卡死”现场

现象：系统在某个子任务上反复执行，队列数量不降，日志显示同一task_id不停流转。

排查方法：先看Reviewer返回的comment。如果comment是空泛的“不够好”“需要提升”，大概率是prompt中缺少“可执行修改意见”的硬性约束。我在实际调试时会把Reviewer的输出打印出来，发现模型有时候根本不按要求输出JSON，而是回到自然语言。这个问题的根子是模型“不知道该给什么”，此时给一个修改模板，例如“请补充XX方面的数据，至少300字，并在结尾总结”，循环会明显收敛。

另一个原因是某个Agent的system prompt过于宽松，导致无论收到什么输入都输出同一类结果。这时需要检查各角色的prompt是否互相冲突。

4.2 上下文溢出和信息丢失

扇出汇聚模式下常见的坑：运气好时，任务细分到十几路并行；但汇总时Aggregator要把十几段结果全部放进prompt，一次请求就超长了。

解决方案有两个方向：

• 截断策略：设置每段输入的最大长度，超出部分摘要化
• 分段汇总：聚合器分层，先小聚合，再大汇总，避免单次调用吞太多内容

实际操作里我更偏向“分段汇总”，因为摘要化会损失细节。比如50个子结果，先分成5组，每组10个，生成5个中间汇总；再汇总这5个中间结果，最终输出。这样每次调用上下文都在可控范围内。

4.3 数据质量下降，但不报错

多Agent跑一段时间后，质量会悄悄下降，表现为输出变得平淡且模式化。这个问题的根源通常是自我反馈机制失效了——Reviewer开始习惯性放行。

我的应对办法是“交叉审校”：每隔几个任务轮换一次Reviewer的system prompt，或者在系统中加入随机抽查角色，随机对通过的任务重新质检。这个“抽查者”角色的prompt可以这样写：“你现在不是审核员，而是一个怀疑一切的第三方。寻找三位不同的独立的证据支持或反驳这段输出。” 这样既能打破审美疲劳，又能有效拦截质量滑坡。

4.4 排查清单速查表

我给自己的代码里加了一个简易的“通过率监控”模块，每五十个任务统计一次Reviewer平均通过率。如果通过率长期高于90%，说明审校已经变成走过场；低于30%，说明执行Agent和审校Agent出现系统性矛盾。这两个信号都需要人工介入。

5. 评估、调优与成本控制的实战经验

5.1 从哪些维度评估一套agency-agents系统

不要只看“任务完成率”。我建议至少记录这四个维度：

• 成功率：任务在最大轮次内完成并输出的比例
• 返工率：需要Reviewer打回修改的任务占比
• 单任务交互轮数：平均每个任务在Agent之间传递的次数
• 上下文吞吐量：每个任务消耗的token总量，按角色拆分统计

这四个维度结合起来才能反映系统是否健康。返工率不是越低越好，适当返工会提升最终质量；但轮数太多就说明拆解阶段做得不好。

5.2 成本优化的典型方案

token统计通常会发现一个惊人事实：Reviewer角色消耗的token往往比执行者还多，因为它每次都要读全文，有时还要反复读。降低成本的办法按优先级排列：

• 任务粗粒度化：减少不必要的子任务拆分
• 结构化输出：强制JSON输出，避免模型啰嗦解释
• 局部审校：不让Reviewer看全文，只传关键段落
• 混合模型策略：执行Agent用较强模型，审校Agent可以用相对低成本模型

最后这一点在实践中很管用。比如规划任务和内容生成使用高能力模型；审校阶段可以先用中等能力模型做一轮粗筛，只有标记了“通过”或“存疑”的结果再升级到高能力模型复评。实测这样的混合策略能在保持质量的同时减少约30%到40%的高规格调用量。

5.3 一个实测中的调优记录

在某个模拟项目X的内容生产系统中，初始配置的通关体验很差：一个任务平均要跑七轮才结束，成本也超预算。定位后发现Planner把“内容生成”拆得太细，连续拆出“标题生成”“导语生成”“正文生成”“结尾生成”四个子任务，而每个都要调用一次模型，成本高且冗余。后调整拆解策略：把相关度高的步骤合并，三段式结构（开头—主体—结尾）合并为一个任务内完成。子任务总数从7个降到4个，平均轮数从7降到3，成本下降约50%，而人工评分显示最终质量没有下降。

这个经验说明：agency-agents系统的瓶颈往往不在模型，而在任务建模。任务建模越干净，系统越省力。

5.4 可观测性：把每一次Agent行为都记录下来

多Agent系统调试最痛苦的是“不知道中间发生了什么”。强烈建议从第一天开始就记录结构化日志，至少包含：任务ID、Agent名称、输入摘要、输出摘要、耗时、token用量、审校结论。开发的时候可以把每次调用都存到本地JSON文件；生产环境则直接放到日志系统里。

日志里多一个“为什么”字段：每个Agent在返回结果时，必须额外提供一两条理由，比如“因为用户要求包含数据，所以我补充了图表”。这样排查时能快速判断是执行错误还是目标理解偏差。这个小习惯帮我省了大量排查时间。

6. 从“能跑”到“好用”的最后一公里

代码能跑通只算完成了40%。剩下的是工程化的活儿：提示词版本管理、Agent配置中心、灰度发布。我现在会把所有角色的system prompt放在一个配置目录里，用Git管理版本。每次改动Prompt，必须顺手跑一遍回归集，把前后结果对比，防止“改好了一个角色，搞垮了另一个”。

在实际落地过程中，我发现最值得投入的是评估数据集。准备二十到三十个典型任务，固定好输入，作为回归基线。任何Prompt修改、任何模型切换、任何编排策略调整，先跑一遍基线，看成功率、返工率和成本三个指标是否恶化。一个没有基线的多Agent系统，改来改去纯靠感觉，最终一定会在线上突然崩掉。

关于“agency-agents”这个方向，我自己踩过最大的坑是早期过于迷信模型的推理能力，把所有决策全部交给模型，结果发现模型的随机性会导致系统行为不可预期。后来逐渐把关键决策逻辑用代码固化下来，比如调度、重试、超时、降级、轮次控制这些都不让模型判断，只有内容生成和语义判断才交给模型。这套“代码控制流程、模型控制内容”的混合架构，是我做了多个项目之后最认可的形态。以后如果要做复杂业务接入，还有两件事值得投入：一个是可插拔的角色注册中心，让业务方能在界面上配置新Agent；另一个是更细粒度的预算控制，让每个任务在启动前就明确自己的token上限。这些细节会让系统从“实验室玩具”真正变成“生产工具”，也希望你少走一些我当年走过的弯路。