企业官网建设流程全解析

这篇倒不是我做过的最复杂的Agent项目，但绝对是我带新团队时最常用来演示“Agent协作到底是怎么跑起来”的一个案例。标题里的“搭建多Agent协作框架，实现核心流程”听起来挺大，其实落地下来就三件事：把任务拆明白、把Agent之间的分工和沟通方式定清楚、再把状态流转用代码固定住。这篇就按我当时从零搭一个多Agent协作框架的全过程来写，从思路到踩坑都放进来。

1. 为什么必须引入多Agent协作框架，而不是写一个大Agent

先聊一个最基础的问题：单Agent能做的事，为什么要拆成多个？很多刚接触Agent开发的朋友上来就写一个超大的Prompt，把所有任务都塞给一个Agent，结果运行时要么上下文爆掉，要么角色混乱，要么一个分支出错全部重来。我在实际项目里吃过这个亏，后来才彻底转向多Agent架构。

1.1 单Agent的天花板：上下文、专注度与故障隔离

一个Agent本质上是一个“大模型+Prompt+工具集+记忆”的组合体。模型本身的上下文窗口是有限的，即便现在窗口已经不小，但塞进去的内容越杂，模型提取关键信息的难度就越大，输出质量下降得很快。这有点像让一个人同时干产品经理、后端开发、测试和运维的活，不是不可能，但效率和质量都会打折。

单Agent还有一个隐患是故障隔离能力差。如果Agent在某一步产生幻觉或者工具调用出错，错误会沿着同一段上下文继续传播，最终输出一个看起来合理但完全不能用结果。多个Agent协作时，每个Agent只处理自己那一段，出错可以被定位到具体的Agent和步骤，修复成本低很多。

1.2 多Agent协作的本质：把“一个人”变成“一支团队”

多Agent协作框架的本质，是用多个各有专长的Agent模拟一支团队。团队的协作效率取决于两件事：分工是否清晰，沟通是否顺畅。分工清晰意味着每个Agent有独立的职责描述、工具集和输出约束；沟通顺畅意味着Agent之间传递的消息结构统一、状态流转明确，不会出现A输出的字段B看不懂的情况。

我第一次搭这个框架时，犯过一个典型错误：把任务拆给多个Agent之后，却没有给它们定义统一的输出格式，结果下游Agent拿到上游的结果，解析半天都拿不到关键字段。后来我花了整整一天把每个Agent的输出Schema统一成JSON结构，所有问题迎刃而解。这件事让我意识到，多Agent框架的核心不是“多模型”，而是“多角色+标准接口”。

1.3 什么场景真正需要多Agent，什么场景不需要

不是所有项目都适合上多Agent。我的判断标准很简单：如果你要处理的任务需要多种不同专业能力，或者任务的某个环节需要独立验证、独立执行，那就值得拆。最典型的例子是内容生产流水线——策划、写作、配图、校对、发布，每个环节技能差异大，拆成独立Agent非常自然。

反过来，如果任务本身很简单，比如“把这段文本翻译成英文”，硬拆成多个Agent只会增加延迟和复杂度，性能反而更差。在技术选型上，我遵循一个原则：先单后多，能单不硬多，拆多必定义接口。

2. 核心流程的设计思路与整体架构拆解

这个项目的核心流程定位是“任务拆解→Agent执行→结果汇合”。听起来简单，真正落地时涉及的细节远比想象中多。我在这里把当时的思路完整拆开讲。

2.1 四种主流协作模式，我为什么选了“协同群组”

多Agent协作的架构模式大体有四种：管线模式、监督者模式、协同群组模式、网状自治模式。管线模式适合流程固定的任务，比如“A→B→C”串行处理；监督者模式适合任务需要动态决策分配的场景，由一个“老板”Agent调度其他Agent；协同群组模式是多个Agent通过一个共享的消息总线协作，各自认领任务；网状自治模式则更自由，Agent之间可以直接对话。

