企业官网建设流程全解析

1. 这套系统到底在解决什么问题

多人多AI协同，说白了就是让多个用户和多个AI代理在同一个任务空间里干活，而且AI代理不只是被动应答，而是能主动代表用户去跟其他AI或其他人交互。我最早接触这个方向是因为一个很具体的痛点：团队里三个人同时用各自的AI助手处理同一份需求文档，结果三个AI给出的方案互相矛盾，版本对不上，最后还得人工去对齐。那时候我就在想，能不能让AI代理之间先协商好，再把人拉进来做决策？

这个系统的核心价值在于：把“人跟AI一对一聊天”升级成“人跟AI代理团队协作”。适合谁来参考？如果你正在做多智能体编排、企业级AI工作流、或者需要多个AI角色分工完成复杂任务的场景，这套架构思路可以直接拿去改。如果你只是想让一个AI帮你写写邮件，那确实用不上，但理解一下代理间协商的机制也没坏处。

我实测下来，这套架构最直接解决的问题有三个：第一，多个AI代理之间的状态同步和冲突消解；第二，人类用户如何在不被信息淹没的前提下介入关键决策；第三，整个交互过程的内容合规性怎么在架构层面保证，而不是靠事后过滤。

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

2.1 为什么是“代理代为交互”而不是“人直接指挥每个AI”

一开始我也想过，让用户直接给每个AI下指令不就行了？但实际操作中你会发现，当AI代理数量超过三个，用户根本管不过来。就像你不可能同时跟五个下属分别交代任务还保证他们互相不打架。所以核心思路是：用户只需要跟一个“主代理”交互，主代理负责把任务拆解、分发给其他专业代理，并在代理之间出现分歧时进行仲裁。

这个设计的关键在于代理层级。我采用的是三层结构：用户层、协调层、执行层。用户层只有一个入口，协调层负责意图理解和任务分解，执行层是多个专业代理各司其职。协调层的代理我称之为“调度代理”，它不直接干活，而是决定谁干活、怎么干、干到什么程度算完。

注意：调度代理的权限必须明确边界，否则它会变成一个瓶颈。我的做法是给调度代理设置最大递归深度和最大代理调用数，防止它无限拆解任务导致系统卡死。

2.2 多AI协同的通信机制选型

代理之间怎么说话，这是架构里最容易踩坑的地方。我试过三种方案：共享内存、消息队列、以及基于事件总线的发布订阅。最后选了事件总线，原因很直接：共享内存虽然快，但多个代理同时读写同一块状态时，锁竞争严重，而且调试极其困难；消息队列适合点对点，但多代理广播场景下要维护大量队列，运维成本高。

事件总线的优势在于解耦。每个代理只管往总线发事件，谁关心谁订阅。比如“任务分解完成”这个事件，执行代理订阅了就开始干活，日志代理订阅了就开始记录，合规检查代理订阅了就开始扫描内容。新增一个代理不需要改其他代理的代码，插上就能用。

具体实现上，我用的是基于Redis Stream的事件总线，每个代理有自己的消费者组。为什么不用Kafka？因为这套系统的消息量级没那么大，Redis Stream足够，而且部署简单，本地开发环境跑一个Redis实例就行。消息格式统一用JSON，包含事件类型、发起代理ID、目标代理ID（可选，为空表示广播）、时间戳、负载数据。

2.3 人类介入点的设计逻辑

多人多AI协同里，人不是旁观者，而是关键决策的触发者。但什么时候该让人介入？我的经验是设置置信度阈值和冲突检测两个触发条件。当某个代理对某个决策的置信度低于阈值（我设的是0.7），或者两个代理对同一问题给出了矛盾结论，系统就暂停自动流程，向人类用户推送一个“决策请求”。

这个决策请求不是简单地把问题抛给人，而是附带上下文：谁提出的、依据是什么、其他代理怎么看、如果选A会怎样、选B会怎样。用户只需要点一下或者简短回复，系统就继续跑。实测下来，一个复杂任务里人类介入次数控制在3到5次是比较舒服的，太多会烦，太少又容易跑偏。

3. 核心模块的实操细节与避坑要点

3.1 代理身份与权限管理

每个AI代理必须有唯一身份标识，这个标识不只是名字，还包括它的能力范围、可访问的数据权限、以及它能调用的工具列表。我见过有人图省事，所有代理共用一个API Key，结果一个代理误操作把另一个代理的数据删了，排查了半天。

