企业官网建设流程全解析

我们团队最近在做一个内部知识库问答系统，本来用单个大模型跑得好好的，但随着需求越堆越多——既要查资料、又要生成周报、还要自动归档邮件——单个Agent开始顾此失彼。于是在一次迭代中我们引入了多Agent协作架构，把任务拆给不同角色并行处理，再通过任务调度来统一协调。项目代号22.7，是一套完整落地、能从零搭建的复杂AI协同任务方案。

如果你也在纠结"多个Agent到底怎么配合""任务调度用什么策略""协作架构怎么选"，这篇文章就是从实际项目中总结出来的完整方案，包含架构选型、调度机制、完整实操步骤和一堆踩坑经验，适合已经跑通单Agent、准备进阶到多Agent协作的同学参考。

1. 多Agent协作架构的本质与设计思路

1.1 为什么要从单Agent走向多Agent协作

单个大模型Agent的能力边界其实很清晰：上下文窗口有限、单次任务专注度有限、工具调用链路过长容易出错。拿我们之前的单Agent方案举例，用户提交一个问题后，这个Agent要依次完成意图识别、资料检索、文档分析、报告生成，最后还要自查一遍。整个过程是一条长链路，任何一个环节出错，后面全部白费。

多Agent协作的思路是把这条长链路拆成多个短链路，每个Agent只负责一个环节。比如专门的检索Agent只用处理"查什么、去哪查、怎么查"，报告Agent只管"怎么组织语言、怎么排版"。每个Agent的提示词可以做得非常精简，上下文也不会被无关内容撑爆。实测下来，单Agent的完整任务成功率达到90%左右就已经不错了，拆成多Agent协作后，整体成功率反而能稳定在95%以上，因为错误被局部化了。

1.2 协作架构的三种基本模型

多Agent协作架构没有标准答案，我们实践下来最常用的是三种模型。

第一种是流水线模型，最适合任务有明显先后顺序的场景。Agent A的输出作为Agent B的输入，A做完B接，B做完C接，跟工厂流水线一样。优点是结构简单、容易调试，缺点是整体耗时等于所有Agent耗时之和，而且上游出错会直接阻塞下游。

第二种是编排器-执行器模型，也是我们22.7方案最常采用的。一个中央调度Agent负责任务分解和结果汇总，多个执行Agent负责具体干活。调度Agent本身不做重活，但必须具备很强的规划力和判断力。它的优势是灵活性极高，某一环节失败后调度器可以重新分配；缺点是调度Agent容易成为瓶颈，如果它的规划能力不够，整个系统的上限就会被拉低。

第三种是黑板模型，多个Agent共享一块"黑板"（公共消息区），各自读取感兴趣的讯息处理后再写回结果。这种模型适合探索性、开放性任务，比如头脑风暴或多方案比较，但现在工程落地还不多，原因后面会讲。

1.3 为什么我们选择了编排器-执行器模型

在22.7项目中我们最终选了编排器-执行器模型，主要是因为这套模型对业务场景的覆盖最全面。我们的知识库问答任务既有明确的流水线环节（检索→分析→生成），又有需要动态变化的场景（用户追问、问题发散），纯流水线模型处理发散场景非常笨拙，纯黑板模型又难以保证结果质量和收敛速度。

编排器-执行器还有一个隐藏优势：便于做权限和能力隔离。执行Agent可以分别绑定不同的工具、不同的知识库、甚至不同的模型。比如检索Agent用一个速度快、价格低的模型就够了，报告Agent则用更强的大模型保证输出质量，整体成本比所有任务都用同一个大模型省了约四成。

2. 任务调度的核心机制与策略选择

2.1 任务分解：从粗粒度到细粒度的拆解规则

任务调度第一步是任务分解，这也是整个多Agent系统里最体现"功底"的部分。把一个复杂问题拆成多个子任务并不难，难的是拆出来的子任务边界清晰、依赖关系明确、粒度大小适中。

