正文

MetaGPT 自定义开发完全指南

基于辩论（Debate）案例的实战教程

📚 目录

核心概念
基础组件详解
自定义开发流程
常见应用场景
完整示例解析
高级技巧

核心概念

MetaGPT 架构三要素

┌─────────────────────────────────────┐
│         Team (团队)                  │
│  ┌─────────────────────────────┐   │
│  │   Environment (环境)        │   │
│  │  ┌─────────┐  ┌─────────┐  │   │
│  │  │ Role 1  │  │ Role 2  │  │   │
│  │  │┌───────┐│  │┌───────┐│  │   │
│  │  ││Action ││  ││Action ││  │   │
│  │  │└───────┘│  │└───────┘│  │   │
│  │  └─────────┘  └─────────┘  │   │
│  └─────────────────────────────┘   │
└─────────────────────────────────────┘

1. Action（动作） - 最小执行单元

定义：智能体的具体行为，如”编写代码”、”发言”、”分析数据”
作用：封装 LLM 调用逻辑和 Prompt 模板
特点：可复用、可组合

2. Role（角色） - 智能体

定义：拥有特定能力和职责的 AI 代理
作用：决定”能做什么”（actions）和”关注什么”（watch）
特点：有记忆、可协作

3. Environment（环境） - 通信中枢

定义：角色之间的消息传递平台
作用：管理消息队列、路由消息、记录历史
特点：支持广播和定向发送

4. Team（团队） - 编排器

定义：管理多个角色的协作
作用：初始化环境、启动流程、控制轮次
特点：处理角色间依赖关系

基础组件详解

1. Action - 动作定义

基本结构

from metagpt.actions import Action

class SpeakAloud(Action):
    """自定义动作：在辩论中发言"""

    # 1. Prompt 模板（核心）
    PROMPT_TEMPLATE: str = """
    ## 背景
    假设你是 {name}，你正在与 {opponent_name} 进行辩论。

    ## 辩论历史
    {context}

    ## 你的回应
    请基于上述信息，给出你的观点...
    """

    # 2. 动作名称
    name: str = "SpeakAloud"

    # 3. 执行逻辑
    async def run(self, context: str, name: str, opponent_name: str):
        # 填充模板
        prompt = self.PROMPT_TEMPLATE.format(
            context=context,
            name=name,
            opponent_name=opponent_name
        )

        # 调用 LLM
        rsp = await self._aask(prompt)
        return rsp

关键方法

高级特性

class AdvancedAction(Action):
    """高级动作示例"""

    # 配置项
    max_tokens: int = 2000
    temperature: float = 0.7

    async def run(self, input_data: dict):
        # 1. 前置处理
        processed = self.preprocess(input_data)

        # 2. 调用 LLM
        result = await self._aask(
            prompt=processed['prompt'],
            system=processed['system']
        )

        # 3. 后处理
        output = self.postprocess(result)
        return output

    def preprocess(self, data):
        """数据预处理"""
        return {
            'prompt': f"处理 {data}",
            'system': "你是一个专业助手"
        }

    def postprocess(self, result):
        """结果后处理"""
        return result.strip()

2. Role - 角色定义

基本结构

from metagpt.roles import Role
from typing import Any

class Debator(Role):
    """辩论者角色"""

    # 1. 角色属性
    name: str = ""           # 角色名称
    profile: str = ""        # 角色简介
    opponent_name: str = ""  # 对手名称（自定义属性）

    # 2. 初始化
    def __init__(self, **data: Any):
        super().__init__(**data)

        # 配置能做什么（添加动作）
        self.set_actions([SpeakAloud])

        # 配置关注什么（监听消息类型）
        self._watch([UserRequirement, SpeakAloud])

    # 3. 观察阶段（可选重写）
    async def _observe(self) -> int:
        await super()._observe()

        # 自定义消息过滤
        self.rc.news = [
            msg for msg in self.rc.news
            if self.name in msg.send_to or "<all>" in msg.send_to
        ]
        return len(self.rc.news)

    # 4. 行动阶段（可选重写）
    async def _act(self) -> Message:
        # 获取要执行的动作
        todo = self.rc.todo  # SpeakAloud 实例

        # 获取记忆上下文
        memories = self.get_memories()
        context = "\n".join(
            f"{msg.sent_from}: {msg.content}"
            for msg in memories
        )

        # 执行动作
        rsp = await todo.run(
            context=context,
            name=self.name,
            opponent_name=self.opponent_name
        )

        # 构造消息
        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=type(todo),
            sent_from=self.name,
            send_to=self.opponent_name,
        )

        # 保存到记忆
        self.rc.memory.add(msg)

        return msg

