当前位置： 首页 > news >正文

AutoGen-AgentChat-13-多智能体相互辩论

news 2025/7/18 6:40:26 
import os
from dotenv import load_dotenv
load_dotenv()

True

多智能体辩论

多智能体辩论是一种多智能体设计模式，它模拟多轮交互，在每一轮中，智能体相互交换响应，并根据其他智能体的响应完善自己的响应。

此示例展示了GSM8K 基准中用于解决数学问题的多智能体辩论模式的实现。

此模式中有两种类型的代理：求解器代理和聚合器代理。求解器代理按照“ 使用稀疏通信拓扑改进多智能体辩论”中描述的技术以稀疏方式连接。求解器代理负责求解数学问题并相互交换答案。聚合器代理负责将数学问题分发给求解器代理，等待它们的最终答案，并聚合答案以获得最终答案。

该模式的工作原理如下：

  • 用户向聚合代理发送一个数学问题。
  • 聚合代理将问题分发给求解代理。
  • 每个求解代理处理该问题，并向其邻居发布响应。
  • 每个求解器代理使用来自其邻居的响应来改进其响应，并发布新的响应。
  • 重复步骤 4，进行固定轮次。在最后一轮中，每个求解器代理都会发布最终响应。
  • 聚合代理采用多数投票的方式汇总所有求解代理的最终响应，得到最终答案，并发布该答案。

我们将使用广播 API，即 ，publish_message()并使用主题和订阅来实现通信拓扑。

import re
from dataclasses import dataclass
from typing import Dict, List# ----------------------------
# 导入 AutoGen 核心模块
# ----------------------------
from autogen_core import (DefaultTopicId,                  # 默认的对话主题标识符（用于消息归属）MessageContext,                  # 消息上下文对象（包含消息历史等信息）RoutedAgent,                     # 可进行消息路由的智能体（Agent），用于构建多 Agent 系统SingleThreadedAgentRuntime,     # 单线程 Agent 运行时环境（不并发，适合基础开发）TypeSubscription,               # 用于指定消息订阅类型的结构default_subscription,           # 默认的订阅设置（所有类型均接受）message_handler,                # 消息处理装饰器，用于注册处理函数
)# ----------------------------
# 导入 AutoGen 的消息模型定义
# ----------------------------
from autogen_core.models import (AssistantMessage,               # 表示 AI 助手的回复消息（role="assistant"）ChatCompletionClient,          # 聊天模型客户端的抽象基类（用于自定义模型接入）LLMMessage,                     # 通用大语言模型消息（可表示 assistant/user/system）SystemMessage,                 # 表示系统级提示词消息（role="system"）UserMessage,                   # 表示用户发送的消息（role="user"）
)# ----------------------------
# 导入 AutoGen 扩展模型：OpenAI Chat 接入封装
# ----------------------------
from autogen_ext.models.openai import OpenAIChatCompletionClient
# 这是 OpenAI 接口的封装类，支持 GPT-3.5 / GPT-4 / GPT-4o 等模型，兼容 AutoGen 架构

消息协议

首先，我们定义代理使用的消息。 IntermediateSolverResponse是每轮求解器代理之间交换的消息，FinalSolverResponse是最后一轮求解器代理发布的消息。

from dataclasses import dataclass# 表示一个用户提问
@dataclass
class Question:content: str  # 问题的具体内容# 表示一个回答内容
@dataclass
class Answer:content: str  # 回答的具体内容# 表示发给求解器（Solver）的请求数据
@dataclass
class SolverRequest:content: str   # 请求的完整文本内容（可能包含上下文或指令）question: str  # 需要解答的问题文本（原始问题）# 表示求解过程中的中间结果响应
@dataclass
class IntermediateSolverResponse:content: str   # 本轮求解器输出的详细内容（可能包含解释、分析）question: str  # 当前处理的问题answer: str    # 当前推理得出的中间答案round: int     # 当前是第几轮推理（用于多轮链式思考）# 表示求解器的最终结果
@dataclass
class FinalSolverResponse:answer: str    # 最终得出的答案（只保留核心结论）

求解器代理

