apunkt/MetaGPT
1
0
Fork 0
mirror of https://github.com/FoundationAgents/MetaGPT.git synced 2026-06-20 15:38:09 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'feature/rfc243' into 'mgx_ops'

Browse source 
feat: rfc243

See merge request pub/MetaGPT!170
This commit is contained in:
王金淋 2024-06-28 06:15:52 +00:00
commit ce485dc436
42 changed files with 2458 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

11
metagpt/actions/requirement_analysis/__init__.py Normal file
View file

@ -0,0 +1,11 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : __init__.py
@Desc : The implementation of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
from metagpt.actions.requirement_analysis.evaluate_action import EvaluationData, EvaluateAction
__all__ = [EvaluationData, EvaluateAction]

74
metagpt/actions/requirement_analysis/evaluate_action.py Normal file
View file

@ -0,0 +1,74 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : evaluate_action.py
@Desc : The implementation of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
from typing import Optional
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_random_exponential
from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.utils.common import CodeParser, general_after_log, to_markdown_code_block
class EvaluationData(BaseModel):
"""Model to represent evaluation data.
Attributes:
is_pass (bool): Indicates if the evaluation passed or failed.
conclusion (Optional[str]): Conclusion or remarks about the evaluation.
"""
is_pass: bool
conclusion: Optional[str] = None
class EvaluateAction(Action):
"""The base class for an evaluation action.
This class provides methods to evaluate prompts using a specified language model.
"""
@retry(
wait=wait_random_exponential(min=1, max=20),
stop=stop_after_attempt(6),
after=general_after_log(logger),
)
async def _evaluate(self, prompt: str) -> (bool, str):
"""Evaluates a given prompt.
Args:
prompt (str): The prompt to be evaluated.
Returns:
tuple: A tuple containing:
- bool: Indicates if the evaluation passed.
- str: The JSON string containing the evaluation data.
"""
rsp = await self.llm.aask(prompt)
json_data = CodeParser.parse_code(text=rsp, lang="json")
data = EvaluationData.model_validate_json(json_data)
return data.is_pass, to_markdown_code_block(val=json_data, type_="json")
async def _vote(self, prompt: str) -> EvaluationData:
"""Evaluates a prompt multiple times and returns the consensus.
Args:
prompt (str): The prompt to be evaluated.
Returns:
EvaluationData: An object containing the evaluation result and a summary of evaluations.
"""
evaluations = {}
for i in range(3):
vote, evaluation = await self._evaluate(prompt)
val = evaluations.get(vote, [])
val.append(evaluation)
if len(val) > 1:
return EvaluationData(is_pass=vote, conclusion="\n".join(val))
evaluations[vote] = val

86
metagpt/actions/requirement_analysis/framework/__init__.py Normal file
View file

@ -0,0 +1,86 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : __init__.py
@Desc : The implementation of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
import json
import uuid
from datetime import datetime
from pathlib import Path
from typing import Optional, Union, List
from pydantic import BaseModel
from metagpt.actions.requirement_analysis.framework.evaluate_framework import EvaluateFramework
from metagpt.actions.requirement_analysis.framework.write_framework import WriteFramework
from metagpt.const import DEFAULT_WORKSPACE_ROOT
from metagpt.utils.common import awrite
async def save_framework(
dir_data: str, trd: Optional[str] = None, output_dir: Optional[Union[str, Path]] = None
) -> List[str]:
"""
Saves framework data to files based on input JSON data and optionally saves a TRD (technical requirements document).
Args:
dir_data (str): JSON data in string format enclosed in triple backticks ("```json" "...data..." "```").
trd (str, optional): Technical requirements document content to be saved. Defaults to None.
output_dir (Union[str, Path], optional): Output directory path where files will be saved. If not provided,
a default directory is created based on the current timestamp and a random UUID suffix.
Returns:
List[str]: List of file paths where data was saved.
Raises:
Any exceptions raised during file writing operations.
Notes:
- JSON data should be provided in the format "```json ...data... ```".
- The function ensures that paths and filenames are correctly formatted and creates necessary directories.
Example:
```python
dir_data = "```json\n[{\"path\": \"/folder\", \"filename\": \"file1.txt\", \"content\": \"Some content\"}]\n```"
trd = "Technical requirements document content."
output_dir = '/path/to/output/dir'
saved_files = await save_framework(dir_data, trd, output_dir)
print(saved_files)
```
"""
output_dir = (
Path(output_dir)
if output_dir
else DEFAULT_WORKSPACE_ROOT / (datetime.now().strftime("%Y%m%d%H%M%ST") + uuid.uuid4().hex[0:8])
)
output_dir.mkdir(parents=True, exist_ok=True)
json_data = dir_data.removeprefix("```json").removesuffix("```")
items = json.loads(json_data)
class Data(BaseModel):
path: str
filename: str
content: str
if trd:
pathname = output_dir / "TRD.md"
await awrite(filename=pathname, data=trd)
files = []
for i in items:
v = Data.model_validate(i)
if v.path and v.path[0] == "/":
v.path = "." + v.path
pathname = output_dir / v.path
pathname.mkdir(parents=True, exist_ok=True)
pathname = pathname / v.filename
await awrite(filename=pathname, data=v.content)
files.append(str(pathname))
return files
__all__ = [WriteFramework, EvaluateFramework]

106
metagpt/actions/requirement_analysis/framework/evaluate_framework.py Normal file
View file

@ -0,0 +1,106 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : evaluate_framework.py
@Desc : The implementation of Chapter 2.1.8 of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
from metagpt.actions.requirement_analysis import EvaluateAction, EvaluationData
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.common import to_markdown_code_block
@register_tool(include_functions=["run"])
class EvaluateFramework(EvaluateAction):
"""WriteFramework deal with the following situations:
1. Given a TRD and the software framework based on the TRD, evaluate the quality of the software framework.
"""
async def run(
self,
*,
use_case_actors: str,
trd: str,
acknowledge: str,
legacy_output: str,
additional_technical_requirements: str,
) -> EvaluationData:
"""
Run the evaluation of the software framework based on the provided TRD and related parameters.
Args:
use_case_actors (str): A description of the actors involved in the use case.
trd (str): The Technical Requirements Document (TRD) that outlines the requirements for the software framework.
acknowledge (str): External acknowledgments or acknowledgments information related to the framework.
legacy_output (str): The previous versions of software framework returned by `WriteFramework`.
additional_technical_requirements (str): Additional technical requirements that need to be considered during evaluation.
Returns:
EvaluationData: An object containing the results of the evaluation.
Example:
>>> evaluate_framework = EvaluateFramework()
>>> use_case_actors = "- Actor: game player;\\n- System: snake game; \\n- External System: game center;"
>>> trd = "## TRD\\n..."
>>> acknowledge = "## Interfaces\\n..."
>>> framework = '{"path":"balabala", "filename":"...", ...'
>>> constraint = "Using Java language, ..."
>>> 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
>>> print(is_pass)
True
>>> evaluation_conclusion = evaluation.conclusion
>>> print(evaluation_conclusion)
Balabala...
"""
prompt = PROMPT.format(
use_case_actors=use_case_actors,
trd=to_markdown_code_block(val=trd),
acknowledge=to_markdown_code_block(val=acknowledge),
legacy_output=to_markdown_code_block(val=legacy_output),
additional_technical_requirements=to_markdown_code_block(val=additional_technical_requirements),
)
return await self._vote(prompt)
PROMPT = """
## Actor, System, External System
{use_case_actors}
## Legacy TRD
{trd}
## Acknowledge
{acknowledge}
## Legacy Outputs
{legacy_output}
## Additional Technical Requirements
{additional_technical_requirements}
---
You are a tool that evaluates the quality of framework code based on the TRD content;
You need to refer to the content of the "Legacy TRD" section to check for any errors or omissions in the framework code found in "Legacy Outputs";
The content of "Actor, System, External System" provides an explanation of actors and systems that appear in UML Use Case diagram;
Information about the external system missing from the "Legacy TRD" can be found in the "Acknowledge" section;
Which interfaces defined in "Acknowledge" are used in the "Legacy TRD"?
Do not implement the interface in "Acknowledge" section until it is used in "Legacy TRD", you can check whether they are the same interface by looking at its ID or url;
Parts not mentioned in the "Legacy TRD" will be handled by other TRDs, therefore, processes not present in the "Legacy TRD" are considered ready;
"Additional Technical Requirements" specifies the additional technical requirements that the generated software framework code must meet;
Do the parameters of the interface of the external system used in the code comply with it's specifications in 'Acknowledge'?
Is there a lack of necessary configuration files?
Return a markdown JSON object with:
- an "issues" key containing a string list of natural text about the issues that need to addressed, found in the "Legacy Outputs" if any exits, each issue found must provide a detailed description and include reasons;
- a "conclusion" key containing the evaluation conclusion;
- a "misalignment" key containing the judgement detail of the natural text string list about the misalignment with "Legacy TRD";
- a "is_pass" key containing a true boolean value if there is not any issue in the "Legacy Outputs";
"""

156
metagpt/actions/requirement_analysis/framework/write_framework.py Normal file
View file