核心方法

角色状态（rc）

class Role:
    rc: RoleContext  # 角色上下文

    # rc 包含：
    # - rc.todo: 待执行的动作
    # - rc.news: 新接收的消息列表
    # - rc.memory: 记忆库
    # - rc.state: 当前状态

3. Environment - 环境

基本用法

from metagpt.environment import Environment

# 创建环境
env = Environment()

# 发布消息到环境
env.publish_message(message)

# 角色从环境接收消息
messages = env.get_messages_for_role(role)

消息传递机制

# 1. 广播消息（所有人可见）
msg = Message(
    content="大家好",
    send_to="<all>"  # 广播标识
)

# 2. 定向消息（指定接收者）
msg = Message(
    content="你好",
    send_to="张三"  # 只有张三能收到
)

# 3. 多人消息
msg = Message(
    content="通知",
    send_to=["张三", "李四"]
)

4. Team - 团队编排

基本流程

from metagpt.team import Team

async def run_team():
    # 1. 创建角色
    role1 = MyRole(name="角色1")
    role2 = MyRole(name="角色2")

    # 2. 创建团队
    team = Team(env=Environment())

    # 3. 招募角色
    team.hire([role1, role2])

    # 4. 设置预算
    team.invest(investment=10.0)

    # 5. 启动项目
    team.run_project(
        idea="项目需求",
        send_to="角色1"  # 指定首发者
    )

    # 6. 运行 N 轮
    await team.run(n_round=5)

控制参数

await team.run(
    n_round=10,           # 最大轮数
    stop_on_no_action=True,  # 无动作时停止
)

自定义开发流程

Step 1: 定义 Action

from metagpt.actions import Action

class MyAction(Action):
    """我的自定义动作"""

    PROMPT_TEMPLATE = """
    你是 {role}，请完成以下任务：
    {task}
    """

    name: str = "MyAction"

    async def run(self, role: str, task: str):
        prompt = self.PROMPT_TEMPLATE.format(
            role=role,
            task=task
        )
        result = await self._aask(prompt)
        return result

Step 2: 定义 Role

from metagpt.roles import Role

class MyRole(Role):
    name: str = "助手"
    profile: str = "AI 助手"

    def __init__(self, **data):
        super().__init__(**data)
        self.set_actions([MyAction])
        self._watch([UserRequirement])

Step 3: 组装 Team

from metagpt.team import Team
from metagpt.environment import Environment

async def main():
    # 创建角色
    assistant = MyRole(name="小助手")

    # 创建团队
    team = Team(env=Environment())
    team.hire([assistant])
    team.invest(5.0)

    # 运行
    team.run_project("帮我写一篇文章")
    await team.run(n_round=3)

# 执行
import asyncio
asyncio.run(main())

常见应用场景

1. 多角色协作系统

应用：软件开发团队

# 角色定义
ProductManager = Role(...)  # 产品经理：撰写需求
Architect = Role(...)       # 架构师：设计架构
Developer = Role(...)       # 开发者：编写代码
Tester = Role(...)         # 测试：编写测试

# 流程
team.hire([ProductManager, Architect, Developer, Tester])
await team.run_project("开发一个博客系统")

特点：

顺序协作（产品 → 架构 → 开发 → 测试）
消息传递（前一个的输出是后一个的输入）

2. 竞争/辩论系统

应用：观点对抗

# 如您的辩论示例
Biden = Debator(name="拜登", opponent_name="特朗普")
Trump = Debator(name="特朗普", opponent_name="拜登")

team.hire([Biden, Trump])
await team.run(n_round=5)

特点：

对抗式交互
双向通信
实时回应

变种：

多方辩论（3+ 角色）
裁判角色（评判输赢）

3. 流水线处理

应用：内容生成管道

# 角色
Writer = Role(...)      # 写作者：生成初稿
Editor = Role(...)      # 编辑：润色优化
Reviewer = Role(...)    # 审核：检查质量

# 流程
team.hire([Writer, Editor, Reviewer])
await team.run_project("写一篇关于AI的文章")

