From e6523779fd2eb5b3dc450301da094443aae913b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E5=BB=BA=E7=94=9F?= Date: Tue, 25 Feb 2025 17:11:56 +0800 Subject: [PATCH] delete useless example code --- examples/reverse_engineering.py | 72 ----------- examples/write_project_framework.py | 194 ---------------------------- 2 files changed, 266 deletions(-) delete mode 100644 examples/reverse_engineering.py delete mode 100644 examples/write_project_framework.py diff --git a/examples/reverse_engineering.py b/examples/reverse_engineering.py deleted file mode 100644 index f80fc09e6..000000000 --- a/examples/reverse_engineering.py +++ /dev/null @@ -1,72 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -import asyncio -import shutil -from pathlib import Path - -import typer - -from metagpt.actions.rebuild_class_view import RebuildClassView -from metagpt.actions.rebuild_sequence_view import RebuildSequenceView -from metagpt.context import Context -from metagpt.llm import LLM -from metagpt.logs import logger -from metagpt.utils.git_repository import GitRepository -from metagpt.utils.project_repo import ProjectRepo - -app = typer.Typer(add_completion=False, pretty_exceptions_show_locals=False) - - -@app.command("", help="Python project reverse engineering.") -def startup( - project_root: str = typer.Argument( - default="", - help="Specify the root directory of the existing project for reverse engineering.", - ), - output_dir: str = typer.Option(default="", help="Specify the output directory path for reverse engineering."), -): - package_root = Path(project_root) - if not package_root.exists(): - raise FileNotFoundError(f"{project_root} not exists") - if not _is_python_package_root(package_root): - raise FileNotFoundError(f'There are no "*.py" files under "{project_root}".') - init_file = package_root / "__init__.py" # used by pyreverse - init_file_exists = init_file.exists() - if not init_file_exists: - init_file.touch() - - if not output_dir: - output_dir = package_root / "../reverse_engineering_output" - logger.info(f"output dir:{output_dir}") - try: - asyncio.run(reverse_engineering(package_root, Path(output_dir))) - finally: - if not init_file_exists: - init_file.unlink(missing_ok=True) - tmp_dir = package_root / "__dot__" - if tmp_dir.exists(): - shutil.rmtree(tmp_dir, ignore_errors=True) - - -def _is_python_package_root(package_root: Path) -> bool: - for file_path in package_root.iterdir(): - if file_path.is_file(): - if file_path.suffix == ".py": - return True - return False - - -async def reverse_engineering(package_root: Path, output_dir: Path): - ctx = Context() - ctx.git_repo = GitRepository(output_dir) - ctx.repo = ProjectRepo(ctx.git_repo) - action = RebuildClassView(name="ReverseEngineering", i_context=str(package_root), llm=LLM(), context=ctx) - await action.run() - - action = RebuildSequenceView(name="ReverseEngineering", llm=LLM(), context=ctx) - await action.run() - - -if __name__ == "__main__": - app() diff --git a/examples/write_project_framework.py b/examples/write_project_framework.py deleted file mode 100644 index 8d23695a7..000000000 --- a/examples/write_project_framework.py +++ /dev/null @@ -1,194 +0,0 @@ -#!/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 metagpt.actions.requirement_analysis.framework import ( - EvaluateFramework, - WriteFramework, - save_framework, -) -from metagpt.actions.requirement_analysis.trd import ( - CompressExternalInterfaces, - DetectInteraction, - EvaluateTRD, - WriteTRD, -) -from metagpt.config2 import Config -from metagpt.const import DEFAULT_WORKSPACE_ROOT -from metagpt.context import Context -from metagpt.logs import logger -from metagpt.utils.common import aread - -app = typer.Typer(add_completion=False) - - -async def _write_trd( - context: Context, actors: Dict[str, str], user_requirements: List[str], available_external_interfaces: str -) -> (str, str): - detect_interaction = DetectInteraction(context=context) - write_trd = WriteTRD(context=context) - evaluate_trd = EvaluateTRD(context=context) - use_case_actors = "".join([f"- {v}: {k}\n" for k, v in actors.items()]) - legacy_user_requirements = [] - legacy_user_requirements_interaction_events = [] - legacy_user_requirements_trd = "" - for ix, r in enumerate(user_requirements): - is_pass = False - evaluation_conclusion = "" - interaction_events = "" - trd = "" - while not is_pass and (context.cost_manager.total_cost < context.cost_manager.max_budget): - interaction_events = await detect_interaction.run( - user_requirements=r, - use_case_actors=use_case_actors, - legacy_interaction_events=interaction_events, - evaluation_conclusion=evaluation_conclusion, - ) - if ix == 0: - trd = await write_trd.run( - user_requirements=r, - use_case_actors=use_case_actors, - available_external_interfaces=available_external_interfaces, - evaluation_conclusion=evaluation_conclusion, - interaction_events=interaction_events, - previous_version_trd=trd, - ) - else: - trd = await write_trd.run( - user_requirements=r, - use_case_actors=use_case_actors, - available_external_interfaces=available_external_interfaces, - evaluation_conclusion=evaluation_conclusion, - interaction_events=interaction_events, - previous_version_trd=trd, - legacy_user_requirements="\n".join(legacy_user_requirements), - legacy_user_requirements_trd=legacy_user_requirements_trd, - legacy_user_requirements_interaction_events="\n".join(legacy_user_requirements_interaction_events), - ) - evaluation = await evaluate_trd.run( - user_requirements=r, - use_case_actors=use_case_actors, - trd=trd, - interaction_events=interaction_events, - legacy_user_requirements_interaction_events="\n".join(legacy_user_requirements_interaction_events), - ) - is_pass = evaluation.is_pass - evaluation_conclusion = evaluation.conclusion - legacy_user_requirements.append(r) - legacy_user_requirements_interaction_events.append(interaction_events) - legacy_user_requirements_trd = trd - - return use_case_actors, legacy_user_requirements_trd - - -async def _write_framework(context: Context, use_case_actors: str, trd: str, acknowledge: str, constraint: str) -> str: - write_framework = WriteFramework(context=context) - evaluate_framework = EvaluateFramework(context=context) - is_pass = False - framework = "" - evaluation_conclusion = "" - while not is_pass and (context.cost_manager.total_cost < context.cost_manager.max_budget): - try: - framework = await write_framework.run( - use_case_actors=use_case_actors, - trd=trd, - acknowledge=acknowledge, - legacy_output=framework, - evaluation_conclusion=evaluation_conclusion, - additional_technical_requirements=constraint, - ) - except Exception as e: - logger.info(f"{e}") - break - evaluation = await evaluate_framework.run( - use_case_actors=use_case_actors, - trd=trd, - acknowledge=acknowledge, - legacy_output=framework, - additional_technical_requirements=constraint, - ) - is_pass = evaluation.is_pass - evaluation_conclusion = evaluation.conclusion - return framework - - -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) - - # Compress acknowledge - compress_acknowledge = CompressExternalInterfaces(context=context) - available_external_interfaces = await compress_acknowledge.run(acknowledge=acknowledge) - - # Write TRD - use_case_actors, trd = await _write_trd( - context=context, - actors=actors, - user_requirements=user_requirements, - available_external_interfaces=available_external_interfaces, - ) - - # Write framework - framework = await _write_framework( - context=context, - use_case_actors=use_case_actors, - trd=trd, - acknowledge=acknowledge, - constraint=technical_constraint, - ) - - # Save - file_list = await save_framework(dir_data=framework, trd=trd, output_dir=output_dir) - logger.info(f"Output:\n{file_list}") - - -@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."), - investment: float = typer.Option(default=15.0, help="Dollar amount to invest in the AI company."), -): - 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) - ctx.cost_manager.max_budget = investment - - asyncio.run( - develop(ctx, user_requirement_filename, actors_filename, acknowledge_filename, constraint_filename, output_dir) - ) - - -if __name__ == "__main__": - app()