我最终选了协同群组模式，原因是这个项目的核心流程虽然复杂，但每个子任务的边界非常清晰，不需要一个中心化的调度者来做太多智能决策。共享消息总线的设计让各个Agent之间解耦，后续想新增Agent只需要实现接口然后注册，不需要改动现有逻辑。项目后期也验证了这个决定的正确性——我们新增了三个Agent，零改动接入现有系统。

2.2 框架选型：为什么没有从零手写编排层

当时市面上已经有不少Agent框架轮子，比如AutoGen、LangGraph、CrewAI等。我简单拆解了它们的共性和差异之后，决定不直接套用，而是参考它们的设计思路，自己搭一个轻量级编排层。原因有三个：第一，通用框架为了适配各种场景，抽象层比较厚，调试和排错时不太直观；第二，项目的核心流程对消息格式有强约束，通用框架未必支持得很好；第三，团队后续有深度定制需求，自研编排层反而可控。

但这不意味着从零开始。底层模型调用、工具库、Memory这些基础能力，我直接复用了成熟组件。自研的部分主要集中在Agent的注册、消息路由、任务状态机这三块。这个“半自研”的策略让我既拿到了成熟组件的稳定性，又保住了业务逻辑的灵活性。

2.3 核心流程的四阶段：拆解、路由、执行、聚合

整个核心流程我拆成了四个阶段。拆解阶段由一个Planner Agent负责，把用户提交的原始目标拆成子任务列表，并为每个子任务标注需要的Agent类型；路由阶段由编排层根据子任务的元信息，把任务投递到对应的Agent队列；执行阶段，每个Agent独立完成自己的子任务，把结果按统一Schema写回消息总线；聚合阶段，一个Reporter Agent负责收集所有子结果，合成最终产出。

这四个阶段看起来常规，但每一步都藏了不少细节。拆解阶段要考虑到子任务的依赖关系，路由阶段要处理队列阻塞，执行阶段要设置超时与重试，聚合阶段要处理部分子任务失败的情况。这些细节我在第3节和第5节会仔细展开。

3. 实操环节：搭建一个最小可用的多Agent协作框架

这一节是全文最核心的部分。我把当时搭建过程中的关键代码、配置、参数选择逻辑都记录下来，你跟着走一遍就能跑通自己的最小闭环。示例代码我用Python来写，因为生态最成熟，测试和调试都方便。

3.1 定义Agent抽象与统一消息结构

第一步是定义所有Agent都必须遵循的接口。我设计了一个Agent基类，核心方法只有一个：process(message) -> message。这个设计参考了Actor模型——Agent之间不直接调用对方的方法，而是通过消息传递来交互。异步、解耦、易于扩展。

from dataclasses import dataclass, field from typing import Any, Callable, Dict, Optional import uuid import json @dataclass class Message: msg_id: str = field(default_factory=lambda: str(uuid.uuid4())) sender: str = "" receiver: str = "" msg_type: str = "task" # task / result / error / heartbeat payload: Dict[str, Any] = field(default_factory=dict) trace_id: str = "" def to_json(self) -> str: return json.dumps(self.__dict__, ensure_ascii=False)

消息结构里我特别加了trace_id字段，这个字段贯穿整个流程，用于全链路追踪。第一步做框架时很多人会忽视这个字段，等到线上排障时才后悔——没有trace_id，你根本无法定位一条消息在多个Agent之间流转的完整路径。

class BaseAgent: def __init__(self, name: str, description: str): self.name = name self.description = description def process(self, message: Message) -> Message: raise NotImplementedError def handle_error(self, message: Message, error: Exception) -> Message: return Message( sender=self.name, receiver=message.sender, msg_type="error", payload={"error_msg": str(error), "original_msg_id": message.msg_id}, trace_id=message.trace_id, )