特点：

单向流转
逐步精炼
质量把控

4. 专家咨询系统

应用：多专家协商

# 专家角色
LegalExpert = Role(...)    # 法律专家
FinanceExpert = Role(...)  # 财务专家
TechExpert = Role(...)     # 技术专家
Coordinator = Role(...)    # 协调者：汇总意见

team.hire([LegalExpert, FinanceExpert, TechExpert, Coordinator])
await team.run_project("评估这个商业计划")

特点：

并行咨询
跨域协作
集成决策

5. 游戏/模拟系统

应用：狼人杀游戏

Villager = Role(...)   # 村民
Werewolf = Role(...)   # 狼人
Seer = Role(...)       # 预言家
Moderator = Role(...)  # 主持人

team.hire([Villager, Werewolf, Seer, Moderator])
await team.run(n_round=10)

特点：

角色隐藏
规则约束
状态管理

完整示例解析

辩论系统剖析

# ========== 1. Action 层 ==========
class SpeakAloud(Action):
    """发言动作"""

    # Prompt 设计要点：
    # - 角色设定：{name}
    # - 对手信息：{opponent_name}
    # - 历史上下文：{context}
    # - 行为约束：100-150字、使用修辞手法

    PROMPT_TEMPLATE = """..."""

    async def run(self, context, name, opponent_name):
        # 1. 构造 Prompt
        prompt = self.PROMPT_TEMPLATE.format(...)

        # 2. 调用 LLM
        rsp = await self._aask(prompt)

        # 3. 返回结果
        return rsp


# ========== 2. Role 层 ==========
class Debator(Role):
    """辩论者"""

    # 自定义属性
    opponent_name: str = ""

    def __init__(self, **data):
        super().__init__(**data)

        # 配置1：能力（做什么）
        self.set_actions([SpeakAloud])

        # 配置2：监听（关注什么）
        self._watch([UserRequirement, SpeakAloud])
        # 意义：
        # - UserRequirement: 监听初始需求
        # - SpeakAloud: 监听对手发言

    async def _observe(self):
        """观察阶段：接收消息"""
        await super()._observe()

        # 过滤消息：只接收发给自己的
        self.rc.news = [
            msg for msg in self.rc.news
            if self.name in msg.send_to or "<all>" in msg.send_to
        ]
        return len(self.rc.news)

    async def _act(self):
        """行动阶段：发言"""
        # 1. 获取记忆
        memories = self.get_memories()
        context = "\n".join(
            f"{msg.sent_from}: {msg.content}"
            for msg in memories
        )

        # 2. 执行动作
        todo = self.rc.todo  # SpeakAloud
        rsp = await todo.run(
            context=context,
            name=self.name,
            opponent_name=self.opponent_name
        )

        # 3. 发送消息
        msg = Message(
            content=rsp,
            sent_from=self.name,
            send_to=self.opponent_name,  # 定向发送
            cause_by=type(todo)
        )
        self.rc.memory.add(msg)

        return msg


# ========== 3. Team 层 ==========
async def debate(idea, investment=3.0, n_round=5):
    # 1. 创建角色
    Biden = Debator(
        name="拜登",
        profile="民主党人",
        opponent_name="特朗普"
    )
    Trump = Debator(
        name="特朗普",
        profile="共和党人",
        opponent_name="拜登"
    )

    # 2. 创建团队和环境
    team = Team(env=Environment())
    team.hire([Biden, Trump])
    team.invest(investment)

    # 3. 启动：让拜登先发言
    team.run_project(idea, send_to="拜登")

    # 4. 运行 5 轮
    await team.run(n_round=n_round)


def main(idea: str, investment: float = 3.0, n_round: int = 10):
    """
    :param idea: 辩论议题，例如 "议题：美国应在应对气候变化方面投入更多" 或 "特朗普：气候变化是一个骗局"
    :param investment: 愿意投入的金额（Token配额等）
    :param n_round: 辩论的最大轮数
    :return:
    """

    asyncio.run(debate(idea, investment, n_round))


if __name__ == "__main__":
    fire.Fire(main)
