MetaGPT-Agent

发表于 2024-12-07 更新于 2025-01-18 分类于 2-深度学习， LLM开发工程师指南， MetaGPT 阅读次数：本文字数： 15k 阅读时长 ≈ 14 分钟

MetaGPT 是多 Agent 框架，通过定义不同的角色及其动作，组件团队去解决问题

如果使用自定义模型，在初始使用时需要在项目根目录下，新建 config 文件夹，并创建 config2.yaml 文件，假设使用 ollama 模型，写入如下模型，如果使用网页搜索、网页浏览、语音转换工具，也需要在这个文件配置 api_key 等信息

llm:
  api_type: 'ollama'
  base_url: 'http://192.168.3.155:11434/api'
  model: 'llama3.1:latest'

配置文件说明

import nest_asyncio
nest_asyncio.apply()
# 查看-修改配置
from metagpt.config2 import Config
config=Config.default()
print(config)

1
2

2024-12-13 10:48:56.569 | INFO     | metagpt.const:get_metagpt_package_root:29 - Package root set to c:\CodeRepos\python\MyCode\Learnning_MetaGPT
extra_fields=None project_path='' project_name='' inc=False reqa_file='' max_auto_summarize_code=0 git_reinit=False llm=LLMConfig(extra_fields=None, api_key='sk-', api_type=<LLMType.OLLAMA: 'ollama'>, base_url='http://192.168.3.155:11434/api', api_version=None, model='qwen2.5:latest', pricing_plan=None, access_key=None, secret_key=None, endpoint=None, app_id=None, api_secret=None, domain=None, max_token=4096, temperature=0.0, top_p=1.0, top_k=0, repetition_penalty=1.0, stop=None, presence_penalty=0.0, frequency_penalty=0.0, best_of=None, n=None, stream=False, logprobs=None, top_logprobs=None, timeout=600, proxy=None, calc_usage=True) proxy='' search=SearchConfig(extra_fields=None, api_type=<SearchEngineType.DUCK_DUCK_GO: 'ddg'>, api_key='', cse_id='', search_func=None, params={'engine': 'google', 'google_domain': 'google.com', 'gl': 'us', 'hl': 'en'}) browser=BrowserConfig(extra_fields=None, engine=<WebBrowserEngineType.PLAYWRIGHT: 'playwright'>, browser_type='chromium') mermaid=MermaidConfig(extra_fields=None, engine='nodejs', path='mmdc', puppeteer_config='', pyppeteer_path='/usr/bin/google-chrome-stable') s3=None redis=None repair_llm_output=False prompt_schema='json' workspace=WorkspaceConfig(extra_fields=None, path=WindowsPath('c:/CodeRepos/python/MyCode/Learnning_MetaGPT/workspace'), use_uid=False, uid='') enable_longterm_memory=False code_review_k_times=2 metagpt_tti_url='' language='zh_cn' redis_key='placeholder' iflytek_app_id='' iflytek_api_secret='' iflytek_api_key='' azure_tts_subscription_key='' azure_tts_region=''

创建 Agent

在 Meta 中，工具 (tool) 和角色 (role) 是核心概念，一个 role 代表一个 agent，以下代码通过创建 role，创建智能体 (agent)

import asyncio
from metagpt.context import Context
from metagpt.roles.product_manager import ProductManager
from metagpt.logs import logger
async def main():
    msg='编写贪吃蛇游戏的产品需求文档'
    context=Context()
    # print('context:\n',context)
    role=ProductManager(context=context)
    while msg:
        msg=await role.run(msg)
        logger.info(str(msg))
asyncio.run(main())