每个Agent还需要注册自己的能力描述，这样Planner才能知道该把任务分给谁。能力描述不是给代码看的，是给大模型看的，所以要用自然语言写清楚。

3.2 实现Planner Agent：把目标拆成子任务

Planner Agent是整个流程的起点。它接收用户的原始需求，输出一个结构化的任务拆解结果。我这里不直接调大模型，而是把拆解过程封装成一个独立的函数，方便测试和替换。

class PlannerAgent(BaseAgent): def __init__(self, name: str, llm_router: Callable): super().__init__(name, "负责将复杂目标拆解为可执行的子任务列表") self.llm_router = llm_router def process(self, message: Message) -> Message: try: tasks = self._decompose(message.payload["goal"]) return Message( sender=self.name, receiver="orchestrator", msg_type="result", payload={"tasks": tasks}, trace_id=message.trace_id, ) except Exception as e: return self.handle_error(message, e) def _decompose(self, goal: str) -> list: prompt = f""" 你是一个任务拆解专家。请把以下目标拆解为3到8个可并行或串行执行的子任务。 每个子任务必须包含： - id: 子任务编号 - desc: 子任务描述 - agent_type: 需要的Agent类型（writer/reviewer/translator/summarizer） - depends_on: 依赖的子任务id列表，没有则填[] 目标：{goal} 只输出JSON数组，不要输出任何其他内容。 """ response = self.llm_router(prompt) return json.loads(response)

这里有一个很关键的工程细节：Prompt末尾强调“不要输出任何其他内容”。大模型默认会输出一大段解释性文字，如果不用强约束，后面解析JSON就会很痛苦。我在实践中还发现，加一个“只输出JSON数组”的约束还不够，最好在解析时做一层容错，用正则提取出类JSON的子串再做解析。

3.3 实现编排器：任务路由与状态机

编排器是整个框架的心脏。它负责接受Planner的输出，把子任务投递给对应的Agent，然后跟踪每个子任务的完成状态。状态机我用一个字典来维护，每个子任务有四个状态：pending、running、completed、failed。

子任务状态机的核心逻辑看起来简单，但处理依赖关系时需要考虑清楚。我的做法是：每轮循环把所有pending状态且依赖全部完成的任务挑出来投递；如果一个任务的依赖中有failed状态，则该任务直接标记为failed，防止下游Agent拿到错误上游数据。

class Orchestrator: def __init__(self): self.agents = {} self.message_bus = [] self.tasks = {} self.status = {} def register_agent(self, agent: BaseAgent): self.agents[agent.name] = agent def submit_goal(self, goal: str) -> str: trace_id = str(uuid.uuid4()) planner = self.agents["planner"] plan_msg = Message(sender="user", receiver="planner", msg_type="task", payload={"goal": goal}, trace_id=trace_id) plan_result = planner.process(plan_msg) tasks = plan_result.payload["tasks"] for task in tasks: task_id = f"{trace_id}-{task['id']}" self.tasks[task_id] = task self.status[task_id] = {"state": "pending", "depends_on": task.get("depends_on", [])} return trace_id def run_pending_tasks(self) -> bool: executed = False for task_id, task_info in self.tasks.items(): st = self.status[task_id] if st["state"] != "pending": continue deps_ok = all( self.status.get(f"{task_id.split('-')[0]}-{dep}", {}).get("state", "failed") == "completed" for dep in task_info.get("depends_on", []) ) if not deps_ok: continue agent_name = task_info.get("agent_type") agent = self.agents.get(agent_name) if not agent: self.status[task_id]["state"] = "failed" continue msg = Message(sender="orchestrator", receiver=agent_name, msg_type="task", payload=task_info, trace_id=task_id.split('-')[0]) self.status[task_id]["state"] = "running" try: result = agent.process(msg) if result.msg_type == "error": self.status[task_id]["state"] = "failed" else: self.status[task_id].update({"state": "completed", "result": result.payload}) except Exception as e: self.status[task_id]["state"] = "failed" self.status[task_id]["error"] = str(e) executed = True return executed

