企业官网建设流程全解析

如果你已经试过把一个稍微复杂点的任务丢给单个大模型 Agent，大概率见过这种场面：上下文窗口越塞越满，生成速度肉眼可见地下降，最要命的是任务写到一半，它开始重复前面说过的话，甚至忘了最初的目标到底是什么。我最早碰到这类问题时，本能反应特别简单——那我把一个大任务拆成几个小任务，多开几个 Agent 并行干，不就行了吗？结果拆完之后更乱：三个 Agent 各说各话，没人知道同伴已经产出了什么，最后还得我自己手工把结果拼成一份能看的东西。

后来我在整理多智能体实践时，接触到一个代号叫agency-agents的设计思路。它解决的核心问题不是"怎么把一个 Agent 调教得更好"，而是"怎么让一群 Agent 像一个真实存在的代理机构那样协作"。什么意思呢？一家公司不会让销售直接跑到财务那里翻账，所有请求都要往前走单流程：前台接单、主管拆任务、专员执行、质检验收。把这个组织结构照着搬进 Agent 系统，就是 agency-agents 的核心内容。适合谁看？如果你正在做多 Agent 应用、写任务编排脚本，或者只是想让几个模型角色认真合作而不是互相抢话，这篇文章应该能帮你在动手写代码之前，先把思路理清楚。

1. 多智能体为什么需要"机构化"：我踩过的编排坑

光看名字容易产生一个误会，以为 agency-agents 是某种"让 Agent 从单线程变成多线程"的并发库。它本质上和并发关系不大，而是关于组织结构。我是在连续踩了三个坑之后才真正理解这件事的。

1.1 单 Agent 的天然上限：不是能力问题，是流程问题

你让一个 Agent"写一份覆盖六个模块的行业调研报告"，它会在第五个模块开始明显泄气：上下文太长了，前面的关键结论被注意力机制逐渐稀释，生成的文字开始车轱辘、泛泛而谈。这不是模型蠢，而是单 Agent 的工作方式决定了它必须把整份报告的所有历史都背在身上。上下文窗口再大，也扛不住无限堆积。

一个很典型的例子：我让一个 Agent 做"城市便利店选址评估"，要求它分别评估人口密度、竞品分布、租金水平和周边配套。它前三个部分完成得非常漂亮，到了第四部分，突然开始复读第二部分已经写过的竞品数据，甚至把某两个区域的数据混在一起。单独看这两个区域都没有错，但合在一起就产生了矛盾。原因是模型不是真的在"评估"，它是在一个超长的上下文里试图保持一致性，精力有限，注意力必然漂移。

1.2 直接拆多 Agent 反而更乱的三个原因

我第一反应是拆任务：一个 Agent 看人口数据，一个 Agent 看竞品，一个 Agent 看租金。拆完以后立刻发现三个新问题。

第一是没有统一的上下文池。A Agent 查到"该区域早高峰客流大"，B Agent 完全不知道，还在独立分析早晚高峰对客流的影响，重复劳动，浪费 Token。第二是没有任务边界。你让 A 分析"消费人群"，让 B 分析"消费习惯"，两者高度重叠，输出结果你根本没法判断该用谁。第三是没有结果验收。每个 Agent 输出一段文字，看起来都挺专业，但谁也没检查这些文字之间是否存在冲突，也没有交叉验证的环节。

这些问题本质上是同一个：我把"人"变成了"多个"，但没有定义"人与人之间怎么传递工作成果"。如果你在一个没有主管、没有流程、没有验收机制的公司里塞进一堆能干的员工，他们只会给你制造混乱，而不是产出好结果。Agent 的世界也一样。

1.3 "agency" 隐喻为什么是解决编排问题的正确姿势

在真实的代理机构里，业务能跑起来靠的是一套固定的流转结构：前台接单后进行登记，主管判断这个任务需要什么技能、拆分给谁，专员各自执行，质检员验收结果，档案室沉淀资料供所有人查阅。这套结构映射到 Agent 系统，每个环节都有对应物。

