MetaGPT智能体开发入门》学习手册 第三章课程任务

2. 第三章课程任务：

• 这个 Agent 拥有三个动作 打印1 打印2 打印3
• 重写有关方法（请不要使用act_by_order，我希望你能独立实现）使得 Agent 顺序执行上面三个动作
• 当上述三个动作执行完毕后，为 Agent 生成新的动作 打印4 打印5 打印6 并顺序执行

from metagpt.actions import Action
from metagpt.roles import Role
from metagpt.schema import Message
import asyncio
from metagpt.logs import logger
import random

class PrintFruit(Action):
    def __init__(self, fruit_name, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.fruit_name = fruit_name

    async def run(self, *args, **kwargs):
        res =  f"Printing fruit: {self.fruit_name}"
        print(res)
        return res


class FruitPrintingAgent(Role):
    def __init__(self, name: str = "FruitPrinter", profile: str = "Fruit Printing Agent", **kwargs):
        super().__init__(name, profile, **kwargs)
        self.fruits = ["Apple", "Banana", "Cherry", "Durian", "Elderberry", "Fig", "Grape", "Honeydew", "Jackfruit"]
        self.current_action_index = 0
        self._init_actions(self._select_random_fruits(3))

    def _select_random_fruits(self, count):
        selected_fruits = random.sample(self.fruits, count)
        return [PrintFruit(fruit) for fruit in selected_fruits]

    async def _think(self) -> None:
        logger.info(self._rc.state)

        if self.current_action_index < len(self._actions):
            self._set_state(self.current_action_index)
        else:
            # 添加新的水果打印动作
            new_fruit_actions = self._select_random_fruits(3)
            self._init_actions(new_fruit_actions)
            self.current_action_index = 0
            self._set_state(self.current_action_index)

    async def _act(self) -> Message:
        todo = self._rc.todo
        # 从记忆中获取最新的一条消息
        msg = self._rc.memory.get(k=1)[0] if self._rc.memory.get(k=1) else None

        # 如果有消息，则将消息内容作为参数传递给当前待执行的动作
        if msg:
            result = await todo.run(msg.content)
            logger.info(result)
        else:
            result = await todo.run("")

        self.current_action_index += 1
        return Message(content=result, role=self.profile)

    async def _react(self) -> Message:
        # 持续执行动作直到所有动作都被执行
        i = 0
        while i < 6:
            await self._think()
            if self._rc.todo is None:
                break
            msg = await self._act()
            i += 1
        return msg


async def main():
    # 初始化消息和智能体
    initial_message = "Start fruit printing"
    role = FruitPrintingAgent()

    # 记录初始消息
    logger.info(initial_message)

    # 启动智能体并等待其执行完毕
    result = await role.run(initial_message)

    # 记录执行结果
    # logger.info(result.content)


asyncio.run(main())
"""
2023-12-14 23:20:49.143 | INFO     | __main__:main:75 - Start fruit printing
2023-12-14 23:20:49.143 | INFO     | __main__:_act:50 - Printing fruit: Fig
2023-12-14 23:20:49.143 | INFO     | __main__:_act:50 - Printing fruit: Apple
2023-12-14 23:20:49.143 | INFO     | __main__:_act:50 - Printing fruit: Durian
2023-12-14 23:20:49.143 | INFO     | __main__:_act:50 - Printing fruit: Honeydew
2023-12-14 23:20:49.143 | INFO     | __main__:_act:50 - Printing fruit: Elderberry
2023-12-14 23:20:49.144 | INFO     | __main__:_act:50 - Printing fruit: Cherry
"""