MetaGPT 学习教程
MetaGPT 学习教程
从零开始掌握 MetaGPT - 多 Agent 协作框架
教程日期：2026-03-26
目录
一、环境准备
1.1 前置知识
学习 MetaGPT 前建议了解：
| 知识领域 | 重要程度 | 说明 |
|---|---|---|
| Python 基础 | ⭐⭐⭐⭐⭐ | Python 3.9+ 异步编程 |
| LLM 基础 | ⭐⭐⭐⭐ | 理解大语言模型调用 |
| Agent 概念 | ⭐⭐⭐⭐ | AI Agent 基本概念 |
| 软件开发流程 | ⭐⭐⭐ | 需求、设计、编码、测试 |
| Pydantic | ⭐⭐⭐ | 数据验证 |
1.2 系统要求
Python 版本：3.9 - 3.11
操作系统：macOS / Linux / Windows
推荐配置：
├── 内存：8GB+
├── 存储：10GB+（用于项目和依赖）
└── 网络：稳定的 API 访问
1.3 安装
# 方式 1: pip 安装（推荐）
pip install --upgrade metagpt
# 方式 2: 从源码安装
git clone https://github.com/FoundationAgents/MetaGPT.git
cd MetaGPT
pip install -e .
# 方式 3: 使用 uv（更快）
uv pip install metagpt
1.4 初始化配置
# 生成默认配置文件
metagpt --init-config
# 配置文件位置
~/.metagpt/config2.yaml
1.5 配置 API Key
方式 1: 环境变量
# OpenAI
export OPENAI_API_KEY="sk-..."
export OPENAI_API_MODEL="gpt-4"
# 或其他提供商
export ANTHROPIC_API_KEY="sk-ant-..."
export DEEPSEEK_API_KEY="..."
方式 2: 配置文件
# ~/.metagpt/config2.yaml
llm:
api_type: "openai"
model: "gpt-4"
api_key: "your-api-key"
base_url: "https://api.openai.com/v1"
temperature: 0.7
max_tokens: 4096
1.6 验证安装
# test_install.py
from metagpt import UserRequirement
from metagpt.roles import ProductManager
async def test():
pm = ProductManager()
result = await pm.run("创建一个简单的计算器")
print(result)
import asyncio
asyncio.run(test())
二、快速开始
2.1 一行命令创建软件
# 使用命令行
metagpt "Create a 2048 game"
# 输出：
# workspace/
# └── 2048_game/
# ├── docs/
# │ ├── prd.md
# │ └── design.md
# ├── src/
# │ └── game.py
# └── tests/
# └── test_game.py
2.2 Python 快速使用
from metagpt.software_company import generate_repo
# 一行代码生成项目
repo = generate_repo("Create a snake game")
print(repo)
2.3 使用 SoftwareCompany
from metagpt.roles import ProductManager, Architect, Engineer
from metagpt.software_company import SoftwareCompany
async def create_software():
# 创建软件公司
company = SoftwareCompany()
# 招聘角色
company.hire([
ProductManager(),
Architect(),
Engineer()
])
# 开始项目
await company.run("Create a todo list app")
import asyncio
asyncio.run(create_software())
2.4 指定输出目录
from metagpt.software_company import SoftwareCompany
from metagpt.config2 import Config
# 配置工作目录
config = Config.default()
config.workspace.path = "./my_project"
company = SoftwareCompany(config=config)
await company.run("Create a blog system")
三、核心概念
3.1 Code = SOP(Team)
MetaGPT 的核心理念：
┌─────────────────────────────────────────────────────────────┐
│ Code = SOP(Team) │
├─────────────────────────────────────────────────────────────┤
│ │
│ Team (团队) │
│ ├── ProductManager: 分析需求，编写 PRD │
│ ├── Architect: 设计架构 │
│ ├── ProjectManager: 分解任务 │
│ ├── Engineer: 编写代码 │
│ └── QaEngineer: 测试验证 │
│ │
│ SOP (标准作业流程) │
│ ├── 需求 → PRD → 设计 → 任务 → 代码 → 测试 │
│ └── 每个角色监听上游输出，执行自己的动作 │
│ │
│ Code (输出) │
│ └── 完整的软件项目 │
│ │
└─────────────────────────────────────────────────────────────┘
3.2 Role（角色）
Role 是 Agent 的抽象，代表团队中的一个角色：
from metagpt.roles import Role
class MyRole(Role):
name: str = "MyRole" # 角色名称
profile: str = "Specialist" # 角色简介
goal: str = "完成任务" # 角色目标
constraints: List[str] = [] # 约束条件
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([SomeAction]) # 监听什么动作
self._init_actions([MyAction]) # 执行什么动作
3.3 Action（动作）
Action 是角色执行的具体操作：
from metagpt.actions import Action
class MyAction(Action):
name: str = "MyAction"
async def run(self, *args, **kwargs) -> str:
"""执行动作"""
result = await self.llm.aask("你的提示词")
return result
3.4 Message（消息）
Message 是角色间通信的载体：
from metagpt.schema import Message
message = Message(
content="消息内容",
role="ProductManager",
cause_by=WritePRD,
send_to={"Architect"}
)
3.5 Environment（环境）
Environment 管理角色交互和消息路由：
from metagpt.environment import Environment
env = Environment()
env.add_role(product_manager)
env.add_role(architect)
await env.run()
四、创建自定义角色
4.1 定义角色类
from metagpt.roles import Role
from metagpt.actions import Action
from metagpt.schema import Message
# 步骤 1: 定义动作
class AnalyzeData(Action):
"""数据分析动作"""
name: str = "AnalyzeData"
async def run(self, data: str) -> str:
prompt = f"""
分析以下数据：
{data}
提供：
1. 数据概览
2. 关键洞察
3. 建议
"""
return await self.llm.aask(prompt)
# 步骤 2: 定义角色
class DataAnalyst(Role):
"""数据分析师角色"""
name: str = "DataAnalyst"
profile: str = "Data Analyst"
goal: str = "提供专业的数据分析"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([UserRequirement]) # 监听用户需求
self._init_actions([AnalyzeData]) # 初始化动作
async def _act(self) -> Message:
"""执行动作"""
data = self.rc.history[-1].content
result = await self._actions[0].run(data)
return Message(content=result, role=self.profile)
4.2 使用自定义角色
from metagpt.environment import Environment
async def run_analyst():
# 创建角色
analyst = DataAnalyst()
# 创建环境
env = Environment()
env.add_role(analyst)
# 发送任务
await env.run(
Message(content="分析销售数据", role="user")
)
import asyncio
asyncio.run(run_analyst())
4.3 角色协作
from metagpt.roles import Role
from metagpt.actions import Action
# 定义动作
class ResearchTopic(Action):
name: str = "ResearchTopic"
async def run(self, topic: str) -> str:
return await self.llm.aask(f"研究主题: {topic}")
class WriteReport(Action):
name: str = "WriteReport"
async def run(self, research: str) -> str:
return await self.llm.aask(f"基于研究写报告: {research}")
# 定义角色
class Researcher(Role):
name: str = "Researcher"
profile: str = "Researcher"
goal: str = "深入研究主题"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([UserRequirement])
self._init_actions([ResearchTopic])
async def _act(self) -> Message:
topic = self.rc.history[-1].content
result = await self._actions[0].run(topic)
return Message(content=result, role=self.profile, send_to={"Writer"})
class Writer(Role):
name: str = "Writer"
profile: str = "Writer"
goal: str = "撰写专业报告"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([ResearchTopic])
self._init_actions([WriteReport])
async def _act(self) -> Message:
research = self.rc.history[-1].content
result = await self._actions[0].run(research)
return Message(content=result, role=self.profile)
4.4 角色目录结构
my_agent/
├── __init__.py
├── my_agent.py # 角色定义
├── actions/
│ ├── __init__.py
│ └── my_action.py # 动作定义
├── prompts/
│ └── my_prompt.py # 提示词模板
└── tests/
└── test_my_agent.py
五、动作系统
5.1 创建动作
from metagpt.actions import Action
from metagpt.schema import Message
from typing import List
class CodeReview(Action):
"""代码审查动作"""
name: str = "CodeReview"
async def run(self, code: str, language: str = "python") -> str:
prompt = f"""
审查以下 {language} 代码：
```{language}
{code}
```
提供：
1. 代码质量评分 (1-10)
2. 发现的问题
3. 改进建议
4. 安全风险
"""
return await self.llm.aask(prompt)
5.2 动作参数
from pydantic import Field
class GenerateDocs(Action):
"""生成文档动作"""
name: str = "GenerateDocs"
# 定义参数
doc_type: str = Field(
default="api",
description="文档类型：api, user, developer"
)
format: str = Field(
default="markdown",
description="输出格式：markdown, html, rst"
)
include_examples: bool = Field(
default=True,
description="是否包含示例"
)
async def run(self, code: str) -> str:
prompt = f"""
为以下代码生成 {self.doc_type} 文档：
格式：{self.format}
包含示例：{self.include_examples}
代码：
{code}
"""
return await self.llm.aask(prompt)
5.3 使用 LLM
from metagpt.actions import Action
class SmartAction(Action):
"""智能动作示例"""
async def run(self, input_text: str) -> str:
# 方式 1: 简单提问
result = await self.llm.aask(input_text)
# 方式 2: 带系统提示
result = await self.llm.aask(
input_text,
system_msgs=["你是一个专业的技术顾问"]
)
# 方式 3: 带历史上下文
result = await self.llm.aask(
input_text,
system_msgs=["你是一个助手"],
history_messages=self.rc.history
)
return result
5.4 动作链
from metagpt.actions import Action
class ActionChain:
"""动作链"""
def __init__(self, actions: List[Action]):
self.actions = actions
async def run(self, initial_input: str) -> str:
result = initial_input
for action in self.actions:
result = await action.run(result)
return result
# 使用
chain = ActionChain([
AnalyzeRequirement(),
DesignSolution(),
ImplementCode(),
WriteTests()
])
result = await chain.run("创建用户登录功能")
六、多 Agent 协作
6.1 创建团队
from metagpt.roles import ProductManager, Architect, ProjectManager, Engineer, QaEngineer
from metagpt.software_company import SoftwareCompany
async def create_team():
# 创建软件公司
company = SoftwareCompany()
# 招聘团队成员
company.hire([
ProductManager(),
Architect(),
ProjectManager(),
Engineer(),
QaEngineer()
])
return company
6.2 定义协作流程
from metagpt.roles import Role
from metagpt.actions import WritePRD, WriteDesign, WriteCode
# ProductManager 监听用户需求，执行 WritePRD
class ProductManager(Role):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([UserRequirement])
self._init_actions([WritePRD])
async def _act(self) -> Message:
prd = await self._actions[0].run(self.rc.history)
return Message(content=prd, send_to={"Architect"})
# Architect 监听 WritePRD，执行 WriteDesign
class Architect(Role):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([WritePRD])
self._init_actions([WriteDesign])
async def _act(self) -> Message:
design = await self._actions[0].run(self.rc.history)
return Message(content=design, send_to={"ProjectManager"})
# Engineer 监听任务，执行 WriteCode
class Engineer(Role):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([WriteTasks])
self._init_actions([WriteCode])
async def _act(self) -> Message:
code = await self._actions[0].run(self.rc.history)
return Message(content=code, send_to={"QaEngineer"})
6.3 协作流程图
┌─────────────────────────────────────────────────────────────────┐
│ 协作流程示例 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 用户需求: "创建一个待办事项应用" │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ ProductManager │ │
│ │ │ │
│ │ _watch: UserReq │ │
│ │ _act: WritePRD │ │
│ └────────┬────────┘ │
│ │ PRD │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Architect │ │
│ │ │ │
│ │ _watch: WritePRD│ │
│ │ _act: WriteDesign │
│ └────────┬────────┘ │
│ │ 设计文档 │
│ ▼ │
│ ┌─────────────────┐ │
│ │ ProjectManager │ │
│ │ │ │
│ │ _watch: Design │ │
│ │ _act: WriteTasks│ │
│ └────────┬────────┘ │
│ │ 任务列表 │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Engineer │ │
│ │ │ │
│ │ _watch: Tasks │ │
│ │ _act: WriteCode │ │
│ └────────┬────────┘ │
│ │ 代码 │
│ ▼ │
│ ┌─────────────────┐ │
│ │ QaEngineer │ │
│ │ │ │
│ │ _watch: Code │ │
│ │ _act: WriteTest │ │
│ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
6.4 Environment 管理
from metagpt.environment import Environment
from metagpt.schema import Message
async def run_team():
# 创建环境
env = Environment()
# 添加角色
pm = ProductManager()
arch = Architect()
eng = Engineer()
env.add_role(pm)
env.add_role(arch)
env.add_role(eng)
# 发送初始消息
await env.run(
Message(content="创建一个计算器", role="user")
)
# 获取结果
for role in env.get_roles():
print(f"{role.name}: {role.rc.history}")
七、记忆与知识库
7.1 使用 Memory
from metagpt.memory import Memory
from metagpt.schema import Message
# 创建记忆
memory = Memory(capacity=100)
# 添加消息
memory.add(Message(content="用户需求: 创建登录功能", role="user"))
memory.add(Message(content="PRD: ...", role="ProductManager"), important=True)
# 获取上下文
context = memory.get_context(n=10)
# 搜索记忆
results = memory.search("登录")
7.2 Knowledge Bank
from metagpt.const import DATA_PATH
from metagpt.knowledge import KnowledgeBank
# 创建知识库
kb = KnowledgeBank(path=DATA_PATH / "knowledge")
# 添加文档
kb.add_document(
name="api_docs",
content="API 文档内容..."
)
# 查询
results = kb.query("如何使用用户 API", k=5)
7.3 Experience Pool
from metagpt.exp_pool import ExperiencePool, Experience
# 创建经验池
pool = ExperiencePool()
# 记录经验
experience = Experience(
context="用户登录功能",
action="WriteCode",
result="成功实现 JWT 认证",
success=True,
feedback="代码质量高"
)
pool.add_experience(experience)
# 查询类似经验
similar = pool.get_similar_experiences("用户认证")
八、LLM 配置
8.1 支持的提供商
# OpenAI
config = {
"api_type": "openai",
"model": "gpt-4",
"api_key": "sk-..."
}
# Azure OpenAI
config = {
"api_type": "azure",
"model": "gpt-4",
"api_key": "...",
"base_url": "https://your-resource.openai.azure.com"
}
# Anthropic
config = {
"api_type": "anthropic",
"model": "claude-3-opus",
"api_key": "sk-ant-..."
}
# DeepSeek
config = {
"api_type": "deepseek",
"model": "deepseek-chat",
"api_key": "..."
}
# Ollama (本地)
config = {
"api_type": "ollama",
"model": "llama3",
"base_url": "http://localhost:11434"
}
8.2 配置文件
# ~/.metagpt/config2.yaml
llm:
api_type: "openai"
model: "gpt-4"
api_key: "${OPENAI_API_KEY}" # 支持环境变量
base_url: "https://api.openai.com/v1"
temperature: 0.7
max_tokens: 4096
top_p: 1.0
# 速率限制
rate_limit:
rpm: 60
tpm: 90000
# 重试配置
retry:
max_retries: 3
delay: 1.0
8.3 多模型配置
llms:
# 默认模型
default:
api_type: "openai"
model: "gpt-4"
# 编码专用
coding:
api_type: "openai"
model: "gpt-4-turbo"
# 分析专用
analysis:
api_type: "anthropic"
model: "claude-3-opus"
from metagpt.provider import LLMFactory
from metagpt.config2 import Config
config = Config.from_yaml("config.yaml")
# 使用默认模型
default_llm = LLMFactory.create(config.llm)
# 使用编码模型
coding_llm = LLMFactory.create(config.llms["coding"])
8.4 代码配置
from metagpt.config2 import Config
from metagpt.provider import OpenAILLM, AnthropicLLM
# 方式 1: 使用配置对象
config = Config.default()
config.llm.api_key = "sk-..."
config.llm.model = "gpt-4"
# 方式 2: 直接创建 LLM
llm = OpenAILLM(
api_key="sk-...",
model="gpt-4",
temperature=0.7
)
# 方式 3: 切换模型
from metagpt.roles import Role
class MyRole(Role):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# 为角色指定特定模型
self.llm = AnthropicLLM(
api_key="sk-ant-...",
model="claude-3-opus"
)
九、RAG 集成
9.1 文档解析
from metagpt.rag import DocumentParser, TextSplitter
# 创建解析器
parser = DocumentParser()
# 解析文档
documents = parser.parse("./docs/api.pdf")
# 文档分块
splitter = TextSplitter(
chunk_size=500,
chunk_overlap=50
)
chunks = splitter.split(documents)
9.2 RAG 引擎
from metagpt.rag import RAGEngine
# 创建 RAG 引擎
rag = RAGEngine(
embedding_model="text-embedding-ada-002",
vector_store="chroma"
)
# 索引文档
await rag.index_documents(chunks)
# 检索
results = await rag.retrieve("如何使用 API", k=5)
# 增强提示词
augmented = await rag.augment_prompt(
query="用户认证流程",
system_prompt="你是一个技术顾问"
)
9.3 在角色中使用 RAG
from metagpt.roles import Role
from metagpt.rag import RAGEngine
class RAGRole(Role):
"""带 RAG 能力的角色"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.rag = RAGEngine()
async def _act(self):
query = self.rc.history[-1].content
# 使用 RAG 增强上下文
augmented_prompt = await self.rag.augment_prompt(
query=query,
system_prompt="基于文档回答问题"
)
result = await self.llm.aask(augmented_prompt)
return Message(content=result, role=self.profile)
十、生产部署
10.1 Docker 部署
# Dockerfile
FROM python:3.10-slim
WORKDIR /app
# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制代码
COPY . .
# 环境变量
ENV PYTHONUNBUFFERED=1
ENV METAGPT_WORKDIR=/app/workspace
# 运行
CMD ["python", "-m", "metagpt"]
# docker-compose.yml
version: '3.8'
services:
metagpt:
build: .
container_name: metagpt
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
volumes:
- ./workspace:/app/workspace
- ./config:/app/config
ports:
- "8000:8000"
10.2 配置管理
from metagpt.config2 import Config
from pydantic import BaseSettings
class Settings(BaseSettings):
"""应用设置"""
openai_api_key: str
openai_model: str = "gpt-4"
workspace_path: str = "./workspace"
log_level: str = "INFO"
class Config:
env_prefix = "METAGPT_"
# 使用
settings = Settings()
config = Config.default()
config.llm.api_key = settings.openai_api_key
10.3 日志配置
import logging
from metagpt.logs import logger
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# 使用
logger.info("开始执行任务")
logger.debug(f"处理消息: {message}")
logger.error(f"执行失败: {error}")
10.4 监控和指标
from metagpt.metrics import TokenMonitor, ExecutionTimer
# Token 监控
token_monitor = TokenMonitor(max_tokens_per_day=1000000)
if token_monitor.check_limit(1000):
result = await llm.aask(prompt)
token_monitor.record_usage(1000)
# 执行计时
timer = ExecutionTimer()
with timer.measure("code_generation"):
code = await engineer.run(task)
print(f"耗时: {timer.get_duration('code_generation')}s")
十一、最佳实践
11.1 角色设计
# 好的做法：单一职责
class CodeReviewer(Role):
"""只负责代码审查"""
name: str = "CodeReviewer"
goal: str = "发现代码问题和改进点"
# 不好的做法：职责过多
class SuperRole(Role):
"""负责所有事情"""
# 不要这样做
11.2 提示词设计
# 好的做法：结构化提示词
PROMPT_TEMPLATE = """
你是一个专业的 {role}。
任务：{task}
输入：
{input}
要求：
1. {requirement_1}
2. {requirement_2}
输出格式：
{output_format}
"""
# 不好的做法：模糊提示词
prompt = "帮我写代码"
11.3 错误处理
from metagpt.exceptions import LLMError, AgentError
import asyncio
async def safe_run(agent, task):
"""安全执行"""
try:
result = await agent.run(task)
return result
except LLMError as e:
print(f"LLM 错误: {e}")
# 重试逻辑
for _ in range(3):
await asyncio.sleep(1)
try:
return await agent.run(task)
except:
continue
except AgentError as e:
print(f"Agent 错误: {e}")
return None
11.4 资源管理
from metagpt.utils import ResourceMonitor
# 监控资源使用
monitor = ResourceMonitor()
# 检查内存
if monitor.memory_usage() > 0.8:
print("内存使用过高")
# 清理记忆
agent.memory.clear()
# 检查 Token 使用
if monitor.token_usage() > 0.9:
print("Token 配额即将耗尽")
十二、常见问题
12.1 安装问题
Q: 安装失败，提示依赖冲突
# 使用虚拟环境
python -m venv venv
source venv/bin/activate
pip install metagpt
# 或使用 uv
uv venv
source .venv/bin/activate
uv pip install metagpt
Q: API Key 配置无效
# 检查环境变量
echo $OPENAI_API_KEY
# 或检查配置文件
cat ~/.metagpt/config2.yaml
12.2 运行问题
Q: Agent 不响应
# 检查是否正确设置了监听
class MyRole(Role):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([UserRequirement]) # 确保设置了监听
Q: 输出格式不符合预期
# 使用结构化输出
from pydantic import BaseModel
class OutputFormat(BaseModel):
title: str
content: str
tags: List[str]
class MyAction(Action):
async def run(self, input_text: str) -> OutputFormat:
result = await self.llm.aask(input_text)
return OutputFormat.parse_raw(result)
12.3 性能问题
Q: 执行速度慢
# 并行执行
import asyncio
async def parallel_run():
tasks = [agent.run(f"任务{i}") for i in range(5)]
results = await asyncio.gather(*tasks)
return results
# 使用更快的模型
config.llm.model = "gpt-4-turbo"
Q: Token 消耗过高
# 减少历史上下文
memory.get_context(n=5) # 只保留最近 5 条
# 使用更短的提示词
# 使用更便宜的模型处理简单任务
参考资源
官方资源
- GitHub: https://github.com/FoundationAgents/MetaGPT
- 文档: https://docs.deepwisdom.ai/main/en/
- MGX 平台: https://mgx.dev/
社区资源
- Discord: https://discord.gg/DYn29wFk9z
- Twitter: https://twitter.com/MetaGPT_
- HuggingFace: https://huggingface.co/spaces/deepwisdom/MetaGPT-SoftwareCompany
学术论文
- ICLR 2024: MetaGPT: Meta Programming for A Multi-Agent Collaborative Framework
- ICLR 2025 Oral: AFlow: Automating Agentic Workflow Generation
教程生成时间：2026-03-26