MetaGPT 深度调研报告

基于 GitHub 开源项目 FoundationAgents/MetaGPT 的深度技术调研

调研日期：2026-03-26

一、项目概述

1.1 基本信息

1.2 一句话介绍

MetaGPT 是一个多 Agent 框架，将 一行需求 作为输入，输出用户故事、竞品分析、需求文档、数据结构、API 文档等完整软件开发产物。

1.3 核心理念

Code = SOP(Team)

解释： - Team: 由不同角色的 Agent 组成的团队 - SOP: Standard Operating Procedures（标准作业流程） - Code: 团队按照 SOP 执行产出的代码

1.4 发展历程

2023 年
├── 2023-06-30: 项目创建
├── 2023-08: 发布技术论文
└── 社区快速增长

2024 年
├── ICLR 2024 论文发表
├── v0.8.0 版本发布
└── 功能持续增强

2025 年
├── 2025-02-19: 发布 MGX (MetaGPT X)
├── 2025-03-04: Product Hunt 当日第一
├── 2025-03-10: Product Hunt 本周第一
└── 2025-01-22: AFlow 论文被 ICLR 2025 接收 (Oral, top 1.8%)

2026 年
├── 持续维护更新
└── Stars 突破 66,000

二、核心架构

2.1 整体架构图

┌─────────────────────────────────────────────────────────────────────────┐
│                         MetaGPT Architecture                             │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │                    User Requirement (一行需求)                   │  │
│   └─────────────────────────────┬───────────────────────────────────┘  │
│                                 │                                       │
│                                 ▼                                       │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │                      SoftwareCompany                             │  │
│   │  ┌─────────────────────────────────────────────────────────┐   │  │
│   │  │                    Environment                           │   │  │
│   │  │  • 消息路由                                              │   │  │
│   │  │  • Agent 协调                                            │   │  │
│   │  │  • 状态管理                                              │   │  │
│   │  └─────────────────────────────────────────────────────────┘   │  │
│   │                              │                                   │  │
│   │          ┌───────────────────┼───────────────────┐              │  │
│   │          │                   │                   │              │  │
│   │          ▼                   ▼                   ▼              │  │
│   │  ┌─────────────┐     ┌─────────────┐     ┌─────────────┐      │  │
│   │  │ProductManager│ ──► │  Architect  │ ──► │ProjectManager│     │  │
│   │  │   (Alice)   │     │   (Bob)     │     │   (Eve)     │      │  │
│   │  │             │     │             │     │             │      │  │
│   │  │ WritePRD    │     │ WriteDesign │     │ WriteTasks  │      │  │
│   │  └─────────────┘     └─────────────┘     └─────────────┘      │  │
│   │                                                 │               │  │
│   │                                                 ▼               │  │
│   │                                          ┌─────────────┐       │  │
│   │                                          │  Engineer   │       │  │
│   │                                          │   (Alex)    │       │  │
│   │                                          │             │       │  │
│   │                                          │ WriteCode   │       │  │
│   │                                          └──────┬──────┘       │  │
│   │                                                 │               │  │
│   │                                                 ▼               │  │
│   │                                          ┌─────────────┐       │  │
│   │                                          │ QaEngineer  │       │  │
│   │                                          │  (Edward)   │       │  │
│   │                                          │             │       │  │
│   │                                          │ WriteTest   │       │  │
│   │                                          └─────────────┘       │  │
│   └─────────────────────────────────────────────────────────────────┘  │
│                                                                         │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │                    Shared Components                             │  │
│   │  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌───────────┐ │  │
│   │  │   Memory    │ │  Knowledge  │ │ Experience  │ │    LLM    │ │  │
│   │  │   System    │ │    Bank     │ │    Pool     │ │  Provider │ │  │
│   │  └─────────────┘ └─────────────┘ └─────────────┘ └───────────┘ │  │
│   └─────────────────────────────────────────────────────────────────┘  │
│                                                                         │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │                      Output                                      │  │
│   │  PRD → System Design → Task List → Code → Tests → Documents     │  │
│   └─────────────────────────────────────────────────────────────────┘  │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

2.2 核心抽象

Message（消息）

@dataclass
class Message:
    """Agent 通信的核心消息结构"""
    content: str                    # 消息内容
    instruct_content: BaseModel     # 结构化指令
    role: str = "user"              # 发送者角色
    cause_by: Type[Action] = UserRequirement  # 触发动作
    sent_from: str = ""             # 发送者名称
    send_to: Set[str] = field(default_factory=set)  # 接收者集合

Role（角色）

