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, ProjectReporepo: ProjectRepo = generate_repo("创建一个 2048 游戏") # 或 ProjectRepo("<路径>")
print(repo)
运行结束后在项目的workspace文件夹下可以看到game_2048项目文件夹，里面包含相关代码文件
具体使用
定一个自己的智能体需要三步，定义活动，定义角色，运行角色，本质就是创建多个类，然后相互调用
定义活动
创建一个 Action 类的子类，活动类里面prompt，并解析返回结果
import refrom metagpt.actions import Actionclass 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@staticmethoddef parse_code(rsp):# @staticmethod 定义静态方法直接调用# r 定义原始字符串，忽略其中的所有转义字符pattern = r"```python(.*)```"match = re.search(pattern, rsp, re.DOTALL)code_text = match.group(1) if match else rspreturn code_text
定义角色
创建一个类继承Role类，然后重写_act函数，角色调用活动， 拥有记忆、思考并采用各种策略行动
from demo.test.DemoAction import DemoAction
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Messageclass 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 asynciofrom demo.test.DemoRole import DemoRole
from metagpt.context import Context
from metagpt.logs import loggerasync 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 subprocessfrom metagpt.actions import Action
from metagpt.logs import loggerclass 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.stdoutlogger.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 Messageclass 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的执行结果收大模型，最后返回的可能是一个不是特别合适的消息。