求解器代理负责求解数学问题并与其他求解器代理交换答案。收到 后SolverRequest，求解器代理会使用 LLM 生成答案。然后，它会根据轮数发布IntermediateSolverResponse 或。FinalSolverResponse

求解器代理被赋予一个主题类型，用于指示代理应向哪个主题发布中间响应。该主题由其邻居订阅，以接收来自该代理的响应——我们稍后将展示如何实现这一点。

我们用来default_subscription()让求解器代理订阅默认主题，聚合器代理使用该默认主题来收集来自求解器代理的最终响应。

# 数学求解智能体：支持多轮交互与邻居协作
@default_subscription
class MathSolver(RoutedAgent):def __init__(self, model_client: ChatCompletionClient, topic_type: str, num_neighbors: int, max_round: int) -> None:# 调用父类构造函数，传入默认名称super().__init__("一名辩手。")  self._topic_type = topic_type         # 消息所属主题类型（用于发布消息时的分类）self._model_client = model_client     # 用于调用大语言模型的客户端self._num_neighbors = num_neighbors   # 同一主题下协作的邻居代理数量self._history: List[LLMMessage] = []  # 存储对话历史记录（系统+用户+助手）self._buffer: Dict[int, List[IntermediateSolverResponse]] = {}  # 存储每一轮邻居的中间回复# 系统提示词（用于引导模型行为）self._system_messages = [SystemMessage(content=("你是一位擅长数学与逻辑推理的助手，请帮我解答一个数学推理题。""请尽可能清晰简洁地说明思路，总字数限制在 100 字以内。""最终答案请以 {{数字}} 的形式呈现，例如：'答案是 {{42}}。'"))]self._round = 0               # 当前推理轮次self._max_round = max_round  # 最大允许的推理轮次# 接收求解请求：处理 SolverRequest 类型的消息@message_handlerasync def handle_request(self, message: SolverRequest, ctx: MessageContext) -> None:# 添加用户问题到历史记录self._history.append(UserMessage(content=message.content, source="user"))# 调用大模型进行推理model_result = await self._model_client.create(self._system_messages + self._history)assert isinstance(model_result.content, str)# 将模型返回的回答加入历史self._history.append(AssistantMessage(content=model_result.content, source=self.metadata["type"]))# 打印当前轮次的解答内容print(f"{'-'*80}\n求解器 {self.id} 第 {self._round} 轮回答：\n{model_result.content}")# 从模型回答中提取形如 {{42}} 的答案match = re.search(r"\{\{(\-?\d+(\.\d+)?)\}\}", model_result.content)if match is None:raise ValueError("模型回答中未检测到有效答案格式 {{number}}")answer = match.group(1)  # 提取数字部分# 推理轮次 +1self._round += 1# 如果已达到最大轮次，发布最终答案if self._round == self._max_round:await self.publish_message(FinalSolverResponse(answer=answer),topic_id=DefaultTopicId())else:# 否则发布中间结果，供其他代理参考await self.publish_message(IntermediateSolverResponse(content=model_result.content,question=message.question,answer=answer,round=self._round,),topic_id=DefaultTopicId(type=self._topic_type),  # 指定主题类型)# 接收邻居的中间结果：处理 IntermediateSolverResponse 类型的消息@message_handlerasync def handle_response(self, message: IntermediateSolverResponse, ctx: MessageContext) -> None:# 按轮次将邻居的回复加入缓存 bufferself._buffer.setdefault(message.round, []).append(message)# 如果收齐所有邻居的回复if len(self._buffer[message.round]) == self._num_neighbors:print(f"{'-'*80}\n求解器 {self.id} 第 {message.round} 轮："f"已收到所有 {self._num_neighbors} 位邻居的回应。")# 构造新的 prompt，整合邻居的回答prompt = "以下是其他代理对这个数学问题的解答结果：\n"for resp in self._buffer[message.round]:prompt += f"某代理的解答：{resp.content}\n"prompt += ("请结合这些解答作为参考，重新给出你对该数学问题的解法与答案。\n"f"原始问题是：{message.question}。\n""请在回答结尾用 {{数字}} 的形式给出你最终的答案。")# 将新的 prompt 发送给自己，作为下一轮输入await self.send_message(SolverRequest(content=prompt, question=message.question),self.id)# 清空当前轮次的缓存self._buffer.pop(message.round)