class Role:
    """Agent 角色基类"""
    name: str                       # 角色名称
    profile: str                    # 角色简介
    goal: str                       # 角色目标
    constraints: List[str]          # 约束条件

    def _watch(self, actions: List[Type[Action]]):
        """监听特定动作"""

    def _init_actions(self, actions: List[Action]):
        """初始化可执行动作"""

    async def _observe(self) -> int:
        """观察并处理消息"""

    async def _act(self) -> Message:
        """执行动作"""

Environment（环境）

class Environment:
    """管理 Agent 交互和消息路由"""

    def __init__(self, context: Context = None):
        self.roles: Dict[str, Role] = {}    # 角色映射
        self.history: List[Message] = []     # 消息历史

    def put_message(self, message: Message):
        """路由消息到接收者"""

    async def run(self, message: Message = None):
        """运行环境循环"""

2.3 项目结构

metagpt/
├── actions/              # 动作定义
│   ├── project_management.py
│   ├── write_code.py
│   ├── write_prd.py
│   └── write_test.py
├── base/                 # 基类
├── configs/              # 配置管理
├── document_store/       # 文档存储
├── environment/          # Agent 环境
├── exp_pool/             # 经验池
├── learn/                # 学习模块
├── management/           # 管理工具
├── memory/               # 记忆系统
├── prompts/              # 提示词模板
├── provider/             # LLM 提供商
├── rag/                  # RAG 检索增强
├── roles/                # Agent 角色定义
│   ├── product_manager.py
│   ├── architect.py
│   ├── project_manager.py
│   ├── engineer.py
│   ├── qa_engineer.py
│   └── ...
├── skills/               # 技能模块
├── strategy/             # 策略模式
├── tools/                # 工具集成
├── utils/                # 工具函数
└── software_company.py   # 主入口

三、Agent 角色系统

3.1 内置角色

3.2 角色定义示例

ProductManager

class ProductManager(Role):
    """产品经理 - 负责需求分析和 PRD 编写"""

    name: str = "Alice"
    profile: str = "Product Manager"
    goal: str = "高效创建全面的产品需求文档"
    constraints: List[str] = []

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._watch([UserRequirement])  # 监听用户需求
        self._init_actions([WritePRD])   # 初始化 PRD 编写动作

    async def _act(self) -> Message:
        """执行产品管理动作"""
        prd = await self._actions[0].run(self.rc.history)
        return Message(content=prd, role=self.profile, send_to={"Architect"})

Architect

class Architect(Role):
    """系统架构师 - 负责系统架构设计"""

    name: str = "Bob"
    profile: str = "Architect"
    goal: str = "设计健壮、可扩展的系统架构"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._watch([WritePRD])      # 监听 PRD 完成
        self._init_actions([WriteDesign])  # 初始化设计动作

    async def _act(self) -> Message:
        """创建系统设计"""
        design = await self._actions[0].run(self.rc.history)
        return Message(content=design, role=self.profile, send_to={"ProjectManager"})

Engineer

class Engineer(Role):
    """软件工程师 - 负责代码实现"""

    name: str = "Alex"
    profile: str = "Engineer"
    goal: str = "编写干净、高效、可维护的代码"

    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, role=self.profile, send_to={"QaEngineer"})

3.3 协作流程

用户需求 "创建一个 2048 游戏"
         │
         ▼
    ┌─────────────────┐
    │ ProductManager  │
    │    (Alice)      │
    │                 │
    │ 分析需求        │
    │ 编写 PRD        │
    └────────┬────────┘
             │ PRD 文档
             ▼
    ┌─────────────────┐
    │   Architect     │
    │     (Bob)       │
    │                 │
    │ 设计架构        │
    │ 定义组件        │
    └────────┬────────┘
             │ 系统设计
             ▼
    ┌─────────────────┐
    │ ProjectManager  │
    │     (Eve)       │
    │                 │
    │ 分解任务        │
    │ 制定计划        │
    └────────┬────────┘
             │ 任务列表
             ▼
    ┌─────────────────┐
    │    Engineer     │
    │     (Alex)      │
    │                 │
    │ 编写代码        │
    │ 实现功能        │
    └────────┬────────┘
             │ 代码
             ▼
    ┌─────────────────┐
    │   QaEngineer    │
    │    (Edward)     │
    │                 │
    │ 编写测试        │
    │ 验证功能        │
    └─────────────────┘
             │
             ▼
        完整项目

四、动作系统

4.1 Action 基类