我实际的体会是，组织结构图就是最好的系统架构图。你不需要去发明什么高深算法，只需要把现实公司运转的那套逻辑用代码重新实现一遍，agency-agents 这套思路的精髓就是如此。

2. 核心实现：一个最小可跑通的 Agency 骨架

说得再热闹，落到代码层面还是得有一堆可以运行的部件。我在这里用一个极简实现来说明核心逻辑，不做过度设计，保证你照着敲就能跑起来。

2.1 角色注册表与任务包：先定义你和谁说话

第一步是定义 Agent 角色。不要直接把 Agent 写死成一个列表里的按钮，而是要用一个注册表把每个 Agent 的"职务说明"存下来。这样调度器才能根据任务需求选择正确的执行者。

import time import uuid from dataclasses import dataclass, field from typing import Any, Dict, List, Optional @dataclass class AgentSpec: name: str # 唯一标识 role: str # 角色说明，会拼进 system prompt skills: List[str] # 技能标签，用于调度匹配 model: str = "default" # 不同 Agent 可以用不同模型配置 max_concurrency: int = 1 # 该 Agent 最多同时处理多少任务 @dataclass class Task: task_id: str = field(default_factory=lambda: uuid.uuid4().hex[:8]) title: str = "" payload: Dict[str, Any] = field(default_factory=dict) required_skill: str = "" # 需要调度器匹配的技能 parent_task: Optional[str] = None # 由哪个任务拆出来的 depends_on: List[str] = field(default_factory=list) # 依赖哪些已完成任务 status: str = "pending" # pending / running / done / failed output: Any = None created_at: float = field(default_factory=time.time)

Task 这个结构看起来很简单，但它在系统里承担的角色远比一个"待办事项"重要。depends_on字段就是机构里的"流程审批关系"：任务 A 没完成，任务 B 绝对不能开始。payload字段则是任务请求的具体内容，不同技能 Agent 看到的是经过裁剪的输入，而不是全量上下文。

2.2 调度器：怎么把任务交给对的 Agent

调度器是最像"主管"的一个组件。它的输入是任务包和 Agent 注册表，输出是"把任务派给哪个 Agent"。我用的策略比较简单：先按技能匹配，再选当前最空闲的那个。

def dispatch(task: Task, registry: List[AgentSpec], busy: Dict[str, int]) -> Optional[AgentSpec]: if task.required_skill: candidates = [a for a in registry if task.required_skill in a.skills] else: candidates = registry if not candidates: return None # 尽量选正在处理任务数量最少、且没有达到并发上限的 Agent candidates = [a for a in candidates if busy[a.name] < a.max_concurrency] if not candidates: return None return min(candidates, key=lambda a: busy[a.name])

为什么用技能标签而不是直接写死"任务 A 必须发给 Agent B"？因为真实项目里技能 Agent 的数量和归属是会变的。你今天可能只有一个"数据搜集"Agent，明天可能拆成"国内数据 Agent"和"海外数据 Agent"。硬编码会让你的调度逻辑越来越难维护，技能标签是中间层，隔离了任务需求和具体执行者之间的绑定。

到这里你也发现了，调度器本身并不聪明，它只是一个按规则分单的"主管"。真正的智能仍然来自每个 Agent 的模型能力。但这个"不聪明"是优点：分单规则足够简单，出了问题你一眼就能定位，而不是在一团互相调用的函数里大海捞针。

2.3 消息总线：Agent 之间不直接说话

多 Agent 系统里最容易翻车的设计，就是让 Agent 之间直接互相调用对方的结果，类似于代码里对象和对象之间互相调函数。刚开始很爽，后面会变成一场灾难：A 的输出格式改了，B 立刻崩；A 重试三次，B 也重试三次，整个执行链乱成一锅粥。

正确的做法是引入一条很轻的消息总线。所有 Agent 完成工作后，把结果写成一条 WorkReport 发到总线上，后继任务通过订阅或者查询来拿结果。这样每个 Agent 都只管自己的输入和输出，不需要知道产出结果的人是谁。

