MetaGPT/examples/debate.py

"""
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.2.1 and 2.2.2 of RFC 116, modify the data
        type of the `cause_by` value in the `Message` to a string, and utilize the new message distribution
        feature for message filtering.
"""
import asyncio
import platform

import fire

from metagpt.actions import Action, BossRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.software_company import SoftwareCompany


class ShoutOut(Action):
    """Action: Shout out loudly 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="ShoutOut", 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 Trump(Role):
    def __init__(
        self,
        name: str = "Trump",
        profile: str = "Republican",
        **kwargs,
    ):
        super().__init__(name, profile, **kwargs)
        self._init_actions([ShoutOut])
        self._watch([ShoutOut])
        self.name = "Trump"
        self.opponent_name = "Biden"

    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.is_recipient({self.name})]
        return len(self._rc.news)

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self._rc.todo}")

        msg_history = self._rc.memory.get_by_actions([ShoutOut])
        context = []
        for m in msg_history:
            context.append(str(m))
        context = "\n".join(context)

        rsp = await ShoutOut().run(context=context, name=self.name, opponent_name=self.opponent_name)

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=ShoutOut,
            tx_from=self.name,
            tx_to=self.opponent_name,
        )

        return msg


class Biden(Role):
    def __init__(
        self,
        name: str = "Biden",
        profile: str = "Democrat",
        **kwargs,
    ):
        super().__init__(name, profile, **kwargs)
        self._init_actions([ShoutOut])
        self._watch([BossRequirement, ShoutOut])
        self.name = "Biden"
        self.opponent_name = "Trump"

    async def _observe(self) -> int:
        await super()._observe()
        # accept the very first human instruction (the debate topic) or messages sent (from opponent) to self,
        # disregard own messages from the last round
        message_filter = {BossRequirement, self.name}
        self._rc.news = [msg for msg in self._rc.news if msg.is_recipient(message_filter)]
        return len(self._rc.news)

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self._rc.todo}")

        msg_history = self._rc.memory.get_by_actions([BossRequirement, ShoutOut])
        context = []
        for m in msg_history:
            context.append(str(m))
        context = "\n".join(context)

        rsp = await ShoutOut().run(context=context, name=self.name, opponent_name=self.opponent_name)

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=ShoutOut,
            tx_from=self.name,
            tx_to=self.opponent_name,
        )

        return msg


async def startup(
    idea: str, investment: float = 3.0, n_round: int = 5, code_review: bool = False, run_tests: bool = False
):
    """We reuse the startup paradigm for roles to interact with each other.
    Now we run a startup of presidents and watch they quarrel. :)"""
    company = SoftwareCompany()
    company.hire([Biden(), Trump()])
    company.invest(investment)
    company.start_project(idea)
    await company.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(startup(idea, investment, n_round))


if __name__ == "__main__":
    fire.Fire(main)