refactor WriteTRD.run

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()

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,
                 )

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,
         )