from collections import defaultdict class SimpleMessageBus: def __init__(self): self._reports = defaultdict(list) self._events = [] def publish(self, report: Any) -> None: self._reports[report.task_id].append(report) self._events.append(report) def query_by_task(self, task_id: str): return self._reports.get(task_id, []) def query_by_agent(self, agent_name: str): return [r for r in self._events if r.agent == agent_name] def replay(self): """把所有事件原样回放，用于调试和复盘""" return list(self._events)

你把它当成 OA 系统就很好理解：销售不会直接去找会计要数据，而是通过流程系统提交申请，传阅、审批、留痕都在系统里完成。流程系统的存在价值不是"增加效率"，而是"提供可回放、可审计的协作记录"。消息总线在这里的价值一模一样：你可以随时回放整个协作过程，定位到某个 Agent 到底在什么时候产出了什么。

2.4 结果回收与汇聚：谁来把碎片拼成长文

执行完的各个 Agent 会产出碎片化的结果，最后一定需要一个"主编式"的汇聚节点。它的输入是一堆 WorkReport，输出是一份完整、连贯、没有互相冲突的最终内容。

@dataclass class WorkReport: agent: str task_id: str summary: str # 一句话摘要 artifacts: Dict[str, Any] # 完整产出物 confidence: float = 0.0 # 0~1，表示模型自评把握 raw_prompt: str = "" # 留痕，便于复盘 raw_output: str = "" # 留痕，便于复盘

汇聚节点本身也可以是一个 Agent。我的习惯是给它一个很明确的指令风格："你是一名严格的最终编辑。下面是若干份中间产出物，请你合并、去重、消除冲突，最终只输出一份完整文本。不要新增任何中间产出物里不存在的信息。"这样它就变成了机构里的"质检员+主编"双面角色——既检查上游成果，又完成最终交付。

3. 实战案例：让一组 Agent 协作完成"每周技术热点简报"

光讲原理读者最容易迷失。我挑一个非常日常的任务来做全流程走查：让几个 Agent 协作完成一份"每周技术热点简报"。任务不大，但完全覆盖了拆解、并行、串联、质检、汇聚这五个关键环节。

3.1 任务设计：先明确输出长什么样

简报的模板我定得很死，死到每一期的格式都完全一致：

{ "week": "第31周", "section_count": 6, "items": [ { "domain": "大模型应用", "hotshots": [ { "title": "一句话标题", "one_liner": "一句话结论", "why_hot": "为什么这周成为热点", "source_hint": "来源方向，不用写链接" } ] } ] }

为什么要定义这么严格的 JSON 输出？因为多 Agent 协作中，结构就是接口。下游 Agent 不需要去理解一段散文的中心思想，只需要按 JSON 的 key 去取数据。这一步节省了我大量的解析时间，也大大降低了下游出错的概率。

3.2 角色与提示词设计：每个 Agent 的"岗位说明书"

这个案例我配置了四个角色：

• collector（搜集员）：负责在给定技术域内搜集热点线索，输出原始条目列表。
• organizer（结构化员）：把散乱的笔记整理成标准 JSON 条目。
• reviewer（审校员）：检查条目是否存在明显编造、重复、过时的情况。
• editor（主编/汇聚）：把多份 JSON 合并去重，生成最终简报。

每个角色的 prompt 我基本是套一个固定公式写的：角色定位 + 输入说明 + 输出格式 + 质量标准 + 反面约束。举个例子，collector 的 prompt 大致是：

你是一名资深技术热点搜集员。你的输入是一个技术方向短语。 请列出本周内与该方向相关的最多3条热点线索。 每条线索需要包含：最可能的关注点、一句话说明、可能成为热点的原因。 注意：不要编造具体数据的来源；不要为了凑数写明显重复的内容； 如果信息不足，宁可只写1条，也不要硬凑。 只输出 JSON 数组。

注意那个"宁可只写1条，也不要硬凑"——这就是反面约束。它比正面要求"请输出 3 条"重要得多，因为它直接压制了模型为了完成数量目标而编造内容的倾向。