2024-12-13 10:49:04.332 | INFO     | metagpt.roles.role:_act:391 - Alice(Product Manager): to do PrepareDocuments(PrepareDocuments)
2024-12-13 10:49:04.715 | INFO     | metagpt.utils.file_repository:save:57 - save to: C:\CodeRepos\python\MyCode\Learnning_MetaGPT\workspace\20241213104904\docs\requirement.txt
2024-12-13 10:49:04.717 | INFO     | __main__:main:16 - Alice(Product Manager): {'root_path': 'docs', 'filename': 'requirement.txt', 'content': '编写贪吃蛇游戏的产品需求文档'}
2024-12-13 10:49:04.718 | INFO     | metagpt.roles.role:_act:391 - Alice(Product Manager): to do WritePRD(WritePRD)
2024-12-13 10:49:04.720 | INFO     | metagpt.actions.write_prd:run:86 - New requirement detected: 编写贪吃蛇游戏的产品需求文档
[CONTENT]
{
    "Language": "zh_cn",
    "Programming Language": "Python",
    "Original Requirements": "编写贪吃蛇游戏的产品需求文档",
    "Project Name": "snake_game",
    "Product Goals": [
        "提供流畅的游戏体验",
        "提高可访问性，确保响应迅速",
        "设计美观的用户界面"
    ],
    "User Stories": [
        "作为玩家，我希望可以选择不同的难度级别",
        "作为玩家，我希望在游戏结束后可以看到得分",
        "作为玩家，我希望在失败后可以重新开始游戏",
        "作为玩家，我希望看到一个美观且舒适的UI",
        "作为玩家，我希望可以在手机上玩这款游戏"
    ],
    "Competitive Analysis": [
        "贪吃蛇游戏A：界面简单，缺乏响应性功能",
        "SnakeGame.com: 美观且响应迅速的UI，显示最高得分",
        "SnakeOnline.net: 响应迅速的UI，显示最高得分，但有大量广告",
        "SnakeWorld.org: 丰富的关卡设计和奖励系统，但界面较为复杂",
        "SnakeChallenge.io: 高度定制化的游戏体验，但操作较难上手"
    ],
    "Competitive Quadrant Chart": "quadrantChart\n    title \"用户参与度与吸引力分析\"\n    x-axis \"低吸引力\" --> \"高吸引力\"\n    y-axis \"低参与度\" --> \"高参与度\"\n    quadrant-1 \"我们应扩大市场\"\n    quadrant-2 \"需要推广\"\n    quadrant-3 \"重新评估\"\n    quadrant-4 \"可能改进\"\n    \"游戏A\": [0.3, 0.6]\n    \"游戏B\": [0.5, 0.4]\n    \"游戏C\": [0.7, 0.8]\n    \"游戏D\": [0.2, 0.1]\n    \"游戏E\": [0.9, 0.3]\n    \"我们的目标产品\": [0.65, 0.7]",
    "Requirement Analysis": "",
    "Requirement Pool": [
        ["P0", "实现基本的游戏逻辑"],
        ["P0", "设计游戏界面和交互"],
        ["P1", "添加不同难度级别的选择"],
        ["P2", "实现得分显示功能"],
        ["P2", "加入重新开始按钮"]
    ],
    "UI Design draft": "设计简洁的界面布局，使用清新舒适的色彩
2024-12-13 10:49:13.369 | INFO     | metagpt.utils.cost_manager:update_cost:57 - Total running cost: $0.003 | Max budget: $10.000 | Current cost: $0.003, prompt_tokens: 948, completion_tokens: 546
2024-12-13 10:49:13.377 | INFO     | metagpt.utils.git_repository:rename_root:203 - Delete directory C:\CodeRepos\python\MyCode\Learnning_MetaGPT\workspace\snake_game
搭配。",
    "Anything UNCLEAR": ""
}
[/CONTENT]
2024-12-13 10:49:13.459 | WARNING  | metagpt.utils.git_repository:rename_root:214 - Move C:\CodeRepos\python\MyCode\Learnning_MetaGPT\workspace\20241213104904 to C:\CodeRepos\python\MyCode\Learnning_MetaGPT\workspace\snake_game error: [WinError 32] 另一个程序正在使用此文件，进程无法访问。: 'C:\\CodeRepos\\python\\MyCode\\Learnning_MetaGPT\\workspace\\20241213104904'
2024-12-13 10:49:13.460 | INFO     | metagpt.utils.git_repository:rename_root:219 - Rename directory C:\CodeRepos\python\MyCode\Learnning_MetaGPT\workspace\20241213104904 to C:\CodeRepos\python\MyCode\Learnning_MetaGPT\workspace\snake_game
2024-12-13 10:49:13.722 | INFO     | metagpt.utils.file_repository:save:57 - save to: C:\CodeRepos\python\MyCode\Learnning_MetaGPT\workspace\snake_game\docs\prd\20241213104913.json
2024-12-13 10:49:13.823 | WARNING  | metagpt.utils.mermaid:mermaid_to_file:35 - RUN `npm install -g @mermaid-js/mermaid-cli` to install mmdc,or consider changing engine to `playwright`, `pyppeteer`, or `ink`.
2024-12-13 10:49:13.826 | INFO     | metagpt.utils.file_repository:save:57 - save to: C:\CodeRepos\python\MyCode\Learnning_MetaGPT\workspace\snake_game\resources\prd\20241213104913.md
2024-12-13 10:49:13.827 | INFO     | __main__:main:16 - Alice(Product Manager): {'docs': {'20241213104913.json': {'root_path': 'docs\\prd', 'filename': '20241213104913.json', 'content': '{"Language":"zh_cn","Programming Language":"Python","Original Requirements":"编写贪吃蛇游戏的产品需求文档","Project Name":"snake_game","Product Goals":["提供流畅的游戏体验","提高可访问性，确保响应迅速","设计美观的用户界面"],"User Stories":["作为玩家，我希望可以选择不同的难度级别","作为玩家，我希望在游戏结束后可以看到得分","作为玩家，我希望在失败后可以重新开始游戏","作为玩家，我希望看到一个美观且舒适的UI","作为玩家，我希望可以在手机上玩这款游戏"],"Competitive Analysis":["贪吃蛇游戏A：界面简单，缺乏响应性功能","SnakeGame.com: 美观且响应迅速的UI，显示最高得分","SnakeOnline.net: 响应迅速的UI，显示最高得分，但有大量广告","SnakeWorld.org: 丰富的关卡设计和奖励系统，但界面较为复杂","SnakeChallenge.io: 高度定制化的游戏体验，但操作较难上手"],"Competitive Quadrant Chart":"quadrantChart\\n    title \\"用户参与度与吸引力分析\\"\\n    x-axis \\"低吸引力\\" --> \\"高吸引力\\"\\n    y-axis \\"低参与度\\" --> \\"高参与度\\"\\n    quadrant-1 \\"我们应扩大市场\\"\\n    quadrant-2 \\"需要推广\\"\\n    quadrant-3 \\"重新评估\\"\\n    quadrant-4 \\"可能改进\\"\\n    \\"游戏A\\": [0.3, 0.6]\\n    \\"游戏B\\": [0.5, 0.4]\\n    \\"游戏C\\": [0.7, 0.8]\\n    \\"游戏D\\": [0.2, 0.1]\\n    \\"游戏E\\": [0.9, 0.3]\\n    \\"我们的目标产品\\": [0.65, 0.7]","Requirement Analysis":"","Requirement Pool":"P0","实现基本的游戏逻辑"],["P0","设计游戏界面和交互"],["P1","添加不同难度级别的选择"],["P2","实现得分显示功能"],["P2","加入重新开始按钮","UI Design draft":"设计简洁的界面布局，使用清新舒适的色彩搭配。","Anything UNCLEAR":""}'}}}
2024-12-13 10:49:13.829 | INFO     | __main__:main:16 - None