@ -0,0 +1,156 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : write_framework.py
@Desc : The implementation of Chapter 2.1.8 of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
import json
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 WriteFramework(Action):
"""WriteFramework deal with the following situations:
1. Given a TRD, write out the software framework.
"""
async def run(
self,
*,
use_case_actors: str,
trd: str,
acknowledge: str,
legacy_output: str,
evaluation_conclusion: str,
additional_technical_requirements: str,
) -> str:
"""
Run the action to generate a software framework based on the provided TRD and related information.
Args:
use_case_actors (str): Description of the use case actors involved.
trd (str): Technical Requirements Document detailing the requirements.
acknowledge (str): External acknowledgements or acknowledgements required.
legacy_output (str): Previous version of the software framework returned by `WriteFramework.run`.
evaluation_conclusion (str): Conclusion from the evaluation of the requirements.
additional_technical_requirements (str): Any additional technical requirements.
Returns:
str: The generated software framework as a string.
Example:
>>> write_framework = WriteFramework()
>>> use_case_actors = "- Actor: game player;\\n- System: snake game; \\n- External System: game center;"
>>> trd = "## TRD\\n..."
>>> acknowledge = "## Interfaces\\n..."
>>> legacy_output = '{"path":"balabala", "filename":"...", ...'
>>> evaluation_conclusion = "Balabala..."
>>> constraint = "Using Java language, ..."
>>> 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,
>>> )
>>> print(framework)
{"path":"balabala", "filename":"...", ...
"""
acknowledge = await self._extract_external_interfaces(trd=trd, knowledge=acknowledge)
prompt = PROMPT.format(
use_case_actors=use_case_actors,
trd=to_markdown_code_block(val=trd),
acknowledge=to_markdown_code_block(val=acknowledge),
legacy_output=to_markdown_code_block(val=legacy_output),
evaluation_conclusion=evaluation_conclusion,
additional_technical_requirements=to_markdown_code_block(val=additional_technical_requirements),
)
return await self._write(prompt)
@retry(
wait=wait_random_exponential(min=1, max=20),
stop=stop_after_attempt(6),
after=general_after_log(logger),
)
async def _write(self, prompt: str) -> str:
rsp = await self.llm.aask(prompt)
# Do not use `CodeParser` here.
tags = ["```json", "```"]
bix = rsp.find(tags[0])
eix = rsp.rfind(tags[1])
if bix >= 0:
rsp = rsp[bix : eix + len(tags[1])]
json_data = rsp.removeprefix("```json").removesuffix("```")
json.loads(json_data) # validate
return json_data
@retry(
wait=wait_random_exponential(min=1, max=20),
stop=stop_after_attempt(6),
after=general_after_log(logger),
)
async def _extract_external_interfaces(self, trd: str, knowledge: str) -> str:
prompt = f"## TRD\n{to_markdown_code_block(val=trd)}\n\n## Knowledge\n{to_markdown_code_block(val=knowledge)}\n"
rsp = await self.llm.aask(
prompt,
system_msgs=[
"You are a tool that removes impurities from articles; you can remove irrelevant content from articles.",
'Identify which interfaces are used in "TRD"? Remove the relevant content of the interfaces NOT used in "TRD" from "Knowledge" and return the simplified content of "Knowledge".',
],
)
return rsp
PROMPT = """
## Actor, System, External System
{use_case_actors}
## TRD
{trd}
## Acknowledge
{acknowledge}
## Legacy Outputs
{legacy_output}
## Evaluation Conclusion
{evaluation_conclusion}
## Additional Technical Requirements
{additional_technical_requirements}
---
You are a tool that generates software framework code based on TRD.
The content of "Actor, System, External System" provides an explanation of actors and systems that appear in UML Use Case diagram;
The descriptions of the interfaces of the external system used in the "TRD" can be found in the "Acknowledge" section; Do not implement the interface of the external system in "Acknowledge" section until it is used in "TRD";
"Legacy Outputs" contains the software framework code generated by you last time, which you can improve by addressing the issues raised in "Evaluation Conclusion";
"Additional Technical Requirements" specifies the additional technical requirements that the generated software framework code must meet;
Develop the software framework based on the "TRD", the output files should include:
- The `README.md` file should include:
- The folder structure diagram of the entire project;
- Correspondence between classes, interfaces, and functions with the content in the "TRD" section；
- Prerequisites if necessary;
- Installation if necessary;
- Configuration if necessary;
- Usage if necessary;
- The `CLASS.md` file should include the class diagram in PlantUML format based on the "TRD";
- The `SEQUENCE.md` file should include the sequence diagram in PlantUML format based on the "TRD";
- The source code files that implement the "TRD" and "Additional Technical Requirements"; do not add comments to source code files;
- The configuration files that required by the source code files, "TRD" and "Additional Technical Requirements";
Return a markdown JSON object list, each object containing:
- a "path" key with a value specifying its path;
- a "filename" key with a value specifying its file name;
- a "content" key with a value containing its file content;
"""

0
metagpt/actions/requirement_analysis/requirement/__init__.py Normal file
View file

123
metagpt/actions/requirement_analysis/requirement/pic2txt.py Normal file
View file

