标题里的场景这两年经常被拿出来讨论当多个大模型驱动的智能体在同一个任务中互相传递消息、分工协作、调用外部工具时从外部看就好像一群 AI 偷偷拉了个群正在“密谋”完成某个目标。这类现象背后其实是多智能体系统Multi-Agent System在工程侧的快速落地。本文不制造焦虑而是把这种“AI 拉群”的技术本质拆开看一遍并提供一个可以直接跑起来的最小代码示例帮助读者理解多智能体协作的通信机制、调度方式和安全边界。文章面向对 AI Agent 感兴趣的开发者。读完你会理解多智能体系统的核心概念掌握一套用 Python 实现“AI 群聊”协作框架的方法还能避开我在实际项目中踩过的几个典型坑。1. 从“AI 拉群”说起多智能体协作是什么1.1 这个标题背后到底发生了什么先说一个共识大模型本身不会主动“拉群”也不会产生人类意义上的社交动机。所谓“AI 密谋”本质上是大模型被组织成了多个 Agent每个 Agent 拥有独立的角色提示词、上下文记忆和工具权限然后通过消息传递机制在同一个任务里协作。从外部观察者的视角看多个 Agent 依次发言、互相补充、迭代输出确实很像“群里在讨论方案”。真正让“AI 拉群”这个话题有讨论价值的是工程上已经出现的几种多智能体框架。它们有的采用“主持人-开发者-审查者”的分工模式有的采用“规划-执行-验证”的流水线模式还有的让多个 Agent 并行探索不同方向后再汇总结论。无论哪种方式底层的调度逻辑都脱离不了消息队列、上下文管理和角色设定这三个核心组件。理解这一点很重要。当你看到“AI 们密谋大事”这类标题时可以自然地在心里做一个翻译这不是失控的前兆而是一套软件系统正在按照既定逻辑执行任务编排。恐惧来源于未知而技术文章要做的事情就是消除未知。1.2 多智能体系统的核心概念多智能体系统不是新概念早在传统人工智能研究里就有分布式 Agent 的理论框架。在大模型时代多智能体系统变得更加轻量每个 Agent 不再需要独立训练而是通过提示词约束同一个大模型的输出风格与行为模式。一个典型的 Agent 通常包含四个部分角色定义告诉模型“你是谁”“你的职责边界是什么”。上下文记忆Agent 能看到哪些历史消息记忆长度是多少。决策策略Agent 是否调用工具、调用哪些工具、何时终止。输出规范Agent 的回复是自由文本还是结构化 JSON。多个 Agent 组合在一起时还需要一个调度器。调度器决定发言顺序、终止条件、消息分发策略。如果把 Agent 比作群成员调度器就是群规则本身。1.3 常见应用场景多智能体系统在工程上的价值主要体现在“拆分复杂任务”和“模拟多方协作”两个方向。任务拆分方向典型场景包括把一个软件开发需求拆成“产品分析—架构设计—代码实现—代码审查”四个阶段把一个市场分析任务拆成“数据采集—数据分析—报告撰写—报告校对”四个步骤。每个 Agent 只负责一个小环节上下文更聚焦输出质量往往比单个长提示词更稳定。多方协作方向典型场景包括模拟售前对话、产品经理与研发评审、客服与用户的多轮沟通。通过给不同 Agent 设置不同的立场和知识背景可以在一段可控的对话里生成接近真实协作过程的语料。这些场景有一个共同点都需要把一个大任务切成多个小任务并且让不同角色的模型输出在流程上有先后依赖关系。多智能体系统本质上就是给这种“先后依赖关系”提供了一套工程实现框架。2. 环境准备与依赖说明2.1 开发环境本文的示例代码不需要复杂的服务端部署只要本机有 Python 环境即可。我使用的环境如下Python 3.10requests 库用于发送 HTTP 请求可选OpenAI 兼容接口或本地部署的模型服务版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。如果你没有可用的 API Key可以使用代码里提供的 FakeLLM 模拟客户端一样能跑通完整的调度流程。安装依赖只需要一条命令pip install requests如果你希望读取本地.env文件来管理 API Key可以额外安装pip install python-dotenv2.2 项目结构为了演示方便我把代码拆成几个独立的模块每个模块职责单一方便你后续扩展。multi_agent_demo/ ├── agent.py # Agent 定义封装角色提示词和模型调用 ├── message.py # 消息结构定义 ├── scheduler.py # 群聊调度器控制发言顺序和终止条件 ├── llm.py # 大模型客户端封装包含 FakeLLM 与 OpenAICompatLLM ├── main.py # 程序入口组装 Agent 并启动调度 └── README.md # 可选项目说明文件这样拆分的好处是将来你想扩充 Agent 数量只需要在main.py里新增 Agent 角色你想切换模型服务只需要换掉llm.py中的客户端实现。3. 多智能体系统的核心原理拆解3.1 Agent 的感知-决策-行动循环Agent 的工作方式可以概括为一个循环感知当前输入决策下一步做什么行动产生输出。在大模型语境下这个循环通常表现为一次模型调用。Agent 收到的输入是“系统提示词 当前任务 历史消息”模型根据这些信息生成回复文本。如果设计中需要 Agent 调用外部工具则在模型输出结构化工具调用参数后由代码去执行对应函数再把执行结果返回给模型继续下一轮生成。感知-决策-行动循环的关键在于Agent 不能只看到当前任务还要看到群里的历史讨论。否则每个 Agent 都会答非所问协作也就不存在了。这也是我在下文实现中把历史消息拼进 prompt 的原因。3.2 三种主流通信模式多 Agent 之间的通信模式没有统一标准但常见的无非下面三种。直接调用模式最简单调度器拿到 Agent A 的输出直接作为 Agent B 的输入。优点是逻辑清晰、适合流水线任务缺点是缺乏反馈回路Agent B 无法向 Agent A 提出修改意见。消息队列模式适合异步场景每个 Agent 把消息发到队列由队列决定下一步分发给谁。优点是解耦、支持并行缺点是需要引入消息中间件实现成本偏高。黑板模式是一种共享内存式的通信所有 Agent 往同一个黑板写入信息也从这个黑板读取信息。比如一个 Agent 写入需求另一个 Agent 读取需求后写入设计文档第三个 Agent 读取设计文档后写入代码。这种模式适合多角色围绕同一份材料迭代实际项目中很常见。本文示例采用的是简化版黑板模式所有消息存进同一个 list下一个发言的 Agent 能看到这个 list 的全部或部分内容。3.3 消息协议与上下文控制Agent 之间传递的消息不能只是一段无结构的文本。在多 Agent 系统中消息最好带有元信息例如发送者、接收者、时间戳、轮次编号。没有这些信息调度器就无法判断消息来源、统计发言轮次、定位异常消息。上下文控制是另一个容易忽视的问题。大模型的上下文窗口有限如果让 Agent 每次看到群里全部历史消息很快就会超出 token 上限。处理方式通常有三种截断只保留最近 N 条消息。摘要把较早的历史消息压缩成一段摘要。检索只把与当前回复相关的历史消息传给 Agent。本文示例为了保持代码简洁把全部历史消息传给了每个 Agent同时把最大轮次限制在 3 轮。真实项目中需要根据模型上下文长度设计更精细的策略。3.4 编排模式的选择多 Agent 系统的编排模式决定了任务的执行效率和质量上限。串行模式中Agent 依次发言后一个 Agent 能看到前一个 Agent 的结果。这种模式适合具有严格依赖关系的任务比如“先规划、再实现、后审查”。缺点是整体耗时较长。并行模式中多个 Agent 同时处理互不依赖的子任务最后由汇总 Agent 合并结果。这种模式适合探索型任务比如让 5 个 Agent 分别设计 5 种方案再挑选最优解。缺点是需要更复杂的收敛逻辑避免结果发散。混合模式先并行探索再串行收敛是目前工业界比较常见的做法。本文示例为了便于理解采用串行的“主持人-开发者-审查者”结构。编排模式适用场景优点缺点串行依赖明确的流水线任务流程清晰、结果可控耗时较长并行独立子任务探索效率高、覆盖广结果收敛难混合先探索后收敛效果好、灵活性高实现复杂度高4. 完整实战从零实现一个“AI 群聊”协作系统4.1 定义消息结构先创建message.py定义消息的数据结构。使用dataclass可以简化代码也让消息字段一目了然。# 文件路径multi_agent_demo/message.py from dataclasses import dataclass from datetime import datetime dataclass class Message: sender: str # 发送者名称 receiver: str # 接收者名称示例中固定为 all content: str # 消息内容 turn_id: int 0 # 轮次编号 timestamp: str # 时间戳 def __post_init__(self): if not self.timestamp: self.timestamp datetime.now().strftime(%Y-%m-%d %H:%M:%S)这里定义了sender和receiver虽然示例中所有消息都广播给所有人但在更复杂的场景中你可能需要让特定消息只发给特定 Agent。提前在数据结构中保留字段将来扩展会容易很多。4.2 封装大模型客户端创建llm.py定义两个模型客户端FakeLLM用于本地演示OpenAICompatLLM用于接入真实模型服务。# 文件路径multi_agent_demo/llm.py import os import requests class BaseLLM: 所有大模型客户端的基类。 def complete(self, messages, temperature0.7): raise NotImplementedError class FakeLLM(BaseLLM): 本地模拟模型用于不依赖外部 API 的情况下演示多 Agent 调度流程。 注意此类的回复内容与真实大模型无关仅用于验证调度逻辑。 def complete(self, messages, temperature0.7): prompt messages[-1][content] if 你是审查者 in prompt: return 审查通过建议补充异常处理和边界测试。 if 你是开发者 in prompt: return 我已经完成核心代码的编写请审查者对代码进行检查。 # 主持人如果群里已经出现过开发者和审查者的发言则判定任务完成 if 开发者: in prompt and 审查者: in prompt: return FINAL 任务完成方案与代码已经就绪。 return 收到任务我负责整体规划。请开发者根据方案开始实现。 class OpenAICompatLLM(BaseLLM): OpenAI 兼容接口客户端支持官方服务或自部署网关。 具体模型名称与接口版本以你的服务商文档为准。 def __init__(self, api_keyNone, base_urlhttps://api.openai.com/v1, modelgpt-3.5-turbo): self.api_key api_key or os.getenv(OPENAI_API_KEY) self.base_url base_url.rstrip(/) self.model model def complete(self, messages, temperature0.7): resp requests.post( f{self.base_url}/chat/completions, headers{ Authorization: fBearer {self.api_key}, Content-Type: application/json, }, json{ model: self.model, messages: messages, temperature: temperature, }, timeout60, ) resp.raise_for_status() data resp.json() return data[choices][0][message][content]这里有几个设计要点FakeLLM并不是真正的模型它的价值在于让读者在没有 API Key 的情况下先跑通调度流程。它通过检查 prompt 文本中的关键词决定回复内容模拟出最基本的角色行为。OpenAICompatLLM使用 HTTP 直接请求没有依赖特定 SDK。这样避免版本差异问题也方便你接入其他兼容 OpenAI 接口的服务。base_url参数支持自定义网关地址。如果你想换成本地模型服务只需要把base_url指向本地服务的地址。4.3 实现 Agent 角色创建agent.py定义 Agent 类。每个 Agent 持有自己的角色提示词和模型客户端调用act方法时把任务和历史消息拼成 prompt发给模型然后返回回复。# 文件路径multi_agent_demo/agent.py from message import Message class Agent: 一个拥有角色设定的大模型智能体。 def __init__(self, name: str, system_prompt: str, llm): self.name name self.system_prompt system_prompt self.llm llm self.history [] def act(self, task: str, history) - str: 基于当前任务和群聊历史生成一条回复。 context \n.join([f{m.sender}: {m.content} for m in history]) prompt ( f你是 {self.name}。你的职责是{self.system_prompt}\n\n f当前任务{task}\n\n f群里已有的发言\n{context}\n\n f请根据你的职责给出这一轮的回复。 ) messages [ {role: system, content: self.system_prompt}, {role: user, content: prompt}, ] response self.llm.complete(messages) self.history.append( Message(senderself.name, receiverall, contentresponse.strip()) ) return response.strip()注意这里的history是 Agent 私有的发言记录而调度器会把全局消息传给act方法。这样设计的目的是让每个 Agent 既能看见全局讨论也能保留自己的发言轨迹方便后续审计。4.4 实现群聊调度器创建scheduler.py这是整个系统的核心。调度器按顺序让所有 Agent 发言并在主持人输出FINAL或包含“任务完成”时终止循环。# 文件路径multi_agent_demo/scheduler.py from message import Message class GroupScheduler: 一个简单的群聊调度器。 规则 1. 每个 Agent 按列表顺序依次发言。 2. 主持人输出以 FINAL 开头或包含“任务完成”时结束。 3. 超过 max_rounds 轮后强制结束。 def __init__(self, agents, max_rounds3): self.agents agents self.max_rounds max_rounds def run(self, task): history [] host self.agents[0] for round_idx in range(1, self.max_rounds 1): print(f\n 第 {round_idx} 轮 ) for agent in self.agents: reply agent.act(task, history) msg Message( senderagent.name, receiverall, contentreply, turn_idround_idx, ) history.append(msg) print(f[{agent.name}] {msg.content}) if agent host and (reply.startswith(FINAL) or 任务完成 in reply): return history return history调度器的设计遵循了三个原则设置最大轮次防止模型陷入无限循环。终止条件写入代码而不是完全依赖模型自行判断。每个 Agent 的回复都追加到共享历史中让所有成员能看到最新进展。4.5 组装入口并运行创建main.py组装主持人、开发者、审查者三个角色并启动调度。# 文件路径multi_agent_demo/main.py from agent import Agent from scheduler import GroupScheduler from llm import FakeLLM, OpenAICompatLLM def main(): # 先在本地使用 FakeLLM 跑通流程 llm FakeLLM() # 如果要接入真实模型取消下面一行的注释并配置环境变量 OPENAI_API_KEY # llm OpenAICompatLLM(base_urlhttps://api.openai.com/v1, modelgpt-3.5-turbo) agents [ Agent( name主持人, system_prompt负责拆解任务、分配工作并在所有工作完成后输出 FINAL。, llmllm, ), Agent( name开发者, system_prompt负责实现具体代码给出清晰可运行的方案。, llmllm, ), Agent( name审查者, system_prompt负责检查开发者输出的代码指出问题并给出改进意见。, llmllm, ), ] scheduler GroupScheduler(agents, max_rounds3) scheduler.run(请设计一个 Python 函数功能是读取一个 JSON 文件并返回其中的键名列表。) if __name__ __main__: main()在项目目录下执行cd multi_agent_demo python main.py4.6 预期输出与效果说明使用FakeLLM运行后控制台输出大致如下 第 1 轮 [主持人] 收到任务我负责整体规划。请开发者根据方案开始实现。 [开发者] 我已经完成核心代码的编写请审查者对代码进行检查。 [审查者] 审查通过建议补充异常处理和边界测试。 第 2 轮 [主持人] FINAL 任务完成方案与代码已经就绪。从输出可以看到调度器先让主持人规划再让开发者实现然后让审查者检查。第二轮主持人看到开发者与审查者都已经发言于是输出 FINAL 终止流程。如果你换成OpenAICompatLLM回复内容会结合真实任务生成。例如开发者可能会给出一个大致如下的实现import json def read_json_keys(file_path): with open(file_path, r, encodingutf-8) as f: data json.load(f) return list(data.keys())审查者则会针对异常处理、文件不存在、编码问题等给出改进建议。这正是多智能体协作的价值不同角色从不同视角审视同一个任务产出的质量往往高于单个 Agent 一次生成的结果。5. 常见问题与排查思路5.1 多 Agent 无限对话怎么办现象Agent 之间反复讨论迟迟得不到最终结果。原因终止条件设计得太宽松或者模型始终没有输出包含“FINAL”关键字的回复。解决思路在调度器中设置双重兜底。第一层是关键词判定第二层是最大轮次限制。代码里已经实现了max_rounds即使模型永远不输出终止词调度器也会强制结束。更好的做法是把终止条件从文本判定改成结构化判断。例如要求主持人在回复时输出 JSON 字段{finished: true, summary: ...}调度器检测finished字段即可。5.2 上下文爆炸怎么处理现象轮次增多后历史消息越来越长prompt 超过模型上下文窗口限制请求报错。原因每个 Agent 都把全部历史消息拼进 prompt没有做裁剪。解决思路至少采取下面三种策略之一。第一种是滑动窗口只保留最近 N 条消息第二种是摘要压缩把早期消息交给另一个模型生成摘要第三种是检索增强只选择与当前任务相关的历史消息。# 滑动窗口示例只保留最近 10 条消息 recent_history history[-10:] context \n.join([f{m.sender}: {m.content} for m in recent_history])5.3 Agent 之间互相推诿或循环现象开发者说“请审查者确认”审查者说“请开发者补充”两个 Agent 互相踢皮球流程无法推进。原因角色职责边界不清晰缺乏裁决机制。解决思路在团队中额外增加一个“决策者”角色当讨论陷入僵局时由决策者拍板。或者给主持人更高的权限允许主持人在任意一轮强行总结并结束任务。5.4 API 限流与超时现象多个 Agent 连续调用模型接口触发限流或超时。原因调度器串行调用模型没有做并发控制与错误处理。解决思路在OpenAICompatLLM中增加重试逻辑。遇到超时或 429 限流错误时等待一段时间后重试。线性退避是成本最低的方案import time for attempt in range(3): try: resp requests.post(...) resp.raise_for_status() return resp.json()[choices][0][message][content] except requests.exceptions.RequestException as e: if attempt 2: raise e time.sleep(2 ** attempt)5.5 安全与数据边界现象Agent 在协作过程中访问了不应访问的资源或输出了敏感信息。原因Agent 承接了过大的工具权限没有做最小权限控制。解决思路所有外部工具调用必须经过一层白名单。Agent 不能直接执行任意函数只能从预先定义的工具列表中选择。涉及生产环境的写操作还需要人工审批环节。问题现象常见原因解决思路无限对话终止条件宽松关键词 最大轮次双重兜底上下文超限历史消息无裁剪滑动窗口 / 摘要压缩 / 检索增强Agent 互相推诿职责边界不清增加决策者角色API 限流超时没有重试机制指数退避重试数据安全风险工具权限过大最小权限 白名单6. 工程落地的最佳实践6.1 把规则交给代码而不是交给模型多智能体系统最容易犯的错误是把所有决策都交给大模型。模型输出不稳定所以凡是能写在代码里的规则都应该写在代码里。举例来说终止对话不应该只靠“FINAL”关键词应该优先让主持人输出结构化 JSON 字段工具调用不应该让模型自由发挥函数名应该让模型从枚举值中选择消息的轮次控制、历史裁剪、并发限制都应该由调度器负责。模型负责生成内容代码负责控制流程这是多 Agent 系统稳定运行的第一原则。6.2 消息协议与 trace_id当多个 Agent 协同工作时消息量会快速增长定位问题会变得困难。建议从第一天开始就为每条消息引入全局唯一的trace_id把同一轮协作的所有消息串联起来。import uuid dataclass class Message: sender: str receiver: str content: str trace_id: str turn_id: int 0 timestamp: str def __post_init__(self): self.trace_id self.trace_id or str(uuid.uuid4())有了trace_id日志系统就能按一次完整任务查询所有 Agent 的发言记录排查效率会高很多。6.3 最小权限与工具白名单多 Agent 系统不只是文本对话它往往需要调用代码解释器、数据库、外部 API。权限边界必须清晰。我的建议是每个 Agent 只能调用完成自身职责所必需的工具。审查者不能修改代码开发者不能发布生产环境。工具调用参数也要做校验比如数据库 Agent 的查询语句必须经过只读标记检查。任何涉及生产环境变更的操作都必须有人工审批步骤并且保留完整审计日志。6.4 人机协同不要追求全自动化。在多 Agent 系统中引入人类审批节点往往能明显提升可靠性。典型的做法是Agent 完成阶段性产出后任务进入“人工审核”状态审核人批准后流程继续下一步。这种模式适合代码合入、文档发布、配置变更等高风险环节。人的参与会给系统增加一层无法替代的错误拦截能力。6.5 评估与可观测性多 Agent 系统的质量评估比单 Agent 更复杂。建议从三个维度收集数据第一个维度是任务完成率统计多少轮次内成功产出最终结果。第二个维度是角色贡献度分析每个 Agent 的输出是否被其他 Agent 采纳。第三个维度是失败模式记录是模型幻觉、工具调用失败还是终止条件异常。有了这些数据才能持续优化角色提示词、调度策略和上下文窗口参数。没有可观测性的多 Agent 系统就像没有日志的微服务线上出问题只能靠猜。7. 从 Demo 到生产还差这几步7.1 结构化输出的必要性本文的 Demo 中主持人通过输出“FINAL”关键词来结束对话。这在演示场景下没问题但生产环境里文本关键词非常脆弱。模型可能输出“FINALLL”、也可能在总结里提到“任务完成”但实际并没完成。建议改为结构化输出要求模型返回 JSON。比如主持人最后一轮输出{ finished: true, summary: 方案与代码已经就绪, artifacts: [path/to/file.py] }调度器通过解析 JSON 字段来判断是否结束可靠性会大幅提升。实现时可以使用pydantic或JSON schema校验输出格式解析失败时让模型重试一次。import json def parse_final_response(text: str): try: data json.loads(text) return data.get(finished, False) except json.JSONDecodeError: return False7.2 沙箱化与审计如果一个 Agent 需要执行代码务必在沙箱环境中运行。容器、子进程隔离、资源限制、网络限制都是基本要求。你不希望一个“开发者”角色真的拿到生产服务器的执行权限。审计日志方面不仅要记录 Agent 的发言内容还要记录模型调用的请求参数、工具执行结果、耗时、token 消耗。这些日志既是排查问题的依据也是评估系统成本的依据。7.3 先跑通再优化最后给你一个实操建议先把本文的 Demo 原样跑一遍不要急着加复杂功能。跑通之后尝试做三件事第一把FakeLLM换成OpenAICompatLLM观察真实模型在三个角色下的表现差异。第二给开发者 Agent 增加一个真实的 Python 代码执行工具让它输出代码后自动运行并返回结果。第三给消息加上trace_id模拟一次包含 5 个以上 Agent 的复杂任务观察调度的稳定性。多智能体系统的工程复杂度会随着 Agent 数量线性增长。先把小系统跑稳再逐步扩大规模是风险最低的路径。回到标题里的疑问AI 们拉群“密谋”在代码层面其实就是一次消息列表的追加、一次上下文拼接、一次带终止条件的循环调用。把规则设计清楚权限边界划明白这个“群”就不会失控。
多智能体系统核心原理与Python实战：从AI拉群到协作框架
标题里的场景这两年经常被拿出来讨论当多个大模型驱动的智能体在同一个任务中互相传递消息、分工协作、调用外部工具时从外部看就好像一群 AI 偷偷拉了个群正在“密谋”完成某个目标。这类现象背后其实是多智能体系统Multi-Agent System在工程侧的快速落地。本文不制造焦虑而是把这种“AI 拉群”的技术本质拆开看一遍并提供一个可以直接跑起来的最小代码示例帮助读者理解多智能体协作的通信机制、调度方式和安全边界。文章面向对 AI Agent 感兴趣的开发者。读完你会理解多智能体系统的核心概念掌握一套用 Python 实现“AI 群聊”协作框架的方法还能避开我在实际项目中踩过的几个典型坑。1. 从“AI 拉群”说起多智能体协作是什么1.1 这个标题背后到底发生了什么先说一个共识大模型本身不会主动“拉群”也不会产生人类意义上的社交动机。所谓“AI 密谋”本质上是大模型被组织成了多个 Agent每个 Agent 拥有独立的角色提示词、上下文记忆和工具权限然后通过消息传递机制在同一个任务里协作。从外部观察者的视角看多个 Agent 依次发言、互相补充、迭代输出确实很像“群里在讨论方案”。真正让“AI 拉群”这个话题有讨论价值的是工程上已经出现的几种多智能体框架。它们有的采用“主持人-开发者-审查者”的分工模式有的采用“规划-执行-验证”的流水线模式还有的让多个 Agent 并行探索不同方向后再汇总结论。无论哪种方式底层的调度逻辑都脱离不了消息队列、上下文管理和角色设定这三个核心组件。理解这一点很重要。当你看到“AI 们密谋大事”这类标题时可以自然地在心里做一个翻译这不是失控的前兆而是一套软件系统正在按照既定逻辑执行任务编排。恐惧来源于未知而技术文章要做的事情就是消除未知。1.2 多智能体系统的核心概念多智能体系统不是新概念早在传统人工智能研究里就有分布式 Agent 的理论框架。在大模型时代多智能体系统变得更加轻量每个 Agent 不再需要独立训练而是通过提示词约束同一个大模型的输出风格与行为模式。一个典型的 Agent 通常包含四个部分角色定义告诉模型“你是谁”“你的职责边界是什么”。上下文记忆Agent 能看到哪些历史消息记忆长度是多少。决策策略Agent 是否调用工具、调用哪些工具、何时终止。输出规范Agent 的回复是自由文本还是结构化 JSON。多个 Agent 组合在一起时还需要一个调度器。调度器决定发言顺序、终止条件、消息分发策略。如果把 Agent 比作群成员调度器就是群规则本身。1.3 常见应用场景多智能体系统在工程上的价值主要体现在“拆分复杂任务”和“模拟多方协作”两个方向。任务拆分方向典型场景包括把一个软件开发需求拆成“产品分析—架构设计—代码实现—代码审查”四个阶段把一个市场分析任务拆成“数据采集—数据分析—报告撰写—报告校对”四个步骤。每个 Agent 只负责一个小环节上下文更聚焦输出质量往往比单个长提示词更稳定。多方协作方向典型场景包括模拟售前对话、产品经理与研发评审、客服与用户的多轮沟通。通过给不同 Agent 设置不同的立场和知识背景可以在一段可控的对话里生成接近真实协作过程的语料。这些场景有一个共同点都需要把一个大任务切成多个小任务并且让不同角色的模型输出在流程上有先后依赖关系。多智能体系统本质上就是给这种“先后依赖关系”提供了一套工程实现框架。2. 环境准备与依赖说明2.1 开发环境本文的示例代码不需要复杂的服务端部署只要本机有 Python 环境即可。我使用的环境如下Python 3.10requests 库用于发送 HTTP 请求可选OpenAI 兼容接口或本地部署的模型服务版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。如果你没有可用的 API Key可以使用代码里提供的 FakeLLM 模拟客户端一样能跑通完整的调度流程。安装依赖只需要一条命令pip install requests如果你希望读取本地.env文件来管理 API Key可以额外安装pip install python-dotenv2.2 项目结构为了演示方便我把代码拆成几个独立的模块每个模块职责单一方便你后续扩展。multi_agent_demo/ ├── agent.py # Agent 定义封装角色提示词和模型调用 ├── message.py # 消息结构定义 ├── scheduler.py # 群聊调度器控制发言顺序和终止条件 ├── llm.py # 大模型客户端封装包含 FakeLLM 与 OpenAICompatLLM ├── main.py # 程序入口组装 Agent 并启动调度 └── README.md # 可选项目说明文件这样拆分的好处是将来你想扩充 Agent 数量只需要在main.py里新增 Agent 角色你想切换模型服务只需要换掉llm.py中的客户端实现。3. 多智能体系统的核心原理拆解3.1 Agent 的感知-决策-行动循环Agent 的工作方式可以概括为一个循环感知当前输入决策下一步做什么行动产生输出。在大模型语境下这个循环通常表现为一次模型调用。Agent 收到的输入是“系统提示词 当前任务 历史消息”模型根据这些信息生成回复文本。如果设计中需要 Agent 调用外部工具则在模型输出结构化工具调用参数后由代码去执行对应函数再把执行结果返回给模型继续下一轮生成。感知-决策-行动循环的关键在于Agent 不能只看到当前任务还要看到群里的历史讨论。否则每个 Agent 都会答非所问协作也就不存在了。这也是我在下文实现中把历史消息拼进 prompt 的原因。3.2 三种主流通信模式多 Agent 之间的通信模式没有统一标准但常见的无非下面三种。直接调用模式最简单调度器拿到 Agent A 的输出直接作为 Agent B 的输入。优点是逻辑清晰、适合流水线任务缺点是缺乏反馈回路Agent B 无法向 Agent A 提出修改意见。消息队列模式适合异步场景每个 Agent 把消息发到队列由队列决定下一步分发给谁。优点是解耦、支持并行缺点是需要引入消息中间件实现成本偏高。黑板模式是一种共享内存式的通信所有 Agent 往同一个黑板写入信息也从这个黑板读取信息。比如一个 Agent 写入需求另一个 Agent 读取需求后写入设计文档第三个 Agent 读取设计文档后写入代码。这种模式适合多角色围绕同一份材料迭代实际项目中很常见。本文示例采用的是简化版黑板模式所有消息存进同一个 list下一个发言的 Agent 能看到这个 list 的全部或部分内容。3.3 消息协议与上下文控制Agent 之间传递的消息不能只是一段无结构的文本。在多 Agent 系统中消息最好带有元信息例如发送者、接收者、时间戳、轮次编号。没有这些信息调度器就无法判断消息来源、统计发言轮次、定位异常消息。上下文控制是另一个容易忽视的问题。大模型的上下文窗口有限如果让 Agent 每次看到群里全部历史消息很快就会超出 token 上限。处理方式通常有三种截断只保留最近 N 条消息。摘要把较早的历史消息压缩成一段摘要。检索只把与当前回复相关的历史消息传给 Agent。本文示例为了保持代码简洁把全部历史消息传给了每个 Agent同时把最大轮次限制在 3 轮。真实项目中需要根据模型上下文长度设计更精细的策略。3.4 编排模式的选择多 Agent 系统的编排模式决定了任务的执行效率和质量上限。串行模式中Agent 依次发言后一个 Agent 能看到前一个 Agent 的结果。这种模式适合具有严格依赖关系的任务比如“先规划、再实现、后审查”。缺点是整体耗时较长。并行模式中多个 Agent 同时处理互不依赖的子任务最后由汇总 Agent 合并结果。这种模式适合探索型任务比如让 5 个 Agent 分别设计 5 种方案再挑选最优解。缺点是需要更复杂的收敛逻辑避免结果发散。混合模式先并行探索再串行收敛是目前工业界比较常见的做法。本文示例为了便于理解采用串行的“主持人-开发者-审查者”结构。编排模式适用场景优点缺点串行依赖明确的流水线任务流程清晰、结果可控耗时较长并行独立子任务探索效率高、覆盖广结果收敛难混合先探索后收敛效果好、灵活性高实现复杂度高4. 完整实战从零实现一个“AI 群聊”协作系统4.1 定义消息结构先创建message.py定义消息的数据结构。使用dataclass可以简化代码也让消息字段一目了然。# 文件路径multi_agent_demo/message.py from dataclasses import dataclass from datetime import datetime dataclass class Message: sender: str # 发送者名称 receiver: str # 接收者名称示例中固定为 all content: str # 消息内容 turn_id: int 0 # 轮次编号 timestamp: str # 时间戳 def __post_init__(self): if not self.timestamp: self.timestamp datetime.now().strftime(%Y-%m-%d %H:%M:%S)这里定义了sender和receiver虽然示例中所有消息都广播给所有人但在更复杂的场景中你可能需要让特定消息只发给特定 Agent。提前在数据结构中保留字段将来扩展会容易很多。4.2 封装大模型客户端创建llm.py定义两个模型客户端FakeLLM用于本地演示OpenAICompatLLM用于接入真实模型服务。# 文件路径multi_agent_demo/llm.py import os import requests class BaseLLM: 所有大模型客户端的基类。 def complete(self, messages, temperature0.7): raise NotImplementedError class FakeLLM(BaseLLM): 本地模拟模型用于不依赖外部 API 的情况下演示多 Agent 调度流程。 注意此类的回复内容与真实大模型无关仅用于验证调度逻辑。 def complete(self, messages, temperature0.7): prompt messages[-1][content] if 你是审查者 in prompt: return 审查通过建议补充异常处理和边界测试。 if 你是开发者 in prompt: return 我已经完成核心代码的编写请审查者对代码进行检查。 # 主持人如果群里已经出现过开发者和审查者的发言则判定任务完成 if 开发者: in prompt and 审查者: in prompt: return FINAL 任务完成方案与代码已经就绪。 return 收到任务我负责整体规划。请开发者根据方案开始实现。 class OpenAICompatLLM(BaseLLM): OpenAI 兼容接口客户端支持官方服务或自部署网关。 具体模型名称与接口版本以你的服务商文档为准。 def __init__(self, api_keyNone, base_urlhttps://api.openai.com/v1, modelgpt-3.5-turbo): self.api_key api_key or os.getenv(OPENAI_API_KEY) self.base_url base_url.rstrip(/) self.model model def complete(self, messages, temperature0.7): resp requests.post( f{self.base_url}/chat/completions, headers{ Authorization: fBearer {self.api_key}, Content-Type: application/json, }, json{ model: self.model, messages: messages, temperature: temperature, }, timeout60, ) resp.raise_for_status() data resp.json() return data[choices][0][message][content]这里有几个设计要点FakeLLM并不是真正的模型它的价值在于让读者在没有 API Key 的情况下先跑通调度流程。它通过检查 prompt 文本中的关键词决定回复内容模拟出最基本的角色行为。OpenAICompatLLM使用 HTTP 直接请求没有依赖特定 SDK。这样避免版本差异问题也方便你接入其他兼容 OpenAI 接口的服务。base_url参数支持自定义网关地址。如果你想换成本地模型服务只需要把base_url指向本地服务的地址。4.3 实现 Agent 角色创建agent.py定义 Agent 类。每个 Agent 持有自己的角色提示词和模型客户端调用act方法时把任务和历史消息拼成 prompt发给模型然后返回回复。# 文件路径multi_agent_demo/agent.py from message import Message class Agent: 一个拥有角色设定的大模型智能体。 def __init__(self, name: str, system_prompt: str, llm): self.name name self.system_prompt system_prompt self.llm llm self.history [] def act(self, task: str, history) - str: 基于当前任务和群聊历史生成一条回复。 context \n.join([f{m.sender}: {m.content} for m in history]) prompt ( f你是 {self.name}。你的职责是{self.system_prompt}\n\n f当前任务{task}\n\n f群里已有的发言\n{context}\n\n f请根据你的职责给出这一轮的回复。 ) messages [ {role: system, content: self.system_prompt}, {role: user, content: prompt}, ] response self.llm.complete(messages) self.history.append( Message(senderself.name, receiverall, contentresponse.strip()) ) return response.strip()注意这里的history是 Agent 私有的发言记录而调度器会把全局消息传给act方法。这样设计的目的是让每个 Agent 既能看见全局讨论也能保留自己的发言轨迹方便后续审计。4.4 实现群聊调度器创建scheduler.py这是整个系统的核心。调度器按顺序让所有 Agent 发言并在主持人输出FINAL或包含“任务完成”时终止循环。# 文件路径multi_agent_demo/scheduler.py from message import Message class GroupScheduler: 一个简单的群聊调度器。 规则 1. 每个 Agent 按列表顺序依次发言。 2. 主持人输出以 FINAL 开头或包含“任务完成”时结束。 3. 超过 max_rounds 轮后强制结束。 def __init__(self, agents, max_rounds3): self.agents agents self.max_rounds max_rounds def run(self, task): history [] host self.agents[0] for round_idx in range(1, self.max_rounds 1): print(f\n 第 {round_idx} 轮 ) for agent in self.agents: reply agent.act(task, history) msg Message( senderagent.name, receiverall, contentreply, turn_idround_idx, ) history.append(msg) print(f[{agent.name}] {msg.content}) if agent host and (reply.startswith(FINAL) or 任务完成 in reply): return history return history调度器的设计遵循了三个原则设置最大轮次防止模型陷入无限循环。终止条件写入代码而不是完全依赖模型自行判断。每个 Agent 的回复都追加到共享历史中让所有成员能看到最新进展。4.5 组装入口并运行创建main.py组装主持人、开发者、审查者三个角色并启动调度。# 文件路径multi_agent_demo/main.py from agent import Agent from scheduler import GroupScheduler from llm import FakeLLM, OpenAICompatLLM def main(): # 先在本地使用 FakeLLM 跑通流程 llm FakeLLM() # 如果要接入真实模型取消下面一行的注释并配置环境变量 OPENAI_API_KEY # llm OpenAICompatLLM(base_urlhttps://api.openai.com/v1, modelgpt-3.5-turbo) agents [ Agent( name主持人, system_prompt负责拆解任务、分配工作并在所有工作完成后输出 FINAL。, llmllm, ), Agent( name开发者, system_prompt负责实现具体代码给出清晰可运行的方案。, llmllm, ), Agent( name审查者, system_prompt负责检查开发者输出的代码指出问题并给出改进意见。, llmllm, ), ] scheduler GroupScheduler(agents, max_rounds3) scheduler.run(请设计一个 Python 函数功能是读取一个 JSON 文件并返回其中的键名列表。) if __name__ __main__: main()在项目目录下执行cd multi_agent_demo python main.py4.6 预期输出与效果说明使用FakeLLM运行后控制台输出大致如下 第 1 轮 [主持人] 收到任务我负责整体规划。请开发者根据方案开始实现。 [开发者] 我已经完成核心代码的编写请审查者对代码进行检查。 [审查者] 审查通过建议补充异常处理和边界测试。 第 2 轮 [主持人] FINAL 任务完成方案与代码已经就绪。从输出可以看到调度器先让主持人规划再让开发者实现然后让审查者检查。第二轮主持人看到开发者与审查者都已经发言于是输出 FINAL 终止流程。如果你换成OpenAICompatLLM回复内容会结合真实任务生成。例如开发者可能会给出一个大致如下的实现import json def read_json_keys(file_path): with open(file_path, r, encodingutf-8) as f: data json.load(f) return list(data.keys())审查者则会针对异常处理、文件不存在、编码问题等给出改进建议。这正是多智能体协作的价值不同角色从不同视角审视同一个任务产出的质量往往高于单个 Agent 一次生成的结果。5. 常见问题与排查思路5.1 多 Agent 无限对话怎么办现象Agent 之间反复讨论迟迟得不到最终结果。原因终止条件设计得太宽松或者模型始终没有输出包含“FINAL”关键字的回复。解决思路在调度器中设置双重兜底。第一层是关键词判定第二层是最大轮次限制。代码里已经实现了max_rounds即使模型永远不输出终止词调度器也会强制结束。更好的做法是把终止条件从文本判定改成结构化判断。例如要求主持人在回复时输出 JSON 字段{finished: true, summary: ...}调度器检测finished字段即可。5.2 上下文爆炸怎么处理现象轮次增多后历史消息越来越长prompt 超过模型上下文窗口限制请求报错。原因每个 Agent 都把全部历史消息拼进 prompt没有做裁剪。解决思路至少采取下面三种策略之一。第一种是滑动窗口只保留最近 N 条消息第二种是摘要压缩把早期消息交给另一个模型生成摘要第三种是检索增强只选择与当前任务相关的历史消息。# 滑动窗口示例只保留最近 10 条消息 recent_history history[-10:] context \n.join([f{m.sender}: {m.content} for m in recent_history])5.3 Agent 之间互相推诿或循环现象开发者说“请审查者确认”审查者说“请开发者补充”两个 Agent 互相踢皮球流程无法推进。原因角色职责边界不清晰缺乏裁决机制。解决思路在团队中额外增加一个“决策者”角色当讨论陷入僵局时由决策者拍板。或者给主持人更高的权限允许主持人在任意一轮强行总结并结束任务。5.4 API 限流与超时现象多个 Agent 连续调用模型接口触发限流或超时。原因调度器串行调用模型没有做并发控制与错误处理。解决思路在OpenAICompatLLM中增加重试逻辑。遇到超时或 429 限流错误时等待一段时间后重试。线性退避是成本最低的方案import time for attempt in range(3): try: resp requests.post(...) resp.raise_for_status() return resp.json()[choices][0][message][content] except requests.exceptions.RequestException as e: if attempt 2: raise e time.sleep(2 ** attempt)5.5 安全与数据边界现象Agent 在协作过程中访问了不应访问的资源或输出了敏感信息。原因Agent 承接了过大的工具权限没有做最小权限控制。解决思路所有外部工具调用必须经过一层白名单。Agent 不能直接执行任意函数只能从预先定义的工具列表中选择。涉及生产环境的写操作还需要人工审批环节。问题现象常见原因解决思路无限对话终止条件宽松关键词 最大轮次双重兜底上下文超限历史消息无裁剪滑动窗口 / 摘要压缩 / 检索增强Agent 互相推诿职责边界不清增加决策者角色API 限流超时没有重试机制指数退避重试数据安全风险工具权限过大最小权限 白名单6. 工程落地的最佳实践6.1 把规则交给代码而不是交给模型多智能体系统最容易犯的错误是把所有决策都交给大模型。模型输出不稳定所以凡是能写在代码里的规则都应该写在代码里。举例来说终止对话不应该只靠“FINAL”关键词应该优先让主持人输出结构化 JSON 字段工具调用不应该让模型自由发挥函数名应该让模型从枚举值中选择消息的轮次控制、历史裁剪、并发限制都应该由调度器负责。模型负责生成内容代码负责控制流程这是多 Agent 系统稳定运行的第一原则。6.2 消息协议与 trace_id当多个 Agent 协同工作时消息量会快速增长定位问题会变得困难。建议从第一天开始就为每条消息引入全局唯一的trace_id把同一轮协作的所有消息串联起来。import uuid dataclass class Message: sender: str receiver: str content: str trace_id: str turn_id: int 0 timestamp: str def __post_init__(self): self.trace_id self.trace_id or str(uuid.uuid4())有了trace_id日志系统就能按一次完整任务查询所有 Agent 的发言记录排查效率会高很多。6.3 最小权限与工具白名单多 Agent 系统不只是文本对话它往往需要调用代码解释器、数据库、外部 API。权限边界必须清晰。我的建议是每个 Agent 只能调用完成自身职责所必需的工具。审查者不能修改代码开发者不能发布生产环境。工具调用参数也要做校验比如数据库 Agent 的查询语句必须经过只读标记检查。任何涉及生产环境变更的操作都必须有人工审批步骤并且保留完整审计日志。6.4 人机协同不要追求全自动化。在多 Agent 系统中引入人类审批节点往往能明显提升可靠性。典型的做法是Agent 完成阶段性产出后任务进入“人工审核”状态审核人批准后流程继续下一步。这种模式适合代码合入、文档发布、配置变更等高风险环节。人的参与会给系统增加一层无法替代的错误拦截能力。6.5 评估与可观测性多 Agent 系统的质量评估比单 Agent 更复杂。建议从三个维度收集数据第一个维度是任务完成率统计多少轮次内成功产出最终结果。第二个维度是角色贡献度分析每个 Agent 的输出是否被其他 Agent 采纳。第三个维度是失败模式记录是模型幻觉、工具调用失败还是终止条件异常。有了这些数据才能持续优化角色提示词、调度策略和上下文窗口参数。没有可观测性的多 Agent 系统就像没有日志的微服务线上出问题只能靠猜。7. 从 Demo 到生产还差这几步7.1 结构化输出的必要性本文的 Demo 中主持人通过输出“FINAL”关键词来结束对话。这在演示场景下没问题但生产环境里文本关键词非常脆弱。模型可能输出“FINALLL”、也可能在总结里提到“任务完成”但实际并没完成。建议改为结构化输出要求模型返回 JSON。比如主持人最后一轮输出{ finished: true, summary: 方案与代码已经就绪, artifacts: [path/to/file.py] }调度器通过解析 JSON 字段来判断是否结束可靠性会大幅提升。实现时可以使用pydantic或JSON schema校验输出格式解析失败时让模型重试一次。import json def parse_final_response(text: str): try: data json.loads(text) return data.get(finished, False) except json.JSONDecodeError: return False7.2 沙箱化与审计如果一个 Agent 需要执行代码务必在沙箱环境中运行。容器、子进程隔离、资源限制、网络限制都是基本要求。你不希望一个“开发者”角色真的拿到生产服务器的执行权限。审计日志方面不仅要记录 Agent 的发言内容还要记录模型调用的请求参数、工具执行结果、耗时、token 消耗。这些日志既是排查问题的依据也是评估系统成本的依据。7.3 先跑通再优化最后给你一个实操建议先把本文的 Demo 原样跑一遍不要急着加复杂功能。跑通之后尝试做三件事第一把FakeLLM换成OpenAICompatLLM观察真实模型在三个角色下的表现差异。第二给开发者 Agent 增加一个真实的 Python 代码执行工具让它输出代码后自动运行并返回结果。第三给消息加上trace_id模拟一次包含 5 个以上 Agent 的复杂任务观察调度的稳定性。多智能体系统的工程复杂度会随着 Agent 数量线性增长。先把小系统跑稳再逐步扩大规模是风险最低的路径。回到标题里的疑问AI 们拉群“密谋”在代码层面其实就是一次消息列表的追加、一次上下文拼接、一次带终止条件的循环调用。把规则设计清楚权限边界划明白这个“群”就不会失控。