这段代码里把task_id设计成trace_id-子任务编号的格式，好处是不需要额外维护关联表，直接通过字符串操作就能找到依赖任务。这种小技巧在工程里的价值远比想象中高，简化了很多状态同步逻辑。所有依赖都完成后，整个流程就收敛到“所有任务completed或failed”的终态，Reporter Agent就可以开始聚合了。

3.4 实现Reporter Agent：把零散结果汇总成最终交付物

Reporter Agent负责收尾。它从编排器拿到所有子任务的执行结果，按用户需求的格式合成最终输出。实操时我对Reporter的要求是：不仅要汇总，还要做质量检查——如果某个子任务的状态是failed，Reporter不能无视它，必须在最终结果中明确提示缺失内容。

class ReporterAgent(BaseAgent): def __init__(self, name: str, llm_router: Callable): super().__init__(name, "负责聚合所有子任务结果并生成最终交付") self.llm_router = llm_router def generate_final_report(self, trace_id: str, orchestrator: Orchestrator) -> str: task_results = [] failed_tasks = [] for tid, status in orchestrator.status.items(): if not tid.startswith(trace_id): continue if status["state"] == "completed": task_results.append(status["result"]) elif status["state"] == "failed": failed_tasks.append(tid) prompt = f""" 你有以下子任务的执行结果： {json.dumps(task_results, ensure_ascii=False, indent=2)} 涉及失败任务，需要明确标注：{json.dumps(failed_tasks, ensure_ascii=False)} 请将这些结果整合成一份结构清晰的最终报告，不要虚构任何内容。 """ return self.llm_router(prompt)

关于Reporter，我一直坚持一个原则：Agent只能基于已有结果做组织和润色，绝不能自行脑补缺失内容。大模型天然有“补全倾向”，面对不完整的上下文，它会倾向于生成一个看起来完整的答案。所以Prompt里必须强制约束“不要虚构任何内容”，这是我在多次踩坑后总结出的经验。

3.5 核心参数的选择：模型温度、超时与重试策略

这个框架里涉及大模型调用的地方主要是Planner和Reporter，我给了这两个Agent不同的温度参数。Planner的任务是拆解，需要稳定和准确，温度设置为0.1；Reporter的任务是整合润色，稍微需要一点语言灵活性，温度设置为0.3。执行Agent如果只做确定性操作，温度直接设为0。

超时设置上，我给每个Agent的执行时间上限是30秒，超过就触发重试，最多重试2次。这个时间参数是根据当时所用模型的平均响应速度定的，如果你的模型更慢，需要自行上调。重试策略上我推荐指数退避——第一次重试等1秒，第二次等2秒，避免Agent同时在等待时产生惊群效应。消息总线队列方面，我设置了最大队列长度100，超过则拒绝新消息并让发送方重试，防止某个Agent处理不过来时内存暴涨。

4. 工具选型与底层支撑组件解析

框架搭起来之后，还需要几个底层组件支撑。我把这部分单拎出来讲，是因为很多教程只讲Agent逻辑，不讲基础设施，结果读者自己实现时在“工具怎么接入”“记忆怎么存”“上游并发怎么扛”这些问题上卡住。

4.1 工具注册机制：让Agent具备调用外部能力

Agent的价值不只是聊天，更重要的是调用工具完成任务。我在框架里设计了一个简单的工具注册表，每个工具是一个函数，附带名称和描述。Agent在执行任务时，如果需要调用工具，就从工具注册表里按名称查找并调用。

工具调用的设计有两个细节值得注意。第一是工具函数的返回结果必须标准化——统一返回JSON结构，包含是否成功、数据和错误信息三个字段。这样Agent（和后续的人工排查）才能一致地处理工具返回。第二是工具注册表要有权限控制，不是所有Agent都能调用所有工具，比如计费相关工具只允许财务Agent调用。我把这个控制放在注册表内部，Agent调用时自动校验它的角色权限。

