超越AutoGPT：用MetaGPT构建具备自我修正能力的代码生成系统

如果你曾尝试过AutoGPT或类似的单智能体代码生成工具，可能会遇到这样的困境：生成的代码看似合理，但缺乏测试验证，更别提代码质量审查了。当项目复杂度上升时，这种“一次性生成”的局限性就暴露无遗——要么代码无法运行，要么存在隐藏的逻辑缺陷，最终仍需人工介入调试。

这正是多智能体框架MetaGPT的突破点所在。它不再依赖单一智能体完成所有工作，而是模拟真实软件开发团队的协作模式，通过角色分工、流程标准化和闭环反馈机制，构建了一个能够自我修正的代码生成系统。今天，我将带你深入这个系统的核心，看看如何利用SimpleCoder、SimpleTester和SimpleReviewer三个角色，打造一个从需求到可运行代码再到测试验证的完整工作流。

1. 从单智能体到多智能体：为什么需要角色分工？

在传统的单智能体代码生成中，我们通常面临几个关键问题：

上下文过载：单个智能体需要同时理解需求、设计架构、编写代码、考虑测试用例，这往往超出了当前大语言模型的上下文处理能力。就像让一个开发者同时扮演产品经理、架构师、工程师和测试员，每个角色的专业深度都会受到影响。

缺乏验证机制：生成的代码是否正确？边界情况是否考虑周全？单智能体系统通常没有内置的验证环节，用户需要手动测试和调试，这抵消了自动化生成的价值。

错误累积：在复杂任务中，前一步的错误会直接影响后续步骤，而单智能体缺乏自我检测和修正的能力，导致错误层层传递。

MetaGPT的多智能体架构正是为了解决这些问题而生。它通过角色专业化分工，让每个智能体专注于特定领域：

角色	核心职责	专业优势
SimpleCoder	将自然语言需求转换为Python代码	专注于代码逻辑实现，理解编程范式
SimpleTester	为生成的代码编写测试用例	专注于测试覆盖率和边界情况
SimpleReviewer	审查测试用例的质量和覆盖率	专注于代码质量和最佳实践

这种分工不仅减轻了单个智能体的认知负担，更重要的是建立了闭环反馈机制。当Reviewer发现测试用例不足时，可以要求Tester补充；当Tester发现代码逻辑问题时，可以间接影响Coder的后续输出。这种动态调整能力，是单智能体系统难以实现的。

2. 构建核心组件：定义智能体的“技能”

在MetaGPT中，每个智能体的能力由其“动作”（Action）定义。让我们从最基础的代码生成动作开始，逐步构建完整的技能体系。

2.1 代码生成动作：SimpleWriteCode

代码生成是系统的起点。我们需要设计一个能够理解自然语言指令并输出Python代码的动作。关键在于输出标准化——确保生成的代码格式统一，便于后续处理。

from metagpt.actions import Action
import re

class SimpleWriteCode(Action):
    """将自然语言指令转换为Python代码"""
    
    PROMPT_TEMPLATE: str = """
    请根据以下指令编写Python函数：
    {instruction}
    
    要求：
    1. 函数命名清晰，符合Python命名规范
    2. 包含必要的参数检查和类型提示（如果适用）
    3. 代码简洁高效，避免不必要的复杂度
    4. 在函数末尾添加简单的使用示例
    
    请将代码包裹在```python和```标记中，不要添加额外解释。
    """
    
    name: str = "SimpleWriteCode"
    
    async def run(self, instruction: str) -> str:
        """执行代码生成"""
        prompt = self.PMPT_TEMPLATE.format(instruction=instruction)
        response = await self._aask(prompt)
        
        # 从响应中提取代码块
        return self._extract_code(response)
    
    @staticmethod
    def _extract_code(response: str) -> str:
        """从模型响应中提取Python代码块"""
        pattern = r"```python\n(.*?)\n```"
        match = re.search(pattern, response, re.DOTALL)
        
        if match:
            return match.group(1).strip()
        else:
            # 如果没有找到代码块，尝试提取看起来像代码的部分
            lines = response.split('\n')
            code_lines = [line for line in lines if line.strip() and not line.strip().startswith('#')]
            return '\n'.join(code_lines)

