企业官网建设流程全解析

这几个月的业余时间，我几乎都搭在这个代号叫agency-agents的内部项目上。刚开始立项的时候，团队里对“要不要上多智能体”吵了很久，毕竟单个大模型 Agent 跑 Demo 确实惊艳，可一放进真实业务流程里就问题不断。我们最后决定用一套非常务实的多智能体协作框架来解决这些痛点，把复杂的任务拆给不同的 Agent 去认领，再用统一的消息协议把它们串成一条完整的闭环。这篇文章不聊虚的，直接讲清楚我们当时为什么选这个架构、核心模块怎么设计、代码层面怎么落地，以及跑生产环境时踩过哪些坑。正在做 Agent 应用选型，或者已经被单 Agent 上下文和容错问题折磨过的开发者，可以重点看一下。

1. 为什么放弃单 Agent：从“全干将”到“小团队作战”

1.1 单 Agent 方案的天花板在哪里

很多团队刚开始做 Agent 应用时，习惯把所有能力都塞给一个大 Agent，让它既当规划器，又当执行器，还要兼任记忆管理。这种做法在 Demo 阶段很爽，因为一个 Agent 就能应对所有问题，对话体验也连贯。但一旦接进生产，几个问题就会逼着你重新思考。

首先是上下文窗口的限制。即便现在主流模型支持很长的上下文，但真实业务里塞进来的日志、历史会话、知识库片段、工具返回结果是惊人的。我见过一个很典型的单 Agent 场景，客服工单处理过程中，Agent 需要同时记住用户基本信息、当前工单描述、历史处理记录、售后政策匹配结果和最终操作日志，一个环节判断错就把前面的关键信息冲掉了，导致答非所问。

其次是任务串行执行带来的低效。单 Agent 架构下，如果任务里包含“先查库存、再查物流、再查优惠券规则”这种可并行执行的子任务，它也只能老老实实一步步来，耗时呈线性增长。换了多智能体协作架构后，查库存、查物流、查优惠券的 Agent 可以并行拉取数据，主流程只等最终汇总结果。

第三个问题是容错性。单 Agent 一旦在某一步出现幻觉或工具调用异常，整个链路的后续步骤都会崩掉，没有任何中间层可以去兜底。而多智能体架构天然有职责边界和重试空间——一个执行 Agent 挂了，编排器可以把同一个任务重新分配给另一个同类 Agent，主流程感知不到这次故障。

1.2 多智能体的三种常见协作范式

选型之前我梳理了目前社区里常见的多智能体协作方式，大致可以分成三类。

第一类是编排式（Orchestrator-Worker）。由一个中心编排器负责任务拆解、结果汇总和流程控制，Worker Agent 只负责执行。优点是流程可控、容易排查问题，适合业务流程相对固定的场景。缺点就是编排器本身可能成为性能瓶颈，而且对复杂动态流程的扩展性一般。

第二类是自治式（Peer-to-Peer）。各 Agent 之间没有固定中心节点，通过共享消息总线自由协商任务归属。这种范式灵活、扩展性强，但容易失控，Agent 之间可能出现循环对话、消息风暴，不适合对结果准确性要求极高的业务场景。

第三类是混合式，也是我们最终采用的方案。核心业务流程走编排式，保证主链路可控；同时引入自治式的消息广播做“旁路协同”，让一些非关键的子任务可以动态分配给合适的 Agent 去处理。这套组合在控制力和灵活性上找到了平衡点。

1.3 我们这次项目的定位

项目代号 agency-agents 的定位其实很明确：做一个内部通用的多智能体任务协作平台，不绑定某一个特定业务。第一期的重点是让客服工单自动处理场景能够完整跑通，同时沉淀出可复用的 Agent 编排能力，后续再横向扩展到运营内容审核、销售线索清洗等场景。

选这个切入点的原因很实际——工单场景对流程可追踪性的要求很高，每一步处理都要能回溯，而且要支持人工兜底。这正好能逼着我们把消息协议、状态管理、可观测性这些基础设施做扎实，而不是只搭一个看起来很炫但没法上生产的壳子。

