1. WriteTasks、ProjectManager类

from metagpt.schema import Message
from metagpt.logs import logger
from metagpt.roles import ProjectManager
from tests.metagpt.roles.mock import MockMessages


from metagpt.actions import WriteTasks
from metagpt.actions.design_api import WriteDesign
from metagpt.roles.role import Role


import json
from typing import Optional

from metagpt.actions.action import Action
from metagpt.actions.action_output import ActionOutput
from metagpt.actions.project_management_an import PM_NODE, REFINED_PM_NODE
from metagpt.const import PACKAGE_REQUIREMENTS_FILENAME
from metagpt.logs import logger
from metagpt.schema import Document, Documents

NEW_REQ_TEMPLATE = """
### Legacy Content
{old_task}

### New Requirements
{context}
"""


class WriteTasks(Action):
    name: str = "CreateTasks"
    i_context: Optional[str] = None

    async def run(self, with_messages):
        changed_system_designs = self.repo.docs.system_design.changed_files
        changed_tasks = self.repo.docs.task.changed_files
        change_files = Documents()
        # Rewrite the system designs that have undergone changes based on the git head diff under
        # `docs/system_designs/`.
        for filename in changed_system_designs:
            task_doc = await self._update_tasks(filename=filename)
            change_files.docs[filename] = task_doc

        # Rewrite the task files that have undergone changes based on the git head diff under `docs/tasks/`.
        for filename in changed_tasks:
            if filename in change_files.docs:
                continue
            task_doc = await self._update_tasks(filename=filename)
            change_files.docs[filename] = task_doc

        if not change_files.docs:
            logger.info("Nothing has changed.")
        # Wait until all files under `docs/tasks/` are processed before sending the publish_message, leaving room for
        # global optimization in subsequent steps.
        return ActionOutput(content=change_files.model_dump_json(), instruct_content=change_files)

    async def _update_tasks(self, filename):
        system_design_doc = await self.repo.docs.system_design.get(filename)
        task_doc = await self.repo.docs.task.get(filename)
        if task_doc:
            task_doc = await self._merge(system_design_doc=system_design_doc, task_doc=task_doc)
            await self.repo.docs.task.save_doc(doc=task_doc, dependencies={system_design_doc.root_relative_path})
        else:
            rsp = await self._run_new_tasks(context=system_design_doc.content)
            task_doc = await self.repo.docs.task.save(
                filename=filename,
                content=rsp.instruct_content.model_dump_json(),
                dependencies={system_design_doc.root_relative_path},
            )
        await self._update_requirements(task_doc)
        return task_doc

    async def _run_new_tasks(self, context):
        node = await PM_NODE.fill(context, self.llm, schema=self.prompt_schema)
        return node

    async def _merge(self, system_design_doc, task_doc) -> Document:
        context = NEW_REQ_TEMPLATE.format(context=system_design_doc.content, old_task=task_doc.content)
        node = await REFINED_PM_NODE.fill(context, self.llm, schema=self.prompt_schema)
        task_doc.content = node.instruct_content.model_dump_json()
        return task_doc

    async def _update_requirements(self, doc):
        m = json.loads(doc.content)
        packages = set(m.get("Required packages", set()))
        requirement_doc = await self.repo.get(filename=PACKAGE_REQUIREMENTS_FILENAME)
        if not requirement_doc:
            requirement_doc = Document(filename=PACKAGE_REQUIREMENTS_FILENAME, root_path=".", content="")
        lines = requirement_doc.content.splitlines()
        for pkg in lines:
            if pkg == "":
                continue
            packages.add(pkg)
        await self.repo.save(filename=PACKAGE_REQUIREMENTS_FILENAME, content="\n".join(packages))


class ProjectManager(Role):
    """
    Represents a Project Manager role responsible for overseeing project execution and team efficiency.

    Attributes:
        name (str): Name of the project manager.
        profile (str): Role profile, default is 'Project Manager'.
        goal (str): Goal of the project manager.
        constraints (str): Constraints or limitations for the project manager.
    """

    name: str = "Eve"
    profile: str = "Project Manager"
    goal: str = (
        "break down tasks according to PRD/technical design, generate a task list, and analyze task "
        "dependencies to start with the prerequisite modules"
    )
    constraints: str = "use same language as user requirement"

    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)

        self.set_actions([WriteTasks])
        self._watch([WriteDesign])

2. 数据准备

from metagpt.utils.git_repository import GitRepository
from metagpt.utils.project_repo import ProjectRepo
from metagpt.const import DEFAULT_WORKSPACE_ROOT
from metagpt.context import Context
import uuid

ctx = Context()
ctx.git_repo = GitRepository(local_path=DEFAULT_WORKSPACE_ROOT / f"unittest/{uuid.uuid4().hex}")
ctx.repo = ProjectRepo(ctx.git_repo)

from metagpt.utils.common import any_to_str, awrite

PRDS_FILE_REPO = "docs/prd"

