Metagpt定义自己的Agent

admin管理员组

文章数量:1567915

文章目录

• 创建自己的项目文件

---

创建自己的项目文件

想要让Agent按自己的需求运行，需要在Metagpt的根目录下自定义一个py文件，官方教程使用的Metagpt版本太旧，如果使用新版Metagpt运行则需要对某些变量名进行修改。本文使用0.8版本进行演示

---

目前为止我们设计的所有思考模式都可以总结为是链式的思考（chain of thought），
能否利用 MetaGPT 框架实现树结构的思考（tree of thought），图结构的思考（graph of thought）？
试着实现让 ai 生成树结构的动作列表，并按照树的遍历方式执行他们。

import asyncio
from enum import Enum

from metagpt.actions import Action
from metagpt.llm import LLM
from metagpt.roles.role import Role
from metagpt.logs import logger


class TraveralMode(str, Enum):
    PRE_ORDER = "pre_order"
    IN_ORDER = "in_order"
    POST_ORDER = "post_order"

    @classmethod
    def values(cls):
        return [item.value for item in cls]


class PrintAction(Action):
    """Action: Print"""

    def __init__(self, name: str = "PrintAction1", number: int = 0, context=None, llm: LLM = None):
        super().__init__(name=name, context=context, llm=llm)
        self._number = number

    async def run(self, *args, **kwargs):
        logger.info(self._number)
        return "DONE"


class MyAgent(Role):
    """Role: MyAgent"""
    # name = "MyAgent"
    # profile = "Test MetaGPT"
    # goal = "Print number"
    # constraints = "No constraints"
    # desc = "TODO"
    # is_human = False
    # traveral_mode = TraveralMode.IN_ORDER
    def __init__(self, name="MyAgent", profile="Test MetaGPT", goal="Print number",
                 constraints="No constraints", desc="TODO", is_human=False,
                 traveral_mode=TraveralMode.IN_ORDER):
        super().__init__(name=name, profile=profile, goal=goal, constraints=constraints, desc=desc, is_human=is_human)
        # [1,2,3,-1,4,5,6]
        # 创建二叉树
        #        1
        #       / \
        #      2   3
        #    / \  / \
        #  -1  4 5  6
        self.set_actions([PrintAction(number=1), PrintAction(number=2), PrintAction(number=3),
                            PrintAction(number=-1), PrintAction(number=4), PrintAction(number=5),
                            PrintAction(number=6)])
        self.rc.max_react_loop = len(self.states)
        self._plan = None
        self._i = 0
        self._traveral_mode = traveral_mode

    # async def _think(self) -> None:
    #     """Determine the next action to be taken by the role."""
    #     logger.info(f"current state={self._rc.state} state length is {len(self._states)}")

    #     if self._rc.todo is None:
    #         self._set_state(0)
    #         return

    #     if self._rc.state + 1 < len(self._states):
    #         self._set_state(self._rc.state + 1)
    #     else:
    #         self._rc.todo = None

    # 前序遍历 ：根节点 -> 左子树 -> 右子树
    def _pre_order_traversal(self, root_index: int = 0) -> list:
        _result = []
        if root_index < len(self.states) and self.actions[root_index]._number != -1:
            _result.append(root_index);
            _result.extend(self._pre_order_traversal(root_index=2 * root_index + 1))
            _result.extend(self._pre_order_traversal(root_index=2 * root_index + 2));
        return _result

    # 中序遍历 ：左子树 -> 根节点 -> 右子树
    def _in_order_traversal(self, root_index: int = 0) -> list:
        _result = []
        if root_index < len(self.states) and self.actions[root_index]._number != -1:
            _result.extend(self._in_order_traversal(root_index=2 * root_index + 1))
            _result.append(root_index);
            _result.extend(self._in_order_traversal(root_index=2 * root_index + 2));
        return _result

    # 后序遍历 ：左子树 -> 右子树 -> 根节点
    def _post_order_traversal(self, root_index: int = 0) -> list:
        _result = []
        if root_index < len(self.states) and self.actions[root_index]._number != -1:
            _result.extend(self._post_order_traversal(root_index=2 * root_index + 1))
            _result.extend(self._post_order_traversal(root_index=2 * root_index + 2));
            _result.append(root_index);
        return _result

    async def _think(self) -> None:
        """Determine the next action to be taken by the role."""
        if self._plan is None:
            logger.info(f"start plan action")
            if self._traveral_mode == TraveralMode.PRE_ORDER:
                self._plan = self._pre_order_traversal(0)
            elif self._traveral_mode == TraveralMode.IN_ORDER:
                self._plan = self._in_order_traversal(0)
            elif self._traveral_mode == TraveralMode.POST_ORDER:
                self._plan = self._post_order_traversal(0)
            numbers = []
            for i in self._plan:
                numbers.append(str(self.actions[i]._number))
            logger.info(f"plan is {'->'.join(numbers)}")
        logger.info(f"{self._i} round state={self.rc.state}")

        if self._i >= len(self._plan):
            self.rc.todo = None
        else:
            next_state = self._plan[self._i]
            self._set_state(next_state)
            self._i += 1


async def main():
    msg = "Print numbers in order"
    role = MyAgent(traveral_mode=TraveralMode.IN_ORDER)
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)


asyncio.run(main())

本文标签： 自己的定义MetaGPTAgent