使用MetaGPT 创建智能体（1）入门

metagpt一个多智能体框架

官网：MetaGPT | MetaGPT

智能体

在大模型领域，智能体通常指一种基于大语言模型（LLM）构建的自主决策系统，能够通过理解环境、规划任务、调用工具、迭代反馈等方式完成复杂目标。具备主动推理能力和多步行动策略的智能实体，简单点讲就是 智能体 = 大语言模型（LLM） + 观察 + 思考 + 行动 + 记忆

其中大模型作为大脑， 负责核心的自然语言理解、逻辑推理和生成能力，例如分析用户指令、分解任务步骤、生成代码或决策依据。

理解指令后，开始规划任务， 将复杂问题拆解为可执行的子任务 ， 常用方法：思维链（Chain-of-Thought）、思维树（Tree-of-Thoughts）等 。

执行任务的时候调用外部工具，来弥补大模型的不足，比如调用API获得实时数据。

执行任务期间可以存储相关数据，例如知识库，用户偏好等。

执行任务后获得反馈，根据反馈迭代优化结果。

例如用户要求“分析某公司近三年股价趋势并预测未来走势”，智能体的行动可能包括：
① 调用财经API获取历史数据 → ② 用Python代码绘制图表 → ③ 基于统计模型预测 → ④ 生成图文并茂的报告

思维链：大模型提示技术，目的是提升模型解决复杂问题的能力， 显式引导模型“像人类一样逐步思考”，将问题拆解为多个中间推理步骤，最终得出答案。

例子：小明有5个苹果，他吃了2个，又买了3包苹果，每包有4个。他现在有多少个苹果

• 初始苹果：5个
• 吃掉后剩余：5 - 2 = 3个
• 购买的苹果：3包 × 4个/包 = 12个
• 总数：3 + 12 = 15个

思维树： 大模型提示技术， 通过模拟人类“多角度思考→评估→决策”的认知过程，显著提升复杂问题的解决能力，并行探索多条路径，形成树状结构

多智能体

Multi-Agent System, MAS，多个自主智能体（Agent） 通过协作、竞争或混合交互完成复杂任务的分布式系统。 每个智能体具备独立感知、决策和执行能力，同时通过通信、协商或共享环境状态与其他智能体交互，最终实现全局目标。

常见交互模式有协作，竞争，混合三种。

在MetaGPT中，多智能体 = 智能体 + 环境 + 标准流程（SOP） + 通信 + 经济

其中环境可以理解为上下文，

标准流程是管理智能体行动和交互的既定程序 ，

通信时多个智能体里面多个智能体交互的过程，

经济指的是多智能体环境中的价值交换系统，决定资源分配和任务优先级 （这个经济不太理解，后面学完再来补充）

环境搭建

GitHub - geekan/MetaGPT: 🌟 The Multi-Agent Framework: First AI Software Company, Towards Natural Language Programming 先下载到本地，然后PyCharm打开，当然直接用PyCharm也行。

docs/README_CN.md就是中文文档，可以作为参考

修改config2.xml配置文件，这里使用的是ollama

llm:
  api_type: "ollama"  # or azure / ollama / groq etc.
  model: "gemma2:2b"  # or 模型名称
  base_url: "http://localhost:11434/api"  # or llm地址
  api_key: "ollama"

安装ollama，Ollama 是一个开源的本地化工具，作用是方便部署运行大模型，命令一键部署，这个的安装没太多说的，Download Ollama on macOS 官网下载，然后下一步下一步就好，安装好后运行命令启动

ollama run gemma2:2b

这里模型是gemma2，也可以使用其他的大模型，需要注意别忘了换配置文件里面的模型名称

新建一个t1.py文件，这是一个最简单的小案例，运行文件后就可以看到控制台在打印相关文件

from metagpt.software_company import generate_repo, ProjectRepo

repo: ProjectRepo = generate_repo("创建一个 2048 游戏")  # 或 ProjectRepo("<路径>")
print(repo)