PRD = {
	"Language": "en_us",
	"Programming Language": "Python",
	"Original Requirements": "开发一个贪吃蛇游戏",
	"Project Name": "snake_game",
	"Product Goals": ["Create an engaging and intuitive user experience", "Ensure the game is scalable and performs well on various devices", "Implement a high-quality UI/UX design"],
	"User Stories": ["As a player, I want to easily navigate the game controls to play the game", "As a player, I want to see my score and high scores displayed clearly on the screen", "As a player, I want the ability to pause and resume the game at any time", "As a player, I want to have the option to restart the game from the beginning", "As a player, I want the game to be visually appealing and responsive on different screen sizes"],
	"Competitive Analysis": ["Snake Game A: Basic gameplay, lacks advanced features and customization", "Snake Game B: Offers a variety of themes and power-ups, but can be slow on older devices", "Snake Game C: Features a simple and clean UI, but lacks multiplayer functionality"],
	"Competitive Quadrant Chart": "quadrantChart\n    title \"Performance and User Engagement\"\n    x-axis \"Low Performance\" --> \"High Performance\"\n    y-axis \"Low Engagement\" --> \"High Engagement\"\n    quadrant-1 \"We should expand\"\n    quadrant-2 \"Need to promote\"\n    quadrant-3 \"Re-evaluate\"\n    quadrant-4 \"May be improved\"\n    \"Game A\": [0.2, 0.4]\n    \"Game B\": [0.5, 0.6]\n    \"Game C\": [0.3, 0.5]\n    \"Our Target Product\": [0.7, 0.7]",
	"Requirement Analysis": "The game should be designed to be accessible to players of all skill levels, with a focus on ease of use and a high-quality visual experience. The game should also be optimized for performance on a range of devices, from low-end to high-end.",
	"Requirement Pool": [
		["P0", "Develop the core gameplay logic for the snake movement and food generation"],
		["P0", "Implement a user-friendly interface with clear score tracking and game controls"],
		["P1", "Add features such as pause, resume, and restart functionality"],
		["P1", "Optimize the game for performance on various devices"],
		["P2", "Design and implement a high-quality UI/UX"]
	],
	"UI Design draft": "A simple and intuitive UI with a clear score display, easy-to-use controls, and a responsive design that adapts to different screen sizes.",
	"Anything UNCLEAR": "It is unclear whether there are specific design preferences or branding requirements for the game."
}


filename = uuid.uuid4().hex + ".json"

json_data = json.dumps(PRD, ensure_ascii=False, indent=4)
await awrite(ctx.repo.workdir / PRDS_FILE_REPO / filename, data=f"{json_data}")

SYSTEM_DESIGN_FILE_REPO = "docs/system_design"

SYSTEM_DESIGN = {
	"Implementation approach": "To create a concise, usable, and complete software system for the snake_game, we will use Python with the following open-source libraries: Pygame for game development, Flask for a simple web server if we want to deploy it online, and Pillow for image handling. The architecture will be modular, separating the game logic, UI/UX design, and server-side code to ensure scalability and maintainability.",
	"File list": ["main.py", "game.py", "ui.py", "server.py"],
	"Data structures and interfaces": "\nclassDiagram\n    class Game {\n        -score int\n        -game_over bool\n        +start_game() void\n        +update_game() void\n        +handle_input() void\n        +render() void\n    }\n    class UI {\n        -score_display str\n        -high_score_display str\n        +update_score(score: int) void\n        +update_high_score(high_score: int) void\n        +render_ui() void\n    }\n    class Server {\n        -game_state dict\n        +start_server() void\n        +handle_client_requests() void\n        +send_game_state() void\n    }\n    Game --> UI\n    Server --> Game\n",
	"Program call flow": "\nsequenceDiagram\n    participant M as Main\n    participant G as Game\n    participant U as UI\n    participant S as Server\n    M->>G: start_game()\n    G->>U: update_score(score)\n    G->>U: update_high_score(high_score)\n    G->>U: render_ui()\n    M->>S: start_server()\n    S->>G: handle_client_requests()\n    G->>S: send_game_state()\n",
	"Anything UNCLEAR": "It is unclear whether the game should be a standalone application or a web-based game. If it is a web-based game, we need to decide on the front-end technology to use."
}

filename = uuid.uuid4().hex + ".json"
json_data = json.dumps(SYSTEM_DESIGN, ensure_ascii=False, indent=4)
await awrite(ctx.repo.workdir / SYSTEM_DESIGN_FILE_REPO / filename, data=f"json_data")

3. 代码运行

system_design = Message(role="Architect", content=f"{SYSTEM_DESIGN}", cause_by=WriteDesign)
project_manager = ProjectManager(context=ctx)
rsp = await project_manager.run(system_design)
logger.info(rsp)

2024-12-18 16:26:55.452 | INFO     | metagpt.roles.role:_act:403 - Eve(Project Manager): to do WriteTasks(WriteTasks)


