就是要摸鱼

MetaGPT解读

发表于 分类于 ，
本文字数： 6.7k 阅读时长 ≈ 10 分钟
MetaGPT解读，与openManus同属团队

MetaGPT

事前准备工作

版本：v0.8.1

按照官网配置好api和model https://docs.deepwisdom.ai/main/zh/guide/get_started/configuration/tools.html

在公司运行需要改aiohttp源码，取消ssl校验

在项目目录下pip install -e . 必须执行，否则会出现改动代码导致日志行号不对

一些说明：

  • SerializeAsAny的目的是为了保证序列化的时候，根据实际类别去序列化，如果是子类，不会丢失父类的属性
  • 在role.py的_react函数中，rsp = Message(content=”No actions taken yet”, cause_by=Action) # will be overwritten after Role _act cause_by是一个str类型，为什么可以用Action这个类去传参？因为Message做了field_validator，会自动将cause_by转换为str，在项目中大量使用了Pydantic的validator功能以及一些装饰器，可以提前了解一下装饰器
  • react过程都比较统一规范，参考之前的openManus

官方的简单例子，调用链路

已有的agent

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 可导入任何角色，初始化它，用一个开始的消息运行它，完成！
import asyncio

from metagpt.roles.product_manager import ProductManager
from metagpt.logs import logger

async def main():
    msg = "Write a PRD for a snake game"
    role = ProductManager()
    result = await role.run(msg)
    logger.info(result.content[:100])

if __name__ == '__main__':
    asyncio.run(main())
role = ProductManager()
    -->role.run(msg)
        -->self.put_message(msg)
        -->self._observe()   注意这里是子类再调父类
            -->self.rc.msg_buffer.pop_all()  news
        -->self.react()   根据不同的mode去执行不同的act
            -->self._react()  rsp
                -->while actions_taken < self.rc.max_react_loop:
                    --> self._think()   
                        -->self._set_state   设置接下来调用哪个Action
                    --> self._act()          调用Action的run函数，直白点就是工具调用
                        -->self.rc.todo.run
                        -->self.rc.memory.add(msg)
        -->return rsp

ProductManager类的构造和_think

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def __init__(self, **kwargs) -> None:
    super().__init__(**kwargs)

    self.set_actions([PrepareDocuments, WritePRD])
    self._watch([UserRequirement, PrepareDocuments])
    self.todo_action = any_to_name(PrepareDocuments)

async def _think(self) -> bool:
    """Decide what to do"""
    if self.git_repo and not self.config.git_reinit:
        self._set_state(1)
    else:
        self._set_state(0)
        self.config.git_reinit = False
        self.todo_action = any_to_name(WritePRD)
    return bool(self.rc.todo)

_think的逻辑其实和构造函数是相关的，PrepareDocuments是去初始化一个目录，在workspace目录下，同时在logs下会记录日志。如果是已经创建了目录，那就把接下来的的Action设置为WritePRD，默认是PrepareDocuments。后面我们会看到，我们创建自定义Agent，其实就是需要我们①去继承Action写”工具”，提供run函数，②继承Role，去重写_act。（可以把Action理解为工具提供方，act理解为工具执行者？）

单Action

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import re
from metagpt.actions import Action
from metagpt.schema import Message
from metagpt.logs import logger

class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction} and provide two runnnable test cases.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """

    name: str = "SimpleWriteCode"

    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)

        rsp = await self._aask(prompt)

        code_text = SimpleWriteCode.parse_code(rsp)

        return code_text

    @staticmethod
    def parse_code(rsp):
        pattern = r"```python(.*)```"
        match = re.search(pattern, rsp, re.DOTALL)
        code_text = match.group(1) if match else rsp
        return code_text
    
from metagpt.roles import Role

class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "SimpleCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteCode])

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo  # todo will be SimpleWriteCode()

        msg = self.get_memories(k=1)[0]  # find the most recent messages
        code_text = await todo.run(msg.content)
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg
    
import asyncio

async def main():
    msg = "write a function that calculates the sum of a list"
    role = SimpleCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)

asyncio.run(main())

多Action

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
"""
Filename: MetaGPT/examples/build_customized_agent.py
Created Date: Tuesday, September 19th 2023, 6:52:25 pm
Author: garylin2099
"""
import asyncio
import re
import subprocess

import fire

from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message

class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction} and provide two runnable test cases.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """

    name: str = "SimpleWriteCode"

    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)

        rsp = await self._aask(prompt)

        code_text = SimpleWriteCode.parse_code(rsp)

        return code_text

    @staticmethod
    def parse_code(rsp):
        pattern = r"```python(.*)```"
        match = re.search(pattern, rsp, re.DOTALL)
        code_text = match.group(1) if match else rsp
        return code_text

class SimpleRunCode(Action):
    name: str = "SimpleRunCode"

    async def run(self, code_text: str):
        result = subprocess.run(["python3", "-c", code_text], capture_output=True, text=True)
        code_result = result.stdout
        logger.info(f"{code_result=}")
        return code_result

class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "SimpleCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteCode])

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo  # todo will be SimpleWriteCode()

        msg = self.get_memories(k=1)[0]  # find the most recent messages
        code_text = await todo.run(msg.content)
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg

class RunnableCoder(Role):
    name: str = "Alice"
    profile: str = "RunnableCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteCode, SimpleRunCode])
        self._set_react_mode(react_mode=RoleReactMode.BY_ORDER.value)

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        # By choosing the Action by order under the hood
        # todo will be first SimpleWriteCode() then SimpleRunCode()
        todo = self.rc.todo

        msg = self.get_memories(k=1)[0]  # find the most k recent messages
        result = await todo.run(msg.content)

        msg = Message(content=result, role=self.profile, cause_by=type(todo))
        self.rc.memory.add(msg)
        return msg

def main(msg="write a function that calculates the product of a list and run it"):
    # role = SimpleCoder()
    role = RunnableCoder()
    logger.info(msg)
    result = asyncio.run(role.run(msg))
    logger.info(result)

if __name__ == "__main__":
    fire.Fire(main)

多Action相比单Action除了数量的区别，Role实例化的时候会设置self._set_react_mode(react_mode=RoleReactMode.BY_ORDER.value)，这样会自动依次执行事先排列好的Action

多Agent

https://docs.deepwisdom.ai/v0.8/zh/guide/tutorials/multi_agent_101.html

原理与上面差不多，但是metaGPT不适配MCP，后面重点关注openmanus和openai-agent类似的支持MCP协议的框架