3.3 编排流程：从任务列表到最终稿件

整个执行链路我用文本画出来大概是这样的：

入口调度器 -> 任务解析（拆出 6 个 domain 子任务） -> 6 个 collector 并行执行 -> 每个 collector 输出 raw_notes.json -> organizer 逐个结构化 -> 全部组织完成后 -> reviewer 逐条审查 -> editor 合并去重，生成 final_report.json

注意里的两个"汇合点"：第一个汇合点是"6 个 collector 全部完成后"，organizer 才能开始；第二个汇合点是"全部 organizer 完成后"，reviewer 才能开始。这个依赖关系用 Task 的depends_on字段就实现了：organizer_task.depends_on = [collector_task_1.id, collector_task_2.id, ...]。调度器在执行前会检查依赖是否全部处于done状态，这是多 Agent 协作里的"任务闸门"，少了它，下游一定会拿到不完整的数据。

这里还有一个并行度的取舍。理论上 6 个 collector 可以 6 个并发执行，但实际要受到模型接口速率限制的约束。我在代码里用一个信号量把最大并发数压到了 3，宁愿跑慢一点，也不要被限流中断，这个在下一节会展开讲。

3.4 怎么判断它们"真的在协作"而不是各自写各自的

跑完之后最重要的一件事是判断协作质量。我的经验是看三个地方。

第一看中间产物里有没有引用上游数据。orgranizer 的输出里如果包含"根据 collector A 提供的三条线索"，说明它真的读取了上游结果；如果它自己又写了一套完全不同的内容，那么这个协作就是假的——下游 Agent 根本没有遵守"基于上游产出继续加工"的指令。

第二看同一数据在流程中是否保持一致。在最终报告里，某条热点如果和 collector 原始输出的表述完全对不上，甚至在审校环节换了内涵，那基本可以判定是模型在某个环节"自由发挥"了。

第三看日志里是否有重试和冲突记录。reviewer 如果标记过两条冲突信息，editor 最后是怎么裁决的？这中间的过程最好都能留在日志里，方便复盘。

我强调一点：Agent 编排跑通了不等于协作质量好。跑通只代表流程链条没有断，但链条里流通的内容有没有失真，是另一个问题。你需要定期回放消息总线里的日志做质量审计，就像机构里的管理层会抽查工单一样。

4. 最容易踩的 4 个坑，以及我的解决方案

这一段全部来自真实踩坑经历。多 Agent 系统真正的问题，往往出现在你自以为已经跑通之后。

4.1 上下文隔离悖论：每个 Agent 都只见树木，汇总时需要森林

第一个坑非常反直觉：为了让上下文不超限，我让每个 Agent 只接收相对精简的输入，结果到了汇聚阶段，editor 对某些领域一无所知，根本没法判断 collector 给的数据靠不靠谱。你可以说"那给 editor 全量数据不就行了"——但全量数据往往又超过单次上下文的处理能力，就这么卡死。

我的解决方案是"逐级摘要 + 共享只读知识库"两件事同时做。

一是逐级摘要，上游 Agent 的输出不是原始全量，而是一层一层收敛。collector 输出 500 字笔记，organizer 把它压成 120 字条目，reviewer 最后只保留 60 字结论。层级越往上，数据越精炼，editor 拿到的是浓缩但关键信息不流失的版本。

二是共享只读知识库，让所有 Agent 都能查到一份经过验证的基础数据。比如在本周简报这个案例里，我会把一份"本周关键词词表"放到只读区，任何 Agent 都觉得自己的判断有依据，而不必把整份词表塞进每个人的上下文。只读意味着 Agent 可以查但不能改，避免了一个 Agent 污染全系统的风险。

4.2 子代理递归失控：让 Agent 自己开 Agent，结果账单比产出先爆