聚合代理

聚合代理负责处理用户问题并将数学问题分发给求解代理。

聚合器使用 订阅默认主题 default_subscription()。默认主题用于接收用户问题，接收来自求解器代理的最终响应，并将最终答案发布回用户。

在更复杂的应用程序中，当您想要将多代理辩论隔离到子组件中时，您应该使用 type_subscription()为聚合器-求解器通信设置特定的主题类型，并让求解器和聚合器发布和订阅该主题类型。
用户提问
↓
[MathAggregator]
↓
┌────┬─────┬─────┐
↓ ↓ ↓ ↓
Solver1 Solver2 … SolverN
↓ ↓ ↓ ↓
→→→ 聚合最终回答 →→→ 输出最终答案

# MathAggregator：数学聚合器智能体
# 用于收集多个求解器（Solver）的最终结果，输出最终答案
@default_subscription
class MathAggregator(RoutedAgent):def __init__(self, num_solvers: int) -> None:super().__init__("数学聚合器")  # 设定智能体名称self._num_solvers = num_solvers       # 需要等待的求解器数量self._buffer: List[FinalSolverResponse] = []  # 用于缓存所有求解器的最终回答# 处理用户提问：广播给所有 Solver 智能体@message_handlerasync def handle_question(self, message: Question, ctx: MessageContext) -> None:# 打印收到问题的提示print(f"{'-'*80}\n聚合器 {self.id} 收到问题：\n{message.content}")# 构造提示词（提示每个求解器回答数学问题）prompt = (f"请你解答下面这道数学题：\n{message.content}\n""请解释你的推理过程，最终答案请以 {{数字}} 的形式呈现，""例如：'答案是 {{42}}。'")# 打印发布任务提示print(f"{'-'*80}\n聚合器 {self.id} 正在发布初始求解请求。")# 发布求解请求消息（发送给所有 Solver）await self.publish_message(SolverRequest(content=prompt, question=message.content),topic_id=DefaultTopicId())# 处理求解器最终返回的答案@message_handlerasync def handle_final_solver_response(self, message: FinalSolverResponse, ctx: MessageContext) -> None:# 将求解器的答案加入缓存列表self._buffer.append(message)# 如果收到了所有求解器的最终回答if len(self._buffer) == self._num_solvers:print(f"{'-'*80}\n聚合器 {self.id} 已收到所有 {self._num_solvers} 个求解器的最终答案。")# 统计每个答案出现的次数，选择最多的作为最终结果（多数投票机制）answers = [resp.answer for resp in self._buffer]majority_answer = max(set(answers), key=answers.count)# 发布聚合后的答案await self.publish_message(Answer(content=majority_answer),topic_id=DefaultTopicId())# 清空缓存，为下一轮任务做准备self._buffer.clear()# 打印最终答案print(f"{'-'*80}\n聚合器 {self.id} 发布最终答案：\n{majority_answer}")

发起辩论

现在，我们将建立一个由 4 个求解器代理和 1 个聚合器代理组成的多智能体辩论系统。求解器代理将以稀疏方式连接，如下图所示：
A — B
| |
| |
D — C
每个 Solver 代理都与另外两个 Solver 代理相连。例如，代理 A 连接到代理 B 和代理 C。

让我们首先创建一个运行时并注册代理类型。
┌────────────────────┐
│ MathAggregator │◄── 聚合最终答案
└────────────────────┘
▲
┌──────────────┬───────┼──────────────┬──────────────┐
▼ ▼ ▼ ▼ ▼
┌────────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐
│MathSolverA │ │MathSolverB │ │MathSolverC │ │MathSolverD │
└────────────┘ └────────────┘ └────────────┘ └────────────┘
▲ ▲ ▲ ▲
↖↙邻居监听 ↖↙邻居监听 ↖↙邻居监听 ↖↙邻居监听（每个监听2个邻居）

