基础流程

1. 导入模块

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

这些导入语句引入了多个与任务管理、设计写作、日志记录等相关的模块，主要用于处理文档、项目管理和任务自动化。

2. WriteTasks 类

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

WriteTasks 继承自 Action，代表一个操作任务，它的目标是根据项目需求和设计文件创建任务。这个类定义了任务生成的主要逻辑。

run 方法

async def run(self, with_messages):changed_system_designs = self.repo.docs.system_design.changed_fileschanged_tasks = self.repo.docs.task.changed_fileschange_files = Documents()

run 方法的主要作用是：

• 检查系统设计和任务文档中有哪些文件发生了变化。
• 如果文件有变化，重新生成相关的任务文档。
• 它通过遍历变动的文件并调用 _update_tasks 方法，来更新任务文件，并返回包含修改后的任务文件的 ActionOutput。

_update_tasks 方法

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

这个方法会：

• 获取相关的系统设计文档和任务文档。
• 如果任务文档已存在，则通过调用 _merge 方法将系统设计与任务文档合并。
• 如果任务文档不存在，则根据系统设计创建新的任务文档。
• 最后，更新任务文档中的需求。

_merge 方法

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

_merge 方法将系统设计文档和任务文档合并，创建新的任务文档。它使用一个模板将旧任务与新需求进行合并，并更新任务文档的内容。

_update_requirements 方法

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 == "":continuepackages.add(pkg)await self.repo.save(filename=PACKAGE_REQUIREMENTS_FILENAME, content="
".join(packages))

这个方法更新了项目的包需求文件，确保所有相关的包都列出，并保存到相应的文件中。

3. ProjectManager 类

class ProjectManager(Role):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"

ProjectManager 继承自 Role 类，代表项目经理的角色。它的目标是根据项目需求文档（PRD）和技术设计将任务拆解成更小的单元，并分析任务之间的依赖关系。项目经理需要生成任务列表并确保使用与用户要求相同的语言。

4. 项目上下文初始化

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 uuidctx = Context()
ctx.git_repo = GitRepository(local_path=DEFAULT_WORKSPACE_ROOT / f"unittest/{uuid.uuid4().hex}")
ctx.repo = ProjectRepo(ctx.git_repo)

这一部分代码初始化了一个 Context，并在其中创建了一个 Git 仓库和项目仓库实例，用于存储和管理项目的文件。

5. 生成 PRD 和 SYSTEM_DESIGN 文档

PRD 和 SYSTEM_DESIGN 分别是项目需求文档和系统设计文档。这些文档用 JSON 格式定义，包含了项目的功能需求、用户故事、竞争分析、技术实现等信息。文档内容通过 json.dumps 方法转换为 JSON 字符串，并通过 awrite 方法写入磁盘。

6. 执行任务生成

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)

通过 Message 类创建一个消息，模拟一个系统设计文档被 WriteDesign 操作所生成。
创建一个 ProjectManager 实例，并调用 run 方法执行任务生成。run 方法会根据给定的设计文档生成任务并返回响应。
最后，将响应内容记录到日志中。

总结：

这段代码实现了一个自动化的项目管理系统，能够根据设计文档和需求文档生成任务。项目经理角色负责协调任务的拆解与依赖分析，并且通过使用 metagpt 框架，系统能够智能地创建和更新任务，生成项目文档，并管理项目的包依赖。这使得项目的管理和执行更加高效和自动化。

完整代码

1. WriteTasks、ProjectManager类

from metagpt.schema import Message
from metagpt.logs import logger
from metagpt.roles import ProjectManager
from tests.metagpt.roles.mock import MockMessagesfrom metagpt.actions import WriteTasks
from metagpt.actions.design_api import WriteDesign
from metagpt.roles.role import Roleimport json
from typing import Optionalfrom 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, DocumentsNEW_REQ_TEMPLATE = """
### Legacy Content
{old_task}### New Requirements
{context}
"""class WriteTasks(Action):name: str = "CreateTasks"i_context: Optional[str] = Noneasync def run(self, with_messages):changed_system_designs = self.repo.docs.system_design.changed_fileschanged_tasks = self.repo.docs.task.changed_fileschange_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:continuetask_doc = await self._update_tasks(filename=filename)change_files.docs[filename] = task_docif 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_docasync def _run_new_tasks(self, context):node = await PM_NODE.fill(context, self.llm, schema=self.prompt_schema)return nodeasync 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_docasync 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 == "":continuepackages.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 uuidctx = 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, awritePRDS_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."}'}}}