2. 整体架构设计：三大核心抽象与一条总线的平衡

2.1 注册表、任务队列与消息总线

agency-agents 的运行时架构可以拆成三个核心组件：Agent 注册表、任务队列和消息总线。

Agent 注册表是一个轻量级服务发现组件，存着所有 Agent 的类型、能力描述、当前健康状态、路由权重等信息。注册表存在的意义是让编排器不感知具体 Agent 实例，而是按能力名称去查询可用节点。比如我们系统里有一个“工单分类 Agent”，注册表里就存着它的能力标签ticket_classify、地址和当前负载情况。这样后续如果要把某类 Agent 扩容成多个实例，编排器只需要改轮询策略，业务代码不用动。

任务队列承担的是异步削峰的作用。因为多个 Agent 执行任务耗时长短不一，如果所有请求都同步阻塞等待，会有大量空转时间浪费。我们当时用了一个分层队列的设计，优先级高的工单走快速通道，普通工单走默认队列，这样既保证了核心用户体验，又提高了整体吞吐。

消息总线是所有 Agent 之间通信的唯一通道。每条消息都遵循统一的消息协议——包含来源 Agent ID、目标 Agent ID、任务类型、消息体、时间戳和追踪 ID。追踪 ID 这个字段太关键了，没有它的排查工作量至少翻一倍。

2.2 为什么消息协议要“窄接口”而不是“大对象”

在定义消息协议的时候，团队讨论过两种方案。一种是直接把业务对象整个传过去，消息里可以包含用户信息、订单详情、操作记录等等；另一种是我们最终采用的“窄接口”，消息体只包含当前任务必要的结构化字段，其他数据通过统一的存储服务去拉取。

窄接口的优势在复杂业务流程里会越来越明显。如果消息体里什么都塞，每个 Agent 都要处理大量冗余字段，既浪费模型 Token 开销，又容易出现字段更新的数据不一致问题。比如工单状态在某流程节点变了，其他 Agent 手里拿到的还是旧消息体里的快照，就会产生误判。我们改成窄接口后，处理指定任务时，Agent 通过消息体里的业务对象 ID 去查询最新的业务状态，保证了每次拿到的都是准确数据。

当然窄接口的代价也很直观——Agent 之间需要约定好业务对象的查询接口，多了些接口设计的成本。但实际跑下来这几个额外的接口定义是非常值得的，后续加新业务场景的时候复用性特别高。

2.3 会话状态与业务状态的分层管理

多智能体系统里最难搞的就是状态管理。我们把状态分成了两层。

会话状态，也就是 Agent 在完成一次任务过程中的上下文记忆，比如当前对话的中间结果、已经执行过的工具调用轨迹。这一层遵循“谁处理谁短暂持有”的原则，任务完成后集中写回持久化存储，不在各 Agent 本地停留过久。

业务状态则是系统要长期保证一致性的数据，比如工单状态、库存数量、审核结果。这一层全部通过业务存储接口管理，由编排器统一协调写操作，避免多个 Agent 同时修改同一条记录造成冲突。

这种分层设计相当于把 Agent 当成“无状态工人”：它们随时可以崩溃、重启、替换，只要任务队列里的消息还在，业务状态存储还完整，整个流程就能继续推进。这也是多智能体架构相比单 Agent 方案最大的可靠性红利。

3. 核心环节实现：手写一个最小可用的多 Agent 调度器

3.1 消息定义与路由规则

直接上代码看核心实现。消息定义我们用的是 Python 的 dataclass，简洁也够用。消息类型先区分任务请求、任务结果、失败回执、心跳事件四类。

# agent_platform/message.py from dataclasses import dataclass, field from typing import Any, Dict from enum import Enum from datetime import datetime class MessageType(str, Enum): TASK_REQUEST = "task.request" # 任务请求 TASK_RESULT = "task.result" # 任务成功结果 TASK_FAILURE = "task.failure" # 任务失败回执 HEARTBEAT = "heartbeat" # 心跳事件 @dataclass class AgentMessage: msg_id: str trace_id: str # 全链路追踪ID，排查问题时救命用 msg_type: MessageType sender: str # 来源Agent名称 recipient: str # 目标Agent名称，支持通配符 task_type: str # 任务类型，比如 ticket_classify payload: Dict[str, Any] # 窄接口消息体 created_at: datetime = field(default_factory=datetime.now)

