diff --git a/examples/mgx_write_project_framework.py b/examples/mgx_write_project_framework.py new file mode 100644 index 000000000..b5d982b72 --- /dev/null +++ b/examples/mgx_write_project_framework.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +@Time : 2024/6/13 +@Author : mashenquan +@File : write_project_framework.py +@Desc : The implementation of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb +""" +import asyncio +import json +import uuid +from pathlib import Path +from typing import Dict, List + +import typer +from pydantic import BaseModel + +from metagpt.config2 import Config +from metagpt.const import DEFAULT_WORKSPACE_ROOT +from metagpt.context import Context +from metagpt.environment import Environment +from metagpt.logs import logger +from metagpt.roles import Architect +from metagpt.roles.di.team_leader import TeamLeader +from metagpt.schema import AIMessage, UserMessage +from metagpt.utils.common import any_to_str, aread, to_markdown_code_block + +app = typer.Typer(add_completion=False) + + +class EnvBuilder(BaseModel): + context: Context + user_requirements: List[str] + actors: Dict[str, str] + acknowledge: str + technical_constraint: str + output_dir: Path + + def build(self) -> Environment: + env = Environment(context=self.context) + team_leader = TeamLeader() + architect = Architect() + + # Prepare context + use_case_actors = "".join([f"- {v}: {k}\n" for k, v in self.actors.items()]) + msg = """ +The content of "Actor, System, External System" provides an explanation of actors and systems that appear in UML Use Case diagram. +## Actor, System, External System +{use_case_actors} + """ + architect.rc.memory.add(AIMessage(content=msg.format(use_case_actors=use_case_actors))) + + # Prepare acknowledge + msg = """ +The descriptions of the interfaces used in TRD can be found in the "Acknowledge" section. +## Acknowledge +{acknowledge} + """ + architect.rc.memory.add(AIMessage(content=msg.format(acknowledge=to_markdown_code_block(val=self.acknowledge)))) + + # Prepare technical requirements + msg = """ +"Additional Technical Requirements" specifies the additional technical requirements that the generated software framework code must meet. +## Additional Technical Requirements +{technical_requirements} +""" + architect.rc.memory.add(AIMessage(content=msg.format(technical_requirements=self.technical_constraint))) + + env.add_roles([team_leader, architect]) + return env + + +async def develop( + context: Context, + user_requirement_filename: str, + actors_filename: str, + acknowledge_filename: str, + constraint_filename: str, + output_dir: str, +): + output_dir = Path(output_dir) if output_dir else DEFAULT_WORKSPACE_ROOT / uuid.uuid4().hex + + v = await aread(filename=user_requirement_filename) + user_requirements = json.loads(v) + v = await aread(filename=actors_filename) + actors = json.loads(v) + acknowledge = await aread(filename=acknowledge_filename) + technical_constraint = await aread(filename=constraint_filename) + env_builder = EnvBuilder( + context=context, + user_requirements=user_requirements, + actors=actors, + acknowledge=acknowledge, + technical_constraint=technical_constraint, + output_dir=output_dir, + ) + env = env_builder.build() + msg = """ +根据"User Requirements"中的用户需求，写TRD +## User Requirements +```json +{user_requirements} +``` + """ + env.publish_message( + UserMessage(content=msg.format(user_requirements=f"{user_requirements}"), send_to=any_to_str(TeamLeader)) + ) + + while not env.is_idle: + await env.run() + + +@app.command() +def startup( + user_requirement_filename: str = typer.Argument(..., help="The filename of the user requirements."), + actors_filename: str = typer.Argument(..., help="The filename of UML use case actors description."), + acknowledge_filename: str = typer.Argument(..., help="External interfaces declarations."), + llm_config: str = typer.Option(default="", help="Low-cost LLM config"), + constraint_filename: str = typer.Option(default="", help="What technical dependency constraints are."), + output_dir: str = typer.Option(default="", help="Output directory."), +): + if llm_config and Path(llm_config).exists(): + config = Config.from_yaml_file(Path(llm_config)) + else: + logger.info("GPT 4 turbo is recommended") + config = Config.default() + ctx = Context(config=config) + + asyncio.run( + develop(ctx, user_requirement_filename, actors_filename, acknowledge_filename, constraint_filename, output_dir) + ) + + +if __name__ == "__main__": + app() diff --git a/examples/write_project_framework.py b/examples/write_project_framework.py index fe015dbdd..968576515 100644 --- a/examples/write_project_framework.py +++ b/examples/write_project_framework.py @@ -5,9 +5,6 @@ @Author : mashenquan @File : write_project_framework.py @Desc : The implementation of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb - -Usage Example: - """ import asyncio import json @@ -64,7 +61,7 @@ async def _write_trd( user_requirements=r, use_case_actors=use_case_actors, available_external_interfaces=available_external_interfaces, - legacy_trd=trd, + previous_version_trd=trd, evaluation_conclusion=evaluation_conclusion, interaction_events=interaction_events, ) diff --git a/metagpt/actions/requirement_analysis/trd/write_trd.py b/metagpt/actions/requirement_analysis/trd/write_trd.py index 98199393d..071de92f0 100644 --- a/metagpt/actions/requirement_analysis/trd/write_trd.py +++ b/metagpt/actions/requirement_analysis/trd/write_trd.py @@ -10,17 +10,23 @@ from tenacity import retry, stop_after_attempt, wait_random_exponential from metagpt.actions import Action from metagpt.logs import logger +from metagpt.tools.tool_registry import register_tool from metagpt.utils.common import general_after_log, to_markdown_code_block +@register_tool(include_functions=["run"]) class WriteTRD(Action): + """WriteTRD deal with the following situations: + 1. Given some new user requirements, write out a new TRD(Technical Requirements Document). + 2. Given some incremental user requirements, update the legacy TRD. + """ + async def run( self, *, user_requirements: str = "", use_case_actors: str, available_external_interfaces: str, - legacy_trd: str = "", evaluation_conclusion: str = "", interaction_events: str = "", legacy_user_requirements: str = "", @@ -30,6 +36,64 @@ class WriteTRD(Action): previous_version_trd: str = "", incremental_user_requirements_interaction_events: str = "", ) -> str: + """ + Handles the writing or updating of a Technical Requirements Document (TRD) based on user requirements. + + Args: + user_requirements (str, optional): New user requirements for creating a new TRD. This value must be not empty if a new TRD is wanted. + use_case_actors (str): Description of the actors involved in the use case. + available_external_interfaces (str): List of available external interfaces. + evaluation_conclusion (str, optional): Conclusion of the evaluation of the requirements. Defaults to an empty string. + interaction_events (str, optional): Events related to user interactions. Defaults to an empty string. + legacy_user_requirements (str, optional): Existing user requirements if updating. Defaults to an empty string. + legacy_user_requirements_trd (str, optional): The TRD associated with the existing user requirements. Defaults to an empty string. + legacy_user_requirements_interaction_events (str, optional): Interaction events related to the existing user requirements. Defaults to an empty string. + incremental_user_requirements (str, optional): New incremental user requirements for updating the TRD. Defaults to an empty string. + previous_version_trd (str, optional): The previous version of the TRD if updating incrementally. Defaults to an empty string. + incremental_user_requirements_interaction_events (str, optional): Interaction events related to the incremental user requirements. Defaults to an empty string. + + Returns: + str: The newly created or updated TRD. + + Example: + >>> # Given a new user requirements, write out a new TRD. + >>> user_requirements = "Write a 'snake game' TRD." + >>> use_case_actors = "- Actor: game player;\\n- System: snake game; \\n- External System: game center;" + >>> available_external_interfaces = "The available external interfaces returned by `CompressExternalInterfaces.run` are ..." + >>> previous_version_trd = "TRD ..." # The last version of the TRD written out if there is. + >>> evaluation_conclusion = "Conclusion ..." # The conclusion returned by `EvaluateTRD.run` if there is. + >>> interaction_events = "Interaction ..." # The interaction events returned by `DetectInteraction.run`. + >>> write_trd = WriteTRD(context=context) + >>> new_version_trd = await write_trd.run( + >>> user_requirements=user_requirements, + >>> use_case_actors=use_case_actors, + >>> available_external_interfaces=available_external_interfaces, + >>> previous_version_trd=previous_version_trd, + >>> evaluation_conclusion=evaluation_conclusion, + >>> interaction_events=interaction_events, + >>> ) + >>> print(new_version_trd) + ## Technical Requirements Document\n ... + + >>> # Given an incremental requirements, update the legacy TRD. + >>> legacy_user_requirements = ["User requirements 1. ...", "User requirements 2. ...", ...] + >>> use_case_actors = "- Actor: game player;\\n- System: snake game; \\n- External System: game center;" + >>> available_external_interfaces = "The available external interfaces returned by `CompressExternalInterfaces.run` are ..." + >>> legacy_user_requirements_trd = "## Technical Requirements Document\\n ..." # The TRD before integrating more user requirements. + >>> legacy_user_requirements_interaction_events = ["The interaction events list of user requirements 1 ...", "The interaction events list of user requiremnts 2 ...", ...] + >>> incremental_user_requirements + >>> new_version_trd = await write_trd.run( + >>> legacy_user_requirements=str(legacy_user_requirements), + >>> use_case_actors=use_case_actors, + >>> available_external_interfaces=available_external_interfaces, + >>> legacy_user_requirements_trd=legacy_user_requirements_trd, + >>> legacy_user_requirements_interaction_events=str(legacy_user_requirements_interaction_events), + >>> incremental_user_requirements=r, + previous_version_trd=trd, + evaluation_conclusion=evaluation_conclusion, + incremental_user_requirements_interaction_events=interaction_events, + ) + """ if incremental_user_requirements: return await self._write_incremental_trd( use_case_actors=use_case_actors, @@ -46,7 +110,7 @@ class WriteTRD(Action): use_case_actors=use_case_actors, original_user_requirement=user_requirements, available_external_interfaces=available_external_interfaces, - legacy_trd=legacy_trd, + legacy_trd=previous_version_trd, evaluation_conclusion=evaluation_conclusion, interaction_events=interaction_events, )