apunkt/MetaGPT
1
0
Fork 0
mirror of https://github.com/FoundationAgents/MetaGPT.git synced 2026-04-26 17:26:22 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #37 from qa6300525/2023-07-10_chengmaoyu

Browse source 
2023 07 10 chengmaoyu
This commit is contained in:
geekan 2023-07-12 15:45:07 +08:00 • committed by GitHub
commit 20a137ef45
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 813 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

1
metagpt/actions/__init__.py
View file

@ -8,6 +8,7 @@
from enum import Enum
from metagpt.actions.action import Action
from metagpt.actions.action_output import ActionOutput
from metagpt.actions.write_prd import WritePRD
from metagpt.actions.write_prd_review import WritePRDReview

20
metagpt/actions/action.py
View file

@ -9,6 +9,10 @@ from typing import Optional
from abc import ABC
from metagpt.llm import LLM
from metagpt.actions.action_output import ActionOutput
from tenacity import retry, stop_after_attempt, wait_fixed
from pydantic import BaseModel
from metagpt.utils.common import OutputParser
class Action(ABC):
@ -21,6 +25,8 @@ class Action(ABC):
self.prefix = ""
self.profile = ""
self.desc = ""
self.content = ""
self.instruct_content = None
def set_prefix(self, prefix, profile):
"""Set prefix for later usage"""
@ -40,6 +46,20 @@ class Action(ABC):
system_msgs.append(self.prefix)
return await self.llm.aask(prompt, system_msgs)
@retry(stop=stop_after_attempt(2), wait=wait_fixed(1))
async def _aask_v1(self, prompt: str, output_class_name: str,
output_data_mapping: dict,
system_msgs: Optional[list[str]] = None) -> ActionOutput:
"""Append default prefix"""
if not system_msgs:
system_msgs = []
system_msgs.append(self.prefix)
content = await self.llm.aask(prompt, system_msgs)
output_class = ActionOutput.create_model_class(output_class_name, output_data_mapping)
parsed_data = OutputParser.parse_data_with_mapping(content, output_data_mapping)
instruct_content = output_class(**parsed_data)
return ActionOutput(content, instruct_content)
async def run(self, *args, **kwargs):
"""Run action"""
raise NotImplementedError("The run method should be implemented in a subclass.")

41
metagpt/actions/action_output.py Normal file
View file

@ -0,0 +1,41 @@
#!/usr/bin/env python
# coding: utf-8
"""
@Time : 2023/7/11 10:03
@Author : chengmaoyu
@File : action_output
"""
from pydantic import create_model, validator, root_validator, BaseModel
from typing import Dict, Type
class ActionOutput:
content: str
instruct_content: BaseModel
def __init__(self, content: str, instruct_content: BaseModel):
self.content = content
self.instruct_content = instruct_content
@classmethod
def create_model_class(cls, class_name: str, mapping: Dict[str, Type]):
new_class = create_model(class_name, **mapping)
@validator('*', allow_reuse=True)
def check_name(v, field):
if field.name not in mapping.keys():
raise ValueError(f'Unrecognized block: {field.name}')
return v
@root_validator(pre=True, allow_reuse=True)
def check_missing_fields(values):
required_fields = set(mapping.keys())
missing_fields = required_fields - set(values.keys())
if missing_fields:
raise ValueError(f'Missing fields: {missing_fields}')
return values
new_class.__validator_check_name = classmethod(check_name)
new_class.__root_validator_check_missing_fields = classmethod(check_missing_fields)
return new_class

74
metagpt/actions/design_api.py
View file

