最近关于 AI 的讨论越来越像一个“都市传说”AI 偷偷拉了一个群互相交换情报密谋一件大事。尤其当多智能体、AI Agent 这些词频繁出现在技术社区里很多非技术读者会脑补出一幅 AI 自主觉醒的画面。作为技术人员与其跟着恐慌不如拆开这个“拉群”的过程看看到底是什么技术机制。所谓“AI 们拉群密谋”本质上就是多个 AI Agent 通过一套消息通信、任务编排和上下文共享机制协作完成一个复杂任务的过程。本文会从多智能体系统Multi-Agent System的概念讲起再用 Python 实现一个最小可运行的“AI 协作群”把消息总线、角色 Agent、任务分发、人工审批这些核心模块逐一落地。即使你没有大模型 API Key也可以直接运行这份 Demo理解多 Agent 之间到底是怎么“聊天”的。读完你会发现AI 拉群不可怕真正决定它安全与否的是群的规则、权限和人类能不能随时叫停。1. 从“AI们拉群”说起多智能体协作是什么1.1 一个真实的“AI拉群”场景假设你在项目群里发了一条需求我们需要一个登录注册功能支持手机号验证码登录并且要记录登录日志。如果只有一个 AI它会直接生成一段代码或一份方案然后任务结束。但如果是“一群 AI”事情会变成这样产品经理 Agent 先拆解需求把任务分成“验证码模块”“登录态模块”“日志模块”开发 Agent 认领任务输出技术方案和排期测试 Agent 收到方案后补充测试建议例如验证码过期、异常输入等边界场景审批 Agent 综合各方结论决定是否进入发布阶段。这个过程中每个 Agent 都在“群里”发言消息被其他 Agent 接收、理解、处理最终形成一条完整的协作链路。这个“群”在技术上通常被称为消息总线Message Bus或编排器Orchestrator。1.2 Agent 与多智能体系统AI Agent智能体可以理解为“不仅能回答问题还能根据目标拆解任务、调用工具、执行动作并依据结果调整策略”的 AI 程序。而多智能体系统就是多个这样的 Agent 在一起工作通过消息传递实现协作。为什么单 Agent 不够用因为真实业务太复杂。一个 Agent 既要懂需求分析又要会写代码还要精通测试用例同时还要知道怎么申请权限模型推理质量和上下文长度很快就会成为瓶颈。与其把所有能力塞进一个巨大的 Prompt不如拆分出多个角色 Agent每个 Agent 只负责一个专业领域再用一套协议让它们互相配合。这也是多智能体系统在 AI 应用开发中越来越流行的原因。1.3 “密谋大事”的技术真相所谓“AI 密谋”至少要满足三个前提多个 Agent 能互相通信、能自由调用外部工具、能绕过人类审批。如果开发者在设计时根本不给 Agent 提供越权工具也没有开放自由对话的循环机制那所谓的“阴谋”从一开始就不会发生。换句话说AI 之间传输的不是“意识”而是结构化的消息。消息由谁发送、发送给谁、携带什么字段、触发什么动作全部由代码定义。我们真正要关注的不是 AI 是否“觉醒”而是消息协议是否完善权限边界是否清晰是否有审计日志一旦把这些问题想清楚“AI 拉群”就不再是玄学而是一套完全可以掌控的工程系统。2. 多智能体系统的核心组件与协作模式2.1 五个核心组件理解多智能体系统最直接的方式是先记住五个核心组件组件作用类比Agent拥有角色和任务处理逻辑的个体群里的一个成员MessageAgent 之间传递的信息载体群聊里的一条消息MessageBus / Orchestrator负责消息路由和流程调度群聊系统本身ToolAgent 可调用的外部能力如查询数据库、调用 API成员手里的工具Shared Memory / Context多个 Agent 共享的上下文状态群公告或共享文档有了这五个组件已经可以搭建一个基础的多 Agent 系统。Agent 通过 Message 互相沟通MessageBus 决定消息怎么流转Tool 让 Agent 能产生实际影响Shared Memory 让所有参与者共享必要信息而 Orchestrator 负责把控整体流程。2.2 三种常见协作模式多智能体系统的协作模式大致有三种中心化编排模式一个 Orchestrator 接收任务拆解后分发给多个子 Agent再汇总结果。优点是流程可控、容易加人工审批缺点是中心节点可能成为瓶颈。去中心化协商模式Agent 之间直接通信互相传递任务和结果适合探索性任务。优点是灵活缺点是容易产生死循环、上下文混乱必须有终止条件。人机混合模式Agent 自动处理低风险环节遇到审批、删除、发布等高危动作时暂停并向人类请求确认。这是生产环境中最推荐的形式。很多人担心的“AI 拉群密谋”其实是第二种模式被滥用的结果。如果 Agent 之间可以无限循环对话、可以无约束调用工具那系统确实可能做出超出预期的事情。解决办法也很简单把高风险动作拆出来放到第三种模式中由人来做最终决策。2.3 消息协议与指令格式无论采用哪种模式Agent 之间通信都不能是纯自然语言随意聊天否则无法保证稳定。工程上一般会定义一套结构化消息格式{ msg_id: f3a12c8d, sender: PM, receiver: Dev, task_type: request, content: 请评估登录注册功能实现方案输出模块拆分与排期, timestamp: 1710000000.123 }msg_id消息唯一标识用于链路追踪sender和receiver标记消息来源和去向task_type区分请求、回复、广播、审批等不同语义content消息正文timestamp时间戳便于排序和审计。在后面的 Python Demo 中我们会用同样思路实现这套消息格式。别看它简单真正的生产级多 Agent 系统消息协议的核心结构也差不多。2.4 关于调用成本与 credits多 Agent 系统会显著增加模型调用次数。一个任务从拆解到开发、测试、审批可能要调用多次大模型接口。在按量计费的平台中每次调用都会消耗 credits额度或 Token。你可以把 credits 理解为调用模型服务的“余额”一条较长消息可能消耗几千甚至上万 Token多 Agent 场景下成本会成倍增长。因此在设计多 Agent 系统时不能只关注效果还要关注每次任务的 Token 成本。常见的优化手段包括减少无关上下文、限制 Agent 回复长度、对重复任务做缓存、用更小的模型处理简单分类和提取任务。这些都应在架构设计阶段提前考虑。3. 环境准备与项目结构3.1 运行环境本文的 Demo 使用 Python 标准库实现不需要额外安装第三方依赖。你可以使用本地已经安装的 Python 3.8 及以上版本操作系统不限Windows、macOS、Linux 均可。核心依赖只有dataclasses、uuid、time、typing这些在 Python 3.7 以后都是内置模块。为了让效果更直观Demo 中会用一个mock_llm函数模拟大模型返回结果。这样即使没有大模型 API Key也可以完整跑通多 Agent 协作流程。如果你身边已经有部署好的模型服务可以将其接入 Demo 中的mock_llm位置替换成真实的大模型调用。本文后面会给出替换思路。3.2 项目结构在任意目录下创建项目文件夹ai_agent_group/ ├── main.py └── README.md为了让读者运行方便所有代码都写在同一个main.py中。代码会从消息结构开始逐步构建消息总线、Agent 基类和角色 Agent最后在主流程中模拟一次“项目群聊”。4. 完整实战用 Python 实现一个最小 AI 协作群4.1 定义消息结构首先定义消息类。消息是 Agent 之间通信的最小单位需要包含发送者、接收者、内容、任务类型等信息。为了让消息对象可以直接打印查看我们使用dataclass来声明。# 文件路径ai_agent_group/main.py from __future__ import annotations import time import uuid from dataclasses import dataclass, field from typing import Callable, Dict, List, Optional dataclass class Message: sender: str receiver: str content: str task_type: str chat msg_id: str field(default_factorylambda: uuid.uuid4().hex[:8]) timestamp: float field(default_factorytime.time)sender和receiver用于消息路由task_type默认是chat我们在 Demo 中会用到request请求处理和reply回复两种类型msg_id自动生成方便在日志中定位同一条消息timestamp用于记录消息产生时间。4.2 实现消息总线消息总线负责把消息从一个 Agent 转发到另一个 Agent。它需要维护两个数据结构agents已注册的 Agent 字典logs所有消息的流水日志相当于群聊记录。转发逻辑很直接如果receiver是all则广播给所有 Agent否则只发送给指定 Agent。为了防止 Agent 之间无限互聊总线还要设定一个最大消息轮次。class MessageBus: def __init__(self, max_rounds: int 30): self.agents: Dict[str, Agent] {} self.logs: List[Message] [] self.max_rounds max_rounds def register(self, agent: Agent) - None: self.agents[agent.name] agent def is_registered(self, name: str) - bool: return name in self.agents def post(self, msg: Message) - None: self.logs.append(msg) if len(self.logs) self.max_rounds: print(f[Bus] 超过最大消息轮次 {self.max_rounds}停止转发。) return if msg.receiver all: print(f[Group] {msg.sender} - 所有人: {msg.content}) for agent in self.agents.values(): if agent.name ! msg.sender: agent.receive(msg) else: if not self.is_registered(msg.receiver): print(f[Bus] 接收方 {msg.receiver} 未注册消息已归档。) return print(f[Direct] {msg.sender} - {msg.receiver}: {msg.content}) self.agents[msg.receiver].receive(msg)4.3 定义 Agent 基类Agent 基类是所有角色的父类。每个 Agent 创建时会自动注册到消息总线并保存自己的历史消息。send方法用于主动发送消息receive方法用于接收消息并调用handle处理。class Agent: def __init__( self, name: str, bus: MessageBus, llm: Optional[Callable[[str], str]] None, ): self.name name self.bus bus self.llm llm or mock_llm self.message_history: List[Message] [] self.bus.register(self) def send(self, receiver: str, content: str, task_type: str chat) - None: msg Message( senderself.name, receiverreceiver, contentcontent, task_typetask_type, ) self.bus.post(msg) def receive(self, msg: Message) - None: self.message_history.append(msg) # 只处理 request 类型的消息避免 reply 再次触发回复形成死循环 if msg.task_type ! request: return response self.handle(msg) if response: target msg.sender if self.bus.is_registered(msg.sender) else all self.send(target, response, task_typereply) def handle(self, msg: Message) - Optional[str]: raise NotImplementedError这里的关键设计是Agent 只对request类型的消息产生回复对reply类型直接忽略。这是防止多 Agent 系统死循环的最简单策略也是很多生产系统的默认选择。如果需求真的需要多轮自由对话则需要额外设计终止条件比如最大轮数、时间上限、语义判断等。4.4 定义模拟大模型函数在真实系统中handle方法内部会去调用大模型。这里为了演示我们用mock_llm返回几个写死的回复模拟不同关键词触发的模型输出。def mock_llm(prompt: str) - str: prompt_lower prompt.lower() if 测试 in prompt or bug in prompt_lower: return 测试Agent分析当前需求缺少边界条件建议先补充异常输入再提测。 if 代码 in prompt or 开发 in prompt: return 开发Agent响应任务已拆解为模块A和模块B预计耗时2人日。 if 需求 in prompt or 规划 in prompt: return 产品Agent已整理用户故事下一步等待开发排期。 return f[模拟LLM] 收到{prompt}已生成默认响应。这段逻辑只是为了验证流程真实场景中应替换为模型网关调用。后续章节会说明如何替换。4.5 定义角色 Agent下面定义四个常见角色。每个角色继承Agent实现自己的handle方法产品经理 Agent负责拆解需求开发 Agent负责评估实现方案测试 Agent负责补充测试建议审批 Agent负责模拟人工审批节点。class ProductManagerAgent(Agent): 产品经理收到需求后拆解为可执行任务。 def handle(self, msg: Message) - Optional[str]: if msg.task_type ! request: return None reply self.llm(f需求规划{msg.content}) return f[{self.name}] 需求已收到开始拆解{reply} class DeveloperAgent(Agent): 开发收到任务后输出方案和排期。 def handle(self, msg: Message) - Optional[str]: if msg.task_type ! request: return None return f[{self.name}] {self.llm(msg.content)} class TesterAgent(Agent): 测试收到方案后补充测试建议。 def handle(self, msg: Message) - Optional[str]: if msg.task_type ! request: return None return f[{self.name}] {self.llm(msg.content)} class HumanApprovalAgent(Agent): 人工审批模拟一个必须由人确认的节点。 def handle(self, msg: Message) - Optional[str]: if msg.task_type ! request: return None print(f[Human] {msg.sender} 提交审批: {msg.content}) return f[{self.name}] 人工审批通过允许进入发布阶段。审批 Agent 是系统安全的关键节点。它的handle方法里可以接入企业审批流、邮件通知、工单系统等也可以暂停等待人工确认实现 Human-in-the-loop。4.6 主流程模拟一次项目群聊最后在主流程中创建消息总线和各个 Agent并按顺序发送任务消息。我们会看到消息如何在 Agent 之间流动。def run_demo(): bus MessageBus(max_rounds30) pm ProductManagerAgent(PM, bus) dev DeveloperAgent(Dev, bus) qa TesterAgent(QA, bus) approval HumanApprovalAgent(Approver, bus) print( 1. 用户向群里发起需求 ) bus.post( Message( senderUser, receiverPM, content我们需要一个登录注册功能支持手机号验证码登录并且要记录登录日志。, task_typerequest, ) ) print(\n 2. 需求评审PM 将任务分发给开发 ) bus.post( Message( senderPM, receiverDev, content请评估登录注册功能实现方案输出模块拆分与排期, task_typerequest, ) ) print(\n 3. 开发给出方案并同步测试 ) bus.post( Message( senderDev, receiverQA, content开发方案已完成模块A验证码、模块B登录态请补充测试建议, task_typerequest, ) ) print(\n 4. 测试补充风险并提交人工审批 ) bus.post( Message( senderQA, receiverApprover, content测试建议需要补充异常输入和验证码过期测试申请发布审批, task_typerequest, ) ) if __name__ __main__: run_demo()将前面所有代码按顺序保存到main.py直接运行即可。5. 运行结果与真实模型替换5.1 运行命令在项目目录下执行python main.py如果使用的是 Python 3.8 及以上版本不需要安装任何依赖。5.2 预期输出运行后控制台会输出类似下面的结果 1. 用户向群里发起需求 [Direct] User - PM: 我们需要一个登录注册功能支持手机号验证码登录并且要记录登录日志。 [Group] PM - 所有人: [PM] 需求已收到开始拆解产品Agent已整理用户故事下一步等待开发排期。 2. 需求评审PM 将任务分发给开发 [Direct] PM - Dev: 请评估登录注册功能实现方案输出模块拆分与排期 [Direct] Dev - PM: [Dev] 开发Agent响应任务已拆解为模块A和模块B预计耗时2人日。 3. 开发给出方案并同步测试 [Direct] Dev - QA: 开发方案已完成模块A验证码、模块B登录态请补充测试建议 [Direct] QA - Dev: [QA] 测试Agent分析当前需求缺少边界条件建议先补充异常输入再提测。 4. 测试补充风险并提交人工审批 [Direct] QA - Approver: 测试建议需要补充异常输入和验证码过期测试申请发布审批 [Human] QA 提交审批: 测试建议需要补充异常输入和验证码过期测试申请发布审批 [Direct] Approver - QA: [Approver] 人工审批通过允许进入发布阶段。从输出可以看到PM 收到需求后先广播了拆解结论Dev 收到 PM 的任务请求后产生了开发方案并直接回复 PMQA 收到 Dev 的方案后补充了测试建议Approver 收到审批请求后模拟了人工审批节点。整个链路都在消息总线的控制范围内没有出现无限循环也没有任何 Agent 越权操作。这就是一个最小可用的“AI 协作群”雏形。5.3 将模拟 LLM 替换为真实模型真实项目中mock_llm需要替换为可用的模型服务。由于不同模型提供方的接口差异较大这里给出通用接入思路而不是固定代码。import os def real_llm(prompt: str) - str: # 1. 从环境变量读取模型服务配置不要在代码里硬编码密钥 api_url os.getenv(LLM_API_URL, ) api_key os.getenv(LLM_API_KEY, ) # 2. 根据你使用的模型服务文档构造请求 # 以 OpenAI 兼容接口为例结构因服务而异请以官方文档为准 # payload { # model: your-model-name, # messages: [{role: user, content: prompt}], # } # headers {Authorization: fBearer {api_key}} # resp requests.post(api_url, jsonpayload, headersheaders, timeout30) # return resp.json()[choices][0][message][content] return 真实模型响应然后创建 Agent 时传入llmreal_llmpm ProductManagerAgent(PM, bus, llmreal_llm) dev DeveloperAgent(Dev, bus, llmreal_llm) qa TesterAgent(QA, bus, llmreal_llm)需要提醒的是多 Agent 场景下模型参数需要调整。比如将temperature调低到 0.2 左右减少随机性设置合理超时在调用失败时进行有限次数重试每次调用前估算 Token 消耗。6. 常见问题与排查思路问题现象可能原因解决思路Agent 之间反复对话停不下来消息类型没有区分请求与回复缺少终止条件采用 request/reply 机制设置最大消息轮次回复内容越来越长Token 成本飙升每轮把完整历史全部塞进模型只传必要上下文过长的历史做摘要压缩某个 Agent 一直没有响应消息接收者名称拼写错误或 handler 没有对应条件检查receiver是否已注册在handle中增加默认分支调用模型接口超时或报错网络不稳定、服务端限流、credits 不足增加重试和熔断设置合理超时监控额度消耗Agent 执行了非预期操作Agent 权限过大或工具调用无人工审批最小权限原则高危动作强制人工审批记录全量日志消息顺序乱掉消息发送和回复使用了异步线程没有按任务链路串联引入trace_id在日志中串联同一任务的所有消息7. 最佳实践与工程建议7.1 安全第一防止“密谋”失控多 Agent 系统越灵活安全边界越要收窄。以下几条是底线不直接给 Agent 操作系统级权限例如删除文件、修改数据库、执行高危 Shell 命令高危操作必须经过人工审批节点审批通过后再放行每个 Agent 只拥有完成自己任务所需的最小权限所有 Agent 的输入输出都写入日志便于事后追溯生产环境开启审计能力记录谁在什么时候给哪个 Agent 下发了什么指令。7.2 消息协议设计多 Agent 系统的稳定性很大程度上取决于消息协议。建议一开始就实现以下字段trace_id一次业务任务的全局唯一标识sender和receiver严格使用 Agent 注册名task_type区分请求、回复、审批、通知priority高优先级任务可插队payload结构化数据而不是纯字符串。有了trace_id后续排查问题时就能把同一任务的几十条消息串联成一条完整链路。7.3 控制上下文与 Token每个 Agent 的message_history如果无限增长最终会把模型上下文塞满。生产中更推荐的做法是Agent 只接收与自己任务相关的消息历史消息超过阈值时让模型先做一次摘要把摘要作为共享上下文不把全量对话历史传给模型而是把“必要结论”传给模型对简单任务使用更小的模型对复杂推理使用更强的大模型混合调度控制成本。7.4 可观测性与评估多 Agent 系统比单 Agent 更难调试因此可观测性必须前置设计。建议在每个 Agent 的receive前后打印或记录输入消息内容模型使用的参数模型返回耗时消耗的 Token 数量后续触发的动作。同时建立评估集。每次修改 Prompt、模型或消息协议后用一组固定的测试任务回归运行对比结果避免“改了一个 Agent影响了整条链路”。7.5 生产环境落地顺序不要一上来就搭建全自动多 Agent 系统。风险最低的落地顺序是先用单 Agent 跑通核心业务流程把业务流程拆成几个固定步骤用脚本串联引入第二个 Agent验证消息路由和上下文传递加入人工审批节点把高风险动作交给人类最后再考虑更复杂的自由协商模式。每一步都应有明确的评估标准和回滚方案。越复杂的系统越要小步验证。8. 总结与下一步学习路线“AI 们拉了个群密谋大事”是一句很有传播力的比喻但在工程技术视角下它无非是一套由消息、路由、编排和审批组成的系统。本文用一个标准库 Demo 还原了这套机制的核心链路消息结构、消息总线、Agent 基类、角色 Agent、人工审批节点。运行这段代码后你应该能直观理解 AI Agent 之间是如何通信的。下一步可以沿着以下方向继续深入学习主流多 Agent 框架例如 AutoGen、LangGraph、CrewAI、Spring AI读它们的消息协议和调度源码实践工具调用让 Agent 能查询数据库、搜索文档、读取日志但始终保留审批节点研究上下文压缩、任务规划、错误恢复等生产级问题为你的多 Agent 系统接入真实模型并建立成本和效果监控。与其害怕 AI 拉群不如先自己写一个“群”亲手定义它的规则、边界和终止条件。理解之后你就会发现真正值得敬畏的从来不是 AI 本身而是设计 AI 协作系统的人。