4.2 记忆与上下文管理：短期工作记忆与长期知识分离

多Agent系统里，记忆管理比单Agent复杂很多。每个Agent既需要短期工作记忆来记住当前任务的关键信息，也可能需要长期记忆来保存跨任务的经验。我这里的做法是把短记忆放在消息结构里，消息的payload中只携带当前任务必要的信息；长记忆则用独立的向量数据库存储，Agent需要时通过检索接口获取。

我特别强调短记忆要“按需携带”，不要一股脑把上游所有输出都塞给下游Agent。这既节省上下文，也避免无关信息干扰模型判断。实际项目中，我把每个子任务的输入输出控制在2000字以内，超过的部分先摘要再传递。这样整套系统的上下文开销是可控的，运行速度也稳定。

4.3 并发控制：多个Agent同时运行时怎么扛住

热词里有一条“ai agent怎么扛并发”，这确实是生产环境必须面对的问题。Agent框架里的并发控制和普通后端服务不太一样，核心在于：每个Agent实例不是无状态的HTTP接口，它有自己的状态和上下文。我采用的方案是“每个Agent类型维护一个工作池”，池里有多个Agent实例，编排器按任务量动态把消息分发给空闲实例。

并发上最怕的是两个问题：共享状态被并发读写导致数据错乱，以及一个Agent占用了大量计算资源导致其他Agent饿死。前者我用每个Agent实例独立状态、只通过消息总线交互来规避；后者我给每个工作池设置了最大并发数，默认是4。合理设置并发上限的比例，可以让整套系统的吞吐量提升3倍以上，实测下来很稳。

4.4 Agent安全：两个必须做的控制点

Agent安全在公共教程里很少被认真讲，但一旦出事就是大事故。我的框架里设了两道安全闸门。第一道是输入过滤，所有进入Agent的文本先用内容安全规则扫描，拦截危险指令；第二道是工具调用审计，Agent每次调用工具都会记录一条审计日志，包含谁调的、什么时间、参数是什么。

另外我强烈建议给Agent设计一个“操作白名单”。只有白名单内的工具才允许调用，新增工具必须走审批流程。别觉得麻烦，线上Agent被注入恶意指令的情况比想象中多得多。后来我看业内一些Agent框架的安全设计，思路也类似，都是从输入过滤、权限控制和审计日志三个维度下手。

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

这部分是全文的精华。我按实际踩坑频率排序，把最常见的问题和排查方式写出来，每一条都是真金白银换来的经验。

5.1 Agent之间消息格式不兼容，下游解析崩溃

这是多Agent系统中最常见、也最隐蔽的问题。表面上看，A Agent输出了JSON，B Agent解析时报错。深入查发现A输出的JSON字段名和B预期的不一致——一个用的是content，一个用的是text。原因往往是两个Agent的Prompt对输出格式的描述不够具体，模型在字段命名上有细微偏差。

解决方案是建立统一的Schema校验层。我在每个Agent的入口加了一个validate_message函数，用JSON Schema严格校验消息结构，不合法直接拒绝。这个校验在开发阶段看起来多余，但上了生产之后，它每天能拦住几十条结构异常的消息。

5.2 多Agent死锁：A在等B，B在等A

死锁问题在串行依赖复杂的情况下会出现。比如任务A依赖任务B，B又间接依赖A，状态机的依赖检测逻辑不完善时，两个任务会永远处于pending状态。

排查方法很简单：在编排器里加一个心跳日志，定期输出所有任务的状态。一旦发现某个任务超过N轮循环还没变化，系统就自动把它标记为failed，避免流程卡死。我设置的是5轮，实测不会有正常任务被误杀，因为正常任务最多两轮就能执行完成。

5.3 上下文污染：前一个任务的信息干扰后一个任务的判断