actionnode:
## context
json_data

-----

## format example
[CONTENT]
{
    "Required packages": [
        "flask==1.1.2",
        "bcrypt==3.2.0"
    ],
    "Required Other language third-party packages": [
        "No third-party dependencies required"
    ],
    "Logic Analysis": [
        [
            "game.py",
            "Contains Game class and ... functions"
        ],
        [
            "main.py",
            "Contains main function, from game import Game"
        ]
    ],
    "Task list": [
        "game.py",
        "main.py"
    ],
    "Full API spec": "openapi: 3.0.0 ...",
    "Shared Knowledge": "`game.py` contains functions shared across the project.",
    "Anything UNCLEAR": "Clarification needed on how to start and initialize third-party libraries."
}
[/CONTENT]

## nodes: "<node>: <type>  # <instruction>"
- Required packages: typing.Optional[typing.List[str]]  # Provide required third-party packages in requirements.txt format.
- Required Other language third-party packages: typing.List[str]  # List down the required packages for languages other than Python.
- Logic Analysis: typing.List[typing.List[str]]  # Provide a list of files with the classes/methods/functions to be implemented, including dependency analysis and imports.
- Task list: typing.List[str]  # Break down the tasks into a list of filenames, prioritized by dependency order.
- Full API spec: <class 'str'>  # Describe all APIs using OpenAPI 3.0 spec that may be used by both frontend and backend. If front-end and back-end communication is not required, leave it blank.
- Shared Knowledge: <class 'str'>  # Detail any shared knowledge, like common utility functions or configuration variables.
- Anything UNCLEAR: <class 'str'>  # Mention any unclear aspects in the project management context and try to clarify them.


## constraint
Language: Please use the same language as Human INPUT.
Format: output wrapped inside [CONTENT][/CONTENT] like format example, nothing else.

## action
Follow instructions of nodes, generate output and make sure it follows the format example.

[CONTENT]
{
    "Required packages": [
        "numpy==1.21.2",
        "pandas==1.3.3",
        "matplotlib==3.4.3"
    ],
    "Required Other language third-party packages": [
        "No third-party dependencies required"
    ],
    "Logic Analysis": [
        [
            "data_processing.py",
            "Contains data processing functions and classes, such as DataProcessor class and data_preparation function"
        ],
        [
            "plotting.py",
            "Contains plotting functions and classes, such as Plotter class and plot_data function"
        ],
        [
            "main.py",
            "Contains the main application logic, which imports DataProcessor and Plotter"
        ]
    ],
    "Task list": [
        "data_processing.py",
        "plotting.py",
        "main.py"
    ],
    "Full API spec": "openapi: 3.0.0 ...",
    "Shared Knowledge": "DataProcessor and Plotter classes are used across the project for data processing and plotting.",
    "Anything UNCLEAR": "Clarification needed on the specific data processing requirements and desired plot types."
}
[/CONTENT]

2024-12-18 16:27:07.350 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model GLM-4-flash not found in TOKEN_COSTS.
2024-12-18 16:27:07.368 | INFO     | metagpt.utils.file_repository:save:57 - save to: D:\llm\MetaGPT\workspace\unittest\315027a13519458c956f9e5db5928532\docs\task\36261056f2014620b7250a3a0e7a623e.json
2024-12-18 16:27:07.368 | INFO     | metagpt.utils.file_repository:save:62 - update dependency: D:\llm\MetaGPT\workspace\unittest\315027a13519458c956f9e5db5928532\docs\task\36261056f2014620b7250a3a0e7a623e.json:{'docs\\system_design\\36261056f2014620b7250a3a0e7a623e.json'}
2024-12-18 16:27:07.377 | INFO     | metagpt.utils.file_repository:save:57 - save to: D:\llm\MetaGPT\workspace\unittest\315027a13519458c956f9e5db5928532\requirements.txt
2024-12-18 16:27:07.377 | INFO     | __main__:<module>:6 - Eve(Project Manager): {'docs': {'36261056f2014620b7250a3a0e7a623e.json': {'root_path': 'docs\\task', 'filename': '36261056f2014620b7250a3a0e7a623e.json', 'content': '{"Required packages":["numpy==1.21.2","pandas==1.3.3","matplotlib==3.4.3"],"Required Other language third-party packages":["No third-party dependencies required"],"Logic Analysis":[["data_processing.py","Contains data processing functions and classes, such as DataProcessor class and data_preparation function"],["plotting.py","Contains plotting functions and classes, such as Plotter class and plot_data function"],["main.py","Contains the main application logic, which imports DataProcessor and Plotter"]],"Task list":["data_processing.py","plotting.py","main.py"],"Full API spec":"openapi: 3.0.0 ...","Shared Knowledge":"DataProcessor and Plotter classes are used across the project for data processing and plotting.","Anything UNCLEAR":"Clarification needed on the specific data processing requirements and desired plot types."}'}}}