Spaces:
Sleeping
Sleeping
| """ | |
| 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 <http://www.gnu.org/licenses/>. | |
| """ | |
| 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) | |
| self.mod_ans = self.mod_ans.replace("json", "") | |
| self.mod_ans = json.loads(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+2} =====\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) | |
| self.mod_ans = self.mod_ans.replace("json", "") | |
| self.mod_ans = json.loads(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 | |