【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'))