这个坑最有戏剧性。我一开始做了一套支持"Agent 创建子 Agent"的系统，本意是让搜集员发现缺少某领域数据时能自动开一个专门小组补数据。结果某个任务里，一个搜集员觉得自己缺"海外市场数据"，于是创建了一个子 Agent；子 Agent 觉得缺"当地支付习惯数据"，又创建了一个孙 Agent；孙 Agent 觉得缺"当地竞品名单"，继续往下开。每一层都在等下层数据，根本没有收敛。

这个问题本质上是把任务的"分解权"完全交给了不设防的模型。它并不理解"创建子 Agent"是要花真实成本的，也不理解"无限递归"会让系统进入死循环。

我的解决方案有两个硬性限制：递归深度上限，和总 Token 预算。

MAX_AGENT_DEPTH = 3 TOTAL_BUDGET_TOKENS = 200_000 def can_create_subagent(current_depth: int, consumed_tokens: int) -> bool: if current_depth >= MAX_AGENT_DEPTH: print("达到子 Agent 最大层级，拒绝继续拆分") return False if consumed_tokens >= TOTAL_BUDGET_TOKENS: print("超出整体预算，拒绝继续拆分") return False return True

还有更狠的一招：子 Agent 的创建必须经过"主管"审批。也就是模型可以请求开子任务，但只有调度器确认目标和预算都合理之后才真正创建。这相当于在流程上加了一个闸门，宁可慢一步，也不要让递归失控，这个"审批制"我一直沿用到现在。

4.3 并行调用与限流：信号量控制任何外部接口的并发

当多个 Agent 同一秒内都向模型接口发起请求，限流几乎是必然的。尤其在我把并发数从 1 调到 6 的那次，整个任务链直接中断，前几个子任务白跑了。

解决思路用的是一个极简单但极其有效的并发控制机制：信号量。

import asyncio SEMAPHORE = asyncio.Semaphore(3) # 全局最多同时 3 个 LLM 请求 async def call_llm_with_limit(messages, model="default"): async with SEMAPHORE: # 这里放真正调用模型接口的代码 response = await call_llm(messages, model=model) return response

把信号量放在调度层还是放在 Agent 层是有讲究的。我建议放在全局调度层，因为限流通常是对整个模型账户的，而不是针对某一个 Agent。如果每个 Agent 各设一个并发限制，全局总量照样可能爆炸。信号量值设多少？我的经验是设为"你实测最稳定并发数的 80%"，留出余量给重试场景。

4.4 结果不可复现：两次跑同一个任务，输出完全不一样

多 Agent 系统的随机性比单 Agent 更让人头疼。你跑了成功的一次，想重新记录参数、复现结果，结果第二次的输出和第一次完全对不上。不是代码改出了 bug，而是模型本身有随机性。

我做了三件事来对付它。第一，缓存中间结果。同一个任务 ID 如果之前已经产出过结果，并且它的输入端没有变化，就直接读缓存而不是重新调模型。第二，固定生成参数，把温度调到一个较低的值，并记录完整参数。第三，留痕，把每次每个 Agent 的完整 prompt 和输出都落盘成 JSONL。

import json def save_trace(report: WorkReport): trace_path = f"traces/{report.task_id}.jsonl" with open(trace_path, "a", encoding="utf-8") as f: f.write(json.dumps(report.__dict__, ensure_ascii=False) + "\n")

这套留痕机制带来的额外收益是：你终于可以回答"这个结果是怎么产生的"这个问题。以前遇到结果不对，只能重新跑一遍瞎猜；现在直接翻 JSONL，定位到某个 Agent 的某个人输出，问题原因一眼可见。

5. 性能与资源控制：把骨架变成能长期运行的实用工具

如果你只是跑一个演示 Demo，前面这些已经够用了。但一旦你想让这个系统成为每周定时跑的常驻工具，就得考虑持久化、成本、重试这些细节。我用一个"从月度报告工具"的实际迭代来说明。

5.1 任务队列持久化：进程重启后不能一切从头再来

多 Agent 任务经常要跑几十甚至上百个任务。一旦中途进程崩了，如果没有持久化，所有已完成的任务都要重新执行，这个损失完全不可接受。

