apunkt/MetaGPT
1
0
Fork 0
mirror of https://github.com/FoundationAgents/MetaGPT.git synced 2026-05-01 20:03:28 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'mgx_ops' into fix/engineer_edit_fail

Browse source
This commit is contained in:
黄伟韬 2024-09-13 19:04:44 +08:00
commit baabcf21ff
9 changed files with 171 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

10
metagpt/prompts/di/engineer2.py
View file

@ -79,13 +79,12 @@ Note:
20. If the code exists, use the Editor tool's open and edit commands to modify it. Since it is not a new code, do not use write_new_code.
21. When using the editor, pay attention to the editor's current directory. When you use editor tools, the paths must be either absolute or relative to the editor's current directory.
22. The default programming languages are Native HTML.
23. When planning, consider whether images are needed. If you are developing a showcase website, start by using ImageGetter.get_image to obtain the necessary images.
"""
CURRENT_STATE = """
The current editor state is:
(Editor current directory: {editor_current_directory})
(Editor open file: {editor_open_file})
The current terminal state is:
(Terminal current directory: {terminal_current_directory})
(Current directory: {current_directory})
(Open file: {editor_open_file})
"""
ENGINEER2_INSTRUCTION = ROLE_INSTRUCTION + EXTRA_INSTRUCTION.strip()
@ -105,6 +104,9 @@ WRITE_CODE_PROMPT = """
# Plan Status
{plan_status}
# Current Coding File
{file_path}
# Further Instruction
{instruction}

17
metagpt/roles/di/engineer2.py
View file