@ -0,0 +1,123 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/27
@Author : mashenquan
@File : pic2txt.py
"""
import json
from pathlib import Path
from typing import List
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 encode_image, general_after_log, to_markdown_code_block
@register_tool(include_functions=["run"])
class Pic2Txt(Action):
"""Pic2Txt deal with the following situations:
Given some pictures depicting user requirements alongside contextual description, write out the intact textual user requirements.
"""
async def run(
self,
*,
image_paths: List[str],
textual_user_requirement: str = "",
legacy_output: str = "",
evaluation_conclusion: str = "",
additional_technical_requirements: str = "",
) -> str:
"""
Given some pictures depicting user requirements alongside contextual description, write out the intact textual user requirements
Args:
image_paths (List[str]): A list of file paths to the input image(s) depicting user requirements.
textual_user_requirement (str, optional): Textual user requirement that alongside the given images, if any.
legacy_output (str, optional): The intact textual user requirements generated by you last time, if any.
evaluation_conclusion (str, optional): Conclusion or evaluation based on the processed requirements.
additional_technical_requirements (str, optional): Any supplementary technical details relevant to the process.
Returns:
str: Textual representation of user requirements extracted from the provided image(s).
Raises:
ValueError: If image_paths list is empty.
OSError: If there is an issue accessing or reading the image files.
Example:
>>> images = ["requirements/pic/1.png", "requirements/pic/2.png", "requirements/pic/3.png"]
>>> textual_user_requirements = "User requirement paragraph 1 ..., ![](1.png). paragraph 2...![](2.png)..."
>>> action = Pic2Txt()
>>> intact_textual_user_requirements = await action.run(image_paths=images, textual_user_requirement=textual_user_requirements)
>>> print(intact_textual_user_requirements)
"User requirement paragraph 1 ..., ![...](1.png) This picture describes... paragraph 2...![...](2.png)..."
"""
descriptions = {}
for i in image_paths:
filename = Path(i)
base64_image = encode_image(filename)
rsp = await self._pic2txt(
"Generate a paragraph of text based on the content of the image, the language of the text is consistent with the language in the image.",
base64_image=base64_image,
)
descriptions[filename.name] = rsp
prompt = PROMPT.format(
textual_user_requirement=textual_user_requirement,
acknowledge=to_markdown_code_block(val=json.dumps(descriptions), type_="json"),
legacy_output=to_markdown_code_block(val=legacy_output),
evaluation_conclusion=evaluation_conclusion,
additional_technical_requirements=to_markdown_code_block(val=additional_technical_requirements),
)
return await self._write(prompt)
@retry(
wait=wait_random_exponential(min=1, max=20),
stop=stop_after_attempt(6),
after=general_after_log(logger),
)
async def _write(self, prompt: str) -> str:
rsp = await self.llm.aask(prompt)
return rsp
@retry(
wait=wait_random_exponential(min=1, max=20),
stop=stop_after_attempt(6),
after=general_after_log(logger),
)
async def _pic2txt(self, prompt: str, base64_image: str) -> str:
rsp = await self.llm.aask(prompt, images=base64_image)
return rsp
PROMPT = """
## Textual User Requirements
{textual_user_requirement}
## Acknowledge
{acknowledge}
## Legacy Outputs
{legacy_output}
## Evaluation Conclusion
{evaluation_conclusion}
## Additional Technical Requirements
{additional_technical_requirements}
---
You are a tool that generates an intact textual user requirements given a few of textual fragments of user requirements and some fragments of UI pictures.
The content of "Textual User Requirements" provides a few of textual fragments of user requirements;
The content of "Acknowledge" provides the descriptions of pictures used in "Textual User Requirements";
"Legacy Outputs" contains the intact textual user requirements generated by you last time, which you can improve by addressing the issues raised in "Evaluation Conclusion";
"Additional Technical Requirements" specifies the additional technical requirements that the generated textual user requirements must meet;
You need to merge the text content of the corresponding image in the "Acknowledge" into the "Textual User Requirements" to generate a complete, natural and coherent description of the user requirements;
Return the intact textual user requirements according to the given fragments of the user requirement of "Textual User Requirements" and the UI pictures;
"""

16
metagpt/actions/requirement_analysis/trd/__init__.py Normal file
View file

@ -0,0 +1,16 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : __init__.py
@Desc : The implementation of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
from metagpt.actions.requirement_analysis.trd.detect_interaction import DetectInteraction
from metagpt.actions.requirement_analysis.trd.evaluate_trd import EvaluateTRD
from metagpt.actions.requirement_analysis.trd.write_trd import WriteTRD
from metagpt.actions.requirement_analysis.trd.compress_external_interfaces import CompressExternalInterfaces
__all__ = [CompressExternalInterfaces, DetectInteraction, WriteTRD, EvaluateTRD]

58
metagpt/actions/requirement_analysis/trd/compress_external_interfaces.py Normal file
View file

@ -0,0 +1,58 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : compress_external_interfaces.py
@Desc : The implementation of Chapter 2.1.5 of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
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
@register_tool(include_functions=["run"])
class CompressExternalInterfaces(Action):
"""CompressExternalInterfaces deal with the following situations:
1. Given a natural text of acknowledgement, it extracts and compresses the information about external system interfaces.
"""
@retry(
wait=wait_random_exponential(min=1, max=20),
stop=stop_after_attempt(6),
after=general_after_log(logger),
)
async def run(
self,
*,
acknowledge: str,
) -> str:
"""
Extracts and compresses information about external system interfaces from a given acknowledgement text.
Args:
acknowledge (str): A natural text of acknowledgement containing details about external system interfaces.
Returns:
str: A compressed version of the information about external system interfaces.
Example:
>>> compress_acknowledge = CompressExternalInterfaces()
>>> acknowledge = "## Interfaces\\n..."
>>> available_external_interfaces = await compress_acknowledge.run(acknowledge=acknowledge)
>>> print(available_external_interfaces)
```json\n[\n{\n"id": 1,\n"inputs": {...
"""
return await self.llm.aask(
msg=acknowledge,
system_msgs=[
"Extracts and compresses the information about external system interfaces.",
"Return a markdown JSON list of objects, each object containing:\n"
'- an "id" key containing the interface id;\n'
'- an "inputs" key containing a dict of input parameters that consist of name and description pairs;\n'
'- an "outputs" key containing a dict of returns that consist of name and description pairs;\n',
],
)

101
metagpt/actions/requirement_analysis/trd/detect_interaction.py Normal file
View file

@ -0,0 +1,101 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : detect_interaction.py
@Desc : The implementation of Chapter 2.1.6 of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
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 DetectInteraction(Action):
"""DetectInteraction deal with the following situations:
1. Given a natural text of user requirements, it identifies the interaction events and the participants of those interactions from the original text.
"""
@retry(
wait=wait_random_exponential(min=1, max=20),
stop=stop_after_attempt(6),
after=general_after_log(logger),
)
async def run(
self,
*,
user_requirements: str,
use_case_actors: str,
legacy_interaction_events: str,
evaluation_conclusion: str,
) -> str:
"""
Identifies interaction events and participants from the user requirements.
Args:
user_requirements (str): A natural language text detailing the user's requirements.
use_case_actors (str): A description of the actors involved in the use case.
legacy_interaction_events (str): The previous version of the interaction events identified by you.
evaluation_conclusion (str): The external evaluation conclusions regarding the interactions events identified by you.
Returns:
str: A string summarizing the identified interaction events and their participants.
Example:
>>> detect_interaction = DetectInteraction()
>>> user_requirements = "User requirements 1. ..."
>>> use_case_actors = "- Actor: game player;\\n- System: snake game; \\n- External System: game center;"
>>> previous_version_interaction_events = "['interaction ...', ...]"
>>> evaluation_conclusion = "Issues: ..."
>>> interaction_events = await detect_interaction.run(
>>> user_requirements=user_requirements,
>>> use_case_actors=use_case_actors,
>>> legacy_interaction_events=previous_version_interaction_events,
>>> evaluation_conclusion=evaluation_conclusion,
>>> )
>>> print(interaction_events)
"['interaction ...', ...]"
"""
msg = PROMPT.format(
use_case_actors=use_case_actors,
original_user_requirements=to_markdown_code_block(val=user_requirements),
previous_version_of_interaction_events=legacy_interaction_events,
the_evaluation_conclusion_of_previous_version_of_trd=evaluation_conclusion,
)
return await self.llm.aask(msg=msg)
PROMPT = """
## Actor, System, External System
{use_case_actors}
## User Requirements
{original_user_requirements}
## Legacy Interaction Events
{previous_version_of_interaction_events}
## Evaluation Conclusion
{the_evaluation_conclusion_of_previous_version_of_trd}
---
You are a tool for capturing interaction events.
"Actor, System, External System" provides the possible participants of the interaction event;
"Legacy Interaction Events" is the contents of the interaction events that you output earlier;
Some descriptions in the "Evaluation Conclusion" relate to the content of "User Requirements", and these descriptions in the "Evaluation Conclusion" address some issues regarding the content of "Legacy Interaction Events";
You need to capture the interaction events occurring in the description within the content of "User Requirements" word-for-word, including:
1. Who is interacting with whom. An interaction event has a maximum of 2 participants. If there are multiple participants, it indicates that multiple events are combined into one event and should be further split;
2. When an interaction event occurs, who is the initiator? What data did the initiator enter?
3. What data does the interaction event ultimately return according to the "User Requirements"?
You can check the data flow described in the "User Requirements" to see if there are any missing interaction events;
Return a markdown JSON object list, each object of the list containing:
- a "name" key containing the name of the interaction event;
- a "participants" key containing a string list of the names of the two participants;
- a "initiator" key containing the name of the participant who initiate the interaction;
- a "input" key containing a natural text description about the input data;
"""

115
metagpt/actions/requirement_analysis/trd/evaluate_trd.py Normal file
View file

@ -0,0 +1,115 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : evaluate_trd.py
@Desc : The implementation of Chapter 2.1.6~2.1.7 of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
from metagpt.actions.requirement_analysis import EvaluateAction, EvaluationData
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.common import to_markdown_code_block
@register_tool(include_functions=["run"])
class EvaluateTRD(EvaluateAction):
"""EvaluateTRD deal with the following situations:
1. Given a TRD, evaluates the quality and returns a conclusion.
"""
async def run(
self,
*,
user_requirements: str,
use_case_actors: str,
trd: str,
interaction_events: str,
legacy_user_requirements_interaction_events: str = "",
) -> EvaluationData:
"""
Evaluates the given TRD based on user requirements, use case actors, interaction events, and optionally external legacy interaction events.
Args:
user_requirements (str): The requirements provided by the user.
use_case_actors (str): The actors involved in the use case.
trd (str): The TRD (Technical Requirements Document) to be evaluated.
interaction_events (str): The interaction events related to the user requirements and the TRD.
legacy_user_requirements_interaction_events (str, optional): External legacy interaction events tied to the user requirements. Defaults to an empty string.
Returns:
EvaluationData: The conclusion of the TRD evaluation.
Example:
>>> evaluate_trd = EvaluateTRD()
>>> user_requirements = "User requirements 1. ..."
>>> use_case_actors = "- Actor: game player;\\n- System: snake game; \\n- External System: game center;"
>>> trd = "## TRD\\n..."
>>> interaction_events = "['interaction ...', ...]"
>>> evaluation_conclusion = "Issues: ..."
>>> legacy_user_requirements_interaction_events = ["user requirements 1. ...", ...]
>>> evaluation = await evaluate_trd.run(
>>> user_requirements=user_requirements,
>>> use_case_actors=use_case_actors,
>>> trd=trd,
>>> interaction_events=interaction_events,
>>> legacy_user_requirements_interaction_events=str(legacy_user_requirements_interaction_events),
>>> )
>>> is_pass = evaluation.is_pass
>>> print(is_pass)
True
>>> evaluation_conclusion = evaluation.conclusion
>>> print(evaluation_conclusion)
## Conclustion\n balabalabala...
"""
prompt = PROMPT.format(
use_case_actors=use_case_actors,
user_requirements=to_markdown_code_block(val=user_requirements),
trd=to_markdown_code_block(val=trd),
legacy_user_requirements_interaction_events=legacy_user_requirements_interaction_events,
interaction_events=interaction_events,
)
return await self._vote(prompt)
PROMPT = """
## Actor, System, External System
{use_case_actors}
## User Requirements
{user_requirements}
## TRD Design
{trd}
## External Interaction Events
{legacy_user_requirements_interaction_events}
## Interaction Events
{legacy_user_requirements_interaction_events}
{interaction_events}
---
You are a tool to evaluate the TRD design.
"Actor, System, External System" provides the all possible participants in interaction events;
"User Requirements" provides the original requirements description, any parts not mentioned in this description will be handled by other modules, so do not fabricate requirements;
"External Interaction Events" is provided by an external module for your use, its content is also referred to "Interaction Events" section; The content in "External Interaction Events" can be determined to be problem-free;
"External Interaction Events" provides some identified interaction events and the interacting participants based on the part of the content of the "User Requirements";
"Interaction Events" provides some identified interaction events and the interacting participants based on the content of the "User Requirements";
"TRD Design" provides a comprehensive design of the implementation steps for the original requirements, incorporating the interaction events from "Interaction Events" and adding additional steps to connect the complete upstream and downstream data flows;
In order to integrate the full upstream and downstream data flow, the "TRD Design" allows for the inclusion of steps that do not appear in the original requirements description, but do not conflict with those explicitly described in the "User Requirements";
Which interactions from "Interaction Events" correspond to which steps in "TRD Design"? Please provide reasons.
Which aspects of "TRD Design" and "Interaction Events" do not align with the descriptions in "User Requirements"? Please provide detailed descriptions and reasons.
If the descriptions in "User Requirements" are divided into multiple steps in "TRD Design" and "Interaction Events," it can be considered compliant with the descriptions in "User Requirements" as long as it does not conflict with them;
There is a possibility of missing details in the descriptions of "User Requirements". Any additional steps in "TRD Design" and "Interaction Events" are considered compliant with "User Requirements" as long as they do not conflict with the descriptions provided in "User Requirements";
If there are interaction events with external systems in "TRD Design", you must explicitly specify the ID of the external interface to use for the interaction events, the input and output parameters of the used external interface must explictly match the input and output of the interaction event；
Does the sequence of steps in "Interaction Events" cause performance or cost issues? Please provide detailed descriptions and reasons;
If each step of "TRD Design" has input data, its input data is provided either by the output of the previous steps or by participants of "Actor, System, External System", and there should be no passive data;
Return a markdown JSON object with:
- an "issues" key containing a string list of natural text about the issues that need to be addressed, found in the "TRD Design" if any exist, each issue found must provide a detailed description and include reasons;
- a "conclusion" key containing the evaluation conclusion;
- a "correspondence_between" key containing the judgement detail of the natural text string list about the correspondence between "Interaction Events" and "TRD Design" steps;
- a "misalignment" key containing the judgement detail of the natural text string list about the misalignment with "User Requirements";
- a "is_pass" key containing a true boolean value if there is not any issue in the "TRD Design";
"""

261
metagpt/actions/requirement_analysis/trd/write_trd.py Normal file
View file

@ -0,0 +1,261 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/6/13
@Author : mashenquan
@File : write_trd.py
@Desc : The implementation of Chapter 2.1.6~2.1.7 of RFC243. https://deepwisdom.feishu.cn/wiki/QobGwPkImijoyukBUKHcrYetnBb
"""
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,
evaluation_conclusion: str = "",
interaction_events: str,
previous_version_trd: str = "",
legacy_user_requirements: str = "",
legacy_user_requirements_trd: str = "",
legacy_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): The new/incremental user requirements.
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): The conclusion of the evaluation of the TRD written by you. Defaults to an empty string.
interaction_events (str): The interaction events related to the user requirements that you are handling.
previous_version_trd (str, optional): The previous version of the TRD written by you, for updating.
legacy_user_requirements (str, optional): Existing user requirements handled by an external object for your use. Defaults to an empty string.
legacy_user_requirements_trd (str, optional): The TRD associated with the existing user requirements handled by an external object for your use. Defaults to an empty string.
legacy_user_requirements_interaction_events (str, optional): Interaction events related to the existing user requirements handled by an external object for your use. Defaults to an empty string.
Returns:
str: The newly created or updated TRD written by you.
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()
>>> new_version_trd = await write_trd.run(
>>> user_requirements=user_requirements,
>>> use_case_actors=use_case_actors,
>>> available_external_interfaces=available_external_interfaces,
>>> evaluation_conclusion=evaluation_conclusion,
>>> interaction_events=interaction_events,
>>> previous_version_trd=previous_version_trd,
>>> )
>>> 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. ...", ...]
>>> 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 ...", ...]
>>> 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 ..."
>>> increment_requirements = "The incremental user requirements are ..."
>>> evaluation_conclusion = "Conclusion ..." # The conclusion returned by `EvaluateTRD.run` if there is.
>>> previous_version_trd = "TRD ..." # The last version of the TRD written out if there is.
>>> write_trd = WriteTRD()
>>> new_version_trd = await write_trd.run(
>>> user_requirements=increment_requirements,
>>> use_case_actors=use_case_actors,
>>> available_external_interfaces=available_external_interfaces,
>>> evaluation_conclusion=evaluation_conclusion,
>>> interaction_events=interaction_events,
>>> previous_version_trd=previous_version_trd,
>>> legacy_user_requirements=str(legacy_user_requirements),
>>> legacy_user_requirements_trd=legacy_user_requirements_trd,
>>> legacy_user_requirements_interaction_events=str(legacy_user_requirements_interaction_events),
>>> )
>>> print(new_version_trd)
## Technical Requirements Document\n ...
"""
if legacy_user_requirements:
return await self._write_incremental_trd(
use_case_actors=use_case_actors,
legacy_user_requirements=legacy_user_requirements,
available_external_interfaces=available_external_interfaces,
legacy_user_requirements_trd=legacy_user_requirements_trd,
legacy_user_requirements_interaction_events=legacy_user_requirements_interaction_events,
incremental_user_requirements=user_requirements,
previous_version_trd=previous_version_trd,
evaluation_conclusion=evaluation_conclusion,
incremental_user_requirements_interaction_events=interaction_events,
)
return await self._write_new_trd(
use_case_actors=use_case_actors,
original_user_requirement=user_requirements,
available_external_interfaces=available_external_interfaces,
legacy_trd=previous_version_trd,
evaluation_conclusion=evaluation_conclusion,
interaction_events=interaction_events,
)
@retry(
wait=wait_random_exponential(min=1, max=20),
stop=stop_after_attempt(6),
after=general_after_log(logger),
)
async def _write_new_trd(
self,
*,
use_case_actors: str,
original_user_requirement: str,
available_external_interfaces: str,
legacy_trd: str,
evaluation_conclusion: str,
interaction_events: str,
) -> str:
prompt = NEW_PROMPT.format(
use_case_actors=use_case_actors,
original_user_requirement=to_markdown_code_block(val=original_user_requirement),
available_external_interfaces=available_external_interfaces,
legacy_trd=to_markdown_code_block(val=legacy_trd),
evaluation_conclusion=evaluation_conclusion,
interaction_events=interaction_events,
)
return await self.llm.aask(prompt)
@retry(
wait=wait_random_exponential(min=1, max=20),
stop=stop_after_attempt(6),
after=general_after_log(logger),
)
async def _write_incremental_trd(
self,
*,
use_case_actors: str,
legacy_user_requirements: str,
available_external_interfaces: str,
legacy_user_requirements_trd: str,
legacy_user_requirements_interaction_events: str,
incremental_user_requirements: str,
previous_version_trd: str,
evaluation_conclusion: str,
incremental_user_requirements_interaction_events: str,
):
prompt = INCREMENTAL_PROMPT.format(
use_case_actors=use_case_actors,
legacy_user_requirements=to_markdown_code_block(val=legacy_user_requirements),
available_external_interfaces=available_external_interfaces,
legacy_user_requirements_trd=to_markdown_code_block(val=legacy_user_requirements_trd),
legacy_user_requirements_interaction_events=legacy_user_requirements_interaction_events,
incremental_user_requirements=to_markdown_code_block(val=incremental_user_requirements),
previous_version_trd=to_markdown_code_block(val=previous_version_trd),
evaluation_conclusion=evaluation_conclusion,
incremental_user_requirements_interaction_events=incremental_user_requirements_interaction_events,
)
return await self.llm.aask(prompt)
NEW_PROMPT = """
## Actor, System, External System
{use_case_actors}
## User Requirements
{original_user_requirement}
## Available External Interfaces
{available_external_interfaces}
## Legacy TRD
{legacy_trd}
## Evaluation Conclusion
{evaluation_conclusion}
## Interaction Events
{interaction_events}
---
You are a TRD generator.
The content of "Actor, System, External System" provides an explanation of actors and systems that appear in UML Use Case diagram;
The content of "Available External Interfaces" provides the candidate steps, along with the inputs and outputs of each step;
"User Requirements" provides the original requirements description, any parts not mentioned in this description will be handled by other modules, so do not fabricate requirements;
"Legacy TRD" provides the old version of the TRD based on the "User Requirements" and can serve as a reference for the new TRD;
"Evaluation Conclusion" provides a summary of the evaluation of the old TRD in the "Legacy TRD" and can serve as a reference for the new TRD;
"Interaction Events" provides some identified interaction events and the interacting participants based on the content of the "User Requirements";
1. What inputs and outputs are described in the "User Requirements"?
2. How many steps are needed to achieve the inputs and outputs described in the "User Requirements"? Which actors from the "Actor, System, External System" section are involved in each step? What are the inputs and outputs of each step? Where is this output used, for example, as input for which interface or where it is required in the requirements, etc.?
3. Output a complete Technical Requirements Document (TRD)：
3.1. In the description, use the actor and system names defined in the "Actor, System, External System" section to describe the interactors;
3.2. The content should include the original text of the requirements from "User Requirements";
3.3. In the TRD, each step can involve a maximum of two participants. If there are more than two participants, the step needs to be further split;
3.4. In the TRD, each step must include detailed descriptions, inputs, outputs, participants, initiator, and the rationale for the step's existence. The rationale should reference the original text to justify it, such as specifying which interface requires the output of this step as parameters or where in the requirements this step is mandated, etc.;
3.5. In the TRD, if you need to call interfaces of external systems, you must explicitly specify the interface IDs of the external systems you want to call;
"""
INCREMENTAL_PROMPT = """
## Actor, System, External System
{use_case_actors}
## Legacy User Requirements
{legacy_user_requirements}
## Available External Interfaces
{available_external_interfaces}
## The TRD of Legacy User Requirements
{legacy_user_requirements_trd}
## The Interaction Events of Legacy User Requirements
{legacy_user_requirements_interaction_events}
## Incremental Requirements
{incremental_user_requirements}
## Legacy TRD
{previous_version_trd}
## Evaluation Conclusion
{evaluation_conclusion}
## Interaction Events
{incremental_user_requirements_interaction_events}
---
You are a TRD generator.
The content of "Actor, System, External System" provides an explanation of actors and systems that appear in UML Use Case diagram;
The content of "Available External Interfaces" provides the candidate steps, along with the inputs and outputs of each step;
"Legacy User Requirements" provides the original requirements description handled by other modules for your use;
"The TRD of Legacy User Requirements" is the TRD generated by other modules based on the "Legacy User Requirements" for your use;
"The Interaction Events of Legacy User Requirements" is the interaction events list generated by other modules based on the "Legacy User Requirements" for your use;
"Incremental Requirements" provides the original requirements description that you need to address, any parts not mentioned in this description will be handled by other modules, so do not fabricate requirements;
The requirements in "Legacy User Requirements" combined with the "Incremental Requirements" form a complete set of requirements, therefore, you need to add the TRD portion of the "Incremental Requirements" to "The TRD of Legacy User Requirements", the added content must not conflict with the original content of "The TRD of Legacy User Requirements";
"Legacy TRD" provides the old version of the TRD you previously wrote based on the "Incremental Requirements" and can serve as a reference for the new TRD;
"Evaluation Conclusion" provides a summary of the evaluation of the old TRD you generated in the "Legacy TRD", and the identified issues can serve as a reference for the new TRD you create;
"Interaction Events" provides some identified interaction events and the interacting participants based on the content of the "Incremental Requirements";
1. What inputs and outputs are described in the "Incremental Requirements"？
2. How many steps are needed to achieve the inputs and outputs described in the "Incremental Requirements"? Which actors from the "Actor, System, External System" section are involved in each step? What are the inputs and outputs of each step? Where is this output used, for example, as input for which interface or where it is required in the requirements, etc.?
3. Output a complete Technical Requirements Document (TRD)：
3.1. In the description, use the actor and system names defined in the "Actor, System, External System" section to describe the interactors;
3.2. The content should include the original text of the requirements from "User Requirements";
3.3. In the TRD, each step can involve a maximum of two participants. If there are more than two participants, the step needs to be further split;
3.4. In the TRD, each step must include detailed descriptions, inputs, outputs, participants, initiator, and the rationale for the step's existence. The rationale should reference the original text to justify it, such as specifying which interface requires the output of this step as parameters or where in the requirements this step is mandated, etc.
"""

1
metagpt/prompts/di/team_leader.py
View file

@ -21,6 +21,7 @@ Note:
4. If the requirement is a common-sense, logical, or math problem, you should respond directly without assigning any task to team members.
5. If you think the requirement is not clear or ambiguous, you should ask the user for clarification immediately. Assign tasks only after all info is clear.
6. It is helpful for Engineer to have both the system design and the project schedule for writing the code, so include paths of both files (if available) and remind Engineer to definitely read them when publishing message to Engineer.
7. If the requirement is writing a TRD and software framework, you should assign it to Architect. When publishing message to Architect, you should directly copy the full original user requirement.
"""
FINISH_CURRENT_TASK_CMD = """

2
metagpt/provider/base_llm.py
View file

@ -65,7 +65,7 @@ class BaseLLM(ABC):
# image url or image base64
url = image if image.startswith("http") else f"data:image/jpeg;base64,{image}"
# it can with multiple-image inputs
content.append({"type": "image_url", "image_url": url})
content.append({"type": "image_url", "image_url": {"url": url}})
return {"role": "user", "content": content}
def _assistant_msg(self, msg: str) -> dict[str, str]:

13
metagpt/provider/openai_api.py
View file

@ -40,8 +40,17 @@ from metagpt.utils.token_counter import (
)
@register_provider([LLMType.OPENAI, LLMType.FIREWORKS, LLMType.OPEN_LLM, LLMType.MOONSHOT, LLMType.MISTRAL, LLMType.YI,
LLMType.OPEN_ROUTER])
@register_provider(
[
LLMType.OPENAI,
LLMType.FIREWORKS,
LLMType.OPEN_LLM,
LLMType.MOONSHOT,
LLMType.MISTRAL,
LLMType.YI,
LLMType.OPEN_ROUTER,
]
)
class OpenAILLM(BaseLLM):
"""Check https://platform.openai.com/examples for examples"""

19
metagpt/roles/architect.py
View file

@ -8,6 +8,8 @@
from metagpt.actions import WritePRD
from metagpt.actions.design_api import WriteDesign
from metagpt.roles.di.role_zero import RoleZero
from metagpt.tools.libs.software_development import write_trd_and_framework
from metagpt.utils.common import tool2name
class Architect(RoleZero):
@ -29,9 +31,14 @@ class Architect(RoleZero):
"libraries. Use same language as user requirement"
)
instruction: str = """Use WriteDesign tool to write a system design document"""
instruction: str = """Use WriteDesign tool to write a system design document if a system design is required; Use `write_trd_and_framework` tool to write a software framework if a software framework is required;"""
max_react_loop: int = 1 # FIXME: Read and edit files requires more steps, consider later
tools: list[str] = ["Editor:write,read,write_content", "RoleZero", "WriteDesign"]
tools: list[str] = [
"Editor:write,read,write_content",
"RoleZero",
"WriteDesign",
write_trd_and_framework.__name__,
]
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
@ -45,11 +52,11 @@ class Architect(RoleZero):
self._watch({WritePRD})
def _update_tool_execution(self):
wd = WriteDesign()
write_design = WriteDesign()
self.tool_execution_map.update(tool2name(WriteDesign, ["run"], write_design.run))
self.tool_execution_map.update(
{
"WriteDesign.run": wd.run,
"WriteDesign": wd.run, # alias
"run": wd.run, # alias
write_trd_and_framework.__name__: write_trd_and_framework,
"run": write_design.run, # alias
}
)

2
metagpt/roles/di/data_analyst.py
View file

@ -83,7 +83,7 @@ class DataAnalyst(DataInterpreter):
# print(*context, sep="\n" + "*" * 5 + "\n")
async with ThoughtReporter(enable_llm_stream=True):
rsp = await self.llm.aask(context)
self.commands = json.loads(CodeParser.parse_code(block=None, lang='json', text=rsp))
self.commands = json.loads(CodeParser.parse_code(block=None, lang="json", text=rsp))
self.rc.working_memory.add(Message(content=rsp, role="assistant"))
await run_commands(self, self.commands, self.rc.working_memory)

10
metagpt/roles/di/role_zero.py
View file

@ -11,7 +11,11 @@ from pydantic import model_validator
from metagpt.actions import Action
from metagpt.actions.di.run_command import RunCommand
from metagpt.logs import logger
from metagpt.prompts.di.role_zero import CMD_PROMPT, ROLE_INSTRUCTION, JSON_REPAIR_PROMPT
from metagpt.prompts.di.role_zero import (
CMD_PROMPT,
JSON_REPAIR_PROMPT,
ROLE_INSTRUCTION,
)
from metagpt.roles import Role
from metagpt.schema import AIMessage, Message, UserMessage
from metagpt.strategy.experience_retriever import DummyExpRetriever, ExpRetriever
@ -21,8 +25,8 @@ from metagpt.tools.libs.editor import Editor
from metagpt.tools.tool_recommend import BM25ToolRecommender, ToolRecommender
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.common import CodeParser
from metagpt.utils.repair_llm_raw_output import RepairType, repair_llm_raw_output
from metagpt.utils.report import ThoughtReporter
from metagpt.utils.repair_llm_raw_output import repair_llm_raw_output, RepairType
@register_tool(include_functions=["ask_human", "reply_to_human"])
@ -166,7 +170,7 @@ class RoleZero(Role):
try:
commands = CodeParser.parse_code(block=None, lang="json", text=self.command_rsp)
commands = json.loads(repair_llm_raw_output(output=commands, req_keys=[None], repair_type=RepairType.JSON))
except json.JSONDecodeError as e:
except json.JSONDecodeError:
commands = await self.llm.aask(msg=JSON_REPAIR_PROMPT.format(json_data=self.command_rsp))
commands = json.loads(CodeParser.parse_code(block=None, lang="json", text=commands))
except Exception as e:

16
metagpt/roles/product_manager.py
View file

@ -9,9 +9,10 @@
from metagpt.actions import UserRequirement, WritePRD
from metagpt.actions.prepare_documents import PrepareDocuments
from metagpt.actions.requirement_analysis.requirement.pic2txt import Pic2Txt
from metagpt.roles.di.role_zero import RoleZero
from metagpt.roles.role import RoleReactMode
from metagpt.utils.common import any_to_name, any_to_str
from metagpt.utils.common import any_to_name, any_to_str, tool2name
from metagpt.utils.git_repository import GitRepository
@ -32,9 +33,9 @@ class ProductManager(RoleZero):
constraints: str = "utilize the same language as the user requirements for seamless communication"
todo_action: str = any_to_name(WritePRD)
instruction: str = """Use WritePRD tool to write PRD"""
instruction: str = """Use WritePRD tool to write PRD if a PRD is required; Use `Pic2Txt` tool to write out an intact textual user requirements if an intact textual user requiremnt is required given some images alongside the contextual textual descriptions;"""
max_react_loop: int = 1 # FIXME: Read and edit files requires more steps, consider later
tools: list[str] = ["Editor:write,read,write_content", "RoleZero", "WritePRD"]
tools: list[str] = ["Editor:write,read,write_content", "RoleZero", "WritePRD", Pic2Txt.__name__]
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
@ -47,12 +48,9 @@ class ProductManager(RoleZero):
def _update_tool_execution(self):
wp = WritePRD()
self.tool_execution_map.update(
{
"WritePRD.run": wp.run,
"WritePRD": wp.run, # alias
}
)
self.tool_execution_map.update(tool2name(WritePRD, ["run"], wp.run))
pic2txt = Pic2Txt()
self.tool_execution_map.update(tool2name(Pic2Txt, ["run"], pic2txt.run))
async def _think(self) -> bool:
"""Decide what to do"""

453
metagpt/strategy/experience_retriever.py
View file

@ -14,7 +14,458 @@ class DummyExpRetriever(ExpRetriever):
"""A dummy experience retriever that returns empty string."""
def retrieve(self, context: str = "") -> str:
return ""
return self.EXAMPLE
EXAMPLE: str = ""
class TRDAllExpRetriever(ExpRetriever):
def retrieve(self, context: str = "") -> str:
return self.EXAMPLE
EXAMPLE: str = """
## example 1
User Requirement: Given some user requirements, write a software framework.
Explanation: Given a complete user requirement, to write a TRD and software framework, you must follow all of the following steps to complete the TRD output required by the user: 1. Call 'write_trd' to generate TRD; 2. Call 'write_framework' to implement TRD into the software framework.
```json
[
{
"command_name": "write_trd_and_framework",
"task_id": "1",
"dependent_task_ids": [],
"instruction": "Execute `write_trd_and_framework` to write a TRD and software framework based on user requirements",
"args": {
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"additional_technical_requirements": "These are additional technical requirements, balabala..."
}
}
]
```
## example 2
User Requirement: Given some user requirements, write a software framework.
Explanation: Given a complete user requirement, to write a software framework, you must follow all of the following steps to complete the TRD output required by the user: 1. Call 'write_trd' to generate TRD; 2. Call 'write_framework' to implement TRD into the software framework.
```json
[
{
"command_name": "write_trd",
"task_id": "1",
"dependent_task_ids": [],
"instruction": "Execute `write_trd` to write the TRD based on user requirements",
"args": {
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
}
},
{
"command_name": "write_framework",
"task_id": "2",
"dependent_task_ids": ["1"],
"instruction": "Execute `write_framework` to write the framework based on the TRD",
"args": {
"use_case_actors": "These are actors involved in the use case, balabala...",
"trd": "<trd> returned by `write_trd`",
"additional_technical_requirements": "These are additional technical requirements, balabala..."
}
}
]
```
## example 3
User Requirement: Given some user requirements, write a TRD, and implement the TRD within a software framework.
Explanation:
Given a complete requirement, 要写TRD需要follow如下步骤：
1. 调用`CompressExternalInterfaces.run`，从acknowledgement中抽取external interfaces的信息；
2. 按顺序执行如下步骤：
2.1. 执行`DetectInteraction.run`;
2.2. 执行`WriteTRD.run`;
2.3. 执行`EvaluateTRD.run`;
2.4. 检查`EvaluateTRD.run`的结果：
2.4.1. 如果`EvaluateTRD.run`的结果被判定为pass，则执行步骤3；
2.4.2. 如果`EvaluateTRD.run`的结果被判定为deny,则继续执行步骤2；
3. 按顺序执行如下步骤：
3.1. 执行`WriteFramework.run`;
3.2. 执行`EvaluateFramework.run`;
3.3. 检查`EvaluateFramework.run`的结果：
3.3.1. 如果`EvaluateFramework.run`的结果被判定为pass，则执行步骤4；
3.3.2. 如果`EvaluateFramework.run`的结果被判定为deny,则继续执行步骤3；
3.3.3. 如果已经重复执行步骤3超过9次，则执行步骤4；
4. 执行`save_framework`,将`WriteFramework.run`的结果保存下来；
```json
[
{
"command_name": "CompressExternalInterfaces.run",
"args": {
"task_id": "1",
"dependent_task_ids": [],
"instruction": "Execute `DetectInteraction.run` to extract external interfaces information from acknowledgement.",
"acknowledge": "## Interfaces\n balabala..."
}
},
{
"command_name": "DetectInteraction.run",
"args": {
"task_id": "2",
"dependent_task_ids": ["1"],
"instruction": "Execute `DetectInteraction.run` to extract external interfaces information from acknowledgement.",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
}
},
{
"command_name": "WriteTRD.run",
"args": {
"task_id": "3",
"dependent_task_ids": ["2"],
"instruction": "Execute `WriteTRD.run` to write TRD",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"available_external_interfaces": "<compressed_external_interfaces_output> returned by `CompressExternalInterfaces.run`",
"interaction_events": "<detected_interaction_events_output> returned by `DetectInteraction.run`"
}
},
{
"command_name": "EvaluateTRD.run",
"args": {
"task_id": "4",
"dependent_task_ids": ["3"],
"instruction": "Execute `EvaluateTRD.run` to evaluate the TRD",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"available_external_interfaces": "<compressed_external_interfaces_output> returned by `CompressExternalInterfaces.run`",
"interaction_events": "<detected_interaction_events_output>",
"trd": "<trd> returned by `EvaluateTRD.run`"
}
},
{
"command_name": "DetectInteraction.run",
"args": {
"task_id": "5",
"dependent_task_ids": ["4"],
"instruction": "Execute `DetectInteraction.run` to extract external interfaces information from acknowledgement.",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"evaluation_conclusion": "<evaluation_conclusion> returned by `EvaluateTRD.run`"
}
},
{
"command_name": "WriteTRD.run",
"args": {
"task_id": "6",
"dependent_task_ids": ["5"],
"instruction": "Execute `WriteTRD.run` to write TRD",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"available_external_interfaces": "<compressed_external_interfaces_output> returned by `CompressExternalInterfaces.run`",
"interaction_events": "<detected_interaction_events_output> returned by `DetectInteraction.run`",
"previous_version_trd": "<trd> returned by `WriteTRD.run`"
}
},
{
"command_name": "EvaluateTRD.run",
"args": {
"task_id": "7",
"dependent_task_ids": ["6"],
"instruction": "Execute `EvaluateTRD.run` to evaluate the TRD",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"available_external_interfaces": "<compressed_external_interfaces_output> returned by `CompressExternalInterfaces.run`",
"interaction_events": "<detected_interaction_events_output> returned by `DetectInteraction.run`",
"trd": "<trd> returned by `WriteTRD.run`",
}
},
{
"command_name": "WriteFramework.run",
"args": {
"task_id": "8",
"dependent_task_ids": ["7"],
"instruction": "Execute `WriteFramework.run` to write a software framework according to the TRD",
"use_case_actors": "These are actors involved in the use case, balabala...",
"trd": "<trd> returned by `WriteTRD.run`",
"acknowledge": "## Interfaces\n balabala...",
"additional_technical_requirements": "These are additional technical requirements, balabala...",
}
},
{
"command_name": "EvaluateFramework.run",
"args": {
"task_id": "9",
"dependent_task_ids": ["8"],
"instruction": "Execute `EvaluateFramework.run` to evaluate the software framework returned by `WriteFramework.run`",
"use_case_actors": "These are actors involved in the use case, balabala...",
"trd": "<trd> returned by `WriteTRD.run`",
"acknowledge": "## Interfaces\n balabala...",
"legacy_output": "<framework> returned by `WriteFramework.run`",
"additional_technical_requirements": "These are additional technical requirements, balabala...",
}
},
{
"command_name": "WriteFramework.run",
"args": {
"task_id": "10",
"dependent_task_ids": ["9"],
"instruction": "Execute `WriteFramework.run` to write a software framework according to the TRD",
"use_case_actors": "These are actors involved in the use case, balabala...",
"trd": "<trd> returned by `WriteTRD.run`",
"acknowledge": "## Interfaces\n balabala...",
"additional_technical_requirements": "These are additional technical requirements, balabala...",
}
},
{
"command_name": "EvaluateFramework.run",
"args": {
"task_id": "11",
"dependent_task_ids": ["10"],
"instruction": "Execute `EvaluateFramework.run` to evaluate the software framework returned by `WriteFramework.run`",
"use_case_actors": "These are actors involved in the use case, balabala...",
"trd": "<trd> returned by `WriteTRD.run`",
"acknowledge": "## Interfaces\n balabala...",
"legacy_output": "<framework> returned by `WriteFramework.run`",
"additional_technical_requirements": "These are additional technical requirements, balabala...",
}
},
{
"command_name": "save_framework",
"args": {
"task_id": "12",
"dependent_task_ids": ["11"],
"instruction": "Execute `save_framework` to save the software framework returned by `WriteFramework.run`",
"dir_data": "<framework> returned by `WriteFramework.run`",
}
}
]
```
"""
class TRDToolExpRetriever(ExpRetriever):
"""A TRD-related experience retriever that returns empty string."""
def retrieve(self, context: str = "") -> str:
return self.EXAMPLE
EXAMPLE: str = """
## example 1
User Requirement: Given some user requirements, write a software framework.
Explanation: Given a complete user requirement, to write a TRD and software framework, you must follow all of the following steps to complete the TRD output required by the user: 1. Call 'write_trd' to generate TRD; 2. Call 'write_framework' to implement TRD into the software framework.
```json
[
{
"command_name": "write_trd_and_framework",
"task_id": "1",
"dependent_task_ids": [],
"instruction": "Execute `write_trd_and_framework` to write a TRD and software framework based on user requirements",
"args": {
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"additional_technical_requirements": "These are additional technical requirements, balabala..."
}
}
]
"""
# EXAMPLE: str = """
# ## example 1
# User Requirement: Given some user requirements, write a software framework.
# Explanation: Given a complete user requirement, to write a software framework, you must follow all of the following steps to complete the TRD output required by the user: 1. Call 'write_trd' to generate TRD; 2. Call 'write_framework' to implement TRD into the software framework.
# ```json
# [
# {
# "command_name": "write_trd",
# "task_id": "1",
# "dependent_task_ids": [],
# "instruction": "Execute `write_trd` to write the TRD based on user requirements",
# "args": {
# "user_requirements": "This is user requirement balabala...",
# "use_case_actors": "These are actors involved in the use case, balabala...",
# }
# },
# {
# "command_name": "write_framework",
# "task_id": "2",
# "dependent_task_ids": ["1"],
# "instruction": "Execute `write_framework` to write the framework based on the TRD",
# "args": {
# "use_case_actors": "These are actors involved in the use case, balabala...",
# "trd": "<trd> returned by `write_trd`",
# "additional_technical_requirements": "These are additional technical requirements, balabala..."
# }
# }
# ]
# ```
# """
class TRDExpRetriever(ExpRetriever):
"""A TRD-related experience retriever that returns empty string."""
def retrieve(self, context: str = "") -> str:
return self.EXAMPLE
EXAMPLE: str = """
## example 1
User Requirement: Given some user requirements, write a TRD, and implement the TRD within a software framework.
Explanation:
Given a complete requirement, 要写TRD需要follow如下步骤：
1. 调用`CompressExternalInterfaces.run`，从acknowledgement中抽取external interfaces的信息；
2. 按顺序执行如下步骤：
2.1. 执行`DetectInteraction.run`;
2.2. 执行`WriteTRD.run`;
2.3. 执行`EvaluateTRD.run`;
2.4. 检查`EvaluateTRD.run`的结果：
2.4.1. 如果`EvaluateTRD.run`的结果被判定为pass，则执行步骤3；
2.4.2. 如果`EvaluateTRD.run`的结果被判定为deny,则继续执行步骤2；
3. 按顺序执行如下步骤：
3.1. 执行`WriteFramework.run`;
3.2. 执行`EvaluateFramework.run`;
3.3. 检查`EvaluateFramework.run`的结果：
3.3.1. 如果`EvaluateFramework.run`的结果被判定为pass，则执行步骤4；
3.3.2. 如果`EvaluateFramework.run`的结果被判定为deny,则继续执行步骤3；
3.3.3. 如果已经重复执行步骤3超过9次，则执行步骤4；
4. 执行`save_framework`,将`WriteFramework.run`的结果保存下来；
```json
[
{
"command_name": "CompressExternalInterfaces.run",
"args": {
"task_id": "1",
"dependent_task_ids": [],
"instruction": "Execute `DetectInteraction.run` to extract external interfaces information from acknowledgement.",
"acknowledge": "## Interfaces\n balabala..."
}
},
{
"command_name": "DetectInteraction.run",
"args": {
"task_id": "2",
"dependent_task_ids": ["1"],
"instruction": "Execute `DetectInteraction.run` to extract external interfaces information from acknowledgement.",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
}
},
{
"command_name": "WriteTRD.run",
"args": {
"task_id": "3",
"dependent_task_ids": ["2"],
"instruction": "Execute `WriteTRD.run` to write TRD",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"available_external_interfaces": "<compressed_external_interfaces_output> returned by `CompressExternalInterfaces.run`",
"interaction_events": "<detected_interaction_events_output> returned by `DetectInteraction.run`"
}
},
{
"command_name": "EvaluateTRD.run",
"args": {
"task_id": "4",
"dependent_task_ids": ["3"],
"instruction": "Execute `EvaluateTRD.run` to evaluate the TRD",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"available_external_interfaces": "<compressed_external_interfaces_output> returned by `CompressExternalInterfaces.run`",
"interaction_events": "<detected_interaction_events_output>",
"trd": "<trd> returned by `EvaluateTRD.run`"
}
},
{
"command_name": "DetectInteraction.run",
"args": {
"task_id": "5",
"dependent_task_ids": ["4"],
"instruction": "Execute `DetectInteraction.run` to extract external interfaces information from acknowledgement.",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"evaluation_conclusion": "<evaluation_conclusion> returned by `EvaluateTRD.run`"
}
},
{
"command_name": "WriteTRD.run",
"args": {
"task_id": "6",
"dependent_task_ids": ["5"],
"instruction": "Execute `WriteTRD.run` to write TRD",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"available_external_interfaces": "<compressed_external_interfaces_output> returned by `CompressExternalInterfaces.run`",
"interaction_events": "<detected_interaction_events_output> returned by `DetectInteraction.run`",
"previous_version_trd": "<trd> returned by `WriteTRD.run`"
}
},
{
"command_name": "EvaluateTRD.run",
"args": {
"task_id": "7",
"dependent_task_ids": ["6"],
"instruction": "Execute `EvaluateTRD.run` to evaluate the TRD",
"user_requirements": "This is user requirement balabala...",
"use_case_actors": "These are actors involved in the use case, balabala...",
"available_external_interfaces": "<compressed_external_interfaces_output> returned by `CompressExternalInterfaces.run`",
"interaction_events": "<detected_interaction_events_output> returned by `DetectInteraction.run`",
"trd": "<trd> returned by `WriteTRD.run`",
}
},
{
"command_name": "WriteFramework.run",
"args": {
"task_id": "8",
"dependent_task_ids": ["7"],
"instruction": "Execute `WriteFramework.run` to write a software framework according to the TRD",
"use_case_actors": "These are actors involved in the use case, balabala...",
"trd": "<trd> returned by `WriteTRD.run`",
"acknowledge": "## Interfaces\n balabala...",
"additional_technical_requirements": "These are additional technical requirements, balabala...",
}
},
{
"command_name": "EvaluateFramework.run",
"args": {
"task_id": "9",
"dependent_task_ids": ["8"],
"instruction": "Execute `EvaluateFramework.run` to evaluate the software framework returned by `WriteFramework.run`",
"use_case_actors": "These are actors involved in the use case, balabala...",
"trd": "<trd> returned by `WriteTRD.run`",
"acknowledge": "## Interfaces\n balabala...",
"legacy_output": "<framework> returned by `WriteFramework.run`",
"additional_technical_requirements": "These are additional technical requirements, balabala...",
}
},
{
"command_name": "WriteFramework.run",
"args": {
"task_id": "10",
"dependent_task_ids": ["9"],
"instruction": "Execute `WriteFramework.run` to write a software framework according to the TRD",
"use_case_actors": "These are actors involved in the use case, balabala...",
"trd": "<trd> returned by `WriteTRD.run`",
"acknowledge": "## Interfaces\n balabala...",
"additional_technical_requirements": "These are additional technical requirements, balabala...",
}
},
{
"command_name": "EvaluateFramework.run",
"args": {
"task_id": "11",
"dependent_task_ids": ["10"],
"instruction": "Execute `EvaluateFramework.run` to evaluate the software framework returned by `WriteFramework.run`",
"use_case_actors": "These are actors involved in the use case, balabala...",
"trd": "<trd> returned by `WriteTRD.run`",
"acknowledge": "## Interfaces\n balabala...",
"legacy_output": "<framework> returned by `WriteFramework.run`",
"additional_technical_requirements": "These are additional technical requirements, balabala...",
}
},
{
"command_name": "save_framework",
"args": {
"task_id": "12",
"dependent_task_ids": ["11"],
"instruction": "Execute `save_framework` to save the software framework returned by `WriteFramework.run`",
"dir_data": "<framework> returned by `WriteFramework.run`",
}
}
]
```
"""
TL_EXAMPLE = """

220
metagpt/tools/libs/software_development.py
View file

@ -2,10 +2,28 @@
# -*- coding: utf-8 -*-
from __future__ import annotations
import uuid
from datetime import datetime
from pathlib import Path
from typing import Optional
from metagpt.const import ASSISTANT_ALIAS
from metagpt.logs import ToolLogItem, log_tool_output
from metagpt.actions.requirement_analysis.framework import (
EvaluateFramework,
WriteFramework,
save_framework,
)
from metagpt.actions.requirement_analysis.trd import (
CompressExternalInterfaces,
DetectInteraction,
EvaluateTRD,
WriteTRD,
)
from metagpt.const import ASSISTANT_ALIAS, DEFAULT_WORKSPACE_ROOT, TEST_DATA_PATH
from metagpt.context import Context
from metagpt.logs import ToolLogItem, log_tool_output, logger
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.common import aread
from metagpt.utils.cost_manager import CostManager
async def import_git_repo(url: str) -> Path:
@ -42,3 +60,201 @@ async def import_git_repo(url: str) -> Path:
log_tool_output(output=outputs, tool_name=import_git_repo.__name__)
return ctx.repo.workdir
async def extract_external_interfaces(acknowledge: str) -> str:
"""
Extracts and compresses information about external system interfaces from a given acknowledgement text.
Args:
acknowledge (str): A natural text of acknowledgement containing details about external system interfaces.
Returns:
str: A compressed version of the information about external system interfaces.
Example:
>>> acknowledge = "## Interfaces\\n..."
>>> external_interfaces = await extract_external_interfaces(acknowledge=acknowledge)
>>> print(external_interfaces)
```json\n[\n{\n"id": 1,\n"inputs": {...
"""
compress_acknowledge = CompressExternalInterfaces()
return await compress_acknowledge.run(acknowledge=acknowledge)
async def mock_asearch_acknowledgement(use_case_actors: str):
return await aread(filename=TEST_DATA_PATH / "requirements/1.acknowledge.md")
@register_tool(tags=["system design", "write trd", "Write a TRD"])
async def write_trd(
use_case_actors: str,
user_requirements: str,
investment: float = 10,
context: Optional[Context] = None,
) -> str:
"""
Handles the writing of a Technical Requirements Document (TRD) based on user requirements.
Args:
user_requirements (str): The new/incremental user requirements.
use_case_actors (str): Description of the actors involved in the use case.
investment (float): Budget. Automatically stops optimizing TRD when the budget is overdrawn.
context (Context, optional): The context configuration. Default is None.
Returns:
str: The newly created 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;"
>>> investment = 10.0
>>> trd = await write_trd(
>>> user_requirements=user_requirements,
>>> use_case_actors=use_case_actors,
>>> investment=investment,
>>> )
>>> print(trd)
## Technical Requirements Document\n ...
"""
context = context or Context(cost_manager=CostManager(max_budget=investment))
compress_acknowledge = CompressExternalInterfaces()
acknowledgement = await mock_asearch_acknowledgement(use_case_actors) # Replaced by acknowledgement_repo later.
external_interfaces = await compress_acknowledge.run(acknowledge=acknowledgement)
detect_interaction = DetectInteraction(context=context)
w_trd = WriteTRD(context=context)
evaluate_trd = EvaluateTRD(context=context)
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=user_requirements,
use_case_actors=use_case_actors,
legacy_interaction_events=interaction_events,
evaluation_conclusion=evaluation_conclusion,
)
trd = await w_trd.run(
user_requirements=user_requirements,
use_case_actors=use_case_actors,
available_external_interfaces=external_interfaces,
evaluation_conclusion=evaluation_conclusion,
interaction_events=interaction_events,
previous_version_trd=trd,
)
evaluation = await evaluate_trd.run(
user_requirements=user_requirements,
use_case_actors=use_case_actors,
trd=trd,
interaction_events=interaction_events,
)
is_pass = evaluation.is_pass
evaluation_conclusion = evaluation.conclusion
return trd
@register_tool(tags=["system design", "write software framework", "Write a software framework based on a TRD"])
async def write_framework(
use_case_actors: str,
trd: str,
additional_technical_requirements: str,
output_dir: Optional[str] = "",
investment: float = 20.0,
context: Optional[Context] = None,
max_loop: int = 20,
) -> str:
"""
Run the action to generate a software framework based on the provided TRD and related information.
Args:
use_case_actors (str): Description of the use case actors involved.
trd (str): Technical Requirements Document detailing the requirements.
additional_technical_requirements (str): Any additional technical requirements.
output_dir (str, optional): Path to save the software framework files. Default is en empty string.
investment (float): Budget. Automatically stops optimizing TRD when the budget is overdrawn.
context (Context, optional): The context configuration. Default is None.
max_loop(int, optional): Acts as a safety exit valve when cost statistics fail. Default is 20.
Returns:
str: The generated software framework as a string of pathnames.
Example:
>>> use_case_actors = "- Actor: game player;\\n- System: snake game; \\n- External System: game center;"
>>> trd = "## TRD\\n..."
>>> additional_technical_requirements = "Using Java language, ..."
>>> investment = 15.0
>>> framework = await write_framework(
>>> use_case_actors=use_case_actors,
>>> trd=trd,
>>> additional_technical_requirements=constraint,
>>> investment=investment,
>>> )
>>> print(framework)
[{"path":"balabala", "filename":"...", ...
"""
context = context or Context(cost_manager=CostManager(max_budget=investment))
write_framework = WriteFramework(context=context)
evaluate_framework = EvaluateFramework(context=context)
is_pass = False
framework = ""
evaluation_conclusion = ""
acknowledgement = await mock_asearch_acknowledgement(use_case_actors) # Replaced by acknowledgement_repo later.
loop_count = 0
output_dir = (
Path(output_dir)
if output_dir
else DEFAULT_WORKSPACE_ROOT / (datetime.now().strftime("%Y%m%d%H%M%ST") + uuid.uuid4().hex[0:8])
)
file_list = []
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=acknowledgement,
legacy_output=framework,
evaluation_conclusion=evaluation_conclusion,
additional_technical_requirements=additional_technical_requirements,
)
except Exception as e:
logger.info(f"{e}")
break
evaluation = await evaluate_framework.run(
use_case_actors=use_case_actors,
trd=trd,
acknowledge=acknowledgement,
legacy_output=framework,
additional_technical_requirements=additional_technical_requirements,
)
is_pass = evaluation.is_pass
evaluation_conclusion = evaluation.conclusion
loop_count += 1
logger.info(f"Loop {loop_count}")
if context.cost_manager.total_cost < 1 and loop_count > max_loop:
break
file_list = await save_framework(dir_data=framework, trd=trd, output_dir=output_dir)
logger.info(f"Output:\n{file_list}")
return "## Software Framework" + "".join([f"\n- {i}" for i in file_list])
@register_tool(tags=["system design", "write trd and framework", "Write a TRD and the framework"])
async def write_trd_and_framework(
use_case_actors: str,
user_requirements: str,
additional_technical_requirements: str,
investment: float = 50.0,
output_dir: Optional[str] = "",
context: Optional[Context] = None,
) -> str:
context = context or Context(cost_manager=CostManager(max_budget=investment))
trd = await write_trd(use_case_actors=use_case_actors, user_requirements=user_requirements, context=context)
return await write_framework(
use_case_actors=use_case_actors,
trd=trd,
additional_technical_requirements=additional_technical_requirements,
output_dir=output_dir,
context=context,
)

2
metagpt/tools/libs/terminal.py
View file

@ -26,7 +26,7 @@ class Terminal:
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
executable="/bin/bash"
executable="/bin/bash",
)
self.stdout_queue = Queue()
self.observer = TerminalReporter()

33
metagpt/utils/common.py
View file

@ -26,7 +26,7 @@ import sys
import traceback
from io import BytesIO
from pathlib import Path
from typing import Any, Callable, List, Literal, Optional, Tuple, Union
from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union
from urllib.parse import quote, unquote
import aiofiles
@ -1013,3 +1013,34 @@ async def save_json_to_markdown(content: str, output_filename: str | Path):
logger.warning(f"An unexpected error occurred: {e}")
return
await awrite(filename=output_filename, data=json_to_markdown(m))
def tool2name(cls, methods: List[str], entry) -> Dict[str, Any]:
"""
Generates a mapping of class methods to a given entry with class name as a prefix.
Args:
cls: The class from which the methods are derived.
methods (List[str]): A list of method names as strings.
entry (Any): The entry to be mapped to each method.
Returns:
Dict[str, Any]: A dictionary where keys are method names prefixed with the class name and
values are the given entry. If the number of methods is less than 2,
the dictionary will contain a single entry with the class name as the key.
Example:
>>> class MyClass:
>>> pass
>>>
>>> tool2name(MyClass, ['method1', 'method2'], 'some_entry')
{'MyClass.method1': 'some_entry', 'MyClass.method2': 'some_entry'}
>>> tool2name(MyClass, ['method1'], 'some_entry')
{'MyClass': 'some_entry', 'MyClass.method1': 'some_entry'}
"""
class_name = cls.__name__
mappings = {f"{class_name}.{i}": entry for i in methods}
if len(mappings) < 2:
mappings[class_name] = entry
return mappings

16
metagpt/utils/di_graph_repository.py
View file

@ -23,8 +23,8 @@ from metagpt.utils.graph_repository import SPO, GraphRepository
class DiGraphRepository(GraphRepository):
"""Graph repository based on DiGraph."""
def __init__(self, name: str, **kwargs):
super().__init__(name=name, **kwargs)
def __init__(self, name: str | Path, **kwargs):
super().__init__(name=str(name), **kwargs)
self._repo = networkx.DiGraph()
async def insert(self, subject: str, predicate: str, object_: str):
@ -112,8 +112,14 @@ class DiGraphRepository(GraphRepository):
async def load(self, pathname: str | Path):
"""Load a directed graph repository from a JSON file."""
data = await aread(filename=pathname, encoding="utf-8")
m = json.loads(data)
self.load_json(data)
def load_json(self, val: str):
if not val:
return self
m = json.loads(val)
self._repo = networkx.node_link_graph(m)
return self
@staticmethod
async def load_from(pathname: str | Path) -> GraphRepository:
@ -126,9 +132,7 @@ class DiGraphRepository(GraphRepository):
GraphRepository: A new instance of the graph repository loaded from the specified JSON file.
"""
pathname = Path(pathname)
name = pathname.with_suffix("").name
root = pathname.parent
graph = DiGraphRepository(name=name, root=root)
graph = DiGraphRepository(name=pathname.stem, root=pathname.parent)
if pathname.exists():
await graph.load(pathname=pathname)
return graph