角色与动作（Roles and Actions）
概述
在 MetaGPT 中，Role（角色） 和 Action（动作） 是构建多智能体系统的两个核心抽象：
- Role：代表一个具有特定职责和能力的智能体
- Action：代表角色可以执行的具体任务
内置角色
MetaGPT 提供了丰富的内置角色，模拟软件开发团队的各个岗位：
ProductManager（产品经理）
负责需求分析和产品规划：
python
from metagpt.roles import ProductManager
pm = ProductManager()
# Actions: WritePRD (编写产品需求文档)输出产物：
- 用户故事
- 竞品分析
- 需求文档
Architect（架构师）
负责系统设计和技术架构：
python
from metagpt.roles import Architect
architect = Architect()
# Actions: WriteDesign (编写系统设计)输出产物：
- 系统架构设计
- API 设计
- 数据结构定义
Engineer（工程师）
负责代码实现：
python
from metagpt.roles import Engineer
engineer = Engineer()
# Actions: WriteCode (编写代码)输出产物：
- 源代码文件
- 模块实现
QaEngineer（质量工程师）
负责测试和质量保证：
python
from metagpt.roles import QaEngineer
qa = QaEngineer()
# Actions: WriteTest (编写测试)输出产物：
- 测试用例
- 测试报告
其他内置角色
| 角色 | 说明 |
|---|---|
ProjectManager | 项目经理，负责任务分解 |
TeamLeader | 团队领导，协调整体工作 |
DataAnalyst | 数据分析师，处理数据相关任务 |
Researcher | 研究员，进行信息检索和研究 |
Teacher | 教师，生成教程内容 |
Assistant | 助手，通用对话 |
自定义角色
基本结构
python
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.actions import Action
class MyRole(Role):
name: str = "MyRole"
profile: str = "A custom role description"
goal: str = "Achieve specific objectives"
constraints: str = "Follow these rules"
def __init__(self, **kwargs):
super().__init__(**kwargs)
# 设置角色可执行的动作
self.set_actions([MyAction1(), MyAction2()])
# 订阅其他角色的动作输出
self._watch([SomeOtherAction])关键属性
| 属性 | 说明 |
|---|---|
name | 角色名称 |
profile | 角色简介 |
goal | 角色目标 |
constraints | 角色约束条件 |
关键方法
| 方法 | 说明 |
|---|---|
set_actions() | 设置角色可执行的动作列表 |
_watch() | 订阅其他动作，触发自身执行 |
_act() | 执行当前动作（可重写） |
_think() | 思考下一步动作（可重写） |
_observe() | 观察环境消息（可重写） |
Action（动作）
基本结构
python
from metagpt.actions import Action
class MyAction(Action):
name: str = "MyAction"
desc: str = "Description of what this action does"
async def run(self, *args, **kwargs) -> str:
"""执行动作的主逻辑"""
# 构建提示词
prompt = self._build_prompt(*args)
# 调用 LLM
result = await self._aask(prompt)
# 处理结果
return self._parse_result(result)LLM 调用方法
python
# 简单问答
result = await self._aask(prompt)
# 带系统提示
result = await self._aask(prompt, system_msgs=["You are a helpful assistant"])
# 带历史上下文
result = await self._aask(prompt, context=previous_messages)实战示例
示例 1：简单翻译角色
python
from metagpt.roles import Role
from metagpt.actions import Action
class TranslateAction(Action):
name: str = "Translate"
PROMPT_TEMPLATE: str = """
Translate the following text to {language}:
{text}
Translation:
"""
async def run(self, text: str, language: str = "Chinese") -> str:
prompt = self.PROMPT_TEMPLATE.format(text=text, language=language)
result = await self._aask(prompt)
return result
class Translator(Role):
name: str = "Translator"
profile: str = "A professional translator"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([TranslateAction()])示例 2：代码审查角色
python
class CodeReviewAction(Action):
name: str = "CodeReview"
PROMPT_TEMPLATE: str = """
Please review the following code and provide feedback:
```{language}
{code}
```
Focus on:
1. Code quality and best practices
2. Potential bugs
3. Performance issues
4. Security concerns
Feedback:
"""
async def run(self, code: str, language: str = "python") -> str:
prompt = self.PROMPT_TEMPLATE.format(code=code, language=language)
return await self._aask(prompt)
class CodeReviewer(Role):
name: str = "CodeReviewer"
profile: str = "Senior code reviewer"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([CodeReviewAction()])
self._watch([WriteCode]) # 监听代码编写完成示例 3：多动作角色
python
class ResearchAction(Action):
name: str = "Research"
async def run(self, topic: str) -> str:
prompt = f"Research the topic: {topic}"
return await self._aask(prompt)
class WriteReportAction(Action):
name: str = "WriteReport"
async def run(self, research: str) -> str:
prompt = f"Write a report based on:\n{research}"
return await self._aask(prompt)
class Researcher(Role):
name: str = "Researcher"
profile: str = "Research and report writer"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([ResearchAction(), WriteReportAction()])
async def _act(self) -> Message:
"""自定义动作执行逻辑"""
# 先研究
research = await ResearchAction().run(self.topic)
# 再写报告
report = await WriteReportAction().run(research)
return Message(content=report, role=self.name)角色协作模式
订阅模式
角色通过 _watch() 订阅其他角色的输出：
python
class Writer(Role):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([WriteArticle()])
class Editor(Role):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([EditArticle()])
self._watch([WriteArticle]) # 等待 Writer 完成消息传递
角色通过 Message 传递信息：
python
from metagpt.schema import Message
# 发送消息
msg = Message(
content="Task completed",
role=self.name,
cause_by=type(self._rc.todo).__name__,
send_to="NextRole"
)
self.publish_message(msg)内存系统
每个角色都有自己的记忆系统：
python
# 添加记忆
self.rc.memory.add(message)
# 获取记忆
memories = self.rc.memory.get()
# 按条件检索
relevant = self.rc.memory.get_by_action(ActionType)最佳实践
1. 单一职责
每个角色应该有明确的单一职责：
python
# 好的设计
class DataCleaner(Role): ...
class DataAnalyzer(Role): ...
# 避免
class DataProcessor(Role): # 职责过多
# 清洗 + 分析 + 可视化 + ...2. 清晰的接口
Action 的输入输出应该明确：
python
class AnalyzeDataAction(Action):
async def run(self, data_path: str) -> dict:
"""
Args:
data_path: 数据文件路径
Returns:
分析结果字典，包含 summary, statistics, insights
"""
...3. 错误处理
在 Action 中处理可能的错误：
python
class MyAction(Action):
async def run(self, *args) -> str:
try:
result = await self._aask(prompt)
return self._parse_result(result)
except Exception as e:
logger.error(f"Action failed: {e}")
return f"Error: {str(e)}"4. 可测试性
编写可测试的角色和动作：
python
# 测试 Action
async def test_translate_action():
action = TranslateAction()
result = await action.run("Hello", "Chinese")
assert "你好" in result or "您好" in result