# 创建一个单线程运行时环境（适合测试与小规模智能体系统）
runtime = SingleThreadedAgentRuntime()# 创建大语言模型客户端（使用 gpt-4o-mini，可替换为 deepseek 等）
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")# 注册第一个数学求解器：MathSolverA
# topic_type 设置为 "MathSolverA"
# num_neighbors 表示该 Solver 会监听其他 2 个 Solver 的中间结果
# max_round 表示最多允许推理 3 轮
await MathSolver.register(runtime,                     # 运行环境"MathSolverA",              # 智能体名称（唯一）lambda: MathSolver(         # 实例化函数（延迟构造）model_client=model_client,topic_type="MathSolverA",num_neighbors=2,max_round=3,),
)# 注册第二个数学求解器：MathSolverB
await MathSolver.register(runtime,"MathSolverB",lambda: MathSolver(model_client=model_client,topic_type="MathSolverB",num_neighbors=2,max_round=3,),
)# 注册第三个数学求解器：MathSolverC
await MathSolver.register(runtime,"MathSolverC",lambda: MathSolver(model_client=model_client,topic_type="MathSolverC",num_neighbors=2,max_round=3,),
)# 注册第四个数学求解器：MathSolverD
await MathSolver.register(runtime,"MathSolverD",lambda: MathSolver(model_client=model_client,topic_type="MathSolverD",num_neighbors=2,max_round=3,),
)# 注册聚合器智能体：MathAggregator
# num_solvers=4 表示将接收 4 个求解器的最终答案并进行聚合
await MathAggregator.register(runtime,"MathAggregator",                    # 聚合器名称lambda: MathAggregator(num_solvers=4)  # 聚合器实例化函数
)

AgentType(type='MathAggregator')

现在我们将使用创建求解器代理拓扑TypeSubscription，它将每个求解器代理的发布主题类型映射到其邻居的代理类型。
(D) ───► A ◄─── (B)
▲ ▲
│ │
© ◄─── B ◄─── A
▲ ▲
│ │
(D) ───► C ◄─── B

# ---------------------------------------------------------------------
# 设置 MathSolverA 的订阅：
# 它将接收来自 MathSolverD 和 MathSolverB 发布的中间推理结果（IntermediateSolverResponse）
# ---------------------------------------------------------------------
await runtime.add_subscription(TypeSubscription("MathSolverA", "MathSolverD"))  # A 监听 D 的消息
await runtime.add_subscription(TypeSubscription("MathSolverA", "MathSolverB"))  # A 监听 B 的消息# ---------------------------------------------------------------------
# 设置 MathSolverB 的订阅：
# 它将接收来自 MathSolverA 和 MathSolverC 的中间推理结果
# ---------------------------------------------------------------------
await runtime.add_subscription(TypeSubscription("MathSolverB", "MathSolverA"))  # B 监听 A
await runtime.add_subscription(TypeSubscription("MathSolverB", "MathSolverC"))  # B 监听 C# ---------------------------------------------------------------------
# 设置 MathSolverC 的订阅：
# 它将接收来自 MathSolverB 和 MathSolverD 的中间推理结果
# ---------------------------------------------------------------------
await runtime.add_subscription(TypeSubscription("MathSolverC", "MathSolverB"))  # C 监听 B
await runtime.add_subscription(TypeSubscription("MathSolverC", "MathSolverD"))  # C 监听 D# ---------------------------------------------------------------------
# 设置 MathSolverD 的订阅：
# 它将接收来自 MathSolverC 和 MathSolverA 的中间推理结果
# ---------------------------------------------------------------------
await runtime.add_subscription(TypeSubscription("MathSolverD", "MathSolverC"))  # D 监听 C
await runtime.add_subscription(TypeSubscription("MathSolverD", "MathSolverA"))  # D 监听 A# ---------------------------------------------------------------------
# 所有求解器（MathSolver）和聚合器（MathAggregator）
# 默认都自动订阅 DefaultTopicId（默认主题）：
# - MathAggregator 会监听 FinalSolverResponse（最终答案）
# - MathSolver 会监听 SolverRequest（初始问题或邻居合成提示）
# ---------------------------------------------------------------------
# 无需显式添加，对 default_subscription 装饰器类会自动订阅默认主题