我的做法是给每个代理分配独立的凭证，并在调度层维护一张权限表。这张表长这样：

这张表在系统启动时加载，运行时不动态修改。如果需要调整权限，走配置更新流程，重启对应代理。这样做虽然不够灵活，但避免了运行时代理越权的问题。

实操心得：代理的“最大并发任务数”这个参数很关键。我一开始没设限制，结果检索代理被同时调用十几次，API直接限流了。后来根据每个代理背后模型或服务的承载能力来设，稳定多了。

3.2 任务分解与依赖编排

调度代理接到用户需求后，第一步是任务分解。这里我用的是“目标-子目标-动作”三层分解法。举个例子，用户说“帮我写一份竞品分析报告”，调度代理会分解成：确定竞品范围、收集竞品信息、分析对比、生成报告。每个子目标再分解成具体动作，比如“收集竞品信息”分解成“检索竞品官网”“检索行业报告”“检索用户评价”。

分解完成后，调度代理会生成一个有向无环图（DAG），节点是动作，边是依赖关系。这个DAG不是给用户看的，而是给执行代理看的。每个执行代理从DAG里认领自己能做的节点，做完之后更新节点状态，触发下游节点。

这里有个坑：DAG的粒度不能太细，否则代理之间通信开销巨大；也不能太粗，否则一个代理干太久，其他代理闲着。我的经验是每个动作节点的预期执行时间控制在30秒到2分钟之间，超过2分钟的就再拆，低于30秒的就合并。

3.3 代理间冲突消解机制

多个代理协同，冲突是必然的。比如检索代理找到的数据和生成代理之前生成的内容矛盾，或者两个执行代理对同一个子任务给出了不同方案。我的冲突消解分三步走：

第一步，自动检测。合规代理和调度代理都会监听事件总线，发现矛盾结论时触发冲突事件。第二步，协商。冲突双方代理各自陈述理由和依据，调度代理根据预设规则（如数据来源权威性、时效性、与用户历史偏好的一致性）给出建议。第三步，如果调度代理无法裁决，或者冲突涉及关键决策，就升级到人类用户。

我实测下来，80%的冲突可以在自动协商阶段解决，剩下20%需要人介入。这个比例跟代理的能力配置有关，如果检索代理的数据源质量高，冲突率会明显下降。

4. 完整实操流程与关键环节实现

4.1 环境准备与基础组件搭建

这套系统对硬件要求不算高，我是在一台16核32G的机器上跑的，同时运行8个代理实例加一个Redis和一个PostgreSQL。如果你只是做原型验证，8核16G也够用。

基础组件清单：

• Redis 7.0+：用作事件总线和代理状态缓存
• PostgreSQL 15+：存储任务元数据、代理配置、审计日志
• Python 3.11+：代理运行时，我用的是asyncio做异步处理
• 一个本地模型或API接入：每个代理背后可以接不同的模型，我测试时用了本地部署的小模型做检索代理，大模型API做生成代理

安装依赖的命令很简单：

pip install redis asyncpg httpx pydantic

Redis的配置需要注意开启Stream功能，默认是开的，但持久化策略要调一下。我建议用AOF持久化，appendfsync设为everysec，兼顾性能和安全。

4.2 代理基类的实现要点

每个代理都继承一个基类，基类负责事件总线的连接、消息的序列化反序列化、心跳上报、以及优雅退出。这个基类我改了大概五版才稳定，核心是处理好异步任务的取消和异常传播。

import asyncio import json import redis.asyncio as redis from pydantic import BaseModel class AgentMessage(BaseModel): event_type: str source_agent: str target_agent: str | None timestamp: float payload: dict class BaseAgent: def __init__(self, agent_id: str, redis_url: str): self.agent_id = agent_id self.redis = redis.from_url(redis_url) self.running = True async def publish(self, event_type: str, payload: dict, target: str = None): msg = AgentMessage( event_type=event_type, source_agent=self.agent_id, target_agent=target, timestamp=asyncio.get_event_loop().time(), payload=payload ) await self.redis.xadd("agent_bus", {"data": msg.model_dump_json()}) async def subscribe(self, consumer_group: str): # 创建消费者组，如果已存在则忽略错误 try: await self.redis.xgroup_create("agent_bus", consumer_group, id="0", mkstream=True) except Exception: pass while self.running: messages = await self.redis.xreadgroup( consumer_group, self.agent_id, {"agent_bus": ">"}, count=10, block=1000 ) for stream, msgs in messages: for msg_id, data in msgs: await self.handle_message(json.loads(data["data"])) await self.redis.xack("agent_bus", consumer_group, msg_id) async def handle_message(self, msg: dict): raise NotImplementedError

