mirror of
https://github.com/FoundationAgents/MetaGPT.git
synced 2026-04-26 17:26:22 +02:00
109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
'''
|
|
Filename: MetaGPT/examples/debate.py
|
|
Created Date: Tuesday, September 19th 2023, 6:52:25 pm
|
|
Author: garylin2099
|
|
'''
|
|
import asyncio
|
|
import platform
|
|
import fire
|
|
|
|
from metagpt.team import Team
|
|
from metagpt.actions import Action, BossRequirement
|
|
from metagpt.roles import Role
|
|
from metagpt.schema import Message
|
|
from metagpt.logs import logger
|
|
|
|
class SpeakAloud(Action):
|
|
"""Action: Speak out aloud in a debate (quarrel)"""
|
|
|
|
PROMPT_TEMPLATE = """
|
|
## BACKGROUND
|
|
Suppose you are {name}, you are in a debate with {opponent_name}.
|
|
## DEBATE HISTORY
|
|
Previous rounds:
|
|
{context}
|
|
## YOUR TURN
|
|
Now it's your turn, you should closely respond to your opponent's latest argument, state your position, defend your arguments, and attack your opponent's arguments,
|
|
craft a strong and emotional response in 80 words, in {name}'s rhetoric and viewpoints, your will argue:
|
|
"""
|
|
|
|
def __init__(self, name="SpeakAloud", context=None, llm=None):
|
|
super().__init__(name, context, llm)
|
|
|
|
async def run(self, context: str, name: str, opponent_name: str):
|
|
|
|
prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)
|
|
# logger.info(prompt)
|
|
|
|
rsp = await self._aask(prompt)
|
|
|
|
return rsp
|
|
|
|
class Debator(Role):
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
profile: str,
|
|
opponent_name: str,
|
|
**kwargs,
|
|
):
|
|
super().__init__(name, profile, **kwargs)
|
|
self._init_actions([SpeakAloud])
|
|
self._watch([BossRequirement, SpeakAloud])
|
|
self.name = name
|
|
self.opponent_name = opponent_name
|
|
|
|
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 msg.send_to == self.name]
|
|
return len(self._rc.news)
|
|
|
|
async def _act(self) -> Message:
|
|
logger.info(f"{self._setting}: ready to {self._rc.todo}")
|
|
todo = self._rc.todo # An instance of SpeakAloud
|
|
|
|
memories = self.get_memories()
|
|
context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
|
|
# print(context)
|
|
|
|
rsp = await todo.run(context=context, name=self.name, opponent_name=self.opponent_name)
|
|
|
|
msg = Message(
|
|
content=rsp,
|
|
role=self.profile,
|
|
cause_by=type(todo),
|
|
sent_from=self.name,
|
|
send_to=self.opponent_name,
|
|
)
|
|
|
|
self._rc.memory.add(msg)
|
|
|
|
return msg
|
|
|
|
async def debate(idea: str, investment: float = 3.0, n_round: int = 5):
|
|
"""Run a team of presidents and watch they quarrel. :) """
|
|
Biden = Debator(name="Biden", profile="Democrat", opponent_name="Trump")
|
|
Trump = Debator(name="Trump", profile="Republican", opponent_name="Biden")
|
|
team = Team()
|
|
team.hire([Biden, Trump])
|
|
team.invest(investment)
|
|
team.start_project(idea, send_to="Biden") # send debate topic to Biden and let him speak first
|
|
await team.run(n_round=n_round)
|
|
|
|
|
|
def main(idea: str, investment: float = 3.0, n_round: int = 10):
|
|
"""
|
|
: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)
|