解决数学问题

现在，我们来运行一场辩论，解决一道数学题。我们将一个帖子发布SolverRequest到默认主题，聚合器代理就会开始辩论。

# 定义数学问题（英文内容不变，以便大模型识别）
question = "李美丽在四月份向她的48位朋友卖出了发夹，然后在五月份又卖出了一半数量的发夹。李美丽在四月和五月一共卖出了多少个发夹？"# 启动运行时环境，开始多智能体协作流程
runtime.start()# 向默认主题（DefaultTopicId）发布问题消息
# - MathAggregator 将收到此问题并分发给多个 MathSolver
await runtime.publish_message(Question(content=question),DefaultTopicId()
)# 等待运行时进入空闲状态（所有任务执行完成）
# - 所有 solver 返回结果后，aggregator 聚合输出，系统才会空闲
await runtime.stop_when_idle()# 关闭与大模型（如 OpenAI GPT）客户端的连接，释放资源
await model_client.close()

--------------------------------------------------------------------------------
聚合器 MathAggregator/default 收到问题：
李美丽在四月份向她的48位朋友卖出了发夹，然后在五月份又卖出了一半数量的发夹。李美丽在四月和五月一共卖出了多少个发夹？
--------------------------------------------------------------------------------
聚合器 MathAggregator/default 正在发布初始求解请求。
--------------------------------------------------------------------------------
求解器 MathSolverC/default 第 0 轮回答：
李美丽在四月份卖出了48个发夹。五月份她又卖出了一半数量，即48 / 2 = 24个发夹。因此，四月份和五月份一共卖出的发夹数量为48 + 24 = 72个。 答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverB/default 第 0 轮回答：
李美丽在四月份卖出48个发夹。五月份卖出的是四月份的一半，即48 / 2 = 24个发夹。总共卖出的发夹数为48 + 24 = 72。 答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverA/default 第 0 轮回答：
李美丽在四月份卖出了48个发夹。五月份卖出的数量是四月份的一半，即48 ÷ 2 = 24个。因此，她在四月和五月总共卖出了48 + 24 = 72个发夹。 答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverD/default 第 0 轮回答：
李美丽在四月份卖出了48个发夹。五月份她又卖出一半，计算为48 ÷ 2 = 24个发夹。四月和五月总共卖出发夹的数量为48 + 24 = 72个发夹。 因此，答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverD/default 第 1 轮：已收到所有 2 位邻居的回应。
--------------------------------------------------------------------------------
求解器 MathSolverB/default 第 1 轮：已收到所有 2 位邻居的回应。
--------------------------------------------------------------------------------
求解器 MathSolverC/default 第 1 轮：已收到所有 2 位邻居的回应。
--------------------------------------------------------------------------------
求解器 MathSolverA/default 第 1 轮：已收到所有 2 位邻居的回应。
--------------------------------------------------------------------------------
求解器 MathSolverB/default 第 1 轮回答：
李美丽在四月份卖出48个发夹。五月份卖出的是四月份数量的一半，即48 ÷ 2 = 24个发夹。因此，四月和五月的总销售量为48 + 24 = 72个发夹。答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverD/default 第 1 轮回答：
李美丽在四月份卖出了48个发夹。五月份她卖出了四月份数量的一半，即48 ÷ 2 = 24个发夹。将两个月的销售数量相加，48 + 24 = 72个发夹。因此，李美丽在四月和五月共卖出了72个发夹。答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverC/default 第 2 轮：已收到所有 2 位邻居的回应。
--------------------------------------------------------------------------------
求解器 MathSolverA/default 第 2 轮：已收到所有 2 位邻居的回应。
--------------------------------------------------------------------------------
求解器 MathSolverC/default 第 1 轮回答：
李美丽在四月份卖出48个发夹。五月份她卖出的是四月份数量的一半，因此为48 / 2 = 24个发夹。将两个月的销售数量相加，得出总数为48 + 24 = 72个发夹。 答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverA/default 第 1 轮回答：
李美丽在四月份卖出了48个发夹。五月份她售出的数量是四月份的一半，即48 ÷ 2 = 24个发夹。因此，她在四月和五月共卖出的发夹总数为48 + 24 = 72个。答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverD/default 第 2 轮：已收到所有 2 位邻居的回应。
--------------------------------------------------------------------------------
求解器 MathSolverB/default 第 2 轮：已收到所有 2 位邻居的回应。
--------------------------------------------------------------------------------
求解器 MathSolverB/default 第 2 轮回答：
李美丽在四月份卖出了48个发夹。五月份她卖出的是四月份数量的一半，即48 / 2 = 24个发夹。将两个月的销售数量相加，得出总数为48 + 24 = 72个发夹。答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverD/default 第 2 轮回答：
李美丽在四月份卖出了48个发夹。五月份则卖出了四月份数量的一半，即48 ÷ 2 = 24个发夹。将两个月的销售量相加，得到48 + 24 = 72个发夹。因此，李美丽在四月和五月总共卖出了72个发夹。答案是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverC/default 第 2 轮回答：
李美丽在四月份卖出48个发夹。五月份卖出的是四月份数量的一半，即48 ÷ 2 = 24个发夹。将四月和五月的销售数量相加，48 + 24 = 72个发夹。 因此，总共卖出的发夹数量是 {{72}}。
--------------------------------------------------------------------------------
求解器 MathSolverA/default 第 2 轮回答：
首先，李美丽在四月份卖出了48个发夹。五月份，她卖出的数量是四月份的一半，即48 ÷ 2 = 24个发夹。将两个月的销售数量相加，可以得到48 + 24 = 72个发夹。所以，李美丽在四月和五月共卖出了72个发夹。答案是 {{72}}。
--------------------------------------------------------------------------------
聚合器 MathAggregator/default 已收到所有 4 个求解器的最终答案。
--------------------------------------------------------------------------------
聚合器 MathAggregator/default 发布最终答案：
72
http://www.dtcms.com/a/284121.html