这个动作有几个关键设计点：

1. 明确的提示模板：不仅要求生成代码，还指定了代码质量要求（命名规范、参数检查等）
2. 结构化输出：要求使用```python标记包裹代码，便于后续解析
3. 容错处理：即使模型没有严格按照格式输出，也能尝试提取代码内容

提示：在实际项目中，你可能需要根据具体需求调整提示模板。例如，如果需要生成特定框架的代码（如FastAPI、Django），可以在模板中添加相关约束。

2.2 测试生成动作：SimpleWriteTest

代码生成后，我们需要验证其正确性。测试生成动作负责为给定代码创建测试用例，重点是覆盖率和实用性。

class SimpleWriteTest(Action):
    """为Python代码生成单元测试"""
    
    PROMPT_TEMPLATE: str = """
    请为以下Python代码编写{k}个单元测试：
    
    {context}
    
    测试要求：
    1. 使用pytest框架
    2. 包含正常情况测试
    3. 包含边界情况测试（如空输入、极值等）
    4. 包含异常情况测试（如无效输入类型）
    5. 测试函数命名清晰，体现测试目的
    
    请将测试代码包裹在```python和```标记中。
    """
    
    name: str = "SimpleWriteTest"
    
    async def run(self, context: str, k: int = 5) -> str:
        """生成测试用例"""
        prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)
        response = await self._aask(prompt)
        return self._extract_code(response)
    
    @staticmethod
    def _extract_code(response: str) -> str:
        """提取测试代码（复用SimpleWriteCode的提取逻辑）"""
        return SimpleWriteCode._extract_code(response)

测试生成的关键在于多样性。通过指定测试数量（k参数）和测试类型要求，我们可以确保生成的测试覆盖不同场景：

• 正常路径测试：验证函数在预期输入下的行为
• 边界条件测试：检查输入边界和极端情况
• 异常处理测试：确保函数能妥善处理错误输入

2.3 代码审查动作：SimpleWriteReview

测试用例本身也需要质量保证。审查动作负责评估测试的完整性和质量，提供改进建议。

class SimpleWriteReview(Action):
    """审查测试用例的质量"""
    
    PROMPT_TEMPLATE: str = """
    请审查以下测试用例的质量：
    
    {context}
    
    审查要点：
    1. 测试覆盖率：是否覆盖了主要功能点和边界情况？
    2. 测试质量：测试用例是否有效？能否真正检测出代码问题？
    3. 代码规范：测试代码是否符合Python和pytest最佳实践？
    4. 改进建议：提出具体的改进建议
    
    请以结构化格式提供审查意见：
    - 覆盖率评分（1-10分）
    - 质量评分（1-10分）
    - 主要问题列表
    - 具体改进建议
    """
    
    name: str = "SimpleWriteReview"
    
    async def run(self, context: str) -> str:
        """执行代码审查"""
        prompt = self.PROMPT_TEMPLATE.format(context=context)
        return await self._aask(prompt)

审查动作的输出是结构化反馈，而不是简单的通过/失败判断。这种设计使得反馈更具操作性，能够指导测试生成动作进行改进。

3. 角色定义与协作：构建智能体团队

有了动作定义，接下来需要创建执行这些动作的角色，并建立它们之间的协作关系。

3.1 编码角色：SimpleCoder

SimpleCoder负责接收用户需求并生成代码。它的核心是观察用户输入并触发代码生成动作。

from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.actions import UserRequirement

class SimpleCoder(Role):
    """编码角色：将需求转换为代码"""
    
    name: str = "Alice"
    profile: str = "SimpleCoder"
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 观察用户需求消息
        self._watch([UserRequirement])
        # 设置可用动作
        self.set_actions([SimpleWriteCode])
    
    async def _act(self) -> Message:
        """执行编码动作"""
        # 获取最新的用户需求
        latest_message = self.get_memories(k=1)[0]
        user_instruction = latest_message.content
        
        # 执行代码生成
        todo = self.rc.todo  # 当前待执行的动作
        generated_code = await todo.run(user_instruction)
        
        # 创建输出消息
        return Message(
            content=generated_code,
            role=self.profile,
            cause_by=type(todo)
        )

SimpleCoder的设计相对简单，因为它只需要处理一种输入（用户需求）和一种输出（生成的代码）。但这里有一个重要细节：它使用_watch([UserRequirement])来订阅用户需求消息。这意味着只有当环境中出现用户需求时，SimpleCoder才会被激活。

3.2 测试角色：SimpleTester

SimpleTester的角色更复杂一些。它需要观察编码输出，并基于完整的历史上下文生成测试用例。

class SimpleTester(Role):
    """测试角色：为生成的代码创建测试"""
    
    name: str = "Bob"
    profile: str = "SimpleTester"
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 设置可用动作
        self.set_actions([SimpleWriteTest])
        # 观察编码动作的输出
        self._watch([SimpleWriteCode])
    
    async def _act(self) -> Message:
        """执行测试生成"""
        todo = self.rc.todo
        
        # 获取所有相关记忆作为上下文
        # 这里使用所有记忆，而不仅仅是最近的一条
        all_memories = self.get_memories()
        context = "\n".join([msg.content for msg in all_memories])
        
        # 生成测试用例，指定数量为5个
        test_cases = await todo.run(context, k=5)
        
        return Message(
            content=test_cases,
            role=self.profile,
            cause_by=type(todo)
        )

SimpleTester的关键设计决策：

1. 使用完整上下文：self.get_memories()获取所有历史消息，而不仅仅是最近的一条。这确保了测试生成时能考虑到完整的交互历史。
2. 指定测试数量：通过k=5参数，控制生成的测试用例数量，平衡覆盖率和效率。
3. 观察编码输出：通过_watch([SimpleWriteCode])，确保只有在代码生成完成后才触发测试生成。

3.3 审查角色：SimpleReviewer

SimpleReviewer负责质量把关。它观察测试输出并提供审查意见，形成闭环反馈。

class SimpleReviewer(Role):
    """审查角色：评估测试用例质量"""
    
    name: str = "Charlie"
    profile: str = "SimpleReviewer"
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 设置可用动作
        self.set_actions([SimpleWriteReview])
        # 观察测试动作的输出
        self._watch([SimpleWriteTest])
    
    async def _act(self) -> Message:
        """执行审查"""
        todo = self.rc.todo
        
        # 获取测试相关的上下文
        # 这里可以优化为只获取与测试相关的消息
        test_messages = [msg for msg in self.get_memories() 
                        if msg.cause_by == SimpleWriteTest]
        context = "\n".join([msg.content for msg in test_messages])
        
        # 如果找不到测试内容，使用所有记忆作为后备
        if not context:
            context = "\n".join([msg.content for msg in self.get_memories()])
        
        review_feedback = await todo.run(context)
        
        return Message(
            content=review_feedback,
            role=self.profile,
            cause_by=type(todo)
        )

审查角色的一个高级用法是启用人类介入。通过设置is_human=True，可以让真实用户代替AI执行审查：

# 在团队创建时启用人类审查
team.hire([
    SimpleCoder(),
    SimpleTester(),
    SimpleReviewer(is_human=True)  # 人类介入
])

当启用人类介入时，系统会在需要审查时暂停，等待用户输入。这在实际项目中非常有用，特别是在需要领域专家判断的场景。

4. 团队协作与流程控制

单个角色定义完成后，需要将它们组织成团队，并定义协作流程。这是MetaGPT多智能体系统的核心优势所在。

4.1 创建团队与角色雇佣

import asyncio
from metagpt.team import Team
from metagpt.logs import logger

async def run_development_team(
    idea: str = "编写一个计算列表乘积的函数",
    investment: float = 3.0,
    n_round: int = 5
):
    """运行开发团队"""
    
    logger.info(f"开始处理需求: {idea}")
    
    # 创建团队
    team = Team()
    
    # 雇佣角色
    team.hire([
        SimpleCoder(),
        SimpleTester(),
        SimpleReviewer(),
    ])
    
    # 设置投资（影响资源分配）
    team.invest(investment=investment)
    
    # 启动项目
    team.run_project(idea)
    
    # 运行指定轮数
    await team.run(n_round=n_round)
    
    # 获取最终结果
    final_messages = team.env.memory.get()
    return final_messages

if __name__ == "__main__":
    # 运行示例
    final_output = asyncio.run(run_development_team(
        idea="编写一个函数，计算斐波那契数列的第n项",
        n_round=3
    ))
    
    # 输出结果
    for i, msg in enumerate(final_output):
        print(f"\n=== 消息 {i+1} ===")
        print(f"角色: {msg.role}")
        print(f"内容:\n{msg.content[:200]}...")  # 只显示前200字符

团队运行的关键参数：

参数	说明	推荐值
idea	用户需求描述	尽可能具体明确
investment	虚拟投资金额	3.0-10.0，影响资源分配
n_round	协作轮数	3-5轮，复杂任务可增加

4.2 协作流程详解

让我们通过一个具体例子，看看三个角色如何协作完成一个任务：

第1轮：代码生成

1. 用户输入："编写一个函数，判断字符串是否为回文"
2. SimpleCoder观察到UserRequirement消息
3. SimpleCoder执行SimpleWriteCode动作，生成代码：

def is_palindrome(s: str) -> bool:
    """判断字符串是否为回文"""
    # 移除空格并转换为小写
    cleaned = ''.join(s.lower().split())
    return cleaned == cleaned[::-1]

# 示例使用
print(is_palindrome("A man a plan a canal Panama"))  # True
print(is_palindrome("hello"))  # False

第2轮：测试生成

1. SimpleTester观察到SimpleWriteCode的输出
2. SimpleTester执行SimpleWriteTest动作，生成测试：

import pytest

def test_is_palindrome_normal():
    """测试正常情况"""
    assert is_palindrome("racecar") == True
    assert is_palindrome("hello") == False

def test_is_palindrome_with_spaces():
    """测试带空格的字符串"""
    assert is_palindrome("A man a plan a canal Panama") == True

def test_is_palindrome_empty():
    """测试空字符串"""
    assert is_palindrome("") == True

def test_is_palindrome_case_insensitive():
    """测试大小写不敏感"""
    assert is_palindrome("RaceCar") == True

def test_is_palindrome_single_char():
    """测试单字符"""
    assert is_palindrome("a") == True

第3轮：测试审查

1. SimpleReviewer观察到SimpleWriteTest的输出
2. SimpleReviewer执行SimpleWriteReview动作，提供反馈：

覆盖率评分：8/10
质量评分：7/10

主要问题：
1. 缺少对非字符串输入的测试（如数字、None）
2. 缺少对包含特殊字符的字符串测试
3. 缺少性能测试（超长字符串）

改进建议：
1. 添加类型检查测试：test_is_palindrome_invalid_input()
2. 添加特殊字符测试：test_is_palindrome_special_chars()
3. 考虑添加边界测试：test_is_palindrome_very_long()

这个流程展示了系统的自我修正潜力。虽然在这个简单示例中，审查反馈没有直接触发新的测试生成，但在更复杂的配置中，我们可以让Tester观察Reviewer的输出，形成真正的闭环。

5. 高级配置与优化技巧

基础系统搭建完成后，让我们看看如何通过高级配置提升系统性能。

5.1 记忆管理优化

默认情况下，每个角色都能访问所有历史消息。但对于大型项目，这可能导致上下文过长。我们可以实现选择性记忆检索：

class OptimizedTester(SimpleTester):
    """优化版测试角色，使用选择性记忆"""
    
    async def _act(self) -> Message:
        todo = self.rc.todo
        
        # 只获取与代码相关的记忆
        code_messages = []
        for msg in self.get_memories():
            if msg.cause_by == SimpleWriteCode:
                code_messages.append(msg)
            elif "def " in msg.content or "import " in msg.content:
                # 启发式判断：包含函数定义或导入语句
                code_messages.append(msg)
        
        if not code_messages:
            # 后备：使用最近的消息
            code_messages = self.get_memories(k=1)
        
        context = "\n".join([msg.content for msg in code_messages])
        test_cases = await todo.run(context, k=5)
        
        return Message(
            content=test_cases,
            role=self.profile,
            cause_by=type(todo)
        )

选择性记忆检索的好处：

• 减少上下文长度：只保留相关信息，提高模型处理效率
• 提高相关性：避免无关信息干扰当前任务
• 可定制性：可以根据不同角色需求设计不同的检索策略

5.2 提示工程优化

动作的提示模板直接影响输出质量。以下是一些优化技巧：

1. 添加示例（Few-shot Learning）

PROMPT_TEMPLATE: str = """
请为以下Python代码编写{k}个单元测试：

{context}

示例（好的测试用例）：
```python
def test_add_positive_numbers():
    assert add(2, 3) == 5