运行结束后在项目的workspace文件夹下可以看到game_2048项目文件夹，里面包含相关代码文件

具体使用

定一个自己的智能体需要三步，定义活动，定义角色，运行角色，本质就是创建多个类，然后相互调用

定义活动

创建一个 Action 类的子类，活动类里面prompt，并解析返回结果

import re

from metagpt.actions import Action

class DemoAction(Action):
    # 提示词的模板
    PROMPT_TEMPLATE: str = """
    编写一个可以{instruction}的java函数，并提供两个可运行的测试用例。
    代码和使用的说明使用中文
    返回“python your_code_here”，没有其他文本， 
    您的代码：
    """

    name: str = "SimpleWriteCode"

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

        # 令 LLM 赋予这个动作能力
        rsp = await self._aask(prompt)
        # 解析返回内容
        code_text = DemoAction.parse_code(rsp)

        return code_text

    @staticmethod
    def parse_code(rsp):
        # @staticmethod 定义静态方法直接调用
        # r 定义原始字符串，忽略其中的所有转义字符
        pattern = r"```python(.*)```"
        match = re.search(pattern, rsp, re.DOTALL)
        code_text = match.group(1) if match else rsp
        return code_text

定义角色

创建一个类继承Role类，然后重写_act函数，角色调用活动， 拥有记忆、思考并采用各种策略行动

from demo.test.DemoAction import DemoAction
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message


class DemoRole(Role):
    name: str = "DemoRole"
    # profile: str = "DemoAction"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([DemoAction]) # 设置执行的活动

    # 重写_act函数， 包含智能体具体行动逻辑
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo  # self.rc.todo 待办事项，执行时会放幕后处理，最后返回一个完整消息

        msg = self.get_memories(k=1)[0]  # 查找最新消息
        code_text = await todo.run(msg.content)
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg

运行

创建角色对象，然后调用run函数运行，在这里创建一个DemoRun.py

import asyncio

from demo.test.DemoRole import DemoRole
from metagpt.context import Context
from metagpt.logs import logger


async def main():
    msg = "编写一个计算列表总和的函数"
    context = Context()
    role = DemoRole(context=context)
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)

if __name__ == '__main__':
    asyncio.run(main())

这是一个简单使用案例，参照官网案例

多个活动运行

首先再定义一个活动

import subprocess

from metagpt.actions import Action
from metagpt.logs import logger


class DemoAction2(Action):

    name: str = "DemoAction2"

    async def run(self, code_text: str):
        # 通过 Python 程序启动并控制子进程
        # ["python3", "-c", code_text] 要执行的命令
        # capture_output 是否捕获输出
        # text 以字符串形式返回输出
        result = subprocess.run(["python3", "-c", code_text], capture_output=True, text=True)
        # stdout 标准输出内容
        code_result = result.stdout
        logger.info(f"{code_result=}")
        return code_result

然后在角色中定义多个活动执行

from demo.test.DemoAction2 import DemoAction2
from demo.test.DemoAction import DemoAction
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message


class DemoRole(Role):
    name: str = "DemoRole"
    profile: str = "DemoRole"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([DemoAction, DemoAction2]) # 设置执行的活动
        self._set_react_mode(react_mode="by_order") # 按照顺序执行action

    # 重写_act函数， 包含智能体具体行动逻辑
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo  # self.rc.todo 待办事项，执行时会放幕后处理，最后返回一个完整消息

        msg = self.get_memories(k=1)[0]  # 查找最新消息
        code_text = await todo.run(msg.content)

        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        self.rc.memory.add(msg) # 添加新的消息，并更新索引

        return msg

运行代码还是原本的，运行后查看打印日志会发现程序首先执行Action1，然后会把Action1的结果传给Action2，然后Action2运行，最后返回结果，需要注意的是Action1的执行结果收大模型，最后返回的可能是一个不是特别合适的消息。