MetaGPT 多角色 Agent¶

pip install metagpt

# 或从源码安装（推荐，获取最新功能）
git clone https://github.com/geekan/MetaGPT.git
cd MetaGPT
pip install -e .

# ~/.metagpt/config2.yaml
llm:
  api_type: "openai"
  model: "gpt-4"
  base_url: "https://api.openai.com/v1"
  api_key: "sk-your-key"

  # 或使用本地模型
  # api_type: "ollama"
  # model: "llama3"
  # base_url: "http://localhost:11434/v1"

# 多模型配置
llm_for_code:
  api_type: "openai"
  model: "gpt-4"
  api_key: "sk-your-key"

llm_for_planning:
  api_type: "openai"  
  model: "gpt-3.5-turbo"
  api_key: "sk-your-key"

import asyncio
from metagpt.roles import ProjectManager, Architect, Engineer
from metagpt.team import Team

async def main():
    team = Team()
    team.hire([
        ProjectManager(),
        Architect(),
        Engineer(),
    ])

    team.invest(investment=5.0)  # 设置预算上限（美元）
    team.run_project("开发一个命令行待办事项管理工具，支持增删改查和优先级排序")

    await team.run(n_round=5)

asyncio.run(main())

import asyncio
from metagpt.roles import Engineer

async def main():
    engineer = Engineer()
    result = await engineer.run("写一个 Python 快速排序实现，包含单元测试")
    print(result)

asyncio.run(main())

workspace/
├── docs/
│   ├── prd.md          # 产品需求文档
│   ├── system_design.md # 系统设计
│   └── api_spec.md     # API 规范
├── src/
│   ├── main.py
│   └── utils.py
├── tests/
│   └── test_main.py
└── requirements.txt

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

class ReviewCode(Action):
    """代码审查动作"""
    name: str = "ReviewCode"

    async def run(self, code: str) -> str:
        prompt = f"""请审查以下代码，指出问题并给出改进建议：

{code}

请从以下角度审查：
1. 代码质量和可读性
2. 潜在 bug
3. 性能问题
4. 安全隐患
"""
        return await self._aask(prompt)

class CodeReviewer(Role):
    """代码审查员角色"""
    name: str = "CodeReviewer"
    profile: str = "Senior Code Reviewer"
    goal: str = "Review code and provide improvement suggestions"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([ReviewCode])
        # 监听 Engineer 的输出
        self._watch([WriteCode])

from metagpt.team import Team
from metagpt.roles import Role

class DataAnalyst(Role):
    name: str = "DataAnalyst"
    profile: str = "Data Analyst"
    goal: str = "Analyze data and produce insights"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([CollectData, AnalyzeData, GenerateReport])

class DataTeam(Team):
    def __init__(self):
        super().__init__()
        self.hire([
            DataAnalyst(),
            Visualizer(),
            ReportWriter(),
        ])

async def run_analysis():
    team = DataTeam()
    team.run_project("分析公司上季度销售数据，找出增长最快的产品类别")
    await team.run(n_round=4)

from metagpt.roles import Role
from metagpt.utils.human_interaction import HumanInteraction

class InteractiveRole(Role):
    async def _act(self):
        # 执行任务
        result = await self.todo.run()

        # 在关键决策点请求人类确认
        if self.needs_confirmation(result):
            human_feedback = await HumanInteraction.ask(
                f"Agent 计划执行以下操作，是否确认？\n{result}"
            )
            if human_feedback == "reject":
                return await self._retry_with_feedback(human_feedback)

        return result

from metagpt.memory import Memory

class SmartRole(Role):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 配置记忆策略
        self.memory = Memory(
            max_length=20,          # 保留最近 20 条消息
            summary_threshold=15,   # 超过 15 条时开始摘要
            importance_threshold=0.7 # 重要性阈值
        )

    async def _observe(self):
        # 自定义消息过滤
        messages = self.rc.memory.get_by_role(self.profile)
        relevant = [m for m in messages if self._is_relevant(m)]
        return relevant

from metagpt.tools import Tool

class WebSearchTool(Tool):
    name = "web_search"
    description = "Search the web for information"

    async def run(self, query: str) -> str:
        # 调用搜索 API
        import aiohttp
        async with aiohttp.ClientSession() as session:
            async with session.get(
                "https://api.search.com",
                params={"q": query}
            ) as resp:
                return await resp.text()

# 在角色中使用
class Researcher(Role):
    tools = [WebSearchTool()]