examples/debate/agent/main.py

import logging
import uuid
from typing import Optional, AsyncGenerator

from aworld.memory.main import MemoryFactory
from examples.debate.agent.base import DebateSpeech
from examples.debate.agent.debate_agent import DebateAgent
from examples.debate.agent.moderator_agent import ModeratorAgent
from aworld.core.common import Observation
from aworld.core.memory import MemoryItem, MemoryConfig
from aworld.output import Output, WorkSpace, Artifact, ArtifactType, CodeArtifact


class DebateArena:
    """
    DebateArena is platform for debate
    """

    affirmative_speaker: DebateAgent
    negative_speaker: DebateAgent

    moderator: Optional[ModeratorAgent]

    speeches: list[DebateSpeech]

    display_panel: str

    def __init__(self,
                 affirmative_speaker: DebateAgent,
                 negative_speaker: DebateAgent,
                 moderator: ModeratorAgent,
                 workspace: WorkSpace,
                 **kwargs
                 ):
        self.affirmative_speaker = affirmative_speaker
        self.negative_speaker = negative_speaker
        self.moderator = moderator
        self.speeches = []
        self.workspace = workspace
        self.affirmative_speaker.set_workspace(workspace)
        self.negative_speaker.set_workspace(workspace)
        self.moderator.set_workspace(workspace)
        self.moderator.memory = MemoryFactory.from_config(MemoryConfig(provider="inmemory", enable_summary=False))

        # Event.register("topic", func= );

    async def async_run(self, topic: str, rounds: int) \
            -> AsyncGenerator[Output, None]:

        """
        Start the debate
        1. debate will start from round 1
        2. each round will have two speeches, one from affirmative_speaker and one from negative_speaker
        3. after all rounds finished, the debate will end

        Args:
            topic: str -> topic of the debate
            affirmative_opinion: str -> affirmative speaker's opinion
            negative_opinion: str -> negative speaker's opinion
            rounds: int -> number of rounds

        Returns: list[DebateSpeech]

        """

        ## 1. generate opinions
        moderator_speech = await self.moderator_speech(topic, rounds)
        if not moderator_speech:
            return
        yield moderator_speech
        await moderator_speech.wait_until_finished()
        self.store_speech(moderator_speech)

        affirmative_opinion = moderator_speech.metadata["affirmative_opinion"]
        negative_opinion = moderator_speech.metadata["negative_opinion"]

        logging.info(f"✈️==================================== opinions =============================================")
        logging.info(f"topic: {topic}")
        logging.info(f"affirmative_opinion: {affirmative_opinion}")
        logging.info(f"negative_opinion: {negative_opinion}")
        logging.info(f"✈️==================================== start... =============================================")

        ## 2. Alternating speeches
        for i in range(1, rounds + 1):
            logging.info(
                f"✈️==================================== round#{i} start =============================================")
            loading_speech = DebateSpeech.from_dict({
                "content": f"\n\n**round#{i} start** \n\n",
                "round": i,
                "type": "loading",
                "stance": "stage",
                "name": "stage",
                "finished": True
            })
            yield loading_speech

            loading_speech = DebateSpeech.from_dict({
                "content": f"\n\n【affirmative】✅：{self.affirmative_speaker.name()}\n Searching ....\n",
                "round": i,
                "type": "loading",
                "stance": "stage",
                "name": "stage",
                "finished": True
            })
            yield loading_speech

            # affirmative_speech
            speech = await self.affirmative_speech(i, topic, affirmative_opinion, negative_opinion)
            yield speech
            await speech.wait_until_finished()
            self.store_speech(speech)

            loading_speech = DebateSpeech.from_dict({
                "content": f"\n\n【negative】❌：{self.negative_speaker.name()}\n Searching ....\n",
                "round": i,
                "type": "loading",
                "stance": "stage",
                "name": "stage",
                "finished": True
            })
            yield loading_speech

            # negative_speech
            speech = await self.negative_speech(i, topic, negative_opinion, affirmative_opinion)
            yield speech
            await speech.wait_until_finished()
            self.store_speech(speech)

            logging.info(
                f"🛬==================================== round#{i} end =============================================")

        ## 3. Summary speeches
        moderator_speech = await self.moderator.summary_speech()
        if not moderator_speech:
            return
        yield moderator_speech
        await moderator_speech.wait_until_finished()
        await self.workspace.add_artifact(
            CodeArtifact.build_artifact(
                artifact_type=ArtifactType.CODE,
                artifact_id="result",
                code_type='html',
                content=moderator_speech.content,
                metadata={
                    "topic": topic
                }
            )
        )
        logging.info(
            f"🛬====================================  total is end =============================================")

    async def moderator_speech(self, topic, rounds) -> DebateSpeech | None:
        results = await self.moderator.async_policy(Observation(content=topic, info={"rounds": rounds}))
        if not results or not results[0] or not results[0].policy_info:
            return None
        return results[0].policy_info

    async def affirmative_speech(self, round: int, topic: str, opinion: str, oppose_opinion: str) -> DebateSpeech:
        """
        affirmative_speaker will start speech
        """

        affirmative_speaker = self.get_affirmative_speaker()

        logging.info(affirmative_speaker.name() + ": " + "start")

        speech = await affirmative_speaker.speech(topic, opinion, oppose_opinion, round, self.speeches)

        logging.info(affirmative_speaker.name() + ":  result: " + speech.content)
        return speech

    async def negative_speech(self, round: int, topic: str, opinion: str, oppose_opinion: str) -> DebateSpeech:
        """
        after affirmative_speaker finished speech, negative_speaker will start speech
        """

        negative_speaker = self.get_negative_speaker()

        logging.info(negative_speaker.name() + ": " + "start")

        speech = await negative_speaker.speech(topic, opinion, oppose_opinion, round, self.speeches)

        logging.info(negative_speaker.name() + ":  result: " + speech.content)
        return speech

    def get_affirmative_speaker(self) -> DebateAgent:
        """
        return the affirmative speaker
        """
        return self.affirmative_speaker

    def get_negative_speaker(self) -> DebateAgent:
        """
        return the negative speaker
        """
        return self.negative_speaker

    def store_speech(self, speech: DebateSpeech):
        self.moderator.memory.add(MemoryItem.from_dict({
            "content": speech.content,
            "metadata": {
                "round": speech.round,
                "speaker": speech.name,
                "type": speech.type
            }
        }))
        self.speeches.append(speech)

    def gen_closing_statement(self):
        pass