class Action:
    """动作基类"""

    name: str = "Action"
    llm: BaseLLM = None

    def __init__(self, name: str = "", llm: BaseLLM = None, **kwargs):
        self.name = name or self.name
        self.llm = llm

    async def run(self, *args, **kwargs) -> str:
        """执行动作 - 子类实现"""
        raise NotImplementedError

    async def aask(self, prompt: str) -> str:
        """向 LLM 发送提示"""
        return await self.llm.aask(prompt)

4.2 内置动作

WritePRD

class WritePRD(Action):
    """编写产品需求文档"""
    name: str = "WritePRD"

    async def run(self, requirements: str) -> str:
        prompt = f"""
        基于以下需求：
        {requirements}

        编写完整的 PRD，包括：
        1. 产品概述
        2. 用户故事
        3. 功能需求
        4. 非功能需求
        5. UI/UX 指南
        """
        return await self.llm.aask(prompt)

WriteDesign

class WriteDesign(Action):
    """编写系统设计"""
    name: str = "WriteDesign"

    async def run(self, prd: str) -> str:
        prompt = f"""
        基于 PRD：
        {prd}

        设计系统架构，包括：
        1. 系统架构图
        2. 组件设计
        3. 数据流
        4. API 设计
        5. 技术选型
        """
        return await self.llm.aask(prompt)

WriteCode

class WriteCode(Action):
    """编写实现代码"""
    name: str = "WriteCode"

    async def run(self, design: str, tasks: str) -> str:
        prompt = f"""
        基于设计：
        {design}

        和任务列表：
        {tasks}

        实现代码：
        1. 遵循 Clean Code 原则
        2. 添加类型注解
        3. 完善注释
        4. 错误处理
        """
        return await self.llm.aask(prompt)

4.3 自定义动作

from metagpt.actions import Action
from metagpt.schema import Message

class CustomAction(Action):
    """自定义动作"""
    name: str = "CustomAction"

    async def run(self, messages: List[Message]) -> str:
        """执行自定义动作"""
        prompt = f"基于: {messages}\n执行自定义操作..."
        result = await self.llm.aask(prompt)
        return result

五、记忆系统

5.1 Memory 结构

class Memory:
    """Agent 记忆系统"""

    def __init__(self, capacity: int = 100):
        self.capacity = capacity
        self.short_term = deque(maxlen=capacity)  # 短期记忆
        self.long_term = []                        # 长期记忆
        self.working_memory = {}                   # 工作记忆

    def add(self, message: Message, important: bool = False):
        """添加消息到记忆"""
        self.short_term.append(message)
        if important:
            self.long_term.append(message)

    def get_context(self, n: int = 10) -> List[Message]:
        """获取最近上下文"""
        return list(self.short_term)[-n:]

    def search(self, query: str) -> List[Message]:
        """搜索相关记忆"""
        return [m for m in self.long_term if query in str(m)]

5.2 记忆架构

┌─────────────────────────────────────────────────────────────────┐
│                    Memory Architecture                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │              Working Memory (工作记忆)                   │   │
│  │  • 当前任务上下文                                        │   │
│  │  • 临时变量和状态                                        │   │
│  │  • 最快访问速度                                          │   │
│  └─────────────────────────────────────────────────────────┘   │
│                              │                                  │
│                              ▼                                  │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │              Short-term Memory (短期记忆)                │   │
│  │  • 最近的对话历史                                        │   │
│  │  • 有限容量 (默认 100)                                   │   │
│  │  • FIFO 淘汰策略                                         │   │
│  └─────────────────────────────────────────────────────────┘   │
│                              │                                  │
│                              ▼                                  │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │              Long-term Memory (长期记忆)                 │   │
│  │  • 重要决策和结果                                        │   │
│  │  • 成功模式记录                                          │   │
│  │  • 持久化存储                                            │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

5.3 Knowledge Bank

class KnowledgeBank:
    """知识存储库"""

    def __init__(self, path: Path = DATA_PATH):
        self.path = path
        self.documents = {}
        self.embeddings = {}

    def add_document(self, name: str, content: str):
        """添加文档到知识库"""
        self.documents[name] = content
        self.embeddings[name] = self._embed(content)

    def query(self, query: str, k: int = 5) -> List[str]:
        """查询知识库"""
        query_embedding = self._embed(query)
        scores = {name: self._similarity(query_embedding, emb)
                  for name, emb in self.embeddings.items()}
        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return [self.documents[name] for name, _ in sorted_docs[:k]]

5.4 Experience Pool

class ExperiencePool:
    """经验池"""

    def __init__(self):
        self.experiences: List[Experience] = []
        self.successful_patterns = {}
        self.failed_patterns = {}