运行后，结果以文件形式保存，保存位置：docs\prd\*. Json

为 Agent 绑定动作

单纯的智能体只能简单地回答问题，我们需要一些能力更强的 Agent，可以调用工具去完成任务，这就涉及 MetaGPT 的另一个核心概念：动作 (Action)
在 MetaGPT 中，类 Action 是动作的逻辑抽象。用户可以通过简单地调用 self._aask 函数让 LLM 拥有这个动作的能力

import nest_asyncio
nest_asyncio.apply()
import re
from metagpt.actions import Action
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

2024-12-13 09:20:49.717 | INFO | metagpt.const:get_metagpt_package_root: 29 - Package root set to c:\CodeRepos\python\MyCode\Learnning_MetaGPT

定义了动作，需要指定能执行动作的角色 (role)，除了前文通过 ProductManager 创建角色外，还可以自定义 role。类似 Action 的定义，role 也有一个抽象类，通过继承这个抽象类，并实现其中的 _act 函数

from metagpt.roles import Role
from metagpt.schema import Message
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
from metagpt.context import Context
async def main():
    msg = "编写一个函数，计算数组的和"
    context = Context()
    role = SimpleCoder(context=context)
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)
asyncio.run(main())

2024-12-13 08:22:17.427 | INFO     | __main__:main:9 - 编写一个函数，计算数组的和
   2024-12-13 08:22:17.428 | INFO     | __main__:_act:13 - Alice(SimpleCoder): to do SimpleWriteCode(SimpleWriteCode)
   ```python
   def calculate_sum(arr):
       return sum(arr)
   # Test case 1: Empty list
   print(calculate_sum([]))  # Expected output: 0
   # Test case 2: List with elements
   print(calculate_sum([1, 2, 3, 4, 5]))
   2024-12-13 08:22:18.405 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model qwen2.5:latest not found in TOKEN_COSTS.
   2024-12-13 08:22:18.406 | INFO     | __main__:main:11 - SimpleCoder: 
   def calculate_sum(arr):
       return sum(arr)
   # Test case 1: Empty list
   print(calculate_sum([]))  # Expected output: 0
   # Test case 2: List with elements
   print(calculate_sum([1, 2, 3, 4, 5]))  # Expected output: 15
     # Expected output: 15
   \```

为 Agent 绑定多个动作

以上例子，agent 只绑定一个动作，智能体对这个动作只有 2 个判断：是否选择使用工具，实际上 agent 可以更智能，能绑定多个动作，此时判断逻辑为：不执行动作 + 选择一个动作执行
前面例子已经定义了生成 Python 代码的动作，下面再为 agent 绑定运行代码的动作

import subprocess
class SimpleRunCode(Action):
    name:str="SimpleRunCode"
    async def run(self,code_text:str):
        result=subprocess.run(["python","-c",code_text],capture_output=True,text=True)
        code_result=result.stdout
        # logger.info(f'{code_result=}')
        return code_result

# 重新定义角色，绑定两个工具
from metagpt.roles import Role
from metagpt.schema import Message
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="by_order")  # 按顺序执行动作
    async def _act(self) -> Message:
        logger.info('---'*5)
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        logger.info('---'*5)
        # By choosing the Action by order under the hood
        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

import asyncio
from metagpt.context import Context
async def main():
    msg = "编写一个函数，计算数组的和"
    context = Context()
    role = RunnableCoder(context=context)
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)
asyncio.run(main())

2024-12-13 09:30:14.197 | INFO     | __main__:main:9 - 编写一个函数，计算数组的和
2024-12-13 09:31:19.557 | INFO     | __main__:_act:15 - ---------------
2024-12-13 09:31:19.559 | INFO     | __main__:_act:16 - Alice(RunnableCoder): to do SimpleWriteCode(SimpleWriteCode)
2024-12-13 09:31:19.561 | INFO     | __main__:_act:17 - ---------------
```python
def calculate_array_sum(arr):
    return sum(arr)