路由规则的实现我们没用复杂的规则引擎，直接基于 task_type 和 Agent 注册表映射关系做分发。每个任务类型可以注册多个处理器，调度器按负载权重做选择。

# agent_platform/router.py class RegistryRouter: def __init__(self, registry): self._registry = registry def route(self, task_type: str): candidates = self._registry.query(task_type) if not candidates: raise NoAvailableAgentError(task_type) # 按当前负载排序，优先选负载最小的节点 active = [c for c in candidates if c.is_healthy()] if not active: raise NoHealthyAgentError(task_type) return min(active, key=lambda ag: ag.current_load)

一个有意思的细节是路由排序。早期我们采用的是固定顺序取第一个健康节点，结果某个节点因为网络抖动导致处理时间莫名变长，工单积压了一大堆。后来改成按当前执行队列长度动态选节点，整体吞吐立刻起来了，这算是一个非常便宜的负载均衡方案。

3.2 编排器的实现思路

编排器是整个框架的“交通警察”。它的职责不是处理业务，而是负责任务拆解、子任务下发、结果回收、失败重试流程。实现思路上核心是维护一个有状态的任务图（TaskGraph），节点是子任务，边是依赖关系。

# agent_platform/orchestrator.py class Orchestrator: def __init__(self, router, bus, storage): self._router = router self._bus = bus self._storage = storage async def run_pipeline(self, pipeline_id: str, task_graph: TaskGraph, init_payload: dict): self._storage.mark_pipeline_started(pipeline_id, task_graph) for ready_task in task_graph.ready_tasks(): await self._dispatch_subtask(pipeline_id, ready_task, init_payload) completed = 0 while completed < task_graph.total_tasks(): msg = await self._bus.recv() if not self._is_relevant(msg): continue self._storage.record_result(pipeline_id, msg) children = task_graph.children(msg.task_type) if children and all(self._storage.check_dependencies_done(pipeline_id, c) for c in children): for child in self._storage.fetch_result(context_id=msg.trace_id): await self._dispatch_subtask(pipeline_id, child, {}) completed += 1 return self._storage.get_pipeline_result(pipeline_id)

注意到这里有一个和很多初学者直觉不一样的设计：编排器不是一次性把所有子任务全部发到消息总线，而是按依赖关系逐步释放。比如“生成工单摘要”这个任务，一定要等“工单分类”和“客户画像拉取”都完成后才能执行，如果一开始就把全部任务发出去，下游 Agent 提前拿到消息后也处理不了，还得等轮询，白白浪费资源。

3.3 观察者模式与回调机制

调度器跑通之后，另一个关键点是回调机制。因为任务队列是异步的，上层业务如果只是干等结果，用户体验并不好。我们引入观察者模式，每个编排任务可以注册相应的事件回调，比如当ticket_classify任务完成时，自动触发另一个 Webhook 通知运营同事。

这块的实现并不复杂，本质上就是订阅发布模型，但它的意义在于让多智能体系统有了“事件驱动”的能力。原本需要业务轮询才能感知的流程变化，现在全部通过事件推送完成，整体交互链路轻了很多。

3.4 幂等与重试：为什么必须处理重复消息

分布式系统里“消息至少一次投递”是常态，多智能体协作也不例外。消息总线可能因为网络分区导致消息重复消费，如果我们的编排器不做幂等处理，同一个工单可能被重复提交给下游 Agent，造成重复扣库存或者重复发券的严重事故。

我们的处理方式是给每条消息分配业务幂等键（Idempotency Key），由存储层实现去重。Agent 在开始执行任务前先检查这个幂等键是否处理过，处理过就直接返回已缓存的结果。

# agent_platform/executor.py class AgentExecutor: async def run(self, msg: AgentMessage): origin_id = self.decorator.get_idempotent_key(msg) if await self.store.already_processed(origin_id): return await self.store.fetch_cached_result(origin_id) result = await self._execute_core(msg) await self.store.cache_result(origin_id, result) return result

