MetaGPT架构深度解析：多智能体软件公司系统设计

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def generate_repo(
    idea,                    # 用户需求描述
    investment=3.0,         # 投资预算
    n_round=5,              # 运行轮次
    code_review=True,       # 是否启用代码审查
    run_tests=False,        # 是否运行测试
    implement=True,         # 是否实现代码
    project_name="",        # 项目名称
    inc=False,              # 增量模式
    project_path="",        # 项目路径
    reqa_file="",          # 质量保证文件
    max_auto_summarize_code=0,  # 自动代码总结次数
    recover_path=None,      # 恢复路径
):
    """运行软件公司的核心逻辑，可从CLI或其他Python脚本调用"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

class Team(BaseModel):
    """
    团队：拥有一个或多个角色（智能体）、SOP（标准操作程序）和用于即时消息的环境，
    专门用于任何多智能体活动，如协作编写可执行代码。
    """
    
    env: Optional[Environment] = None      # 环境实例
    investment: float = Field(default=10.0)  # 投资金额
    idea: str = Field(default="")          # 项目想法
    use_mgx: bool = Field(default=True)    # 是否使用MGX环境

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

class Environment(ExtEnv):
    """
    环境，承载一批角色，角色可以向环境发布消息，可以被其他角色观察到
    """
    
    desc: str = Field(default="")                    # 环境描述
    roles: dict[str, SerializeAsAny[BaseRole]] = Field(default_factory=dict)  # 角色字典
    member_addrs: Dict[BaseRole, Set] = Field(default_factory=dict)  # 成员地址
    history: Memory = Field(default_factory=Memory)  # 历史记录
    context: Context = Field(default_factory=Context)  # 上下文

1
2
3
4
5
6
7

class Environment:
    def publish_message(self, message: Message, peekable: bool = True) -> bool:
        """发布消息给所有订阅的角色"""
        # 通知所有相关角色
        for role in self.roles.values():
            if self._should_notify(role, message):
                role.put_message(message)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

class BaseLLM:
    """LLM提供商的基础接口"""
    async def aask(self, prompt: str, **kwargs) -> str:
        """异步询问LLM"""
        raise NotImplementedError

class OpenAILLM(BaseLLM):
    """OpenAI实现"""
    pass

class AnthropicLLM(BaseLLM):
    """Anthropic实现"""
    pass

1
2
3
4
5
6
7
8
9

def create_role(role_type: str, **kwargs) -> BaseRole:
    """根据类型创建角色实例"""
    role_map = {
        "ProductManager": ProductManager,
        "Architect": Architect,
        "Engineer2": Engineer2,
        "DataAnalyst": DataAnalyst,
    }
    return role_map[role_type](**kwargs)

1
2
3
4
5
6
7
8

async def run(self, n_round=3, idea="", send_to="", auto_archive=True):
    """异步运行团队协作"""
    while n_round > 0:
        if self.env.is_idle:
            break
        n_round -= 1
        self._check_balance()
        await self.env.run()  # 异步执行环境运行

1
2
3
4
5
6
7

def _check_balance(self):
    """检查预算余额"""
    if self.cost_manager.total_cost >= self.cost_manager.max_budget:
        raise NoMoneyException(
            self.cost_manager.total_cost, 
            f"资金不足: {self.cost_manager.max_budget}"
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 标准软件开发环境
env = Environment(context=ctx)

# Android开发环境  
env = AndroidEnv(context=ctx)

# Minecraft游戏环境
env = MinecraftEnv(context=ctx)

# MGX增强环境
env = MGXEnv(context=ctx)

1
2
3
4
5

from metagpt.logs import logger

logger.info(f"投资金额: ${investment}")
logger.debug(f"剩余轮次: {n_round}")
logger.error("预算不足异常")