相关文章：

  • 船舶机械零件的深孔工艺及检测方法 —— 激光频率梳 3D 轮廓检测
  • istio如何自定义重试状态码
  • JAVA面试宝典 -《缓存架构：穿透 / 雪崩 / 击穿解决方案》
  • JVM 内存分配与垃圾回收策略
  • Java学习--JVM（2）
  • Java面试(基础篇) - 第二篇！
  • 如何用 Python + LLM 构建一个智能栗子表格提取工具？
  • Spring,Spring Boot 和 Spring MVC 的关系以及区别
  • 深入解析Hadoop：机架感知算法与数据放置策略
  • #Linux内存管理# vm_normal_page()函数返回的什么样页面的struct page数据结构？为什么内存管理代码中需要这个函数？
  • 【机器学习】第三章 分类算法
  • 如何判断你的EDA工具安装是否真的成功？
  • 数据呈现：让图表说话，从数字到洞察的可视化艺术
  • “显著性”（Saliency）是计算机视觉中的一个重要概念，主要指的是图像或视频中最吸引人注意力的区域或对象
  • Python进阶操作——创建容器
  • Ubuntu开启root用户登陆
  • MyBatis延迟加载（Lazy Loading）之“关联查询”深度解析与实践
  • ros0基础-day13
  • java之json转excel生成
  • Oracle11.2.0.4 RAC迁移升级Oracle19.3 RAC
  • R语言基础| 基本图形绘制（条形图、堆积图、分组图、填充条形图、均值条形图）
  • 规则引擎rule-engine(三)可视化api设计和实现解释
  • 解读一个大学专业——信号与图像处理
  • 2025年Java后端秋招面试的高频八股文+场景题
  • NFS、iSCSI 和lnmp部署操作
  • 深度学习之神经网络（二）
  • 神经网络常见激活函数 15-B-SiLU 函数
  • OCR 赋能档案数字化：让沉睡的档案 “活” 起来
  • 无线充电技术详解：原理、芯片选型与应用场景
  • Android开发中Retrofit使用方法与底层原理详解