我们总结出一套比较实用的拆解规则：每个子任务的产出必须是可验证的，不能是开放式答案；子任务之间的依赖关系尽量控制在DAG（有向无环图）范围内，避免出现循环依赖；单个子任务的耗时控制在十几秒到几十秒之间，太长的任务继续拆，太短的任务合并掉。

以"生成一份项目周报"为例，我们的编排器会拆成以下子任务：拉取本周代码提交记录、汇总各成员的进展反馈、生成周报初稿、校验格式和错别字、输出最终版本。其中前两个子任务互不依赖，可以并行；后面三个必须串行。这种拆解方式让整个周报任务从原本单Agent要几分钟，缩短到几十秒。

2.2 调度策略：并行、串行与条件分支

任务调度不能只会"一个个来"。我们22.7方案里实现了四种调度策略，实际使用频率从高到低排列。

串行调度用得最多，适用于有明确先后依赖关系的任务。调度器维护一个执行队列，一个Agent完成并返回结果后，下一个Agent才启动。并行调度用于无依赖关系的任务组，比如同时检索多个知识库、同时拉取多份数据，调度器用并发控制机制把多个任务同时派发给不同的执行Agent。

条件分支调度稍微复杂一点。调度器根据上一个Agent的输出结果判断下一步走哪条分支。比如用户提问"今天天气怎么样"，检索Agent返回的结果如果命中本地知识库就直接走答案生成分支；如果没命中就额外派一个Agent去调用天气API。这种动态路由能力是单Agent很难做好的。

还有一种是聚合调度，多个Agent各自输出结果后，由一个专门的聚合Agent合并去重、消除矛盾，产出最终答案。这在多来源信息汇总场景里非常好用，但要给聚合Agent明确好处理冲突的规则，比如"以最新数据为准""两处来源矛盾时标注不确定性"。

2.3 上下文管理与消息路由

多Agent协作中最麻烦的其实是上下文管理。每个Agent能看到的上下文不同，相互之间传递什么信息、过滤什么信息，直接影响最终质量。

我们最早犯过一个错误：调度器把完整对话历史广播给所有Agent，结果执行Agent的注意力被无关内容分散，输出质量直线下滑。后来改成"按需投影"策略——调度器根据子任务类型决定给每个Agent只看哪一部分上下文。检索Agent只拿到用户原始问题和知识库访问凭证，报告Agent只拿到检索结果和格式要求，各自所需的上下文最小化。

消息路由也是容易被低估的模块。我们用了一套轻量级的事件总线，所有Agent通过它收发消息，消息带类型标签和优先级。调度器根据不同消息类型投递给不同Agent，高优先级的任务可以插队。这套设计让我们后续加新Agent时不用改动核心代码，只要注册一个新角色ID和消息处理逻辑就能接入。

3. 完整实操：用工人组合构建一个多Agent协同任务

3.1 场景定义与角色配置

下面用一个真实例子完整走一遍22.7的落地过程。场景是"自动生成一篇科技行业早报"，要求每天早上9点前产出：包含AI大模型领域的前三条重要动态、每条的简要分析和一句评论。

我们在这个场景里配置了四个执行Agent和一个编排器。检索Agent负责从预置的RSS源和新闻API拉取当日资讯；筛选Agent负责对新闻做去重、打分、排序，选出与AI大模型领域相关度最高的前五条；分析Agent负责对入选新闻生成深度点评；汇总Agent负责把点评整理成最终早报格式。编排器负责任务调度和全流程监控。

四个Agent的提示词各不同。检索Agent的提示词里只强调"追踪哪些信息源、返回原始链接和时间戳"；筛选Agent的提示词里重点写清楚评分维度，比如"涉及大模型发布/融资/开源动态加5分，涉及泛AI但不聚焦大模型加1分"；分析Agent的提示词则要求必须包含事实描述、背景铺垫和独立观点三个部分。角色边界越清晰，协作效率越高。

3.2 22.7环境配置与依赖安装