def test_add_negative_numbers():
    assert add(-1, -1) == -2

def test_add_mixed_numbers():
    assert add(5, -3) == 2

示例（不好的测试用例）：

def test1():  # 命名不清晰
    result = add(2, 3)
    
def test_all():  # 测试多个功能，不符合单一职责
    assert add(2,3)==5
    assert subtract(5,3)==2

请按照好示例的格式编写测试。 """

**2. 使用结构化输出要求**
```python
PROMPT_TEMPLATE: str = """
请审查以下测试用例，按以下格式提供反馈：

## 覆盖率分析
- 功能覆盖：[已覆盖/部分覆盖/未覆盖]哪些功能点
- 边界覆盖：哪些边界情况已测试，哪些缺失
- 异常覆盖：异常处理是否充分

## 质量问题
- 测试独立性：是否存在测试间依赖
- 断言质量：断言是否明确、有意义
- 代码质量：是否符合Python最佳实践

## 具体建议
1. [具体建议1]
2. [具体建议2]
3. [具体建议3]

测试代码：
{context}
"""

3. 添加领域特定约束

PROMPT_TEMPLATE: str = """
请为以下数据处理函数编写测试：

{context}

特别注意：
1. 数据科学函数通常需要测试NaN值处理
2. 确保测试数据形状一致性
3. 验证数值精度（使用pytest.approx）
4. 测试大数据集性能（如果适用）