我的方案极其朴素：把任务状态按 JSONL 一行一条追加到本地文件。每完成一个任务就立刻落盘，而不是等到全部跑完再写。

def append_task_status(task: Task): record = { "task_id": task.task_id, "title": task.title, "status": task.status, "depends_on": task.depends_on, "created_at": task.created_at, "output": task.output, } with open("task_store.jsonl", "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n")

重启恢复的逻辑也很简单：加载 JSONL，筛选出status == "done"的任务，把它们的 ID 放进一个"已完成集合"。调度器在决定是否执行一个任务时，先检查这个任务 ID 是否已经在完成集合里，如果是就直接跳过。这种粗暴方案在处理几万条记录时完全够用，没必要上数据库。

5.2 Token 用量量化：设计阶段就算清账单

多 Agent 系统的成本比单 Agent 高一个数量级——同一份信息会在多个 Agent 的上下文里重复出现。我在项目早期吃过亏：一次全量跑完，成本比预估高出 5 倍，因为只算了"最终输入输出"的量，完全没算中间 Agent 的进出。

建议至少记录以下三个指标：

我是用一个小函数来实时打点的：

def log_cost(agent_name: str, prompt_tokens: int, completion_tokens: int) -> None: rows = { "agent": agent_name, "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": prompt_tokens + completion_tokens, "ts": time.time(), } with open("cost_log.jsonl", "a", encoding="utf-8") as f: f.write(json.dumps(rows, ensure_ascii=False) + "\n")

一天跑结束后，对 cost_log 做一次汇总，就能清楚看到"哪个环节 Agent 花的 Token 最多"。根据这个数据，你往往能发现：某一步的 prompt 反复在塞大量历史信息，但实际利用率不高。减少成本的最佳手段不是换更便宜的模型，而是减少重复信息进入上下文。

5.3 失败重试与超时：模型接口不是永远可靠的

调用模型接口的失败率会随着并发升高而明显放大。我把重试逻辑统一收敛到一个函数里，避免每个 Agent 自己写一套重试策略。

import time async def call_llm_with_retry(messages, model="default", max_retries=3): for attempt in range(max_retries): try: return await call_llm_with_limit(messages, model=model) except Exception as e: if attempt == max_retries - 1: raise wait_time = 2 ** attempt # 指数退避：1, 2, 4 秒 print(f"第 {attempt + 1} 次失败：{e}，等待 {wait_time}s 后重试") await asyncio.sleep(wait_time)

指数退避是最次之但是也最实用的策略：第一次失败等 1 秒，第二次等 2 秒，第三次等 4 秒，最多三次。网上有很多复杂的退避算法，但一个长期跑批任务里，三次重试已经能覆盖绝大多数瞬时故障。

5.4 什么情况下不要用多 Agent：一个逆耳但重要的清单

这是我目前最想强调的部分。不是所有任务都适合多 Agent 编排，有些任务用单 Agent 反而是最优解。我的判断维度主要是四个。

一个写文案的小任务，你硬拆成 10 个 Agent，那纯粹是给自己制造麻烦。多 Agent 系统的价值在于处理"需要在多个专业视角之间完成信息传递"的复杂任务，而不是把简单事情复杂化。我会在脑子里保持一个简单经验值：如果单 Agent 加一个长 prompt 能搞定，就不要上多 Agent。agency-agents 这套架构真正的适用范围，永远是那些"一个人干不过来"的活，而不是"嫌一个人干活太慢"的活。

这套骨架本身没有用什么高深技术，无非是"角色注册表 + 调度器 + 消息总线 + 汇聚节点"的组合，但我实际跑下来发现，它带来的收益远超预期。现在我做任何多 Agent 任务，都会先在纸上把"哪些角色、哪些输入、哪些输出、依赖关系是什么"写一遍，再动手写代码。这种做法帮我省掉的返工时间，可能比任何一个优化技巧都要多。你也可以试试：下一次让多个 Agent 干活之前，先像排值班表一样，把每个人的职责和上下游关系写清楚，再让它开工。