动手之前先交代环境。我们用的是Python 3.10 + Docker Compose跑的整套服务，模型层接的是本地部署的开源大模型（Qwen系列7B和14B各一个），Agent之间通过Redis Stream传消息，编排器逻辑用Python的asyncio实现，负责调度的核心代码大约三百行。

依赖安装没什么特殊之处，核心的几个包是：langgraph（编排框架）、redis（消息队列）、openai（兼容层，用于统一调用本地和远程模型）、httpx（异步HTTP请求）。

pip install langgraph redis openai httpx

需要说明的是，我们没有全部依赖框架，编排器的调度逻辑是手写的。原因很实在：框架封装的调度策略是通用的，但我们内部的任务类型、消息格式、上下文投影规则都是定制过的，用通用框架反而要花更多时间适配，手写反而更简洁。如果你的任务类型比较标准，直接用langgraph这类框架也可以，不用纠结。

3.3 编排器的核心调度代码

这一节直接放关键代码。我们的编排器里最重要的一个函数是run_schedule，它接收一个任务图（用DAG表示），然后按依赖关系逐层执行。

from dataclasses import dataclass, field from typing import Callable, Dict, List import asyncio @dataclass class TaskNode: name: str handler: Callable dependencies: List[str] = field(default_factory=list) retry_count: int = 3 class Scheduler: def __init__(self): self.nodes: Dict[str, TaskNode] = {} self.results: Dict[str, any] = {} def register(self, node: TaskNode): self.nodes[node.name] = node async def execute(self, node_name: str): if node_name in self.results: return self.results[node_name] node = self.nodes[node_name] for dep in node.dependencies: if dep not in self.results: await self.execute(dep) for attempt in range(node.retry_count): try: dep_results = {dep: self.results[dep] for dep in node.dependencies} self.results[node_name] = await node.handler(dep_results) return self.results[node_name] except Exception as e: if attempt == node.retry_count - 1: raise print(f"Agent {node_name} 第{attempt + 1}次执行失败: {e}") async def run(self, root_node: str): await self.execute(root_node) return self.results[root_node]

这段代码的核心就是递归遍历任务图、按照依赖顺序执行、失败自动重试三次。我们的22.7方案里，所有Agent任务都被封装成TaskNode注册进Scheduler，调度器本身不关心Agent内部逻辑。

实际配置早报任务时，注册代码如下：

scheduler = Scheduler() scheduler.register(TaskNode("feeds", feed_handler)) scheduler.register(TaskNode("filter", filter_handler, dependencies=["feeds"])) scheduler.register(TaskNode("analyze", analyze_handler, dependencies=["filter"])) scheduler.register(TaskNode("collect", collect_handler, dependencies=["analyze"])) result = await scheduler.run("collect")

四个节点串成一条链：feeds做完才能做filter，filter做完才能做analyze，最后汇总。这就是流水线模型的一个典型实现。当然实际项目中我们在外层还包了一层调度器来处理并行分支，但核心逻辑原理一致。

3.4 消息路由与上下文投影的实现

调度器能正常串联Agent，还要靠消息路由和上下文投影两个模块。我们给每个Agent定义了消息的input_schema，例如检索Agent要求输入是一个包含query和max_items的字典；筛选Agent要求输入是包含raw_items和score_rules的字典。

上下文投影的实现思路其实很简单：在Agent启动前做一个轻量级过滤函数，把调度器汇总的结果按需裁剪。

def project_context(task_type: str, full_context: dict) -> dict: if task_type == "retrieval": return {"query": full_context["query"], "content_base": full_context["content_base_id"]} if task_type == "filter": return {"raw_items": full_context["raw_items"][:20], "rules": full_context["filter_rules"]} if task_type == "report": return {"filtered_items": full_context["filtered_items"], "format": "tech_report"} return full_context

这个函数被编排器在每个Agent执行前调用。实测下来，上下文裁剪掉后，每个Agent的prompt长度平均缩短50%以上，模型响应速度明显提升，而且输出质量更稳定。这就是上下文投影的实际好处。

3.5 运行结果与效果对照

整个早报任务运行起来后，效果对照如下表：