生成{k}个全面的测试用例。
"""

5.3 错误处理与重试机制

在实际运行中，可能会遇到各种错误：API调用失败、输出格式错误、逻辑矛盾等。实现健壮的错误处理至关重要。

class RobustSimpleWriteCode(SimpleWriteCode):
    """增强版代码生成动作，包含错误处理和重试"""
    
    MAX_RETRIES = 3
    
    async def run(self, instruction: str) -> str:
        """带重试机制的代码生成"""
        
        for attempt in range(self.MAX_RETRIES):
            try:
                prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
                response = await self._aask(prompt)
                
                extracted_code = self._extract_code(response)
                
                # 验证代码基本语法
                if self._validate_code_syntax(extracted_code):
                    return extracted_code
                else:
                    logger.warning(f"第{attempt+1}次尝试：生成的代码语法无效，重试中...")
                    
            except Exception as e:
                logger.error(f"第{attempt+1}次尝试失败：{str(e)}")
                
                if attempt == self.MAX_RETRIES - 1:
                    # 最后一次尝试失败，返回降级结果
                    return f"# 代码生成失败，请手动实现：{instruction}"
        
        return f"# 经过{self.MAX_RETRIES}次尝试后仍失败：{instruction}"
    
    @staticmethod
    def _validate_code_syntax(code: str) -> bool:
        """简单验证Python代码语法"""
        try:
            # 尝试编译代码（不执行）
            compile(code, '<string>', 'exec')
            return True
        except SyntaxError:
            return False

错误处理策略包括：

• 重试机制：对临时性错误（如API超时）自动重试
• 降级处理：当无法生成理想输出时，提供有用的替代方案
• 验证检查：对输出进行基本验证，确保可用性

5.4 性能监控与调优

对于生产系统，我们需要监控性能并持续优化：

import time
from typing import Dict, Any

class MonitoredRole(Role):
    """带监控功能的角色基类"""
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.metrics = {
            'total_calls': 0,
            'total_time': 0.0,
            'avg_response_time': 0.0,
            'success_rate': 1.0,
            'errors': []
        }
    
    async def _act(self) -> Message:
        """带性能监控的执行"""
        start_time = time.time()
        self.metrics['total_calls'] += 1
        
        try:
            result = await self._perform_action()
            elapsed = time.time() - start_time
            
            self.metrics['total_time'] += elapsed
            self.metrics['avg_response_time'] = (
                self.metrics['total_time'] / self.metrics['total_calls']
            )
            
            return result
            
        except Exception as e:
            self.metrics['errors'].append(str(e))
            self.metrics['success_rate'] = (
                (self.metrics['total_calls'] - len(self.metrics['errors'])) 
                / self.metrics['total_calls']
            )
            raise
    
    async def _perform_action(self) -> Message:
        """子类实现具体的动作逻辑"""
        raise NotImplementedError
    
    def get_metrics(self) -> Dict[str, Any]:
        """获取性能指标"""
        return self.metrics.copy()

监控指标可以帮助我们：

• 识别瓶颈：哪个角色或动作最耗时？
• 评估质量：成功率如何？常见错误是什么？
• 容量规划：根据平均响应时间估算系统负载能力

6. 实际应用场景与扩展

基础的三角色系统可以扩展到更复杂的应用场景。让我们看看几个实际用例。

6.1 完整软件开发流程

在真实项目中，我们可能需要更完整的角色集合：

# 扩展的角色体系
team.hire([
    ProductManager(),      # 产品经理：分析需求，写PRD
    Architect(),           # 架构师：系统设计
    SimpleCoder(),         # 工程师：编码实现
    SimpleTester(),        # 测试工程师：单元测试
    IntegrationTester(),   # 集成测试工程师
    SimpleReviewer(),      # 代码审查员
    DevOpsEngineer(),      # DevOps：部署配置
])

每个角色都有专门的职责和协作关系，形成完整的软件开发流水线。

6.2 特定领域适配

对于不同领域，我们可以定制专门的角色和动作：

数据科学项目

class DataCleaner(Role):
    """数据清洗专家"""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([CleanDataAction, ValidateDataAction])
        self._watch([DataRequirement])

class ModelTrainer(Role):
    """模型训练专家"""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SelectModelAction, TrainModelAction, EvaluateModelAction])
        self._watch([CleanDataAction])

Web开发项目

class APIDesigner(Role):
    """API设计专家"""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([DesignAPIAction, CreateSwaggerAction])
        self._watch([ProductRequirement])

class FrontendDeveloper(Role):
    """前端开发专家"""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([CreateUIAction, ImplementComponentAction])
        self._watch([APIDesignAction])

6.3 与现有工具集成

MetaGPT系统可以集成到现有开发工具链中：

class GitIntegrator(Role):
    """Git集成角色，管理代码版本"""
    
    async def _act(self) -> Message:
        # 检查是否有新生成的代码
        code_messages = [msg for msg in self.get_memories() 
                        if msg.cause_by in [SimpleWriteCode, SimpleWriteTest]]
        
        if code_messages:
            latest_code = code_messages[-1].content
            
            # 提交到Git
            commit_hash = self._git_commit(latest_code)
            
            # 创建CI/CD流水线触发消息
            return Message(
                content=f"代码已提交，提交哈希：{commit_hash}",
                role=self.profile,
                cause_by=type(self.rc.todo)
            )
    
    def _git_commit(self, code: str) -> str:
        """模拟Git提交"""
        # 实际实现中会调用Git命令
        import hashlib
        return hashlib.md5(code.encode()).hexdigest()[:8]

集成可能性包括：

• 版本控制：自动提交生成的代码
• CI/CD：触发自动化测试和部署
• 文档生成：自动生成API文档
• 监控告警：集成到运维监控系统

7. 挑战与解决方案

在实际使用MetaGPT构建自我修正系统时，可能会遇到一些挑战。以下是我在实践中总结的解决方案。

7.1 上下文管理挑战

问题：随着协作轮数增加，上下文越来越长，可能超出模型限制。

解决方案：

1. 选择性记忆：每个角色只关注相关信息
2. 摘要生成：将长上下文压缩为摘要
3. 分阶段处理：将大任务分解为独立子任务

class ContextManager:
    """上下文管理器，优化记忆使用"""
    
    @staticmethod
    def summarize_messages(messages: List[Message], max_tokens: int = 2000) -> str:
        """将消息列表压缩为摘要"""
        # 简单实现：取关键信息
        summary_parts = []
        
        for msg in messages[-10:]:  # 只考虑最近10条消息
            summary = f"{msg.role}: {msg.content[:100]}..."
            summary_parts.append(summary)
        
        return "\n".join(summary_parts)

7.2 角色协调挑战

问题：角色之间可能出现循环依赖或死锁。

解决方案：

1. 超时机制：设置最大执行时间
2. 优先级调度：定义角色执行顺序
3. 冲突解决：引入仲裁角色处理冲突

class Orchestrator(Role):
    """协调者角色，管理多角色协作"""
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.role_priorities = {
            'SimpleCoder': 1,
            'SimpleTester': 2,
            'SimpleReviewer': 3,
            'HumanReviewer': 4  # 人类介入优先级最高
        }
    
    async def _act(self) -> Message:
        """协调角色执行"""
        # 检查当前环境状态
        pending_messages = self.env.memory.get_unread()
        
        # 按优先级排序
        sorted_messages = sorted(
            pending_messages,
            key=lambda msg: self.role_priorities.get(msg.role, 99)
        )
        
        # 触发最高优先级的角色
        if sorted_messages:
            target_role = sorted_messages[0].role
            return Message(
                content=f"请{target_role}处理此消息",
                role=self.profile,
                cause_by=type(self.rc.todo)
            )

7.3 质量评估挑战

问题：如何客观评估生成代码和测试的质量？

解决方案：

1. 自动化指标：计算测试覆盖率、代码复杂度等
2. 交叉验证：使用不同模型或方法验证结果
3. 人工审核：关键节点引入人工检查

class QualityEvaluator:
    """质量评估器"""
    
    @staticmethod
    def evaluate_code_quality(code: str) -> Dict[str, float]:
        """评估代码质量"""
        metrics = {
            'complexity': QualityEvaluator._calculate_complexity(code),
            'readability': QualityEvaluator._calculate_readability(code),
            'testability': QualityEvaluator._calculate_testability(code),
            'maintainability': QualityEvaluator._calculate_maintainability(code),
        }
        return metrics
    
    @staticmethod
    def evaluate_test_quality(tests: str, code: str) -> Dict[str, float]:
        """评估测试质量"""
        metrics = {
            'coverage': QualityEvaluator._estimate_coverage(tests, code),
            'effectiveness': QualityEvaluator._estimate_effectiveness(tests),
            'maintainability': QualityEvaluator._calculate_maintainability(tests),
        }
        return metrics

构建基于MetaGPT的自我修正代码生成系统，核心在于理解多智能体协作的本质——不是简单地将任务分配给多个AI，而是建立角色之间的有机联系和反馈循环。从SimpleCoder到SimpleTester再到SimpleReviewer，每个角色都有明确的职责边界，但又通过消息传递形成完整的工作流。

在实际项目中，我发现这种架构的最大优势是可扩展性。当需要新功能时，不是修改现有角色的复杂逻辑，而是添加新的专门角色。当发现质量问题时，不是调整单个提示词，而是优化角色间的协作机制。这种模块化设计使得系统能够随着需求变化而演进，而不是推倒重来。

当然，没有任何系统是银弹。MetaGPT多智能体系统需要精心设计角色职责、提示模板和协作流程。它可能比单智能体方案需要更多的初始配置，但带来的好处是更可靠的结果、更好的可维护性和更强的适应性。对于中大型项目或需要高质量输出的场景，这种投入通常是值得的。