这个基类里，subscribe方法用的是消费者组模式，保证同一个消息不会被同一个代理重复消费。block=1000表示没有消息时阻塞1秒，避免空转消耗CPU。

注意：xgroup_create在消费者组已存在时会抛异常，所以要用try-except包住。我一开始没处理这个，每次重启代理都报错，虽然不影响运行但日志很乱。

4.3 调度代理的任务分解逻辑

调度代理是整个系统的大脑，它的核心方法decompose_task接收用户原始需求，输出一个DAG。我用的分解策略是“先广度后深度”：先识别需求涉及哪些领域，每个领域生成一个子目标，再对每个子目标做深度分解。

async def decompose_task(self, user_request: str) -> dict: # 第一步：领域识别 domains = await self.identify_domains(user_request) # 第二步：为每个领域生成子目标 sub_goals = [] for domain in domains: goals = await self.generate_sub_goals(user_request, domain) sub_goals.extend(goals) # 第三步：构建DAG dag = {"nodes": [], "edges": []} for i, goal in enumerate(sub_goals): dag["nodes"].append({ "id": f"node_{i}", "goal": goal, "agent_type": self.match_agent_type(goal), "status": "pending" }) # 根据依赖关系添加边 deps = await self.infer_dependencies(goal, sub_goals[:i]) for dep in deps: dag["edges"].append({"from": dep, "to": f"node_{i}"}) return dag

identify_domains和generate_sub_goals都是调用模型完成的，但我在prompt里加了约束：领域数量不超过5个，子目标总数不超过15个。这是为了防止DAG过大导致执行代理过载。

4.4 执行代理的任务认领与执行

执行代理启动后，先向调度代理注册自己的能力标签，然后进入事件循环。当调度代理发布“任务就绪”事件时，执行代理检查DAG中是否有自己可执行的节点（前置依赖都已完成且自己有能力执行），有就认领并执行。

认领机制我用的是乐观锁：执行代理先读取节点状态，如果状态是pending且自己匹配，就尝试用Redis的原子操作把状态改为claimed。改成功了就执行，改失败了说明被别的代理抢了，跳过。

async def try_claim_node(self, node_id: str) -> bool: key = f"node_status:{node_id}" # 使用Redis的SET NX实现原子认领 result = await self.redis.set(key, self.agent_id, nx=True, ex=300) return result is not None

ex=300是认领超时时间，5分钟。如果代理认领后崩溃了，5分钟后其他代理可以重新认领，避免任务永久卡住。

执行完成后，代理更新节点状态为completed，并发布“节点完成”事件。调度代理监听这个事件，检查DAG中是否有新的节点变为可执行状态，有就发布新的“任务就绪”事件。

4.5 人类介入的推送与响应处理

当系统需要人类决策时，调度代理会生成一个决策请求，推送到用户界面。推送渠道我试过邮件、企业微信机器人、以及自建的WebSocket推送。邮件太慢，企业微信机器人适合团队场景，WebSocket适合有自己前端的情况。我最后用的是WebSocket加企业微信机器人双通道，确保用户能及时看到。

决策请求的格式：

{ "request_id": "dec_001", "type": "conflict_resolution", "summary": "检索代理和生成代理对竞品A的市场份额数据存在分歧", "options": [ {"id": "opt_1", "desc": "采用检索代理的数据（来源：2024年Q2行业报告）", "confidence": 0.85}, {"id": "opt_2", "desc": "采用生成代理的数据（来源：模型推理）", "confidence": 0.62} ], "context": "该数据将用于报告第三章的市场分析部分", "deadline": "2025-01-15T18:00:00Z" }

用户回复后，调度代理根据选择更新相关节点的数据，继续执行。如果超时未回复，系统会采用置信度更高的选项，并记录一条“超时自动决策”日志。

5. 内容合规的架构级保障

5.1 为什么合规不能只靠事后过滤