耗时缩短是因为检索和预筛选两个环节实际上走了并行，Token消耗降低是因为每个Agent的上下文都被裁剪过。成功率提升则是因为失败被局部隔离了——单Agent方案里如果某个环节上下文溢出，整个任务必须重来，而多Agent方案里失败只会发生在单环节点上，重试代价小很多。

4. 多Agent协作中的关键问题与实操心法

4.1 死锁与循环依赖的识别与规避

任务调度最怕的就是任务图出现循环依赖，A等B、B等A，两个Agent永远没有结果返回。我们在22.7初期就踩过一次这个坑：某个分析任务的分支逻辑写错了，导致Agent A的结果被送给Agent B后又传回Agent A，调度器不断重跑这两个任务但始终无法收敛。

后来我们加了两个防御手段。第一，任务图构建时做一次DAG校验，发现环直接拒绝启动；第二，所有Agent执行都设超时时间，超过30秒没返回就标记失败并触发降级逻辑。这两个机制帮我们在后续迭代中躲掉了至少三次类似的坑。

检查依赖关系可以用拓扑排序，代码量很小：

from collections import deque def has_cycle(nodes: Dict[str, TaskNode]) -> bool: indeg = {name: 0 for name in nodes} for node in nodes.values(): for dep in node.dependencies: indeg[node.name] += 1 queue = deque([name for name, deg in indeg.items() if deg == 0]) visited = 0 while queue: cur = queue.popleft() visited += 1 for node in nodes.values(): if cur in node.dependencies: indeg[node.name] -= 1 if indeg[node.name] == 0: queue.append(node.name) return visited != len(nodes)

注意：如果任务数量达到几十个，建议直接在构建任务图的时候就做循环检测，不要等到运行时。

4.2 上下文丢失与信息过载的平衡

多Agent协作下，上下文管理是最容易出问题的地方。信息过载导致Agent抓不住重点，信息不足则导致输出粗糙，这个平衡很微妙。

我们的经验是"给Agent能完成当前任务的最少信息，而不是尽可能多的信息"。比如筛选Agent只需要知道"有哪些新闻候选和评分规则"，它就不需要看到用户原始对话；汇总Agent只需要知道"筛选后的内容和格式要求"，它也不需要看评分过程。上下文裁剪得越狠，各Agent的专注度越高，最终质量越稳定。

但这里有一个例外：如果下游Agent需要理解的背景较多（比如做深度分析），上游就应该把关键中间结论和推理摘要一并传递，而不是只给最终结果。我们在分析Agent的输入里附加了筛选Agent的总结评语，效果比只给原始新闻链接好很多。

4.3 成本控制：模型分级与并发限制

多Agent方案比单Agent多了一个优势：可以做模型分级。我们22.7方案的默认配置是：检索类、筛选类Agent用本地7B模型，分析类Agent用14B模型，汇总类Agent使用更强的商用大模型API。整体下来，每任务成本比统一用商用API的方案降低60%以上。

并发限制同样重要。如果编排器把所有子任务一次性并发出去，本地模型的显存会直接被打满，延迟反而飙升。我们实测下来的建议是：本地模型并发数控制在2-4之间，远程API并发数控制在5-8之间，这是性价比比较高的区间。并发控制可以用Python的asyncio.Semaphore轻松实现。

sem = asyncio.Semaphore(4) async def limited_call(agent_func): async with sem: return await agent_func()

4.4 调试多Agent系统最实用的三个手段

多Agent系统的调试比单Agent要难受得多，因为问题可能出在任何一个环节。我们后来总结出三个实用的调试手段。

第一个是全链路日志跟踪。每个Agent执行前打印输入摘要，执行后打印输出摘要，调度器记录每个节点的耗时和重试次数。有了这些日志，绝大多数问题都能定位到具体节点。

第二个是局部复现。发现某个Agent输出质量差时，直接用它的输入样本单独调用它，把它的输出跟完整链路里的输出对比，看是不是环境参数不同导致的。