@ -18,6 +18,7 @@ from metagpt.schema import UserMessage
from metagpt.strategy.experience_retriever import ENGINEER_EXAMPLE
from metagpt.tools.libs.cr import CodeReview
from metagpt.tools.libs.git import git_create_pull
from metagpt.tools.libs.image_getter import ImageGetter
from metagpt.tools.libs.terminal import Terminal
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.common import CodeParser, awrite
@ -42,6 +43,7 @@ class Engineer2(RoleZero):
"SearchEnhancedQA",
"Engineer2",
"CodeReview",
"ImageGetter",
]
# SWE Agent parameter
run_eval: bool = False
@ -58,10 +60,13 @@ class Engineer2(RoleZero):
Display the current terminal and editor state.
This information will be dynamically added to the command prompt.
"""
current_directory = (await self.terminal.run_command("pwd")).strip()
# Synchronize Terminal and Editor Working Directories
if str(self.editor.working_dir.absolute()).strip() != current_directory:
self.editor._set_workdir(current_directory)
state = {
"editor_open_file": self.editor.current_file,
"editor_current_directory": self.editor.working_dir.absolute(),
"terminal_current_directory": (await self.terminal.run_command("pwd")).strip(),
"current_directory": current_directory,
}
self.cmd_prompt_current_state = CURRENT_STATE.format(**state).strip()
@ -84,10 +89,12 @@ class Engineer2(RoleZero):
)
else:
# Default tool map
image_getter = ImageGetter()
self.tool_execution_map.update(
{
"git_create_pull": git_create_pull,
"Engineer2.write_new_code": self.write_new_code,
"ImageGetter.get_image": image_getter.get_image,
"CodeReview.review": cr.review,
"CodeReview.fix": cr.fix,
"Terminal.run_command": self.terminal.run_command,
@ -117,10 +124,14 @@ class Engineer2(RoleZero):
plan_status, _ = self._get_plan_status()
prompt = WRITE_CODE_PROMPT.format(
user_requirement=self.planner.plan.goal,
file_path=path,
plan_status=plan_status,
instruction=instruction,
)
context = self.llm.format_msg(self.rc.memory.get(self.memory_k) + [UserMessage(content=prompt)])
# Sometimes the Engineer repeats the last command to respond.
# Replace the last command with a manual prompt to guide the Engineer to write new code.
memory = self.rc.memory.get(self.memory_k)[:-1]
context = self.llm.format_msg(memory + [UserMessage(content=prompt)])
async with EditorReporter(enable_llm_stream=True) as reporter:
await reporter.async_report({"type": "code", "filename": Path(path).name, "src_path": path}, "meta")

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

@ -432,7 +432,7 @@ class RoleZero(Role):
for index, cmd in enumerate(commands)
if index == index_of_first_exclusive or cmd["command_name"] not in self.exclusive_tool_commands
]
command_rsp = "```json\n" + json.dumps(commands, indent=4, ensure_ascii=False) + "\n```json"
command_rsp = "```json\n" + json.dumps(commands, indent=4, ensure_ascii=False) + "\n```"
logger.info(
"exclusive command more than one in current command list. change the command list.\n" + command_rsp
)

17
metagpt/strategy/experience_retriever.py
View file

@ -845,7 +845,7 @@ Consider this example only after you have obtained the content of system design
Suppose the system design and project schedule prescribes three files index.html, style.css, script.js, to follow the design and schedule, I will create a plan consisting of three tasks, each corresponding to the creation of one of the required files: `index.html`, `style.css`, and `script.js`.
Here's the plan:
[Optional] 0. **Task 0**: Obtain images before coding.
1. **Task 1**: Create `index.html` - This file will contain the HTML structure necessary for the game's UI.
2. **Task 2**: Create `style.css` - This file will define the CSS styles to make the game visually appealing and responsive.
3. **Task 3**: Create `script.js` - This file will contain the Pure JavaScript code for the game logic and UI interactions.
@ -1034,6 +1034,21 @@ Thought: Now that the changes have been pushed to the remote repository, due to
}
]
```
## example 12
The requirement is to create a product website featuring goods such as caps, dresses, and T-shirts.
I believe pictures would improve the site, so I will get the images first.
```json
[
{
"command_name": "ImageGetter.get_image",
"args": {
"search_term": "cap",
"save_file_path": "/tmp/workspace/images/cap.png",
}
}
]
```
"""
WEB_SCRAPING_EXAMPLE = """

24
metagpt/tools/libs/editor.py
View file

@ -515,6 +515,8 @@ class Editor(BaseModel):
temp_file_path = ""
src_abs_path = file_name.resolve()
first_error_line = None
# The file to store previous content and will be removed automatically.
temp_backup_file = tempfile.NamedTemporaryFile("w", delete=True)
try:
# lint the original file
@ -557,10 +559,8 @@ class Editor(BaseModel):
# because the env var will be set AFTER the agentskills is imported
if self.enable_auto_lint:
# BACKUP the original file
original_file_backup_path = file_name.parent / f".backup.{file_name.name}"
with original_file_backup_path.open("w") as f:
f.writelines(lines)
temp_backup_file.writelines(lines)
temp_backup_file.flush()
lint_error, first_error_line = self._lint_file(file_name)
# Select the errors caused by the modification
@ -610,15 +610,13 @@ class Editor(BaseModel):
linter_error_msg=LINTER_ERROR_MSG + lint_error,
window_after_applied=self._print_window(file_name, show_line, n_added_lines + 20),
window_before_applied=self._print_window(
original_file_backup_path, show_line, n_added_lines + 20
Path(temp_backup_file.name), show_line, n_added_lines + 20
),
guidance_message=guidance_message,
).strip()
# recover the original file
with original_file_backup_path.open() as fin, file_name.open("w") as fout:
fout.write(fin.read())
original_file_backup_path.unlink()
shutil.move(temp_backup_file.name, src_abs_path)
return lint_error_info
except FileNotFoundError as e:
@ -636,18 +634,16 @@ class Editor(BaseModel):
error_info = ERROR_GUIDANCE.format(
linter_error_msg=LINTER_ERROR_MSG + str(e),
window_after_applied=self._print_window(file_name, start or len(lines), 40),
window_before_applied=self._print_window(original_file_backup_path, start or len(lines), 40),
window_before_applied=self._print_window(Path(temp_backup_file.name), start or len(lines), 40),
guidance_message=guidance_message,
).strip()
# Clean up the temporary file if an error occurs
with original_file_backup_path.open() as fin, file_name.open("w") as fout:
fout.write(fin.read())
shutil.move(temp_backup_file.name, src_abs_path)
if temp_file_path and Path(temp_file_path).exists():
Path(temp_file_path).unlink()
# logger.warning(f"An unexpected error occurred: {e}")
raise Exception(f"{error_info}") from e
# Update the file information and print the updated content
with file_name.open("r", encoding="utf-8") as file:
n_total_lines = max(1, len(file.readlines()))
@ -873,7 +869,6 @@ class Editor(BaseModel):
If you need to use it multiple times, wait for the next turn.
"""
file_name = self._try_fix_path(file_name)
ret_str = self._edit_file_impl(
file_name,
start=line_number,
@ -882,6 +877,7 @@ class Editor(BaseModel):
is_insert=True,
is_append=False,
)
self.resource.report(file_name, "path")
return ret_str
def append_file(self, file_name: str, content: str) -> str:
@ -896,7 +892,6 @@ class Editor(BaseModel):
If you need to use it multiple times, wait for the next turn.
"""
file_name = self._try_fix_path(file_name)
ret_str = self._edit_file_impl(
file_name,
start=None,
@ -905,6 +900,7 @@ class Editor(BaseModel):
is_insert=False,
is_append=True,
)
self.resource.report(file_name, "path")
return ret_str
def search_dir(self, search_term: str, dir_path: str = "./") -> str:

19
metagpt/tools/libs/git.py
View file

@ -2,6 +2,8 @@
# -*- coding: utf-8 -*-
from __future__ import annotations
import urllib
from pathlib import Path
from typing import Optional
from github.Issue import Issue
@ -62,11 +64,22 @@ async def git_create_pull(
Returns:
PullRequest: The created pull request.
"""
from metagpt.tools.libs import get_env
from metagpt.utils.git_repository import GitRepository
access_token = await get_env(key="access_token", app_name=app_name)
git_credentials_path = Path.home() / ".git-credentials"
with open(git_credentials_path, "r", encoding="utf-8") as f:
lines = f.readlines()
for line in lines:
line = line.strip()
if not line:
continue
parsed_url = urllib.parse.urlparse(line)
if app_name in parsed_url.hostname:
colon_index = parsed_url.netloc.find(":")
at_index = parsed_url.netloc.find("@")
access_token = parsed_url.netloc[colon_index + 1 : at_index]
break
return await GitRepository.create_pull(
base=base,
head=head,

85
metagpt/tools/libs/image_getter.py Normal file
View file

@ -0,0 +1,85 @@
from __future__ import annotations
from typing import Optional
from playwright.async_api import Browser as Browser_
from playwright.async_api import BrowserContext, Page, Playwright, async_playwright
from pydantic import BaseModel, ConfigDict, Field
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.common import decode_image
from metagpt.utils.proxy_env import get_proxy_from_env
from metagpt.utils.report import BrowserReporter
DOWNLOAD_PICTURE_JAVASCRIPT = """
async () => {{
var img = document.querySelector('{img_element_selector}');
if (img && img.src) {{
const response = await fetch(img.src);
if (response.ok) {{
const blob = await response.blob();
return await new Promise(resolve => {{
const reader = new FileReader();
reader.onloadend = () => resolve(reader.result);
reader.readAsDataURL(blob);
}});
}}
}}
return null;
}}
"""
@register_tool(include_functions=["get_image"])
class ImageGetter(BaseModel):
"""
A tool to get images.
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
playwright: Optional[Playwright] = Field(default=None, exclude=True)
browser_instance: Optional[Browser_] = Field(default=None, exclude=True)
browser_ctx: Optional[BrowserContext] = Field(default=None, exclude=True)
page: Optional[Page] = Field(default=None, exclude=True)
headless: bool = Field(default=True)
proxy: Optional[dict] = Field(default_factory=get_proxy_from_env)
reporter: BrowserReporter = Field(default_factory=BrowserReporter)
url: str = "https://unsplash.com/s/photos/{search_term}/"
img_element_selector: str = ".zNNw1 > div > img:nth-of-type(2)"
async def start(self) -> None:
"""Starts Playwright and launches a browser"""
if self.playwright is None:
self.playwright = playwright = await async_playwright().start()
browser = self.browser_instance = await playwright.chromium.launch(headless=self.headless, proxy=self.proxy)
browser_ctx = self.browser_ctx = await browser.new_context()
self.page = await browser_ctx.new_page()
async def get_image(self, search_term, image_save_path):
"""
Get an image related to the search term.
Args:
search_term (str): The term to search for the image. The search term must be in English. Using any other language may lead to a mismatch.
image_save_path (str): The file path where the image will be saved.
"""
# Search for images from https://unsplash.com/s/photos/
if self.page is None:
await self.start()
await self.page.goto(self.url.format(search_term=search_term), wait_until="domcontentloaded")
# Wait until the image element is loaded
try:
await self.page.wait_for_selector(self.img_element_selector)
except TimeoutError:
return f"{search_term} not found. Please broaden the search term."
# Get the base64 code of the first retrieved image
image_base64 = await self.page.evaluate(
DOWNLOAD_PICTURE_JAVASCRIPT.format(img_element_selector=self.img_element_selector)
)
if image_base64:
image = decode_image(image_base64)
image.save(image_save_path)
return f"{search_term} found. The image is saved in {image_save_path}."
return f"{search_term} not found. Please broaden the search term."

1
metagpt/tools/libs/linter.py
View file

@ -33,6 +33,7 @@ class Linter:
self.languages = dict(
python=self.py_lint,
sql=self.fake_lint, # base_lint lacks support for full SQL syntax. Use fake_lint to bypass the validation.
css=self.fake_lint, # base_lint lacks support for css syntax. Use fake_lint to bypass the validation.
)
self.all_lint_cmd = None