¡¾AI AgentϵÁС¿¡¾MetaGPT¶àÖÇÄÜÌåѧϰ¡¿3. ¿ª·¢Ò»¸ö¼òµ¥µÄ¶àÖÇÄÜÌåϵͳ£¬¼æ¿´MetaGPT¶àÖÇÄÜÌåÔËÐлúÖÆ

class Environment(ExtEnv):
    """»·¾³£¬³ÐÔØÒ»Åú½ÇÉ«£¬½ÇÉ«¿ÉÒÔÏò»·¾³·¢²¼ÏûÏ¢£¬¿ÉÒÔ±»ÆäËû½ÇÉ«¹Û²ìµ½
    Environment, hosting a batch of roles, roles can publish messages to the environment, and can be observed by other roles
    """
    model_config = ConfigDict(arbitrary_types_allowed=True)
    desc: str = Field(default="")  # »·¾³ÃèÊö
    roles: dict[str, SerializeAsAny["Role"]] = Field(default_factory=dict, validate_default=True)
    member_addrs: Dict["Role", Set] = Field(default_factory=dict, exclude=True)
    history: str = ""  # For debug
    context: Context = Field(default_factory=Context, exclude=True)

def add_roles(self, roles: Iterable["Role"]):
    """Ôö¼ÓÒ»ÅúÔÚµ±Ç°»·¾³µÄ½ÇÉ«
    Add a batch of characters in the current environment
    """
    for role in roles:
        self.roles[role.profile] = role
    for role in roles:  # setup system message with roles
        role.set_env(self)
        role.context = self.context

def set_env(self, env: "Environment"):
    """Set the environment in which the role works. The role can talk to the environment and can also receive
    messages by observing."""
    self.rc.env = env
    if env:
        env.set_addresses(self, self.addresses)
        self.llm.system_prompt = self._get_prefix()
        self.set_actions(self.actions)  # reset actions to update llm and prefix

def publish_message(self, msg):
    """If the role belongs to env, then the role's messages will be broadcast to env"""
    if not msg:
        return
    if not self.rc.env:
        # If env does not exist, do not publish the message
        return
    self.rc.env.publish_message(msg)

def publish_message(self, message: Message, peekable: bool = True) -> bool:
    """
    Distribute the message to the recipients.
    In accordance with the Message routing structure design in Chapter 2.2.1 of RFC 116, as already planned
    in RFC 113 for the entire system, the routing information in the Message is only responsible for
    specifying the message recipient, without concern for where the message recipient is located. How to
    route the message to the message recipient is a problem addressed by the transport framework designed
    in RFC 113.
    """
    logger.debug(f"publish_message: {message.dump()}")
    found = False
    # According to the routing feature plan in Chapter 2.2.3.2 of RFC 113
    for role, addrs in self.member_addrs.items():
        if is_send_to(message, addrs):
            role.put_message(message)
            found = True
    if not found:
        logger.warning(f"Message no recipients: {message.dump()}")
    self.history += f"\n{message}"  # For debug
    return True

for role, addrs in self.member_addrs.items():
    if is_send_to(message, addrs):
        role.put_message(message)

async def run(self, k=1):
    """´¦ÀíÒ»´ÎËùÓÐÐÅÏ¢µÄÔËÐÐ
    Process all Role runs at once
    """
    for _ in range(k):
        futures = []
        for role in self.roles.values():
            future = role.run()
            futures.append(future)
        await asyncio.gather(*futures)
        logger.debug(f"is idle: {self.is_idle}")

class WritePoem(Action):
    name: str = "WritePoem"
    PROMPT_TEMPLATE: str = """
    Here is the historical conversation record : {msg} .
    Write a poem about the subject provided by human, Return only the content of the generated poem with NO other texts.
    If the teacher provides suggestions about the poem, revise the student's poem based on the suggestions and return.
    your poem:
    """
    async def run(self, msg: str):
        prompt = self.PROMPT_TEMPLATE.format(msg = msg)
        rsp = await self._aask(prompt)
        return rsp
class Student(Role):
    name: str = "xiaoming"
    profile: str = "Student"
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([WritePoem])
        self._watch([UserRequirement, ReviewPoem])
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        todo = self.rc.todo
        msg = self.get_memories()  # »ñÈ¡ËùÓмÇÒä
        # logger.info(msg)
        poem_text = await WritePoem().run(msg)
        logger.info(f'student : {poem_text}')
        msg = Message(content=poem_text, role=self.profile,
                      cause_by=type(todo))
        return msg