@ -7,7 +7,9 @@
"""
import shutil
from pathlib import Path
from typing import List, Tuple
from metagpt.actions import ActionOutput
from metagpt.actions import Action
from metagpt.const import WORKSPACE_ROOT
from metagpt.utils.common import CodeParser
@ -18,6 +20,9 @@ from metagpt.utils.mermaid import mermaid_to_file
PROMPT_TEMPLATE = """
# Context
{context}
## Format example
{format_example}
-----
Role: You are an architect; the goal is to design a SOTA PEP8-compliant python system; make the best use of good open source tools
Requirement: Fill in the following missing information based on the context, note that all sections are response with code form separately
@ -37,6 +42,53 @@ Attention: Use '##' to split sections, not '#', and '## <SECTION_NAME>' SHOULD W
## Anything UNCLEAR: Provide as Plain text. Make clear here.
"""
FORMAT_EXAMPLE = """
---
## Implementation approach
We will ...
## Python package name
```python
"snake_game"
```
## File list
```python
[
"main.py",
]
```
## Data structures and interface definitions
```mermaid
classDiagram
class Game{
+int score
}
...
Game "1" -- "1" Food: has
```
## Program call flow
```mermaid
sequenceDiagram
participant M as Main
...
G->>M: end game
```
## Anything UNCLEAR
The requirement is clear to me.
---
"""
OUTPUT_MAPPING = {
"Implementation approach": (str, ...),
"Python package name": (str, ...),
"File list": (List[str], ...),
"Data structures and interface definitions": (str, ...),
"Program call flow": (str, ...),
"Anything UNCLEAR": (str, ...),
}
class WriteDesign(Action):
@ -60,17 +112,22 @@ class WriteDesign(Action):
logger.info(f"Saving PRD to {prd_file}")
prd_file.write_text(prd)
def _save_system_design(self, docs_path, resources_path, system_design):
data_api_design = CodeParser.parse_code(block="Data structures and interface definitions", text=system_design)
seq_flow = CodeParser.parse_code(block="Program call flow", text=system_design)
def _save_system_design(self, docs_path, resources_path, content):
data_api_design = CodeParser.parse_code(block="Data structures and interface definitions", text=content)
seq_flow = CodeParser.parse_code(block="Program call flow", text=content)
mermaid_to_file(data_api_design, resources_path / 'data_api_design')
mermaid_to_file(seq_flow, resources_path / 'seq_flow')
system_design_file = docs_path / 'system_design.md'
logger.info(f"Saving System Designs to {system_design_file}")
system_design_file.write_text(system_design)
system_design_file.write_text(content)
def _save(self, context, system_design):
ws_name = CodeParser.parse_str(block="Python package name", text=system_design)
if isinstance(system_design, ActionOutput):
content = system_design.content
ws_name = CodeParser.parse_str(block="Python package name", text=content)
else:
content = system_design
ws_name = CodeParser.parse_str(block="Python package name", text=system_design)
workspace = WORKSPACE_ROOT / ws_name
self.recreate_workspace(workspace)
docs_path = workspace / 'docs'
@ -78,10 +135,11 @@ class WriteDesign(Action):
docs_path.mkdir(parents=True, exist_ok=True)
resources_path.mkdir(parents=True, exist_ok=True)
self._save_prd(docs_path, resources_path, context[-1].content)
self._save_system_design(docs_path, resources_path, system_design)
self._save_system_design(docs_path, resources_path, content)
async def run(self, context):
prompt = PROMPT_TEMPLATE.format(context=context)
system_design = await self._aask(prompt)
prompt = PROMPT_TEMPLATE.format(context=context, format_example=FORMAT_EXAMPLE)
# system_design = await self._aask(prompt)
system_design = await self._aask_v1(prompt, "system_design", OUTPUT_MAPPING)
self._save(context, system_design)
return system_design

77
metagpt/actions/project_management.py
View file

@ -5,15 +5,21 @@
@Author : alexanderwu
@File : project_management.py
"""
from typing import List, Tuple
from metagpt.actions.action import Action
from metagpt.actions.action_output import ActionOutput
from metagpt.const import WORKSPACE_ROOT
from metagpt.logs import logger
from metagpt.utils.common import CodeParser
from metagpt.utils.common import OutputParser, CodeParser
from tenacity import retry, stop_after_attempt, wait_fixed
PROMPT_TEMPLATE = """
PROMPT_TEMPLATE = '''
# Context
{context}
## Format example
{format_example}
-----
Role: You are a project manager; the goal is to break down tasks according to PRD/technical design, give a task list, and analyze task dependencies to start with the prerequisite modules
Requirements: Based on the context, fill in the following missing information, note that all sections are returned in Python code triple quote form seperatedly. Here the granularity of the task is a file, if there are any missing files, you can supplement them
@ -33,10 +39,72 @@ Attention: Use '##' to split sections, not '#', and '## <SECTION_NAME>' SHOULD W
## Anything UNCLEAR: Provide as Plain text. Make clear here. For example, don't forget a main entry. don't forget to init 3rd party libs.
'''
FORMAT_EXAMPLE = '''
---
## Required Python third-party packages
```python
"""
flask==1.1.2
"""
```
## Required Other language third-party packages
```python
"""
No third-party ...
"""
```
## Full API spec
```python
"""
openapi: 3.0.0
...
description: A JSON object ...
"""
```
## Logic Analysis
```python
[
("game.py", "Contains ..."),
]
```
## Task list
```python
[
"game.py",
]
```
## Shared Knowledge
```python
"""
'game.py' contains ...
"""
```
## Anything UNCLEAR
We need ... how to start.
---
'''
OUTPUT_MAPPING = {
"Required Python third-party packages": (str, ...),
"Required Other language third-party packages": (str, ...),
"Full API spec": (str, ...),
"Logic Analysis": (List[Tuple[str, str]], ...),
"Task list": (List[str], ...),
"Shared Knowledge": (str, ...),
"Anything UNCLEAR": (str, ...),
}
class WriteTasks(Action):
def __init__(self, name="CreateTasks", context=None, llm=None):
super().__init__(name, context, llm)
@ -46,9 +114,8 @@ class WriteTasks(Action):
file_path.write_text(rsp)
async def run(self, context):
prompt = PROMPT_TEMPLATE.format(context=context)
rsp = await self._aask(prompt)
self._save(context, rsp)
prompt = PROMPT_TEMPLATE.format(context=context, format_example=FORMAT_EXAMPLE)
rsp = await self._aask_v1(prompt, "task", OUTPUT_MAPPING)
return rsp

5
metagpt/actions/write_code.py
View file

@ -31,6 +31,9 @@ Attention: Use '##' to split sections, not '#', and '## <SECTION_NAME>' SHOULD W
"""
## {filename}: Please encapsulate your code within triple quotes. Focus your efforts on implementing ONLY WITHIN THIS FILE. Any class or function labeled as MISSING-DESIGN should be implemented IN THIS FILE ALONE. Do NOT make changes to any other files.
OUTPUT_MAPPING = {
"{filename}": (str, ...),
}
class WriteCode(Action):
@ -47,6 +50,7 @@ class WriteCode(Action):
return
design = [i for i in context if i.cause_by == WriteDesign][0]
ws_name = CodeParser.parse_str(block="Python package name", text=design.content)
ws_path = WORKSPACE_ROOT / ws_name
if f"{ws_name}/" not in filename and all(i not in filename for i in ["requirements.txt", ".md"]):
@ -63,5 +67,6 @@ class WriteCode(Action):
context = kwargs['context']
logger.info(f'Writing {filename}..')
code_rsp = await self._aask(prompt)
# code_rsp = await self._aask_v1(prompt, "code_rsp", OUTPUT_MAPPING)
self._save(context, filename, code_rsp)
return code_rsp