# Test case 1: Empty array
test_case_1 = []
print(calculate_array_sum(test_case_1))  # Expected output: 0
# Test case 2: Array with elements
test_case_2 = [1, 2, 3, 4, 5]
print(calculate_array_sum(test_case_2))  #
2024-12-13 09:31:23.256 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model qwen2.5:latest not found in TOKEN_COSTS.
2024-12-13 09:31:23.261 | INFO     | __main__:_act:15 - ---------------
2024-12-13 09:31:23.263 | INFO     | __main__:_act:16 - Alice(RunnableCoder): to do SimpleRunCode(SimpleRunCode)
2024-12-13 09:31:23.264 | INFO     | __main__:_act:17 - ---------------
 Expected output: 15
\```
2024-12-13 09:32:05.116 | INFO     | __main__:main:11 - RunnableCoder: 0
15

结果显示，role 分别调用 SimpleWriteCode、SimpleRunCode 动作，生成代码并进行运行（由于 logger 在并发编程中的问题，输出部分错乱）
总结：

Action：对应 langchian、llamaIndex 的工具，通过继承 Action 类 + 实现 run 函数，实现自定义
Role：对应 langchian、llamaIndex 的 Agent，通过继承 Role 类 + 实现_act 函数，实现自定义

Role 的执行逻辑：

通过 self. Rc. Todo 获取下一步需要执行的动作
通过 self. Get_memories (k=1)[0] 获取 Role 最近的 K 个 messages
通过 todo.Run (msg. Content) 执行动作，并获得结果

通过调试，发现 role.Run (msg) 背后逻辑如下，即按顺序调用动作

async def _act_by_order(self) -> Message:
    """switch action each time by order defined in _init_actions, i.e. _act (Action1) -> _act (Action2) -> ..."""
    start_idx = self.rc.state if self.rc.state >= 0 else 0  # action to run from recovered state
    rsp = Message(content="No actions taken yet")  # return default message if actions=[]
    for i in range(start_idx, len(self.states)):
        Self._set_state (i)
        Rsp = await self._act ()
    Return rsp  # return output from the last action

关于 Role 内，动作的执行逻辑有三个方向：

React：边思考边执行，不断循环，直到解决问题
Order：按顺序执行
Plan_and_act：先规划，然后按步骤执行

MetaGPT-Agent-20250115192633

如果设置 Role 的相应方式为：RoleReactMode. REACT，那么调用将是以上图片的逻辑，概括如下

Role 观察用户输入 (_observe)
通过 llm 选择执行工具 (_think)，
执行工具 (_act)，得到结果
Llm 判定结果是否回答用户问题，不能回答时，再次执行 2-3，重复多次，直到能回答问题
重复执行 2-3 的过程，就是所谓的 react

MetaGPT-Agent-20250115192552