第三个是**"最小复现用例"思想**。如果整个链路失败但不知道原因，先砍掉调度器，用一套写死的固定输入，逐个单独跑Agent，问题很快就能浮出水面。

5. 常见问题速查与经验补遗

为了让你在实际搭建的时候少走弯路，我把这个过程中比较高频率遇到的问题整理成一个速查参考表，都是我们自己或者朋友团队实际碰到过的：

还有一些经验性的东西，不属于问题但值得知道。比如不要把Agent名字取得太抽象，我们之前管检索Agent叫"AgentAlpha"，调试的时候看到日志完全不知道谁在跑。后来统一改成"内容检索Agent""相关性筛选Agent"这种一目了然的命名，团队沟通效率都提升了不少。

再比如提示词里给Agent一个"人设"往往比给一堆规则更好用。分析Agent的人设是"资深科技记者"，筛选Agent的人设是"严苛的编辑"，生成的风格马上就不一样了。这条经验在很多场景里都很实用。

还有一点跟模型选型相关：多Agent场景对模型的要求和单Agent不完全一样。单Agent强调整体推理能力，多Agent更看重指令遵循能力和输出格式稳定性。我们试过用7B模型做分析类任务，卡在格式上的次数很多；后来14B模型专门跑分析，情况立刻好转。如果你的Agent经常在有格式要求的时候翻车，可以考虑单独给这类Agent升级模型。

6. 从多Agent到复杂AI协同任务的一些心得

6.1 好的编排器不是"会分配任务"，而是"会提前预判"

用了一段时间之后，我越来越觉得编排器这个角色才是多Agent系统的灵魂。它不直接产出内容，但它的规划能力决定了整体产出效率。我们在22.7的第二版迭代里给编排器加了一个前置判断逻辑：先评估用户问题的难度和涉及范围，简单问题直接走轻量级链路（两个Agent搞定），复杂问题才走全链路（四个Agent全力跑）。

这个小改动带来了两倍多的效率提升——大约一半的简单问题不再需要启动全链路，整体资源消耗下来一大截。如果你的多Agent系统还在追求"所有任务都精细拆分"，反而可以考虑一下反方向：能少分就少分，在恰当节点简化链路。

6.2 拆解任务是在"理解业务"，不完全是在"设计技术"

坦白讲，做多Agent系统的过程中花时间最多的不是写代码，而是理解业务。任务拆得好不好，答案很简单：你对这个业务的环节是否真的了解。如果一个业务链条本身模糊不清，再优秀的调度器也调度不出来合理的结果。

我们做早报任务的时候，一开始总是拆得很细——多到十几个Agent。跑了一阵子发现很多Agent重叠严重，该合并的没合并，白白浪费资源。后来重新梳理了业务流程，把十几个Agent压回四个，效果反而更好了。任务不是拆得越细越好，拆的粒度要跟业务的自然阶段对齐。

6.3 Agent之间也需要"团队磨合"

多Agent协作还有一个比较感性的经验：Agent之间需要"磨合"。不是调一次就能稳定，我们会根据运行日志定期调整每个Agent的提示词，有时候只是加一句"注意引用来源"就能让下游汇报Agent的输出质量大幅提升。

这其实跟带团队很像。把每个Agent当成一个有性格的组员，你会更容易理解为什么它输出质量不稳、为什么它偶尔"跑偏"。知道这几个Agent各自的脾气之后，反而更好管理了。

我现在的做法是每周跑一次全量任务集，把每个Agent的输出质量、耗时、失败率拉一张表对比着看。哪个Agent退步了就单独调它的提示词，哪个Agent超时就考虑要不要换模型。这套"运营"思维的引入，让整个系统的稳定性一直在缓慢上升。

如果你正准备做多Agent项目，我的建议是：先搭一个能跑通的骨架，再逐步加细节。不要一上来就追求复杂的编排策略、完美的上下文管理——先让两个Agent协作起来，体会一下协作带来的问题和收益，然后再逐步扩展。多Agent真正难的不是技术，而是你对"任务"本身的理解是否足够透彻。