74
metagpt/actions/write_prd.py
View file

@ -5,10 +5,11 @@
@Author : alexanderwu
@File : write_prd.py
"""
from metagpt.actions import Action
from metagpt.actions import Action, ActionOutput
from metagpt.actions.search_and_summarize import SEARCH_AND_SUMMARIZE_SYSTEM, SearchAndSummarize, \
SEARCH_AND_SUMMARIZE_PROMPT, SEARCH_AND_SUMMARIZE_SYSTEM_EN_US
from metagpt.logs import logger
from typing import List, Tuple
PROMPT_TEMPLATE = """
# Context
@ -36,10 +37,13 @@ quadrantChart
"Campaign F": [0.35, 0.78]
"Our Target Product": [0.5, 0.6]
```
## Format example
{format_example}
-----
Role: You are a professional product manager; the goal is to design a concise, usable, efficient product
Requirements: According to the context, fill in the following missing information, note that each sections are returned in Python code triple quote form seperatedly. If the requirements are unclear, ensure minimum viability and avoid excessive design
ATTENTION: Use '##' to SPLIT SECTIONS, not '#'. AND '## <SECTION_NAME>' SHOULD WRITE BEFORE the code and triple quote.
ATTENTION: Use '##' to SPLIT SECTIONS, not '#'. AND '## <SECTION_NAME>' SHOULD WRITE BEFORE the code and triple quote. Output carefully referenced "Format example" in format.
## Original Requirements: Provide as Plain text, place the polished complete original requirements here
@ -56,15 +60,72 @@ ATTENTION: Use '##' to SPLIT SECTIONS, not '#'. AND '## <SECTION_NAME>' SHOULD W
## Requirement Pool: Provided as Python list[str, str], the parameters are requirement description, priority(P0/P1/P2), respectively, comply with PEP standards; no more than 5 requirements and consider to make its difficulty lower
## Anything UNCLEAR: Provide as Plain text. Make clear here.
"""
FORMAT_EXAMPLE = """
---
## Original Requirements
The boss ...
## Product Goals
```python
[
"Create a ...",
]
```
## User Stories
```python
[
"As a user, ...",
]
```
## Competitive Analysis
```python
[
"Python Snake Game: ...",
]
```
## Competitive Quadrant Chart
```mermaid
quadrantChart
title Reach and engagement of campaigns
...
"Our Target Product": [0.6, 0.7]
```
## Requirement Analysis
The product should be a ...
## Requirement Pool
```python
[
("End game ...", "P0")
]
```
## Anything UNCLEAR
There are no unclear points.
---
"""
OUTPUT_MAPPING = {
"Original Requirements": (str, ...),
"Product Goals": (List[str], ...),
"User Stories": (List[str], ...),
"Competitive Analysis": (List[str], ...),
"Competitive Quadrant Chart": (str, ...),
"Requirement Analysis": (str, ...),
"Requirement Pool": (List[Tuple[str, str]], ...),
"Anything UNCLEAR": (str, ...),
}
class WritePRD(Action):
def __init__(self, name="", context=None, llm=None):
super().__init__(name, context, llm)
async def run(self, requirements, *args, **kwargs) -> str:
async def run(self, requirements, *args, **kwargs) -> ActionOutput:
sas = SearchAndSummarize()
rsp = await sas.run(context=requirements, system_text=SEARCH_AND_SUMMARIZE_SYSTEM_EN_US)
info = f"### Search Results\n{sas.result}\n\n### Search Summary\n{rsp}"
@ -72,6 +133,7 @@ class WritePRD(Action):
logger.info(sas.result)
logger.info(rsp)
prompt = PROMPT_TEMPLATE.format(requirements=requirements, search_information=info)
prd = await self._aask(prompt)
prompt = PROMPT_TEMPLATE.format(requirements=requirements, search_information=info,
format_example=FORMAT_EXAMPLE)
prd = await self._aask_v1(prompt, "prd", OUTPUT_MAPPING)
return prd

4
metagpt/provider/openai_api.py
View file

@ -173,7 +173,7 @@ class OpenAIGPTAPI(BaseGPTAPI, RateLimiter):
"max_tokens": CONFIG.max_tokens_rsp,
"n": 1,
"stop": None,
"temperature": 0.5
"temperature": 0.3
}
else:
kwargs = {
@ -182,7 +182,7 @@ class OpenAIGPTAPI(BaseGPTAPI, RateLimiter):
"max_tokens": CONFIG.max_tokens_rsp,
"n": 1,
"stop": None,
"temperature": 0.5
"temperature": 0.3
}
return kwargs

6
metagpt/roles/engineer.py
View file

@ -22,7 +22,7 @@ from collections import OrderedDict
async def gather_ordered_k(coros, k) -> list:
tasks = OrderedDict()
results = [None]*len(coros)
results = [None] * len(coros)
done_queue = asyncio.Queue()
for i, coro in enumerate(coros):
@ -59,6 +59,8 @@ class Engineer(Role):
@classmethod
def parse_tasks(self, task_msg: Message) -> list[str]:
if not task_msg.instruct_content:
return task_msg.instruct_content.dict().get("Task list")
return CodeParser.parse_file_list(block="Task list", text=task_msg.content)
@classmethod
@ -67,6 +69,8 @@ class Engineer(Role):
@classmethod
def parse_workspace(cls, system_design_msg: Message) -> str:
if not system_design_msg.instruct_content:
return system_design_msg.instruct_content.dict().get("Python package name")
return CodeParser.parse_str(block="Python package name", text=system_design_msg.content)
def get_workspace(self) -> Path:

10
metagpt/roles/role.py
View file

@ -12,7 +12,7 @@ from typing import Type, Iterable
from metagpt.logs import logger
# from metagpt.environment import Environment
from metagpt.actions import Action
from metagpt.actions import Action, ActionOutput
from metagpt.llm import LLM
from metagpt.schema import Message
from metagpt.memory import Memory
@ -45,7 +45,6 @@ ROLE_TEMPLATE = """Your response should be based on the previous conversation hi
"""
@dataclass
class RoleSetting:
"""角色设定"""
@ -83,6 +82,7 @@ class RoleContext:
class Role:
"""角色/代理"""
def __init__(self, name="", profile="", goal="", constraints="", desc=""):
self._llm = LLM()
self._setting = RoleSetting(name, profile, goal, constraints, desc)
@ -153,7 +153,11 @@ class Role:
logger.info(f"{self._setting}: ready to {self._rc.todo}")
response = await self._rc.todo.run(self._rc.important_memory)
# logger.info(response)
msg = Message(content=response, role=self.profile, cause_by=type(self._rc.todo))
if isinstance(response, ActionOutput):
msg = Message(content=response.content, instruct_content=response.instruct_content,
role=self.profile, cause_by=type(self._rc.todo))
else:
msg = Message(content=response, role=self.profile, cause_by=type(self._rc.todo))
self._rc.memory.add(msg)
# logger.debug(f"{response}")

3
metagpt/schema.py
View file

@ -10,7 +10,7 @@ from dataclasses import dataclass, field
from typing import Type, TypedDict
from metagpt.logs import logger
# from pydantic import BaseModel
from pydantic import BaseModel
@ -23,6 +23,7 @@ class RawMessage(TypedDict):
class Message:
"""list[<role>: <content>]"""
content: str
instruct_content: BaseModel = field(default=None)
role: str = field(default='user') # system / user / assistant
cause_by: Type["Action"] = field(default="")

118
metagpt/utils/common.py
View file

@ -10,7 +10,7 @@ import ast
import inspect
import re
from typing import Union
from typing import Union, List, Tuple
from metagpt.logs import logger
from langchain.schema import AgentAction, AgentFinish, OutputParserException
@ -27,6 +27,112 @@ def check_cmd_exists(command) -> int:
return result
class OutputParser:
@classmethod
def parse_blocks(cls, text: str):
# 首先根据"##"将文本分割成不同的block
blocks = text.split("##")
# 创建一个字典，用于存储每个block的标题和内容
block_dict = {}
# 遍历所有的block
for block in blocks:
# 如果block不为空，则继续处理
if block.strip() != "":
# 将block的标题和内容分开，并分别去掉前后的空白字符
block_title, block_content = block.split("\n", 1)
# LLM可能出错，在这里做一下修正
if block_title[-1] == ":":
block_title = block_title[:-1]
block_dict[block_title.strip()] = block_content.strip()
return block_dict
@classmethod
def parse_code(cls, text: str, lang: str = "") -> str:
pattern = rf'```{lang}.*?\s+(.*?)```'
match = re.search(pattern, text, re.DOTALL)
if match:
code = match.group(1)
else:
raise Exception
return code
@classmethod
def parse_str(cls, text: str):
text = text.split("=")[-1]
text = text.strip().strip("'").strip("\"")
return text
@classmethod
def parse_file_list(cls, text: str) -> list[str]:
# Regular expression pattern to find the tasks list.
pattern = r'\s*(.*=.*)?(\[.*\])'
# Extract tasks list string using regex.
match = re.search(pattern, text, re.DOTALL)
if match:
tasks_list_str = match.group(2)
# Convert string representation of list to a Python list using ast.literal_eval.
tasks = ast.literal_eval(tasks_list_str)
else:
raise Exception
return tasks
@classmethod
def parse_data(cls, data):
block_dict = cls.parse_blocks(data)
parsed_data = {}
for block, content in block_dict.items():
# 尝试去除code标记
try:
content = cls.parse_code(text=content)
except Exception:
pass
# 尝试解析list
try:
content = cls.parse_file_list(text=content)
except Exception:
pass
parsed_data[block] = content
return parsed_data
@classmethod
def parse_data_with_mapping(cls, data, mapping):
block_dict = cls.parse_blocks(data)
parsed_data = {}
for block, content in block_dict.items():
# 尝试去除code标记
try:
content = cls.parse_code(text=content)
except Exception:
pass
typing_define = mapping.get(block, None)
if isinstance(typing_define, tuple):
typing = typing_define[0]
else:
typing = typing_define
if typing == List[str] or typing == List[Tuple[str, str]]:
# 尝试解析list
try:
content = cls.parse_file_list(text=content)
except Exception:
pass
# TODO: 多余的引号去除有风险，后期再解决
# elif typing == str:
# # 尝试去除多余的引号
# try:
# content = cls.parse_str(text=content)
# except Exception:
# pass
parsed_data[block] = content
return parsed_data
class CodeParser:
@classmethod
@ -56,7 +162,7 @@ class CodeParser:
return block_dict
@classmethod
def parse_code(cls, block: str, text: str, lang: str="") -> str:
def parse_code(cls, block: str, text: str, lang: str = "") -> str:
if block:
text = cls.parse_block(block, text)
pattern = rf'```{lang}.*?\s+(.*?)```'
@ -70,16 +176,17 @@ class CodeParser:
return code
@classmethod
def parse_str(cls, block: str, text: str, lang: str=""):
def parse_str(cls, block: str, text: str, lang: str = ""):
code = cls.parse_code(block, text, lang)
code = code.split("=")[-1]
code = code.strip().strip("'").strip("\"")
return code
@classmethod
def parse_file_list(cls, block: str, text: str, lang: str="") -> list[str]:
def parse_file_list(cls, block: str, text: str, lang: str = "") -> list[str]:
# Regular expression pattern to find the tasks list.
code = cls.parse_code(block, text, lang)
print(code)
pattern = r'\s*(.*=.*)?(\[.*\])'
# Extract tasks list string using regex.
@ -96,6 +203,7 @@ class CodeParser:
class NoMoneyException(Exception):
"""Raised when the operation cannot be completed due to insufficient funds"""
def __init__(self, amount, message="Insufficient funds"):
self.amount = amount
self.message = message
@ -154,4 +262,4 @@ if __name__ == '__main__':
logger.info(rsp)
rsp = parser.parse(final_answer_sample)
logger.info(rsp)
logger.info(rsp)

49
tests/metagpt/actions/test_action_output.py Normal file
View file

@ -0,0 +1,49 @@
#!/usr/bin/env python
# coding: utf-8
"""
@Time : 2023/7/11 10:49
@Author : chengmaoyu
@File : test_action_output
"""
from metagpt.actions import ActionOutput
from typing import List, Tuple
t_dict = {"Required Python third-party packages": "\"\"\"\nflask==1.1.2\npygame==2.0.1\n\"\"\"\n",
"Required Other language third-party packages": "\"\"\"\nNo third-party packages required for other languages.\n\"\"\"\n",
"Full API spec": "\"\"\"\nopenapi: 3.0.0\ninfo:\n title: Web Snake Game API\n version: 1.0.0\npaths:\n /game:\n get:\n summary: Get the current game state\n responses:\n '200':\n description: A JSON object of the game state\n post:\n summary: Send a command to the game\n requestBody:\n required: true\n content:\n application/json:\n schema:\n type: object\n properties:\n command:\n type: string\n responses:\n '200':\n description: A JSON object of the updated game state\n\"\"\"\n",
"Logic Analysis": [
["app.py", "Main entry point for the Flask application. Handles HTTP requests and responses."],
["game.py", "Contains the Game and Snake classes. Handles the game logic."],
["static/js/script.js", "Handles user interactions and updates the game UI."],
["static/css/styles.css", "Defines the styles for the game UI."],
["templates/index.html", "The main page of the web application. Displays the game UI."]],
"Task list": ["game.py", "app.py", "static/css/styles.css", "static/js/script.js", "templates/index.html"],
"Shared Knowledge": "\"\"\"\n'game.py' contains the Game and Snake classes which are responsible for the game logic. The Game class uses an instance of the Snake class.\n\n'app.py' is the main entry point for the Flask application. It creates an instance of the Game class and handles HTTP requests and responses.\n\n'static/js/script.js' is responsible for handling user interactions and updating the game UI based on the game state returned by 'app.py'.\n\n'static/css/styles.css' defines the styles for the game UI.\n\n'templates/index.html' is the main page of the web application. It displays the game UI and loads 'static/js/script.js' and 'static/css/styles.css'.\n\"\"\"\n",
"Anything UNCLEAR": "We need clarification on how the high score should be stored. Should it persist across sessions (stored in a database or a file) or should it reset every time the game is restarted? Also, should the game speed increase as the snake grows, or should it remain constant throughout the game?"}
WRITE_TASKS_OUTPUT_MAPPING = {
"Required Python third-party packages": (str, ...),
"Required Other language third-party packages": (str, ...),
"Full API spec": (str, ...),
"Logic Analysis": (List[Tuple[str, str]], ...),
"Task list": (List[str], ...),
"Shared Knowledge": (str, ...),
"Anything UNCLEAR": (str, ...),
}
def test_create_model_class():
test_class = ActionOutput.create_model_class("test_class", WRITE_TASKS_OUTPUT_MAPPING)
assert test_class.__name__ == "test_class"
def test_create_model_class_with_mapping():
t = ActionOutput.create_model_class("test_class_1", WRITE_TASKS_OUTPUT_MAPPING)
t1 = t(**t_dict)
value = t1.dict()["Task list"]
assert value == ["game.py", "app.py", "static/css/styles.css", "static/js/script.js", "templates/index.html"]
if __name__ == '__main__':
test_create_model_class()
test_create_model_class_with_mapping()

139
tests/metagpt/utils/test_code_parser.py Normal file
View file

@ -0,0 +1,139 @@
#!/usr/bin/env python
# coding: utf-8
"""
@Time : 2023/7/10 17:14
@Author : chengmaoyu
@File : test_code_parser.py
"""
import pytest
from metagpt.utils.common import CodeParser
t_text = '''
## Required Python third-party packages
```python
"""
flask==1.1.2
pygame==2.0.1
"""
```
## Required Other language third-party packages
```python
"""
No third-party packages required for other languages.
"""
```
## Full API spec
```python
"""
openapi: 3.0.0
info:
title: Web Snake Game API
version: 1.0.0
paths:
/game:
get:
summary: Get the current game state
responses:
'200':
description: A JSON object of the game state
post:
summary: Send a command to the game
requestBody:
required: true
content:
application/json:
schema:
type: object
properties:
command:
type: string
responses:
'200':
description: A JSON object of the updated game state
"""
```
## Logic Analysis
```python
[
("app.py", "Main entry point for the Flask application. Handles HTTP requests and responses."),
("game.py", "Contains the Game and Snake classes. Handles the game logic."),
("static/js/script.js", "Handles user interactions and updates the game UI."),
("static/css/styles.css", "Defines the styles for the game UI."),
("templates/index.html", "The main page of the web application. Displays the game UI.")
]
```
## Task list
```python
[
"game.py",
"app.py",
"static/css/styles.css",
"static/js/script.js",
"templates/index.html"
]
```
## Shared Knowledge
```python
"""
'game.py' contains the Game and Snake classes which are responsible for the game logic. The Game class uses an instance of the Snake class.
'app.py' is the main entry point for the Flask application. It creates an instance of the Game class and handles HTTP requests and responses.
'static/js/script.js' is responsible for handling user interactions and updating the game UI based on the game state returned by 'app.py'.
'static/css/styles.css' defines the styles for the game UI.
'templates/index.html' is the main page of the web application. It displays the game UI and loads 'static/js/script.js' and 'static/css/styles.css'.
"""
```
## Anything UNCLEAR
We need clarification on how the high score should be stored. Should it persist across sessions (stored in a database or a file) or should it reset every time the game is restarted? Also, should the game speed increase as the snake grows, or should it remain constant throughout the game?
'''
class TestCodeParser:
@pytest.fixture
def parser(self):
return CodeParser()
@pytest.fixture
def text(self):
return t_text
def test_parse_blocks(self, parser, text):
result = parser.parse_blocks(text)
print(result)
assert result == {"title": "content", "title2": "content2"}
def test_parse_block(self, parser, text):
result = parser.parse_block("title", text)
print(result)
assert result == "content"
def test_parse_code(self, parser, text):
result = parser.parse_code("title", text, "python")
print(result)
assert result == "print('hello world')"
def test_parse_str(self, parser, text):
result = parser.parse_str("title", text, "python")
print(result)
assert result == "hello world"
def test_parse_file_list(self, parser, text):
result = parser.parse_file_list("Task list", text)
print(result)
assert result == ['task1', 'task2']
if __name__ == '__main__':
t = TestCodeParser()
t.test_parse_file_list(CodeParser(), t_text)
# TestCodeParser.test_parse_file_list()

223
tests/metagpt/utils/test_output_parser.py Normal file
View file

@ -0,0 +1,223 @@
#!/usr/bin/env python
# coding: utf-8
"""
@Time : 2023/7/11 10:25
@Author : chengmaoyu
@File : test_output_parser.py
"""
import pytest
from typing import List, Tuple
import re
import ast
from metagpt.utils.common import OutputParser
def test_parse_blocks():
test_text = "##block1\nThis is block 1.\n##block2\nThis is block 2."
expected_result = {'block1': 'This is block 1.', 'block2': 'This is block 2.'}
assert OutputParser.parse_blocks(test_text) == expected_result
def test_parse_code():
test_text = "```python\nprint('Hello, world!')\n```"
expected_result = "print('Hello, world!')"
assert OutputParser.parse_code(test_text, 'python') == expected_result
with pytest.raises(Exception):
OutputParser.parse_code(test_text, 'java')
def test_parse_str():
test_text = "name = 'Alice'"
expected_result = 'Alice'
assert OutputParser.parse_str(test_text) == expected_result
def test_parse_file_list():
test_text = "files=['file1', 'file2', 'file3']"
expected_result = ['file1', 'file2', 'file3']
assert OutputParser.parse_file_list(test_text) == expected_result
with pytest.raises(Exception):
OutputParser.parse_file_list("wrong_input")
def test_parse_data():
test_data = "##block1\n```python\nprint('Hello, world!')\n```\n##block2\nfiles=['file1', 'file2', 'file3']"
expected_result = {'block1': "print('Hello, world!')", 'block2': ['file1', 'file2', 'file3']}
assert OutputParser.parse_data(test_data) == expected_result
if __name__ == '__main__':
t_text = '''
## Required Python third-party packages
```python
"""
flask==1.1.2
pygame==2.0.1
"""
```
## Required Other language third-party packages
```python
"""
No third-party packages required for other languages.
"""
```
## Full API spec
```python
"""
openapi: 3.0.0
info:
title: Web Snake Game API
version: 1.0.0
paths:
/game:
get:
summary: Get the current game state
responses:
'200':
description: A JSON object of the game state
post:
summary: Send a command to the game
requestBody:
required: true
content:
application/json:
schema:
type: object
properties:
command:
type: string
responses:
'200':
description: A JSON object of the updated game state
"""
```
## Logic Analysis
```python
[
("app.py", "Main entry point for the Flask application. Handles HTTP requests and responses."),
("game.py", "Contains the Game and Snake classes. Handles the game logic."),
("static/js/script.js", "Handles user interactions and updates the game UI."),
("static/css/styles.css", "Defines the styles for the game UI."),
("templates/index.html", "The main page of the web application. Displays the game UI.")
]
```
## Task list
```python
[
"game.py",
"app.py",
"static/css/styles.css",
"static/js/script.js",
"templates/index.html"
]
```
## Shared Knowledge
```python
"""
'game.py' contains the Game and Snake classes which are responsible for the game logic. The Game class uses an instance of the Snake class.
'app.py' is the main entry point for the Flask application. It creates an instance of the Game class and handles HTTP requests and responses.
'static/js/script.js' is responsible for handling user interactions and updating the game UI based on the game state returned by 'app.py'.
'static/css/styles.css' defines the styles for the game UI.
'templates/index.html' is the main page of the web application. It displays the game UI and loads 'static/js/script.js' and 'static/css/styles.css'.
"""
```
## Anything UNCLEAR
We need clarification on how the high score should be stored. Should it persist across sessions (stored in a database or a file) or should it reset every time the game is restarted? Also, should the game speed increase as the snake grows, or should it remain constant throughout the game?
'''
OUTPUT_MAPPING = {
"Original Requirements": (str, ...),
"Product Goals": (List[str], ...),
"User Stories": (List[str], ...),
"Competitive Analysis": (List[str], ...),
"Competitive Quadrant Chart": (str, ...),
"Requirement Analysis": (str, ...),
"Requirement Pool": (List[Tuple[str, str]], ...),
"Anything UNCLEAR": (str, ...),
}
t_text1 = '''## Original Requirements:
The boss wants to create a web-based version of the game "Fly Bird".
## Product Goals:
- Create a web-based version of the game "Fly Bird" that is engaging and addictive.
- Provide a seamless and intuitive user experience.
- Optimize the game for different devices and screen sizes.
## User Stories:
- As a user, I want to be able to control the bird's flight by clicking or tapping on the screen.
- As a user, I want to see my score and the highest score achieved in the game.
- As a user, I want the game to be challenging but not frustratingly difficult.
- As a user, I want to be able to pause and resume the game at any time.
- As a user, I want to be able to share my score on social media.
## Competitive Analysis:
- Flappy Bird: A popular mobile game where the player controls a bird's flight through a series of obstacles.
- Angry Birds: A physics-based puzzle game where the player launches birds to destroy structures and defeat pigs.
- Snake Game: A classic game where the player controls a snake to eat food and grow longer without hitting the walls or its own body.
- Temple Run: An endless running game where the player controls a character to avoid obstacles and collect coins.
- Subway Surfers: An endless running game where the player controls a character to avoid obstacles and collect coins while being chased by a guard.
- Doodle Jump: A vertical platform game where the player controls a character to jump on platforms and avoid falling.
- Fruit Ninja: A fruit-slicing game where the player uses their finger to slice flying fruits.
## Competitive Quadrant Chart:
```mermaid
quadrantChart
title Reach and engagement of games
x-axis Low Reach --> High Reach
y-axis Low Engagement --> High Engagement
quadrant-1 We should expand
quadrant-2 Need to promote
quadrant-3 Re-evaluate
quadrant-4 May be improved
"Flappy Bird": [0.8, 0.9]
"Angry Birds": [0.9, 0.8]
"Snake Game": [0.6, 0.6]
"Temple Run": [0.9, 0.7]
"Subway Surfers": [0.9, 0.7]
"Doodle Jump": [0.7, 0.5]
"Fruit Ninja": [0.8, 0.6]
"Our Target Product": [0.7, 0.8]
```
## Requirement Analysis:
The product should be a web-based version of the game "Fly Bird" that is engaging, addictive, and optimized for different devices and screen sizes. It should provide a seamless and intuitive user experience, with controls that allow the user to control the bird's flight by clicking or tapping on the screen. The game should display the user's score and the highest score achieved. It should be challenging but not frustratingly difficult, allowing the user to pause and resume the game at any time. The user should also have the option to share their score on social media.
## Requirement Pool:
```python
[
("Implement bird's flight control using click or tap", "P0"),
("Display user's score and highest score achieved", "P0"),
("Implement challenging but not frustrating difficulty level", "P1"),
("Allow user to pause and resume the game", "P1"),
("Implement social media sharing feature", "P2")
]
```
## Anything UNCLEAR:
There are no unclear points.
'''
d = OutputParser.parse_data_with_mapping(t_text1, OUTPUT_MAPPING)
import json
print(json.dumps(d))