@dataclass
class Experience:
    """单个经验记录"""
    context: str          # 上下文
    action: Action        # 执行动作
    result: str           # 结果
    success: bool         # 是否成功
    feedback: str         # 反馈

六、LLM 集成

6.1 支持的提供商

6.2 配置示例

# config.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
  top_p: 1.0

  # 速率限制
  rate_limit:
    rpm: 60      # 每分钟请求数
    tpm: 90000   # 每分钟 Token 数

# 多模型配置
llms:
  default:
    api_type: "openai"
    model: "gpt-4"

  coding:
    api_type: "openai"
    model: "gpt-4-turbo"

  analysis:
    api_type: "anthropic"
    model: "claude-3-opus"

6.3 代码配置

from metagpt.config2 import Config
from metagpt.provider import LLMFactory

# 方式 1: 使用配置文件
config = Config.from_yaml("config.yaml")
llm = LLMFactory.create(config.llm)

# 方式 2: 使用环境变量
import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
os.environ["OPENAI_API_MODEL"] = "gpt-4"

# 方式 3: 编程配置
from metagpt.provider import OpenAILLM

llm = OpenAILLM(
    api_key="your-api-key",
    model="gpt-4",
    temperature=0.7
)

七、RAG 集成

7.1 文档解析

from metagpt.rag import DocumentParser, TextSplitter

class DocumentParser:
    """解析各种文档格式"""

    SUPPORTED_FORMATS = [".pdf", ".docx", ".txt", ".md", ".html"]

    def __init__(self):
        self.parsers = {
            ".pdf": PDFParser(),
            ".docx": DocxParser(),
            ".txt": TextParser(),
            ".md": MarkdownParser(),
        }

    def parse(self, file_path: str) -> List[Document]:
        """解析文档为块"""
        ext = Path(file_path).suffix
        parser = self.parsers.get(ext)
        if parser:
            documents = parser.parse(file_path)
            return self._split_documents(documents)

7.2 RAG 引擎

class RAGEngine:
    """RAG 检索引擎"""

    def __init__(self,
                 embedding_model: str = "text-embedding-ada-002",
                 vector_store: str = "chroma"):
        self.embedder = EmbeddingModel(embedding_model)
        self.vector_store = VectorStore(vector_store)

    async def index_documents(self, documents: List[Document]):
        """索引文档"""
        for doc in documents:
            embedding = await self.embedder.embed(doc.content)
            self.vector_store.add(
                embedding=embedding,
                content=doc.content,
                metadata=doc.metadata
            )

    async def retrieve(self, query: str, k: int = 5) -> List[Document]:
        """检索相关文档"""
        query_embedding = await self.embedder.embed(query)
        return self.vector_store.search(query_embedding, k=k)

    async def augment_prompt(self, query: str, system_prompt: str) -> str:
        """增强提示词"""
        relevant_docs = await self.retrieve(query)
        context = "\n".join([doc.content for doc in relevant_docs])

        return f"""
{system_prompt}

相关上下文：
{context}

用户查询：
{query}
"""

八、与其他框架对比

8.1 vs LangChain

8.2 vs AutoGen

8.3 vs Agency Swarm

8.4 vs RuFlo

九、生产部署

9.1 安装

# 基础安装
pip install --upgrade metagpt

# 初始化配置
metagpt --init-config

9.2 Docker 部署

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: .
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./workspace:/app/workspace
    ports:
      - "8000:8000"

9.3 快速使用

# 命令行
metagpt "Create a 2048 game"

# Python
from metagpt.software_company import generate_repo

repo = generate_repo("Create a 2048 game")
print(repo)

十、学术贡献

ICLR 2024 论文

@inproceedings{hong2024metagpt,
  title={Meta{GPT}: Meta Programming for A Multi-Agent Collaborative Framework},
  author={Sirui Hong and Mingchen Zhuge and Jonathan Chen and others},
  booktitle={The Twelfth International Conference on Learning Representations},
  year={2024},
  url={https://openreview.net/forum?id=VtmBAGCN7o}
}

AFlow (ICLR 2025 Oral)

• 论文 "AFlow: Automating Agentic Workflow Generation"
• 被接收为 Oral Presentation (top 1.8%)

参考资源

• GitHub: https://github.com/FoundationAgents/MetaGPT
• 文档: https://docs.deepwisdom.ai/main/en/
• MGX 平台: https://mgx.dev/
• HuggingFace Space: https://huggingface.co/spaces/deepwisdom/MetaGPT-SoftwareCompany
• Discord: https://discord.gg/DYn29wFk9z
• Twitter: https://twitter.com/MetaGPT_

报告生成时间：2026-03-26