1  #!/usr/bin/env python
  2  # -*- coding: utf-8 -*-
  3  """
  4  @Time    : 2024/6/13
  5  @Author  : mashenquan
  6  @File    : write_project_framework.py
  7  @Desc    : The implementation of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
  8  """
  9  import asyncio
 10  import json
 11  import uuid
 12  from json import JSONDecodeError
 13  from pathlib import Path
 14  from typing import Dict, List
 15  
 16  import typer
 17  from pydantic import BaseModel
 18  
 19  from metagpt.config2 import Config
 20  from metagpt.const import DEFAULT_WORKSPACE_ROOT
 21  from metagpt.context import Context
 22  from metagpt.environment import Environment
 23  from metagpt.environment.mgx.mgx_env import MGXEnv
 24  from metagpt.logs import logger
 25  from metagpt.roles import Architect
 26  from metagpt.roles.di.team_leader import TeamLeader
 27  from metagpt.schema import AIMessage, UserMessage
 28  from metagpt.strategy.experience_retriever import TRDToolExpRetriever
 29  from metagpt.utils.common import aread
 30  
 31  app = typer.Typer(add_completion=False)
 32  
 33  
 34  class EnvBuilder(BaseModel):
 35      context: Context
 36      user_requirements: List[str]
 37      actors: Dict[str, str]
 38      technical_constraint: str
 39      output_dir: Path
 40  
 41      def build(self) -> Environment:
 42          env = MGXEnv(context=self.context)
 43          team_leader = TeamLeader()
 44          architect = Architect(experience_retriever=TRDToolExpRetriever())
 45  
 46          # Prepare context
 47          use_case_actors = "".join([f"- {v}: {k}\n" for k, v in self.actors.items()])
 48          msg = """
 49  The content of "Actor, System, External System" provides an explanation of actors and systems that appear in UML Use Case diagram.
 50  ## Actor, System, External System
 51  {use_case_actors}
 52          """
 53          architect.rc.memory.add(AIMessage(content=msg.format(use_case_actors=use_case_actors)))
 54  
 55          # Prepare technical requirements
 56          msg = """
 57  "Additional Technical Requirements" specifies the additional technical requirements that the generated software framework code must meet.
 58  ## Additional Technical Requirements
 59  {technical_requirements}
 60  """
 61          architect.rc.memory.add(AIMessage(content=msg.format(technical_requirements=self.technical_constraint)))
 62  
 63          env.add_roles([team_leader, architect])
 64          return env
 65  
 66  
 67  async def develop(
 68      context: Context,
 69      user_requirement_filename: str,
 70      actors_filename: str,
 71      constraint_filename: str,
 72      output_dir: str,
 73  ):
 74      output_dir = Path(output_dir) if output_dir else DEFAULT_WORKSPACE_ROOT / uuid.uuid4().hex
 75  
 76      v = await aread(filename=user_requirement_filename)
 77      try:
 78          user_requirements = json.loads(v)
 79      except JSONDecodeError:
 80          user_requirements = [v]
 81      v = await aread(filename=actors_filename)
 82      actors = json.loads(v)
 83      technical_constraint = await aread(filename=constraint_filename)
 84      env_builder = EnvBuilder(
 85          context=context,
 86          user_requirements=user_requirements,
 87          actors=actors,
 88          technical_constraint=technical_constraint,
 89          output_dir=output_dir,
 90      )
 91      env = env_builder.build()
 92      msg = """
 93  Given the user requirement of "User Requirements", write out the software framework.
 94  ## User Requirements
 95  {user_requirements}
 96      """
 97      env.publish_message(
 98          UserMessage(content=msg.format(user_requirements="\n".join(user_requirements)), send_to="Bob"),
 99          user_defined_recipient="Bob",
100      )
101  
102      while not env.is_idle:
103          await env.run()
104  
105  
106  @app.command()
107  def startup(
108      user_requirement_filename: str = typer.Argument(..., help="The filename of the user requirements."),
109      actors_filename: str = typer.Argument(..., help="The filename of UML use case actors description."),
110      llm_config: str = typer.Option(default="", help="Low-cost LLM config"),
111      constraint_filename: str = typer.Option(default="", help="What technical dependency constraints are."),
112      output_dir: str = typer.Option(default="", help="Output directory."),
113  ):
114      if llm_config and Path(llm_config).exists():
115          config = Config.from_yaml_file(Path(llm_config))
116      else:
117          logger.info("GPT 4 turbo is recommended")
118          config = Config.default()
119      ctx = Context(config=config)
120  
121      asyncio.run(develop(ctx, user_requirement_filename, actors_filename, constraint_filename, output_dir))
122  
123  
124  if __name__ == "__main__":
125      app()