很多人做AI系统，合规就是最后加一个敏感词过滤。但这套系统里，多个代理互相交互，内容在代理之间流转，如果只在最终输出时过滤，中间过程可能已经产生了不合规的中间结果，甚至被其他代理引用放大。所以我的做法是：合规检查嵌入到事件总线的每一个环节。

具体来说，合规代理订阅所有事件，对每个事件的payload做实时扫描。扫描分三级：第一级是关键词匹配，速度快但误报率高；第二级是语义分析，用一个小模型判断内容是否涉及风险；第三级是人工复核队列，只有前两级都拿不准的才进队列。

5.2 合规检查的性能优化

全量扫描所有事件对性能有影响。我实测过，如果每个事件都走语义分析，系统吞吐量下降约40%。优化方案是分级处理：关键词匹配对所有事件都做，语义分析只对包含文本内容且长度超过50字的事件做，人工复核队列每天清理一次。

另外，合规代理的扫描是异步的，不阻塞主流程。但如果扫描发现高风险内容，会立即发布“合规告警”事件，调度代理收到后暂停相关任务，等待人工确认。

实操心得：合规规则的更新频率很高，我建议把规则配置放在数据库里，合规代理定期拉取最新规则，而不是硬编码在代码里。这样调整规则不需要重启代理。

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

6.1 代理之间消息丢失或重复

这是最常见的问题。消息丢失通常是因为消费者组没有正确ack，或者Redis内存满了被淘汰。我的排查步骤：先看Redis的info memory，确认used_memory是否接近maxmemory；再看消费者组的pending列表，xpending agent_bus consumer_group，如果有大量pending说明代理处理不过来。

消息重复则是因为ack时机不对。我一开始是处理完才ack，但如果处理过程中代理崩溃，消息会被重新投递。后来改成处理前先ack，但这样如果处理失败消息就丢了。最终方案是：处理前ack，但处理失败时把消息重新发布到总线，并记录失败日志。

6.2 调度代理任务分解不合理

有时候调度代理会把一个简单任务拆成十几个子任务，导致执行代理频繁切换上下文，效率反而低。排查方法是看DAG的节点数量和平均执行时间。如果节点数超过15个，或者平均执行时间低于20秒，就说明分解过细。

调整方法是在调度代理的prompt里加约束，或者在调度层加一个后处理步骤：合并那些可以顺序执行且属于同一代理的小节点。我后来加了一个“节点合并”逻辑，把同一代理的连续小节点合并成一个，效果很明显。

6.3 人类介入请求被忽略

用户可能因为忙没看到决策请求，导致任务卡住。我的解决方案是设置多级提醒：第一次推送后5分钟没响应，发第二次提醒；15分钟没响应，发第三次并抄送备用联系人；30分钟没响应，自动采用默认选项并记录。默认选项的选择逻辑是：如果有置信度超过0.9的选项，选它；否则选保守方案（即不引入新数据、不改变现有结论的方案）。

6.4 合规误报导致任务频繁暂停

合规代理的语义分析模型如果太敏感，会把正常内容判为风险，导致任务频繁暂停。我踩过这个坑，后来调整了策略：只有置信度超过0.95的风险才触发暂停，0.8到0.95之间的只记录日志不暂停，低于0.8的忽略。这样既保证了安全，又减少了误报干扰。

7. 这套架构的扩展方向与个人体会

这套系统跑通之后，我陆续做了一些扩展。一个是加了“代理记忆”，每个代理维护自己的短期记忆和长期记忆，短期记忆存在Redis里，长期记忆存PostgreSQL，这样代理在处理相似任务时能复用之前的经验。另一个是加了“代理能力动态注册”，新代理上线时自动向调度代理注册自己的能力标签，不需要改配置文件。

我个人在实际操作中的体会是：多AI协同系统的难点不在AI本身，而在协调机制。代理之间的通信协议、冲突消解规则、人类介入时机，这些才是决定系统好不好用的关键。模型能力可以慢慢升级，但协调机制设计不好，再强的模型也白搭。

最后分享一个小技巧：如果你刚开始做这套系统，不要一上来就搞五六个代理。先从两个代理加一个调度代理开始，把通信和冲突消解跑通，再逐步增加代理数量和类型。我一开始贪多，同时上了八个代理，结果调试了两周才让它们不互相打架。后来砍到三个代理重新跑，一天就通了，然后再慢慢加回去，反而更快。