class ReviewPoem(Action):
    name: str = "ReviewPoem"
    PROMPT_TEMPLATE: str = """
    Here is the historical conversation record : {msg} .
    Check student-created poems about the subject provided by human and give your suggestions for revisions. You prefer poems with elegant sentences and retro style.
    Return only your comments with NO other texts.
    your comments:
    """
    async def run(self, msg: str):
        prompt = self.PROMPT_TEMPLATE.format(msg = msg)
        rsp = await self._aask(prompt)
        return rsp
class Teacher(Role):
    name: str = "laowang"
    profile: str = "Teacher"
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([ReviewPoem])
        self._watch([WritePoem])
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        todo = self.rc.todo
        msg = self.get_memories()  # »ñÈ¡ËùÓмÇÒä
        poem_text = await ReviewPoem().run(msg)
        logger.info(f'teacher : {poem_text}')
        msg = Message(content=poem_text, role=self.profile,
                      cause_by=type(todo))
        return msg

classroom = Environment()
classroom.add_roles([Student(), Teacher()])
classroom.publish_message(
    Message(role="Human", content=topic, cause_by=UserRequirement,
            send_to='' or MESSAGE_ROUTE_TO_ALL),
    peekable=False,
)

import asyncio
from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.environment import Environment
from metagpt.const import MESSAGE_ROUTE_TO_ALL
async def main(topic: str, n_round=3):
    while n_round > 0:
        # self._save()
        n_round -= 1
        logger.debug(f"max {n_round=} left.")
        await classroom.run()
    return classroom.history
asyncio.run(main(topic='wirte a poem about moon'))

import asyncio
from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.environment import Environment
from metagpt.const import MESSAGE_ROUTE_TO_ALL
classroom = Environment()
class WritePoem(Action):
    name: str = "WritePoem"
    PROMPT_TEMPLATE: str = """
    Here is the historical conversation record : {msg} .
    Write a poem about the subject provided by human, Return only the content of the generated poem with NO other texts.
    If the teacher provides suggestions about the poem, revise the student's poem based on the suggestions and return.
    your poem:
    """
    async def run(self, msg: str):
        prompt = self.PROMPT_TEMPLATE.format(msg = msg)
        rsp = await self._aask(prompt)
        return rsp
class ReviewPoem(Action):
    name: str = "ReviewPoem"
    PROMPT_TEMPLATE: str = """
    Here is the historical conversation record : {msg} .
    Check student-created poems about the subject provided by human and give your suggestions for revisions. You prefer poems with elegant sentences and retro style.
    Return only your comments with NO other texts.
    your comments:
    """
    async def run(self, msg: str):
        prompt = self.PROMPT_TEMPLATE.format(msg = msg)
        rsp = await self._aask(prompt)
        return rsp
    
class Student(Role):
    name: str = "xiaoming"
    profile: str = "Student"
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([WritePoem])
        self._watch([UserRequirement, ReviewPoem])
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        todo = self.rc.todo
        msg = self.get_memories()  # »ñÈ¡ËùÓмÇÒä
        # logger.info(msg)
        poem_text = await WritePoem().run(msg)
        logger.info(f'student : {poem_text}')
        msg = Message(content=poem_text, role=self.profile,
                      cause_by=type(todo))
        return msg
class Teacher(Role):
    name: str = "laowang"
    profile: str = "Teacher"
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([ReviewPoem])
        self._watch([WritePoem])
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        todo = self.rc.todo
        msg = self.get_memories()  # »ñÈ¡ËùÓмÇÒä
        poem_text = await ReviewPoem().run(msg)
        logger.info(f'teacher : {poem_text}')
        msg = Message(content=poem_text, role=self.profile,
                      cause_by=type(todo))
        return msg
async def main(topic: str, n_round=3):
    classroom.add_roles([Student(), Teacher()])
    classroom.publish_message(
        Message(role="Human", content=topic, cause_by=UserRequirement,
                send_to='' or MESSAGE_ROUTE_TO_ALL),
        peekable=False,
    )
    while n_round > 0:
        # self._save()
        n_round -= 1
        logger.debug(f"max {n_round=} left.")
        await classroom.run()
    return classroom.history
asyncio.run(main(topic='wirte a poem about moon'))