幂等键建议放在消息头的统一字段里，不要把业务字段拼出来当幂等键用。否则业务字段一变，系统就会误判为新任务，去重逻辑直接失效。

4. 实战演练：如何用这套框架跑通一次工单自动处理

4.1 场景定义与 Agent 角色划分

理论讲再多，不如一个拿来就能用的案例实在。我们拿了客服工单自动处理场景做验证。

工单从提交到关闭大致需要四步：自动分类、内容摘要生成、解决方案匹配、结果质检。对应到 agency-agents 的架构里，我设计了三个 Worker Agent 和一个质检 Agent。

工单分类 Agent 负责判断工单属于哪个领域（账号问题、支付问题、技术故障等），输出一个分类标签和置信度分数。摘要生成 Agent 负责把用户的长描述压缩成一条几十字的工单摘要。解决方案匹配 Agent 会带着分类标签和摘要去知识库检索，返回 Top3 的候选方案。质检 Agent 则专门检查前面的输出是否完整、是否有明显矛盾，必要时发起人工复核。

为什么把质检单独拉出来做成一个 Agent 而不是合在摘要 Agent 里？因为我们测试时发现，同一个 Agent 既做生成又做校验，很容易出现“自己给自己找理由”的情况——生成结果有点小瑕疵，校验时自己就把瑕疵合理化了。拆成独立 Agent 以后，质检通过率立刻有了明显提升，因为生成逻辑和校验逻辑相互独立，幻觉发生的概率都低了不少。

4.2 任务拆解与执行流

三个主要 Agent 中，分类和摘要是并行的，因为它们依赖的信息源不同，互不阻塞；解决方案匹配则必须等分类结果出来后执行；最终质检是最后一个环节。

# scenario_template.py graph = TaskGraph() graph.add_node(TaskNode("ticket_classify", dependencies=[])) graph.add_node(TaskNode("ticket_summarize", dependencies=[])) graph.add_node(TaskNode("solution_match", dependencies=["ticket_classify", "ticket_summarize"])) graph.add_node(TaskNode("quality_check", dependencies=["solution_match"]))

执行流程上，编排器先下发 classify 和 summarize，等两者都完成后，再下发 solution_match，最后 quality_check。这个依赖顺序很好理解，但我在这里特别想强调一个容易被忽略的点——上游 Agent 的输出质量直接影响下游处理效果。

我们遇到过这样的情况：摘要 Agent 把用户一句关键抱怨“账号在异地登录后无法充值”压缩成“用户账号异常”之后，解决方案匹配 Agent 就从知识库里检索出了一堆封号相关方案，完全偏离用户的真实诉求。后来我们把摘要生成 Agent 的 Prompt 里加上强制要求“核心诉求必须出现在摘要前30个字内，禁止泛化描述”，效果立竿见影。

4.3 参数与阈值设置经验

一些关键参数我们也踩出了比较合理的经验值。

置信度阈值：工单分类 Agent 输出的置信度高于 0.85 时，直接走自动流程；低于 0.6 的工单自动转人工；0.6 到 0.85 之间的走“带确认的自动回复”，系统先展示方案，用户确认后才正式回复。这样一个三段式的设法，让早期准确率不高的情况下，自动覆盖率也能保持在比较高的水平。

超时时间：单个 Worker Agent 的单次执行超时我们设的是 30 秒，超过这个时间直接认定失败并触发重试（最多 2 次）。这里有个判断经验——超时设置的过短，会因为模型偶尔的慢响应频繁触发不必要重试，反而加重下游 Agent 的负载；设置的过长，用户等待时间又太久。30 秒对大部分基于中速模型的推理请求来说是一个相对均衡的值。

重试策略：重试采用指数退避，第一次重试前等 2 秒，第二次等 8 秒。这种策略在整体执行链路里的作用是给下游依赖的数据服务时间做一个缓冲，同时峰值时段能明显减少对存储服务的冲击。这里要特别注意：重试必须保证幂等，不然重试带来的收益全部会被重复副作用抵消掉。

