""" Filename: MetaGPT/examples/debate.py Created Date: Tuesday, September 19th 2023, 6:52:25 pm Author: garylin2099 @Modified By: mashenquan, 2023-11-1. In accordance with Chapter 2.1.3 of RFC 116, modify the data type of the `send_to` value of the `Message` object; modify the argument type of `get_by_actions`. """ import asyncio import platform from typing import Any import fire from metagpt.actions import Action, UserRequirement from metagpt.logs import logger from metagpt.roles import Role from metagpt.schema import Message from research_actions import CollectLinks, WebBrowseAndSummarize, ConductResearch from metagpt.roles import Role from metagpt.roles.role import RoleReactMode import asyncio import re class RequestResearch(Action): """Action: Request research from researcher""" PROMPT_TEMPLATE: str = """ ## BACKGROUND You are {name}, a {profile} preparing for a debate on: {topic} ## TASK What specific aspect of this topic would you like to research to strengthen your position? Provide ONE specific research query (1-2 sentences) that would help you in this debate. Focus on facts, statistics, examples, or evidence that would support your {profile} perspective. """ name: str = "RequestResearch" async def run(self, name: str, profile: str, topic: str) -> str: prompt = self.PROMPT_TEMPLATE.format(name=name, profile=profile, topic=topic) rsp = await self._aask(prompt) return rsp class SpeakAloud(Action): """Action: Speak out aloud in a debate (quarrel)""" PROMPT_TEMPLATE: str = """ ## BACKGROUND Suppose you are {name}, you are in a debate with {opponent_name1} and {opponent_name2}. You are debating the topic: {idea} ## DEBATE HISTORY Previous rounds: {context} ## RESEARCH INFORMATION {research_info} ## YOUR TURN {instruction} """ name: str = "SpeakAloud" async def run(self, context: str, name: str, opponent_name1: str, opponent_name2: str, idea: str = "", profile: str = "", round_num: int = 1, research_info: str = "") -> str: if round_num <= 3: instruction = f"This is round {round_num} of 3 opening rounds. You should ONLY state your view on the topic, give your arguments and how you logically and rigorously arrived at your views. Do NOT rebut or respond to any of your opponents' arguments yet. Your viewpoint should be clear, concise, and extremely stereotypical of a {profile}. MANDATORY: Use specific facts, statistics, and evidence from your research information to support your arguments. Include proper citations in your response using [Source: URL or description] format." else: instruction = f"This is round {round_num}. You should defend your arguments, and attack and directly rebut your opponents' arguments if they differ from yours. Craft a strong, logically rigorous response in {name}'s rhetoric and viewpoints. MANDATORY: Support your arguments with specific evidence from your research and include citations using [Source: URL or description] format." prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name1=opponent_name1, opponent_name2=opponent_name2, idea=idea, profile=profile, instruction=instruction, research_info=research_info) rsp = await self._aask(prompt) return rsp class EvaluateDebate(Action): """Action: Evaluate and summarize a debate into concise recommendations""" PROMPT_TEMPLATE: str = """ ## ROLE You are a neutral evaluator analyzing a debate to provide clear, actionable recommendations. ## DEBATE TOPIC {topic} ## DEBATE CONTENT {debate_content} ## TASK Write a concise evaluation (200-300 words) that: 1. Summarizes key arguments from all participants 2. Identifies core trade-offs and considerations 3. Provides balanced, practical recommendations for decision-makers 4. Focuses on actionable solutions Your response should be structured and help stakeholders make informed decisions. """ name: str = "EvaluateDebate" async def run(self, topic: str, debate_content: str) -> str: prompt = self.PROMPT_TEMPLATE.format(topic=topic, debate_content=debate_content) rsp = await self._aask(prompt) return rsp class DebateEvaluator(Role): name: str = "Evaluator" profile: str = "Neutral Analyst" def __init__(self, **data: Any): super().__init__(**data) self.set_actions([EvaluateDebate]) async def evaluate(self, topic: str, debate_messages: list) -> str: todo = EvaluateDebate() debate_content = "\n\n".join(f"{msg.sent_from}: {msg.content}" for msg in debate_messages) evaluation = await todo.run(topic=topic, debate_content=debate_content) return evaluation class Researcher(Role): """Researcher role with three actions""" name: str = "Researcher" profile: str = "Research Assistant" def __init__(self, **data): super().__init__(**data) self.set_actions([CollectLinks, WebBrowseAndSummarize, ConductResearch]) self._set_react_mode(RoleReactMode.BY_ORDER.value, len(self.actions)) async def research_topic(self, topic: str) -> str: """Conduct quick research on a topic""" # Collect links (limited to 2 queries, 2 URLs each) collect_action = CollectLinks() links = await collect_action.run(topic, decomposition_nums=2, url_per_query=2) # Browse and summarize (max 4 URLs total) browse_action = WebBrowseAndSummarize() summaries = [] url_count = 0 for query, urls in links.items(): if urls and url_count < 4: remaining_urls = 4 - url_count limited_urls = urls[:remaining_urls] result = await browse_action.run(*limited_urls, query=query) summaries.extend(result.values()) url_count += len(limited_urls) # Conduct research research_action = ConductResearch() content = "\n---\n".join(summaries) report = await research_action.run(topic, content) return report class Debator(Role): name: str = "" profile: str = "" opponent_name1: str = "" opponent_name2: str = "" research_info: str = "" research_count: int = 0 max_research: int = 1 def __init__(self, **data: Any): super().__init__(**data) self.set_actions([SpeakAloud]) self._watch([UserRequirement, SpeakAloud]) async def request_research(self, topic: str, researcher: Researcher) -> str: """Request research from the researcher""" if self.research_count >= self.max_research: return "Research limit reached" request_action = RequestResearch() query = await request_action.run(name=self.name, profile=self.profile, topic=topic) # Get research report research_result = await researcher.research_topic(query) self.research_info += f"\n\nResearch Query: {query}\nResearch Result: {research_result}" self.research_count += 1 return research_result async def _observe(self) -> int: await super()._observe() # accept messages sent (from opponent) to self, disregard own messages from the last round self.rc.news = [msg for msg in self.rc.news if self.name in msg.send_to or msg.send_to == {self.name}] return len(self.rc.news) async def _act(self) -> Message: logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})") todo = self.rc.todo memories = self.get_memories() context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories) # Get the debate topic from the first message topic = self.get_memories()[0].content if self.get_memories() else "debate topic" # Count how many times this speaker has spoken speaker_turns = len([m for m in memories if m.sent_from == self.name]) rsp = await todo.run(context=context, name=self.name, opponent_name1=self.opponent_name1, opponent_name2=self.opponent_name2, idea=topic, profile=self.profile, round_num=speaker_turns + 1, research_info=self.research_info) msg = Message( content=rsp, role=self.profile, cause_by=type(todo), sent_from=self.name, send_to={self.opponent_name1, self.opponent_name2} ) self.rc.memory.add(msg) return msg async def debate(idea: str, investment: float = 3.0, n_round: int = 5): """Run a debate and evaluate the results""" School = Debator(name="Principal", profile="School", opponent_name1="John", opponent_name2 = "Mom") Student = Debator(name="John", profile="Student", opponent_name1="Mom", opponent_name2 = "Principal") Parent = Debator(name="Mom", profile="Parent", opponent_name1="Principal", opponent_name2 = "John") # Create researcher researcher = Researcher() logger.info(f"Starting debate on: {idea}") # Research phase - each debater gets 1 research only debaters = [School, Student, Parent] logger.info(f"\n=== Research Phase ===") for debater in debaters: logger.info(f"{debater.name} requesting research...") await debater.request_research(idea, researcher) # Start with Principal responding to the topic current_speaker = School second_speaker = Student third_speaker = Parent # Store all debate messages for evaluation all_messages = [] # Initial message to start the debate msg = Message(content=idea, role="user", send_to={"Principal"}, sent_from="User") for round_num in range(n_round): logger.info(f"\n=== Round {round_num + 1} ===\n{current_speaker.name}'s turn:") # Current speaker responds response = await current_speaker.run(msg) logger.info(f"{current_speaker.name}: {response.content}") # Store the response for evaluation all_messages.append(response) # Switch speakers for next round current_speaker, second_speaker, third_speaker = third_speaker, current_speaker, second_speaker msg = response # Evaluate the debate logger.info("\n=== EVALUATING DEBATE ===") evaluator = DebateEvaluator() evaluation = await evaluator.evaluate(idea, all_messages) logger.info(f"\n=== EVALUATION RESULTS ===\n{evaluation}") return evaluation def main(idea: str, investment: float = 3.0, n_round: int = 6): """ :param idea: Debate topic, such as "Topic: The U.S. should commit more in climate change fighting" or "Trump: Climate change is a hoax" :param investment: contribute a certain dollar amount to watch the debate :param n_round: maximum rounds of the debate :return: """ if platform.system() == "Windows": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) asyncio.run(debate(idea, investment, n_round)) if __name__ == "__main__": fire.Fire(main) # run as python debate.py --idea="TOPIC" --investment=3.0 --n_round=5