这个问题的表现是：Agent在处理任务时莫名其妙提到与当前任务无关的内容。原因是消息总线复用了同一个Agent实例，而Agent的上下文窗口里还残留着上一个任务的信息。

修复方案是在任务切换时强制重置Agent实例的上下文。我把每个Agent设计成无状态的消息处理器——它不维护跨任务的记忆，每次process调用都从一个干净的上下文开始。如果确实需要跨任务记忆，走独立的向量数据库，不塞在上下文里。

5.4 大模型输出不稳定：JSON格式飘忽不定

大模型输出不稳定的问题在Agent框架里被放大了，因为Agent之间的通信严重依赖结构化数据。我的解决办法是：所有模型输出先经过一个解析层，做三件事——用正则提取类JSON片段、用宽容模式解析（容忍尾逗号、单双引号混用）、解析失败时自动重试一次。

如果重试还不行，就降级为纯文本传递。这个降级机制虽然不完美，但保证了流程不被卡死。在框架早期版本里，我遇到过连续三次解析失败导致整个流程失败的案例，后来加了解析容错和降级策略，基本就不再出现这种问题了。

5.5 如何快速定位是哪个Agent出了问题

多Agent系统最大的排障难点是定位问题。我的经验是：链路追踪（tracing）能力必须一开始就建好。前面提到的trace_id在这里派上大用场。所有Agent在处理消息时都记录日志，日志格式包含trace_id、agent_name、message_id、处理耗时和结果状态。

排查时只需要拿一个trace_id去日志系统搜索，就能把这条任务从Planner到Reporter的完整路径拉出来，一眼看出哪个Agent耗时异常或返回错误。这个能力让排障时间从小时级降到了分钟级，是我在所有Agent项目里都会保留的基础组件。

6. 这个框架后续还能怎么扩展

多Agent协作框架搭好之后，扩展方向非常明确。我这里分享几个我自己验证过可行的方向，以及一个我认为非常有前景的前沿思路。

第一个方向是接入因果推断能力。热词里提到的“因果强化学习CRL”思路，在实际Agent场景中有一个非常实用的变体——让Agent在执行决策时不仅看“相关性”，还尝试判断“因果性”。比如说，一个Agent发现某种处理方式经常导致下游任务失败，它可以基于这些历史数据做简单的因果分析，不只是记住“这样会导致失败”，而是进一步判断“在什么条件下会导致失败”。这种能力可以让Agent的决策质量上一个台阶。

第二个方向是让Agent具备工具编排能力。现在的框架里，工具调用是Agent自己决定的，但工具与工具之间的编排还是靠代码写死的。扩展方向是可以让Agent动态生成工具调用序列，类似让Agent写一段小流程脚本去执行。这个方向自由度更高，但需要更严格的沙箱和权限控制。

第三个方向是把人工审批节点插入流程。很多实际业务场景里，某些步骤需要人工确认才能继续。我给框架加了一个human_in_the_loop节点，Agent执行到该节点时暂停，等待人工通过接口确认后继续。这个扩展极大提升了框架在真实业务中的可用性。

第四个方向是模型异构。现在框架里所有Agent都调用同一个大模型，但实际不同Agent的任务难度差异很大。Planner任务复杂，适合用能力更强的模型；普通执行Agent任务简单，可以用更轻量的模型。在框架里按Agent类型配置不同的模型Endpoint，能在几乎不损失质量的情况下把成本降一半。

最后分享一个我个人的体会：多Agent框架最关键的瓶颈往往不在技术，而在任务拆解的合理性。拆得过粗，每个Agent的负担就重，退化回单Agent；拆得过细，消息传递开销太大，上下文碎片化严重。最理想的状态是每个Agent都能在一个清晰的边界内，用最少的上下文完成自己的职责。这个平衡没有标准答案，只能根据你的业务场景反复调试。我每接一个新项目，至少会花三分之一的时间在任务拆解的调优上，这个投入非常值得。