下表汇总了我们在工单场景里沉淀的首版参数，你可以直接抄作业再微调：

5. 踩坑记录与排查技巧实录

5.1 Agent 之间循环调用与消息风暴

多智能体架构里最容易翻车的就是消息风暴。我们早期在测试开放式的“旁路协同”功能时，两个 Agent 因为对一个任务归属的判断产生了分歧，A 给 B 发“请处理”，B 发现职责不匹配后回“请转给 A”，A 收到后坚持认为这属于 B，又发回去。几分钟内消息总线上积累了上万条循环消息，直接把消息队列打爆了。

排查时靠的就是追踪 ID。我们抽样提交了大量消息，发现 trace_id 相同的消息来回出现在两个 Agent 之间，很快定位到问题。修复方案是在调度器里加上了跳数限制——每条消息最多转发 5 跳，超过直接丢弃并告警。同时给同一个 trace_id 在单位时间内设定最大消息条数，超出后自动熔断。

这个坑的核心教训是：自治式消息广播虽然灵活，但必须在基础设施层加护栏，不能完全依赖各个 Agent 的自觉。

5.2 共享业务状态的脏读问题

另一个印象深刻的坑是 Agent 并发修改业务状态导致的脏读。在工单场景里，分类 Agent 和质检 Agent 都需要读取工单当前状态，本来没什么问题。但我们后续扩展销售线索清洗场景时，清洗 Agent 和去重 Agent 同时操作同一条线索记录，先是出现 A 改成“已转化”、B 又改回“待跟进”的逻辑冲突，后来又出现两边同时读取到旧版本数据、各自写入一半字段导致记录损坏。

后来统一了状态更新规范，所有业务状态变更都必须通过存储服务提供独占的更新接口，旧的字段更新请求会被版本号校验拒绝。这也是我在第 2.3 节强调状态分层管理的原因——如果当时的执行 Agent 都直接去操作数据库，这个问题的影响面会大一倍。

5.3 上下文截断导致的“失忆”

长会话场景下，多智能体系统的每个 Agent 只能看到当前任务相关的消息片段，这本来是优点。但我们在跑连续多轮会话时发现，主 Agent 因为消息体里塞入了前面几轮的全部历史摘要，上下文超长后被模型静默截断，导致它的“记忆”出现断层——前面会话里的用户偏好，到第三轮时突然完全想不起来了。

这个问题暴露出的真正问题不是模型上下文不够大，而是我们把记忆相关的信息都放进了消息体里，没有按需加载。最终我们改了存储策略，只在消息体里保留长期记忆摘要的索引，调度器根据当前任务类型按需把相关的历史片段注入到上下文里。

5.4 可观测性：追踪一次跨 Agent 调用的完整链路

多智能体系统排障难度远高于单 Agent，因为一次业务请求可能横跨好几个 Agent、好几次消息投递。可观测性绝对不是锦上添花，而是保命手段。

我们把每个传入请求生成一个 trace_id，Webhook 入口、编排器、执行器、消息总线、存储服务全部打印这个 trace_id。实测排查效率提升很明显——之前出了问题要靠人工拼接各种日志片段，现在直接按 trace_id 把整条调用链拉出来看。这里分享一个经验：除了路径追踪，每个 Agent 处理前后的 payload 快照才是最有用的排障信息。很多问题出现不是因为链路的某一步报错，而是某一步悄悄改变了数据结构。

5.5 常见问题速查表

写在最后的一个小技巧

最后聊一个我们在 agency-agents 项目里反复验证出的小技巧。如果你也在设计多智能体协作平台，不要一上来就追求“大而全”的通用编排能力，先挑一个业务场景把闭环跑通。我们最初只做了工单自动处理一个场景，但三个 Agent 之间消息协议、路由规则、状态分层、幂等机制这些核心能力全部沉淀了下来。后面扩新场景时，基本就是注册新 Agent、写新任务图、调阈值参数这三件事，不需要再动框架层的东西。先把地基打牢，后续的业务扩展会轻松非常多。