多智能体辩论框架横向对比

核心代码:6 步辩论主循环

# [1] 正方初判
prompt1_1 = prompts["prompt_1"].replace("{{story}}", story)
outputs = model1.generate(**tokenizer1(prompt1_1, return_tensors="pt").to("cuda"),
                          max_new_tokens=1024, temperature=0.0)
generation1_1 = parse_response(tokenizer1.decode(outputs[0]), "Answer:")

# [3] 正方看到反方答案后给反馈
prompt1_2 = prompts["prompt_2"].replace("{{other_response}}", generation2_1) \
                               .replace("{{your_response}}", generation1_1)

# [5] 正方综合反馈后给最终答案
prompt1_3 = prompts["prompt_3"].replace("{{feedback}}", generation2_2)
generation1_3 = parse_final_answer(generation1_3)  # 归一化为 Yes/No/Neither

line["gemma_final"] = generation1_3
line["seallm_final"] = generation2_3

核心代码:辩论主循环 + 视角切换

def debating_loop(self, seed_message, max_rounds, steps, mode):
    for debater in self.protagonists[1:]:
        debater.append_to_chat(seed_message)
    msg = seed_message
    for _ in range(max_rounds):
        if mode == "sequential":
            debaters = self.protagonists
        elif mode == "random_no_repeat":
            debaters = random.sample(self.protagonists, k=len(self.protagonists))
        for debater in debaters:
            # 回复的 role 是 "assistant",传给下一个 agent 前被改成 "user"
            msg = self.debating_step(msg, debater, steps)

def debating_step(self, input_msg, character, steps):
    input_msg["role"] = "user"        # 视角切换
    output_msg = character.reply(input_msg, steps=steps)
    return output_msg

核心代码:Constitutional AI 自我改进

def reply(self, user_msg, steps=["critique", "revision"]):
    self.chat_step(user_msg)              # 先正常回复
    refined = self.self_improve(steps)    # 再自我批评+修订
    return refined

def self_improve(self, steps):
    for principle in self.principles:
        if "critique" in steps:
            self.chat_step({"role":"user","content": principle["critique"]})
        if "revision" in steps:
            self.chat_step({"role":"user","content": principle["revision"]})
    return self.chat[-1]

核心代码:辩论主循环 + Judge 兜底

def run(self):
    for round in range(self.max_round - 1):
        if self.mod_ans["debate_translation"] != '':
            break  # 主持人已定论,早停
        # 正方读到反方上轮发言后反驳
        self.affirmative.add_event(
            self.save_file['debate_prompt'].replace('##oppo_ans##', self.neg_ans))
        self.aff_ans = self.affirmative.ask()
        self.affirmative.add_memory(self.aff_ans)
        # 反方反驳
        self.negative.add_event(
            self.save_file['debate_prompt'].replace('##oppo_ans##', self.aff_ans))
        self.neg_ans = self.negative.ask()
        self.negative.add_memory(self.neg_ans)
        # 主持人裁决
        self.moderator.add_event(self.save_file['moderator_prompt']
            .replace('##aff_ans##', self.aff_ans)
            .replace('##neg_ans##', self.neg_ans))
        self.mod_ans = eval(self.moderator.ask())  # ⚠️ eval 解析 JSON

    if self.mod_ans["debate_translation"] == '':
        # 终极杀招:Judge 兜底
        judge = DebatePlayer(model_name=self.model_name, name='Judge', ...)
        judge.add_event(self.save_file['judge_prompt_last1']
            .replace('##aff_ans##', aff_ans).replace('##neg_ans##', neg_ans))
        ans = eval(judge.ask())

核心代码:Agent 私有记忆封装

class Agent:
    def __init__(self, model_name, name, temperature, sleep_time=0):
        self.memory_lst = []  # 私有记忆,只含自己视角

    def add_event(self, event):    # 当前轮上下文(user)
        self.memory_lst.append({"role": "user", "content": f"{event}"})
    def add_memory(self, memory):  # 持久记忆(assistant)
        self.memory_lst.append({"role": "assistant", "content": f"{memory}"})

    @backoff.on_exception(backoff.expo, (RateLimitError, APIError), max_tries=20)
    def query(self, messages, max_tokens, api_key, temperature):
        response = openai.ChatCompletion.create(...)  # 旧版 SDK
        return response['choices'][0]['message']['content']

核心代码:Core API 实现两 Agent 辩论

from autogen_core import (AgentId, MessageContext, RoutedAgent,
    SingleThreadedAgentRuntime, message_handler, type_subscription)
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 1. 定义消息协议
@dataclass
class AnswerMessage:
    content: str
@dataclass
class CritiqueMessage:
    content: str

# 2. 定义辩论 Agent
@type_subscription("debate")
class DebateAgent(RoutedAgent):
    def __init__(self, model_client, system_message, max_rounds=3):
        super().__init__("Debate Agent")
        self._model_client = model_client
        self._system_message = system_message
        self._max_rounds = max_rounds
        self._round = 0

    @message_handler
    async def handle_answer(self, message: AnswerMessage, ctx):
        # 收到对方答案 → 生成批评
        self._round += 1
        return [CritiqueMessage(content=critique_text)]

    @message_handler
    async def handle_critique(self, message: CritiqueMessage, ctx):
        if self._round >= self._max_rounds:
            return []  # 结束
        return [AnswerMessage(content=revised_answer)]  # 修正后回击

# 3. 注册并运行
async def main():
    runtime = SingleThreadedAgentRuntime()
    model = OpenAIChatCompletionClient(model="gpt-4o")
    await DebateAgent.register(runtime, "Affirmative",
        lambda: DebateAgent(model, "You argue the affirmative side."))
    await DebateAgent.register(runtime, "Negative",
        lambda: DebateAgent(model, "You argue the negative side."))
    await runtime.start()
    await runtime.send_message(AnswerMessage(content="..."), AgentId("Affirmative","debate"))