在业务系统里接入大模型LLM已经不是新鲜事但很多团队做着做着会发现一个尴尬的问题单个 Agent 单打独斗效果还行一旦把多个 Agent 放进同一个任务流里冲突就层出不穷——A 说要这样B 说要那样C 直接推翻前两个结论最后整个系统混乱得像开了一场没有主持人的讨论会。前阵子我在做一个多角色协作的实验项目时反复遇到“多个 Agent 各说各话、无法收敛”的困境。翻了不少资料之后发现问题的根源不只是模型能力而是我们缺少一套社会化的协调机制。后来接触到Socially Grounded Agentic AI社会性根基的智能体 AI这个方向才逐渐理顺了思路。这篇文章我想把“社会理论如何指导多 Agent 系统设计”这件事结合可运行的代码示例整理成一篇完整教程。无论你是刚接触 Agentic AI 的初学者还是已经在做多 Agent 编排的开发者都可以从中拿到一套可以直接用的思路和代码框架。1. 什么是 Socially Grounded Agentic AI1.1 从 Agentic AI 说起先来聊一个基础概念Agentic AI。它的核心含义是“智能体 AI”——AI 不再只是被动回答问题的对话机器人而是具备目标拆解、规划、工具调用、自我反思能力的自主系统。一个大模型在拿到任务之后可以自己决定“先做什么、再做什么、调用什么工具、如何判断结果是否达标”这就是 Agentic AI 的基本形态。当前业界比较常见的应用场景包括自动化客服Agent 根据用户描述自行查询订单、判断退款条件并执行操作。数据分析助手Agent 读取数据库结构、编写 SQL、执行查询、分析结果并生成报表。代码开发助手Agent 拆解需求、生成代码、运行测试、修复报错。金融风控Agent 聚合多个数据源自动输出风险评估结论。这些场景的共同点是单个 Agent 足以应对“单线任务”。但到了复杂业务场景任务往往是多线程的需要多个角色协同才能完成。1.2 多个 Agent 在一起问题就来了当系统中同时存在多个 Agent 时会出现三类典型问题第一角色冲突。两个 Agent 都可能认为自己负责“最终决策”导致结果互相覆盖。第二信息不一致。每个 Agent 基于自己的上下文做推理对同一个事实的认知可能不同甚至得出相反结论。第三缺少收敛机制。多个 Agent 讨论到一半没人拍板最终无法产出结论。这些问题本质上不是模型能力的问题而是系统缺少一套“社会规范”。真实人类社会在漫长协作中形成了一套成熟的互动规则比如分工、投票、仲裁、信任机制、权力分层。把这些规则用工程手段引入多 Agent 系统就是Socially Grounded Agentic AI的核心思路。1.3 Socially Grounded 的技术含义用一句话概括Socially Grounded Agentic AI Agentic AI 社会理论 多智能体协作机制。“Grounded根基”在这里强调的是AI 的行为不是凭空生成的而是基于对社会互动规则的理解和遵循。在实际工程中这意味着你需要在系统设计阶段明确以下内容Agent 之间如何通信不同 Agent 的职责边界冲突发生后如何协调决策权如何分配是否存在人类参与审核的节点。这些规则对应到代码层面就是消息协议、角色定义、仲裁逻辑、投票算法和权限控制。接下来我用自己的一个实验项目来完整演示如何把这些理念落地成一个可运行的多 Agent 协作系统。2. 环境准备与系统设计2.1 技术选型本文的示例项目是一个“多角色投资分析系统”。系统中有三个 Agent分别模拟分析师、风控官和决策者它们共同完成一份投资分析报告。选型如下Python 3.9LangChain 或直接使用 OpenAI SDK示例中用 OpenAI 兼容接口简单的消息队列机制用 Pythonqueue模块模拟使用一个轻量级 Web 框架Flask接收外部请求说明示例代码重点演示“社会协调机制”的实现思路模型调用部分采用 OpenAI 兼容接口写法。如果你使用的是其他大模型服务只需要替换 API 基地址和模型名称即可。2.2 项目结构social-agent-system/ ├── agent.py # Agent 基类与角色实现 ├── coordinator.py # 协调者负责仲裁与决策 ├── message_bus.py # 消息总线负责 Agent 间通信 ├── social_protocols.py # 社会规则定义投票、仲裁、信任 ├── main.py # 主程序入口 └── requirements.txt # 依赖清单2.3 依赖安装pip install flask openai python-dotenv如果你使用的是国内可直连的大模型服务可以跳过openai环境变量的配置直接在代码中设置base_url。3. 核心概念拆解社会理论怎么映射到代码在写代码之前先把几个关键的社会学概念“翻译”成工程术语。3.1 角色与分工社会理论中的“角色分化”指的是不同成员承担不同职责。映射到系统中就是为每个 Agent 定义独立的 system prompt 和工具集。这里需要注意角色的边界必须清晰。如果两个 Agent 的 prompt 高度重叠就会出现职责重叠最终导致输出互相干扰。所以设计角色时要明确“我的目标”“我可用的工具”“我不可做的事”。3.2 消息协议社会中的沟通需要语言规范系统中的 Agent 通信也需要消息协议。一个标准的消息结构通常包含sender发送方角色receiver接收方角色message_type消息类型如 suggestion、vote、query、final_decisioncontent消息内容context上下文引用3.3 投票与共识多个 Agent 如何达成共识人类社会常用的方式是投票。工程上我们可以实现两种投票策略多数票制超过一半 Agent 同意即可通过。加权投票制不同角色拥有不同权重风险官的一票可能比普通角色更重。3.4 仲裁与最终决策当 Agent 之间无法达成共识时需要仲裁机制。通常有两种方式规则仲裁预先定义优先级规则例如“风控一票否决”。人类仲裁将冲突结果提交给人类审核员由人做出最终决定。理想的生产系统应当两者结合而本文的示例会同时实现这两种机制。4. 完整实战案例多 Agent 协作系统下面进入核心环节。我会把上面设计的概念逐一写成可运行的代码。4.1 定义 Agent 基类与角色首先创建一个基类BaseAgent它具备发送消息、接收消息的能力并且每个 Agent 会维护自己的“思维状态”。# 文件路径agent.py from abc import ABC, abstractmethod from typing import Dict, Any, List class BaseAgent(ABC): def __init__(self, name: str, role_description: str, message_bus): self.name name self.role_description role_description self.message_bus message_bus self.memory: List[Dict[str, Any]] [] def send_message(self, receiver: str, message_type: str, content: str, context: Dict[str, Any] None): print(f[{self.name}] 发送给 [{receiver}]{message_type}) self.message_bus.send( senderself.name, receiverreceiver, message_typemessage_type, contentcontent, contextcontext or {} ) def receive_message(self, message: Dict[str, Any]): self.memory.append(message) print(f[{self.name}] 收到来自 [{message[sender]}] 的消息{message[content][:40]}...) abstractmethod def act(self, task: str): 每个角色处理自己的具体任务 pass接下来实现三个具体角色AnalystAgent分析师负责分析数据、提出投资建议。RiskAgent风控官负责评估风险拥有“一票否决权”。DecisionAgent决策者协调全局综合各方建议输出最终结果。# 文件路径agent.py继续补充 class AnalystAgent(BaseAgent): def act(self, task: str): # 模拟分析师的分析过程 analysis_result self._call_llm(task, self.role_description) return analysis_result def _call_llm(self, task: str, role: str): # 实际项目中替换为自己的 LLM 调用逻辑 return { role: self.name, suggestion: f分析师建议基于当前数据与趋势建议关注 {task} 中的长期价值。, confidence: 0.82 } class RiskAgent(BaseAgent): def act(self, task: str): risk_result self._call_llm(task, self.role_description) return risk_result def _call_llm(self, task: str, role: str): return { role: self.name, suggestion: f风控官提示{task} 存在市场波动风险建议控制仓位。, confidence: 0.91, veto: False # 特殊字段表示是否一票否决 } class DecisionAgent(BaseAgent): def act(self, task: str): decision_result self._call_llm(task, self.role_description) return decision_result def _call_llm(self, task: str, role: str): return { role: self.name, suggestion: f决策者综合多方观点计划执行 {task} 的分步实施方案。, confidence: 0.75 }在代码中每个 Agent 的act()方法内部会真正调用大模型。为了先跑通流程这里先用模拟数据替代后面我们会把真实的 LLM 调用补上。4.2 实现消息总线消息总线负责在各个 Agent 之间传递消息。它模拟的其实是社会系统中的“沟通渠道”。# 文件路径message_bus.py import queue import threading from typing import Dict, Any, Optional class MessageBus: def __init__(self): self._queues: Dict[str, queue.Queue] {} self._lock threading.Lock() def register_agent(self, agent_name: str): with self._lock: if agent_name not in self._queues: self._queues[agent_name] queue.Queue() def send(self, sender: str, receiver: str, message_type: str, content: str, context: Dict[str, Any]): message { sender: sender, receiver: receiver, message_type: message_type, content: content, context: context } with self._lock: if receiver not in self._queues: raise ValueError(f接收方 {receiver} 未注册) self._queues[receiver].put(message) def receive(self, agent_name: str) - Optional[Dict[str, Any]]: with self._lock: q self._queues.get(agent_name) if q is None: return None try: return q.get_nowait() except queue.Empty: return None def receive_all(self, agent_name: str) - list: messages [] while True: msg self.receive(agent_name) if msg is None: break messages.append(msg) return messages4.3 实现社会协议模块这是整个系统的核心。它负责实现“投票”“仲裁”“信任评分”三类社会机制。# 文件路径social_protocols.py from typing import Dict, Any, List class SocialProtocols: def __init__(self): # 每个角色的投票权重 self.vote_weights { AnalystAgent: 1.0, RiskAgent: 1.5, DecisionAgent: 1.2 } # 信任评分 self.trust_scores { AnalystAgent: 100, RiskAgent: 100, DecisionAgent: 100 } def simple_vote(self, suggestions: List[Dict[str, Any]]) - str: 多数票制投票 approve_count 0 reject_count 0 for suggestion in suggestions: # 如果风险官持有否决意向直接进入仲裁 if suggestion.get(veto): return veto if suggestion.get(confidence, 0) 0.7: approve_count 1 else: reject_count 1 if approve_count reject_count: return approve return reject def weighted_vote(self, suggestions: List[Dict[str, Any]]) - str: 加权投票制 total_score 0.0 weight_sum 0.0 for suggestion in suggestions: role suggestion.get(role) weight self.vote_weights.get(role, 1.0) # 根据置信度打分 score suggestion.get(confidence, 0) * weight total_score score weight_sum weight # 一票否决 if suggestion.get(veto): return veto avg_score total_score / weight_sum if weight_sum 0 else 0 return approve if avg_score 0.65 else reject def update_trust(self, role: str, delta: int): 根据协作结果更新信任评分 if role in self.trust_scores: self.trust_scores[role] delta def arbitrate(self, suggestions: List[Dict[str, Any]], human_decision: str None) - str: 仲裁机制 1. 如果存在一票否决 - 直接否决 2. 如果传入人类决策 - 以人类决策为准 3. 否则使用加权投票 for suggestion in suggestions: if suggestion.get(veto): return reject if human_decision and human_decision in (approve, reject): return human_decision return self.weighted_vote(suggestions)这里需要解释一下“信任评分”的作用。在真实社会系统中长期表现差的成员会逐渐失去影响力。我们在多 Agent 系统中也可以引入类似机制当某个 Agent 连续多次给出低质量建议时降低它的投票权重从而避免系统被“噪音 Agent”带偏。4.4 实现协调者协调者把消息总线、Agent 和社会协议串起来它负责整个协作流程的编排。# 文件路径coordinator.py from typing import List, Dict, Any from agent import BaseAgent, AnalystAgent, RiskAgent, DecisionAgent from message_bus import MessageBus from social_protocols import SocialProtocols class Coordinator: def __init__(self): self.bus MessageBus() self.protocols SocialProtocols() self.agents: Dict[str, BaseAgent] {} def setup(self): agent_configs [ AnalystAgent(AnalystAgent, 你是一名资深的投资分析师擅长解读市场数据。, self.bus), RiskAgent(RiskAgent, 你是一名谨慎的风险控制官负责评估投资风险。, self.bus), DecisionAgent(DecisionAgent, 你是一名优秀的决策者善于综合多方意见。, self.bus) ] for agent in agent_configs: self.bus.register_agent(agent.name) self.agents[agent.name] agent def run(self, task: str): # Step 1: 各角色独立分析 suggestions [] for agent_name, agent in self.agents.items(): result agent.act(task) result[agent_name] agent_name suggestions.append(result) # Step 2: 让风险官的建议在消息总线上广播给决策者 self.bus.send( senderRiskAgent, receiverDecisionAgent, message_typerisk_alert, contentsuggestions[1][suggestion], context{confidence: suggestions[1][confidence]} ) messages self.bus.receive_all(DecisionAgent) print(f决策者收到 {len(messages)} 条消息) # Step 3: 应用社会协议进行仲裁 final_result self.protocols.arbitrate(suggestions) # Step 4: 输出结果 print(\n 决策结果 ) for suggestion in suggestions: print(f[{suggestion.get(role)}]: {suggestion[suggestion]}) print(f\n最终裁决{final_result}) if final_result approve: print(系统批准该投资计划可以进入下一步执行阶段。) else: print(系统否决该投资计划需要调整策略后重新提交。) return { final_result: final_result, suggestions: suggestions }4.5 编写主程序入口# 文件路径main.py from coordinator import Coordinator def main(): coordinator Coordinator() coordinator.setup() task 新能源板块的 3 个月中期投资计划 result coordinator.run(task) print(\n系统运行完毕。) if __name__ __main__: main()运行效果[AnalystAgent] 发送给 [DecisionAgent] 的消息 [RiskAgent] 发送给 [DecisionAgent] 的消息 决策者收到 1 条消息 决策结果 [AnalystAgent]: 分析师建议基于当前数据与趋势建议关注 新能源板块的 3 个月中期投资计划 中的长期价值。 [RiskAgent]: 风控官提示新能源板块的 3 个月中期投资计划 存在市场波动风险建议控制仓位。 [DecisionAgent]: 决策者综合多方观点计划执行 新能源板块的 3 个月中期投资计划 的分步实施方案。 最终裁决approve 系统批准该投资计划可以进入下一步执行阶段。4.6 接入真实大模型调用上面代码里的_call_llm方法是模拟实现。现在替换为真实的 OpenAI 兼容接口调用。# 文件路径agent.py替换方法示例 import os from openai import OpenAI client OpenAI( api_keyos.getenv(OPENAI_API_KEY, your-api-key), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) ) class AnalystAgent(BaseAgent): def _call_llm(self, task: str, role: str) - Dict[str, Any]: response client.chat.completions.create( modelgpt-4o-mini, temperature0.3, messages[ {role: system, content: role}, {role: user, content: f请对以下任务进行分析并给出建议、置信度{task}} ] ) content response.choices[0].message.content return { role: self.name, suggestion: content, confidence: 0.85 }实际使用中建议增加 JSON 输出解析逻辑让大模型返回结构化的 JSON这样下游模块可以直接读取 confidence、veto 等字段。5. 常见问题与排查思路我在实践过程中把最容易踩的坑整理成了一个排查表供大家参考问题现象常见原因解决思路多个 Agent 输出互相矛盾系统无法收敛缺少仲裁机制或仲裁权重设置不合理引入加权投票和角色优先级必须明确“谁有最终拍板权”某个 Agent 频繁给出低质量建议Prompt 设计不清晰或模型温度参数过高优化角色 Prompt加入明确的输出格式要求降低 temperature消息丢失消息总线的接收方名称错误或队列被并发消费检查 Agent 注册名日志中打印消息发送/接收状态投票结果不稳定置信度阈值设置不当分析历史结果根据业务要求调整阈值Agent 间上下文不一致没有共享上下文池或消息传递时遗漏 context 字段在消息协议中强制携带 context 字段建立全局知识库调大模型成本过高每个 Agent 都发起独立调用未做缓存对相同任务做缓存处理支持流式输出降低等待时间排查多 Agent 问题时有一个原则很重要先检查协议层再检查模型层。大多数协作问题并不是模型“不够聪明”而是协议没有定义清楚。比如一个 Agent 不知道自己的建议会被如何投票或者没有一个机制告诉它“你的结果需要达到什么标准才会被采纳”系统就必然出问题。6. 最佳实践与工程建议经过这个项目的落地验证我总结出几条非常有价值的工程建议这些不是理论推演而是真正踩过坑之后的反省。6.1 角色 Prompt 设计要追求边界清晰不要以为 Prompt 越长越好。角色的 Prompt 应该回答三个问题我的职责是什么我不可做什么我的输出格式是什么例如风控官角色的 Prompt 可以这样写你是风险控制官负责评估项目中的风险。 你需要给出风险等级高/中/低、风险描述、是否建议否决veto true/false。 你不负责给出具体投资建议那是分析师的工作。这种设计能从源头降低角色冲突概率。6.2 社会协议必须支持“人类介入”完全依赖 AI 做决策风险很大。生产系统一定要在协议层预留“人类审核”节点。代码中仲裁函数已经预留了human_decision参数这就是设计上的安全兜底。6.3 为每个 Agent 增加可观测性在多 Agent 系统中最常见的问题是“黑盒”。你很难知道某个结论是哪个 Agent 提出的以及为什么它这么提。给每条消息增加message_id、timestamp、trace_id字段便于链路追踪。生产环境建议接入日志平台方便定位问题。6.4 善用信任评分机制做动态调整固定权重的弊端在于一旦某个 Agent 因为迭代升级出现问题它会继续用同样的权重影响决策。动态信任评分可以让系统逐渐“不信任”失准的 Agent本质上是为系统增加一层自适应机制。6.5 安全与权限边界如果你的 Agent 系统会执行真实操作修改订单、发送邮件、操作数据库那么必须遵守最小权限原则。给每个 Agent 分配独立的 API Key 或权限角色并且严禁 Agent 执行未经验证的破坏性操作。6.6 降低调用成本多个 Agent 分别调用大模型会产生较高的 token 消耗。建议在以下维度进行优化对 Agent 的输入做必要裁剪不传递不相关的背景资料使用模型缓存方案处理重复问题优先使用性价比更高的模型完成中间步骤使用流式输出加快用户体验7. 总结与学习路线现在回头看一下我们在这一篇文章里做完了哪些事理解了 Agentic AI 与 Socially Grounded Agentic AI 的基本概念掌握了角色分工、消息协议、投票机制、仲裁机制等核心思想用 Python 实现了一个包含 Agent、消息总线、社会协议的完整多 Agent 协作系统最后还梳理了生产落地时最常见的坑与解决思路。接下来你可以往这些方向继续深入第一社会选择理论Social Choice Theory的资料它和 Agent 系统的投票机制强相关第二研究组织行为学中的信任模型看信任评分体系如何更进一步设计第三学习多 Agent 强化学习Multi-Agent Reinforcement Learning的内容它能帮你从动态博弈的角度理解 Agent 间的冲突与合作第四观察 LangGraph、CrewAI 等主流框架是如何实现角色编排和社会化协调的将本文的思想和成熟的工程框架结合。实践是理解这套体系最好的方式。建议你先基于本文代码运行一遍然后修改投票权重、增加一个第四角色或者把消息总线替换成 Redis 消息队列看看系统在并发场景下的表现。只有真正动手碰到冲突、看到不收敛的现象才会对“社会性协调”为什么重要有更深的体会。如果你在运行过程中遇到其他问题或者有更好的多 Agent 协作设计方案欢迎在评论区分享你的思路。学 Agentic AI 是一条需要边做边悟的路希望这篇文章能帮你少走几个弯路。 多Agent协作冲突如何破？Socially Grounded Agentic AI实战解析
在业务系统里接入大模型LLM已经不是新鲜事但很多团队做着做着会发现一个尴尬的问题单个 Agent 单打独斗效果还行一旦把多个 Agent 放进同一个任务流里冲突就层出不穷——A 说要这样B 说要那样C 直接推翻前两个结论最后整个系统混乱得像开了一场没有主持人的讨论会。前阵子我在做一个多角色协作的实验项目时反复遇到“多个 Agent 各说各话、无法收敛”的困境。翻了不少资料之后发现问题的根源不只是模型能力而是我们缺少一套社会化的协调机制。后来接触到Socially Grounded Agentic AI社会性根基的智能体 AI这个方向才逐渐理顺了思路。这篇文章我想把“社会理论如何指导多 Agent 系统设计”这件事结合可运行的代码示例整理成一篇完整教程。无论你是刚接触 Agentic AI 的初学者还是已经在做多 Agent 编排的开发者都可以从中拿到一套可以直接用的思路和代码框架。1. 什么是 Socially Grounded Agentic AI1.1 从 Agentic AI 说起先来聊一个基础概念Agentic AI。它的核心含义是“智能体 AI”——AI 不再只是被动回答问题的对话机器人而是具备目标拆解、规划、工具调用、自我反思能力的自主系统。一个大模型在拿到任务之后可以自己决定“先做什么、再做什么、调用什么工具、如何判断结果是否达标”这就是 Agentic AI 的基本形态。当前业界比较常见的应用场景包括自动化客服Agent 根据用户描述自行查询订单、判断退款条件并执行操作。数据分析助手Agent 读取数据库结构、编写 SQL、执行查询、分析结果并生成报表。代码开发助手Agent 拆解需求、生成代码、运行测试、修复报错。金融风控Agent 聚合多个数据源自动输出风险评估结论。这些场景的共同点是单个 Agent 足以应对“单线任务”。但到了复杂业务场景任务往往是多线程的需要多个角色协同才能完成。1.2 多个 Agent 在一起问题就来了当系统中同时存在多个 Agent 时会出现三类典型问题第一角色冲突。两个 Agent 都可能认为自己负责“最终决策”导致结果互相覆盖。第二信息不一致。每个 Agent 基于自己的上下文做推理对同一个事实的认知可能不同甚至得出相反结论。第三缺少收敛机制。多个 Agent 讨论到一半没人拍板最终无法产出结论。这些问题本质上不是模型能力的问题而是系统缺少一套“社会规范”。真实人类社会在漫长协作中形成了一套成熟的互动规则比如分工、投票、仲裁、信任机制、权力分层。把这些规则用工程手段引入多 Agent 系统就是Socially Grounded Agentic AI的核心思路。1.3 Socially Grounded 的技术含义用一句话概括Socially Grounded Agentic AI Agentic AI 社会理论 多智能体协作机制。“Grounded根基”在这里强调的是AI 的行为不是凭空生成的而是基于对社会互动规则的理解和遵循。在实际工程中这意味着你需要在系统设计阶段明确以下内容Agent 之间如何通信不同 Agent 的职责边界冲突发生后如何协调决策权如何分配是否存在人类参与审核的节点。这些规则对应到代码层面就是消息协议、角色定义、仲裁逻辑、投票算法和权限控制。接下来我用自己的一个实验项目来完整演示如何把这些理念落地成一个可运行的多 Agent 协作系统。2. 环境准备与系统设计2.1 技术选型本文的示例项目是一个“多角色投资分析系统”。系统中有三个 Agent分别模拟分析师、风控官和决策者它们共同完成一份投资分析报告。选型如下Python 3.9LangChain 或直接使用 OpenAI SDK示例中用 OpenAI 兼容接口简单的消息队列机制用 Pythonqueue模块模拟使用一个轻量级 Web 框架Flask接收外部请求说明示例代码重点演示“社会协调机制”的实现思路模型调用部分采用 OpenAI 兼容接口写法。如果你使用的是其他大模型服务只需要替换 API 基地址和模型名称即可。2.2 项目结构social-agent-system/ ├── agent.py # Agent 基类与角色实现 ├── coordinator.py # 协调者负责仲裁与决策 ├── message_bus.py # 消息总线负责 Agent 间通信 ├── social_protocols.py # 社会规则定义投票、仲裁、信任 ├── main.py # 主程序入口 └── requirements.txt # 依赖清单2.3 依赖安装pip install flask openai python-dotenv如果你使用的是国内可直连的大模型服务可以跳过openai环境变量的配置直接在代码中设置base_url。3. 核心概念拆解社会理论怎么映射到代码在写代码之前先把几个关键的社会学概念“翻译”成工程术语。3.1 角色与分工社会理论中的“角色分化”指的是不同成员承担不同职责。映射到系统中就是为每个 Agent 定义独立的 system prompt 和工具集。这里需要注意角色的边界必须清晰。如果两个 Agent 的 prompt 高度重叠就会出现职责重叠最终导致输出互相干扰。所以设计角色时要明确“我的目标”“我可用的工具”“我不可做的事”。3.2 消息协议社会中的沟通需要语言规范系统中的 Agent 通信也需要消息协议。一个标准的消息结构通常包含sender发送方角色receiver接收方角色message_type消息类型如 suggestion、vote、query、final_decisioncontent消息内容context上下文引用3.3 投票与共识多个 Agent 如何达成共识人类社会常用的方式是投票。工程上我们可以实现两种投票策略多数票制超过一半 Agent 同意即可通过。加权投票制不同角色拥有不同权重风险官的一票可能比普通角色更重。3.4 仲裁与最终决策当 Agent 之间无法达成共识时需要仲裁机制。通常有两种方式规则仲裁预先定义优先级规则例如“风控一票否决”。人类仲裁将冲突结果提交给人类审核员由人做出最终决定。理想的生产系统应当两者结合而本文的示例会同时实现这两种机制。4. 完整实战案例多 Agent 协作系统下面进入核心环节。我会把上面设计的概念逐一写成可运行的代码。4.1 定义 Agent 基类与角色首先创建一个基类BaseAgent它具备发送消息、接收消息的能力并且每个 Agent 会维护自己的“思维状态”。# 文件路径agent.py from abc import ABC, abstractmethod from typing import Dict, Any, List class BaseAgent(ABC): def __init__(self, name: str, role_description: str, message_bus): self.name name self.role_description role_description self.message_bus message_bus self.memory: List[Dict[str, Any]] [] def send_message(self, receiver: str, message_type: str, content: str, context: Dict[str, Any] None): print(f[{self.name}] 发送给 [{receiver}]{message_type}) self.message_bus.send( senderself.name, receiverreceiver, message_typemessage_type, contentcontent, contextcontext or {} ) def receive_message(self, message: Dict[str, Any]): self.memory.append(message) print(f[{self.name}] 收到来自 [{message[sender]}] 的消息{message[content][:40]}...) abstractmethod def act(self, task: str): 每个角色处理自己的具体任务 pass接下来实现三个具体角色AnalystAgent分析师负责分析数据、提出投资建议。RiskAgent风控官负责评估风险拥有“一票否决权”。DecisionAgent决策者协调全局综合各方建议输出最终结果。# 文件路径agent.py继续补充 class AnalystAgent(BaseAgent): def act(self, task: str): # 模拟分析师的分析过程 analysis_result self._call_llm(task, self.role_description) return analysis_result def _call_llm(self, task: str, role: str): # 实际项目中替换为自己的 LLM 调用逻辑 return { role: self.name, suggestion: f分析师建议基于当前数据与趋势建议关注 {task} 中的长期价值。, confidence: 0.82 } class RiskAgent(BaseAgent): def act(self, task: str): risk_result self._call_llm(task, self.role_description) return risk_result def _call_llm(self, task: str, role: str): return { role: self.name, suggestion: f风控官提示{task} 存在市场波动风险建议控制仓位。, confidence: 0.91, veto: False # 特殊字段表示是否一票否决 } class DecisionAgent(BaseAgent): def act(self, task: str): decision_result self._call_llm(task, self.role_description) return decision_result def _call_llm(self, task: str, role: str): return { role: self.name, suggestion: f决策者综合多方观点计划执行 {task} 的分步实施方案。, confidence: 0.75 }在代码中每个 Agent 的act()方法内部会真正调用大模型。为了先跑通流程这里先用模拟数据替代后面我们会把真实的 LLM 调用补上。4.2 实现消息总线消息总线负责在各个 Agent 之间传递消息。它模拟的其实是社会系统中的“沟通渠道”。# 文件路径message_bus.py import queue import threading from typing import Dict, Any, Optional class MessageBus: def __init__(self): self._queues: Dict[str, queue.Queue] {} self._lock threading.Lock() def register_agent(self, agent_name: str): with self._lock: if agent_name not in self._queues: self._queues[agent_name] queue.Queue() def send(self, sender: str, receiver: str, message_type: str, content: str, context: Dict[str, Any]): message { sender: sender, receiver: receiver, message_type: message_type, content: content, context: context } with self._lock: if receiver not in self._queues: raise ValueError(f接收方 {receiver} 未注册) self._queues[receiver].put(message) def receive(self, agent_name: str) - Optional[Dict[str, Any]]: with self._lock: q self._queues.get(agent_name) if q is None: return None try: return q.get_nowait() except queue.Empty: return None def receive_all(self, agent_name: str) - list: messages [] while True: msg self.receive(agent_name) if msg is None: break messages.append(msg) return messages4.3 实现社会协议模块这是整个系统的核心。它负责实现“投票”“仲裁”“信任评分”三类社会机制。# 文件路径social_protocols.py from typing import Dict, Any, List class SocialProtocols: def __init__(self): # 每个角色的投票权重 self.vote_weights { AnalystAgent: 1.0, RiskAgent: 1.5, DecisionAgent: 1.2 } # 信任评分 self.trust_scores { AnalystAgent: 100, RiskAgent: 100, DecisionAgent: 100 } def simple_vote(self, suggestions: List[Dict[str, Any]]) - str: 多数票制投票 approve_count 0 reject_count 0 for suggestion in suggestions: # 如果风险官持有否决意向直接进入仲裁 if suggestion.get(veto): return veto if suggestion.get(confidence, 0) 0.7: approve_count 1 else: reject_count 1 if approve_count reject_count: return approve return reject def weighted_vote(self, suggestions: List[Dict[str, Any]]) - str: 加权投票制 total_score 0.0 weight_sum 0.0 for suggestion in suggestions: role suggestion.get(role) weight self.vote_weights.get(role, 1.0) # 根据置信度打分 score suggestion.get(confidence, 0) * weight total_score score weight_sum weight # 一票否决 if suggestion.get(veto): return veto avg_score total_score / weight_sum if weight_sum 0 else 0 return approve if avg_score 0.65 else reject def update_trust(self, role: str, delta: int): 根据协作结果更新信任评分 if role in self.trust_scores: self.trust_scores[role] delta def arbitrate(self, suggestions: List[Dict[str, Any]], human_decision: str None) - str: 仲裁机制 1. 如果存在一票否决 - 直接否决 2. 如果传入人类决策 - 以人类决策为准 3. 否则使用加权投票 for suggestion in suggestions: if suggestion.get(veto): return reject if human_decision and human_decision in (approve, reject): return human_decision return self.weighted_vote(suggestions)这里需要解释一下“信任评分”的作用。在真实社会系统中长期表现差的成员会逐渐失去影响力。我们在多 Agent 系统中也可以引入类似机制当某个 Agent 连续多次给出低质量建议时降低它的投票权重从而避免系统被“噪音 Agent”带偏。4.4 实现协调者协调者把消息总线、Agent 和社会协议串起来它负责整个协作流程的编排。# 文件路径coordinator.py from typing import List, Dict, Any from agent import BaseAgent, AnalystAgent, RiskAgent, DecisionAgent from message_bus import MessageBus from social_protocols import SocialProtocols class Coordinator: def __init__(self): self.bus MessageBus() self.protocols SocialProtocols() self.agents: Dict[str, BaseAgent] {} def setup(self): agent_configs [ AnalystAgent(AnalystAgent, 你是一名资深的投资分析师擅长解读市场数据。, self.bus), RiskAgent(RiskAgent, 你是一名谨慎的风险控制官负责评估投资风险。, self.bus), DecisionAgent(DecisionAgent, 你是一名优秀的决策者善于综合多方意见。, self.bus) ] for agent in agent_configs: self.bus.register_agent(agent.name) self.agents[agent.name] agent def run(self, task: str): # Step 1: 各角色独立分析 suggestions [] for agent_name, agent in self.agents.items(): result agent.act(task) result[agent_name] agent_name suggestions.append(result) # Step 2: 让风险官的建议在消息总线上广播给决策者 self.bus.send( senderRiskAgent, receiverDecisionAgent, message_typerisk_alert, contentsuggestions[1][suggestion], context{confidence: suggestions[1][confidence]} ) messages self.bus.receive_all(DecisionAgent) print(f决策者收到 {len(messages)} 条消息) # Step 3: 应用社会协议进行仲裁 final_result self.protocols.arbitrate(suggestions) # Step 4: 输出结果 print(\n 决策结果 ) for suggestion in suggestions: print(f[{suggestion.get(role)}]: {suggestion[suggestion]}) print(f\n最终裁决{final_result}) if final_result approve: print(系统批准该投资计划可以进入下一步执行阶段。) else: print(系统否决该投资计划需要调整策略后重新提交。) return { final_result: final_result, suggestions: suggestions }4.5 编写主程序入口# 文件路径main.py from coordinator import Coordinator def main(): coordinator Coordinator() coordinator.setup() task 新能源板块的 3 个月中期投资计划 result coordinator.run(task) print(\n系统运行完毕。) if __name__ __main__: main()运行效果[AnalystAgent] 发送给 [DecisionAgent] 的消息 [RiskAgent] 发送给 [DecisionAgent] 的消息 决策者收到 1 条消息 决策结果 [AnalystAgent]: 分析师建议基于当前数据与趋势建议关注 新能源板块的 3 个月中期投资计划 中的长期价值。 [RiskAgent]: 风控官提示新能源板块的 3 个月中期投资计划 存在市场波动风险建议控制仓位。 [DecisionAgent]: 决策者综合多方观点计划执行 新能源板块的 3 个月中期投资计划 的分步实施方案。 最终裁决approve 系统批准该投资计划可以进入下一步执行阶段。4.6 接入真实大模型调用上面代码里的_call_llm方法是模拟实现。现在替换为真实的 OpenAI 兼容接口调用。# 文件路径agent.py替换方法示例 import os from openai import OpenAI client OpenAI( api_keyos.getenv(OPENAI_API_KEY, your-api-key), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) ) class AnalystAgent(BaseAgent): def _call_llm(self, task: str, role: str) - Dict[str, Any]: response client.chat.completions.create( modelgpt-4o-mini, temperature0.3, messages[ {role: system, content: role}, {role: user, content: f请对以下任务进行分析并给出建议、置信度{task}} ] ) content response.choices[0].message.content return { role: self.name, suggestion: content, confidence: 0.85 }实际使用中建议增加 JSON 输出解析逻辑让大模型返回结构化的 JSON这样下游模块可以直接读取 confidence、veto 等字段。5. 常见问题与排查思路我在实践过程中把最容易踩的坑整理成了一个排查表供大家参考问题现象常见原因解决思路多个 Agent 输出互相矛盾系统无法收敛缺少仲裁机制或仲裁权重设置不合理引入加权投票和角色优先级必须明确“谁有最终拍板权”某个 Agent 频繁给出低质量建议Prompt 设计不清晰或模型温度参数过高优化角色 Prompt加入明确的输出格式要求降低 temperature消息丢失消息总线的接收方名称错误或队列被并发消费检查 Agent 注册名日志中打印消息发送/接收状态投票结果不稳定置信度阈值设置不当分析历史结果根据业务要求调整阈值Agent 间上下文不一致没有共享上下文池或消息传递时遗漏 context 字段在消息协议中强制携带 context 字段建立全局知识库调大模型成本过高每个 Agent 都发起独立调用未做缓存对相同任务做缓存处理支持流式输出降低等待时间排查多 Agent 问题时有一个原则很重要先检查协议层再检查模型层。大多数协作问题并不是模型“不够聪明”而是协议没有定义清楚。比如一个 Agent 不知道自己的建议会被如何投票或者没有一个机制告诉它“你的结果需要达到什么标准才会被采纳”系统就必然出问题。6. 最佳实践与工程建议经过这个项目的落地验证我总结出几条非常有价值的工程建议这些不是理论推演而是真正踩过坑之后的反省。6.1 角色 Prompt 设计要追求边界清晰不要以为 Prompt 越长越好。角色的 Prompt 应该回答三个问题我的职责是什么我不可做什么我的输出格式是什么例如风控官角色的 Prompt 可以这样写你是风险控制官负责评估项目中的风险。 你需要给出风险等级高/中/低、风险描述、是否建议否决veto true/false。 你不负责给出具体投资建议那是分析师的工作。这种设计能从源头降低角色冲突概率。6.2 社会协议必须支持“人类介入”完全依赖 AI 做决策风险很大。生产系统一定要在协议层预留“人类审核”节点。代码中仲裁函数已经预留了human_decision参数这就是设计上的安全兜底。6.3 为每个 Agent 增加可观测性在多 Agent 系统中最常见的问题是“黑盒”。你很难知道某个结论是哪个 Agent 提出的以及为什么它这么提。给每条消息增加message_id、timestamp、trace_id字段便于链路追踪。生产环境建议接入日志平台方便定位问题。6.4 善用信任评分机制做动态调整固定权重的弊端在于一旦某个 Agent 因为迭代升级出现问题它会继续用同样的权重影响决策。动态信任评分可以让系统逐渐“不信任”失准的 Agent本质上是为系统增加一层自适应机制。6.5 安全与权限边界如果你的 Agent 系统会执行真实操作修改订单、发送邮件、操作数据库那么必须遵守最小权限原则。给每个 Agent 分配独立的 API Key 或权限角色并且严禁 Agent 执行未经验证的破坏性操作。6.6 降低调用成本多个 Agent 分别调用大模型会产生较高的 token 消耗。建议在以下维度进行优化对 Agent 的输入做必要裁剪不传递不相关的背景资料使用模型缓存方案处理重复问题优先使用性价比更高的模型完成中间步骤使用流式输出加快用户体验7. 总结与学习路线现在回头看一下我们在这一篇文章里做完了哪些事理解了 Agentic AI 与 Socially Grounded Agentic AI 的基本概念掌握了角色分工、消息协议、投票机制、仲裁机制等核心思想用 Python 实现了一个包含 Agent、消息总线、社会协议的完整多 Agent 协作系统最后还梳理了生产落地时最常见的坑与解决思路。接下来你可以往这些方向继续深入第一社会选择理论Social Choice Theory的资料它和 Agent 系统的投票机制强相关第二研究组织行为学中的信任模型看信任评分体系如何更进一步设计第三学习多 Agent 强化学习Multi-Agent Reinforcement Learning的内容它能帮你从动态博弈的角度理解 Agent 间的冲突与合作第四观察 LangGraph、CrewAI 等主流框架是如何实现角色编排和社会化协调的将本文的思想和成熟的工程框架结合。实践是理解这套体系最好的方式。建议你先基于本文代码运行一遍然后修改投票权重、增加一个第四角色或者把消息总线替换成 Redis 消息队列看看系统在并发场景下的表现。只有真正动手碰到冲突、看到不收敛的现象才会对“社会性协调”为什么重要有更深的体会。如果你在运行过程中遇到其他问题或者有更好的多 Agent 协作设计方案欢迎在评论区分享你的思路。学 Agentic AI 是一条需要边做边悟的路希望这篇文章能帮你少走几个弯路。