""" MAD: Multi-Agent Debate with Large Language Models Copyright (C) 2023 The MAD Team This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program. If not, see . """ import os import json import random # random.seed(0) from Agent import Agent os.environ["TOGETHER_API_KEY"] = "48bf2536f85b599c7d5d7f9921cc9ee7056f40ed535fd2174d061e1b9abcf8af" NAME_LIST=[ "Affirmative side", "Negative side", "Moderator", ] class DebatePlayer(Agent): def __init__(self, model_name: str, name: str, temperature:float, openai_api_key: str, sleep_time: float) -> None: """Create a player in the debate Args: model_name(str): model name name (str): name of this player temperature (float): higher values make the output more random, while lower values make it more focused and deterministic openai_api_key (str): As the parameter name suggests sleep_time (float): sleep because of rate limits """ super(DebatePlayer, self).__init__(model_name, name, temperature, sleep_time) self.openai_api_key = openai_api_key class Debate: def __init__(self, model_name: str='Qwen/Qwen1.5-72B-Chat', temperature: float=0, num_players: int=3, openai_api_key: str=os.environ["TOGETHER_API_KEY"], config: dict=None, max_round: int=3, sleep_time: float=0 ) -> None: """Create a debate Args: model_name (str): openai model name temperature (float): higher values make the output more random, while lower values make it more focused and deterministic num_players (int): num of players openai_api_key (str): As the parameter name suggests max_round (int): maximum Rounds of Debate sleep_time (float): sleep because of rate limits """ self.model_name = model_name self.temperature = temperature self.num_players = num_players self.openai_api_key = openai_api_key self.config = config self.max_round = max_round self.sleep_time = sleep_time self.initial_debate = '' self.init_prompt() # creat&init agents self.creat_agents() self.init_agents() def init_prompt(self): def prompt_replace(key): self.config[key] = self.config[key].replace("##debate_topic##", self.config["debate_topic"]) prompt_replace("player_meta_prompt") prompt_replace("moderator_meta_prompt") prompt_replace("affirmative_prompt") prompt_replace("judge_prompt_last2") def creat_agents(self): # creates players self.players = [ DebatePlayer(model_name=self.model_name, name=name, temperature=self.temperature, openai_api_key=self.openai_api_key, sleep_time=self.sleep_time) for name in NAME_LIST ] self.affirmative = self.players[0] self.negative = self.players[1] self.moderator = self.players[2] def init_agents(self): # start: set meta prompt self.affirmative.set_meta_prompt(self.config['player_meta_prompt']) self.negative.set_meta_prompt(self.config['player_meta_prompt']) self.moderator.set_meta_prompt(self.config['moderator_meta_prompt']) # start: first round debate, state opinions print(f"===== Start Debate Round =====\n") self.affirmative.add_event(self.config['affirmative_prompt']) self.aff_ans = self.affirmative.ask() self.affirmative.add_memory(self.aff_ans) self.config['base_answer'] = self.aff_ans affirm_side = "\n\n正方观点：" + self.aff_ans self.initial_debate += affirm_side self.negative.add_event(self.config['negative_prompt'].replace('##aff_ans##', self.aff_ans)) self.neg_ans = self.negative.ask() self.negative.add_memory(self.neg_ans) neg_side = "\n\n反方观点：" + self.neg_ans self.initial_debate += neg_side self.moderator.add_event(self.config['moderator_prompt'].replace('##aff_ans##', self.aff_ans).replace('##neg_ans##', self.neg_ans).replace('##round##', 'first')) self.mod_ans = self.moderator.ask() self.moderator.add_memory(self.mod_ans) self.mod_ans = eval(self.mod_ans) def round_dct(self, num: int): dct = { 1: 'first', 2: 'second', 3: 'third', 4: 'fourth', 5: 'fifth', 6: 'sixth', 7: 'seventh', 8: 'eighth', 9: 'ninth', 10: 'tenth' } return dct[num] def print_answer(self): print("\n\n===== Debate Done! =====") print("\n----- Debate Topic -----") print(self.config["debate_topic"]) print("\n----- Base Answer -----") print(self.config["base_answer"]) print("\n----- Debate Answer -----") print(self.config["debate_answer"]) print("\n----- Debate Reason -----") print(self.config["Reason"]) def broadcast(self, msg: str): """Broadcast a message to all players. Typical use is for the host to announce public information Args: msg (str): the message """ # print(msg) for player in self.players: player.add_event(msg) def speak(self, speaker: str, msg: str): """The speaker broadcast a message to all other players. Args: speaker (str): name of the speaker msg (str): the message """ if not msg.startswith(f"{speaker}: "): msg = f"{speaker}: {msg}" # print(msg) for player in self.players: if player.name != speaker: player.add_event(msg) def ask_and_speak(self, player: DebatePlayer): ans = player.ask() player.add_memory(ans) self.speak(player.name, ans) def run(self): for round in range(self.max_round - 1): print(f"===== Debate Round-{round+1} =====\n") self.affirmative.add_event(self.config['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.config['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.config['moderator_prompt'].replace('##aff_ans##', self.aff_ans).replace('##neg_ans##', self.neg_ans).replace('##round##', self.round_dct(round+2))) self.mod_ans = self.moderator.ask() self.moderator.add_memory(self.mod_ans) self.mod_ans = eval(self.mod_ans) if self.mod_ans["debate_answer"] != '': self.config.update(self.mod_ans) self.config['success'] = True # ultimate deadly technique. else: judge_player = DebatePlayer(model_name=self.model_name, name='Judge', temperature=self.temperature, openai_api_key=self.openai_api_key, sleep_time=self.sleep_time) aff_ans = self.affirmative.memory_lst[2]['content'] neg_ans = self.negative.memory_lst[2]['content'] judge_player.set_meta_prompt(self.config['moderator_meta_prompt']) # extract answer candidates judge_player.add_event(self.config['judge_prompt_last1'].replace('##aff_ans##', aff_ans).replace('##neg_ans##', neg_ans)) ans = judge_player.ask() judge_player.add_memory(ans) # select one from the candidates judge_player.add_event(self.config['judge_prompt_last2']) ans = judge_player.ask() judge_player.add_memory(ans) ans = eval(ans) if ans["debate_answer"] != '': self.config['success'] = True # save file self.config.update(ans) self.players.append(judge_player) self.print_answer() combined_string = ''.join([ self.config["debate_topic"], self.initial_debate, f"\n\n经过{self.max_round}轮辩论......", "\n\n仲裁观点：", self.config["debate_answer"], "\n\n仲裁理由：", self.config["Reason"] ]) return combined_string