# 使用示例：python debate.py --idea="关于人工智能是否会取代人类的辩论" --investment=3.0 --n_round=5
# ========== 执行流程 ==========
# Round 1: 拜登收到 UserRequirement → _act() → 发言 → 消息发给特朗普
# Round 2: 特朗普收到 SpeakAloud → _act() → 发言 → 消息发给拜登
# Round 3: 拜登收到 SpeakAloud → _act() → 发言 → ...
# ...循环直到 n_round

高级技巧

1. 状态管理

class StatefulRole(Role):
    current_state: str = "idle"

    async def _think(self):
        # 根据状态决定行为
        if self.current_state == "idle":
            self.rc.todo = WaitAction()
        elif self.current_state == "working":
            self.rc.todo = WorkAction()

    async def _act(self):
        result = await super()._act()
        # 更新状态
        self.current_state = "idle"
        return result

2. 动态动作选择

class SmartRole(Role):
    def __init__(self, **data):
        super().__init__(**data)
        # 配置多个动作
        self.set_actions([Action1, Action2, Action3])

    async def _think(self):
        # 根据消息内容选择动作
        latest_msg = self.rc.news[-1]

        if "写代码" in latest_msg.content:
            self.rc.todo = Action1()
        elif "测试" in latest_msg.content:
            self.rc.todo = Action2()
        else:
            self.rc.todo = Action3()

3. 记忆管理

# 获取最近 K 条记忆
recent = role.get_memories(k=5)

# 按类型过滤记忆
code_msgs = [
    msg for msg in role.get_memories()
    if msg.cause_by == WriteCode
]

# 清空记忆
role.rc.memory.clear()

4. 条件停止

class SmartTeam(Team):
    async def run(self, n_round=10):
        for i in range(n_round):
            # 执行一轮
            await super()._run_one_round()

            # 自定义停止条件
            if self.check_goal_achieved():
                print("目标达成，提前结束")
                break

    def check_goal_achieved(self):
        # 检查是否完成目标
        last_msg = self.env.get_last_message()
        return "完成" in last_msg.content

5. 并发控制

import asyncio

# 并发执行多个角色的动作
async def parallel_act(roles):
    tasks = [role._act() for role in roles]
    results = await asyncio.gather(*tasks)
    return results

配置清单

必需配置

| 配置项 | 位置 | 说明 | | — | — | — | | LLM API Key | config.yaml 或环境变量 | 调用 LLM 所需 | | Model Name | Config.default() | 使用的模型 |

可选配置

| 配置项 | 默认值 | 说明 | | — | — | — | | max_tokens | 2048 | 最大生成长度 | | temperature | 0.7 | 随机性 | | investment | 10.0 | Token 预算 |

总结

开发检查清单

定义至少一个 Action
定义至少一个 Role
在 Role 中 set_actions([...])
在 Role 中 _watch([...])
创建 Team 和 Environment
team.hire([...])
team.run_project(...) 或发送初始消息
await team.run(n_round=...)

调试技巧

# 1. 打印日志
from metagpt.logs import logger
logger.info(f"当前状态: {role.rc.state}")

# 2. 查看消息
print(f"收到消息: {role.rc.news}")

# 3. 查看记忆
print(f"历史记忆: {role.get_memories()}")

参考资源

MetaGPT 官方文档
GitHub 示例
您的辩论代码：metagpt_3way.py

本文目录

正文

MetaGPT 自定义开发完全指南

📚 目录

核心概念

MetaGPT 架构三要素

1. Action（动作） - 最小执行单元

2. Role（角色） - 智能体

3. Environment（环境） - 通信中枢

4. Team（团队） - 编排器

基础组件详解

1. Action - 动作定义

基本结构

关键方法

高级特性

2. Role - 角色定义

基本结构

核心方法

角色状态（rc）

3. Environment - 环境

基本用法

消息传递机制

4. Team - 团队编排

基本流程

控制参数

自定义开发流程

Step 1: 定义 Action

Step 2: 定义 Role

Step 3: 组装 Team

常见应用场景

1. 多角色协作系统

应用：软件开发团队

2. 竞争/辩论系统

应用：观点对抗

3. 流水线处理

应用：内容生成管道

4. 专家咨询系统

应用：多专家协商

5. 游戏/模拟系统

应用：狼人杀游戏

完整示例解析

辩论系统剖析

高级技巧

1. 状态管理

2. 动态动作选择

3. 记忆管理

4. 条件停止

5. 并发控制

配置清单

必需配置

可选配置

总结

开发检查清单

调试技巧

参考资源