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

Merge branch 'mgx_ops' into dynamic_think

Browse source
This commit is contained in:
yzlin 2024-05-09 10:20:43 +08:00
commit b3b50b3edf
32 changed files with 1064 additions and 350 deletions
Download patch file Download diff file Expand all files Collapse all files

1
examples/di/run_flask.py
View file

@ -2,7 +2,6 @@ import asyncio
from metagpt.roles.di.data_interpreter import DataInterpreter
USE_GOT_REPO_REQ = """
Write a service using Flask, create a conda environment and run it, and call the service's interface for validation.
Notice: Don't write all codes in one response, each time, just write code for one step.

4
metagpt/actions/action_node.py
View file

@ -17,7 +17,7 @@ from pydantic import BaseModel, Field, create_model, model_validator
from tenacity import retry, stop_after_attempt, wait_random_exponential
from metagpt.actions.action_outcls_registry import register_action_outcls
from metagpt.const import USE_CONFIG_TIMEOUT
from metagpt.const import MARKDOWN_TITLE_PREFIX, USE_CONFIG_TIMEOUT
from metagpt.llm import BaseLLM
from metagpt.logs import logger
from metagpt.provider.postprocess.llm_output_postprocess import llm_output_postprocess
@ -113,7 +113,7 @@ Follow format example's {prompt_schema} format, generate output and make sure it
"""
def dict_to_markdown(d, prefix="- ", kv_sep="\n", postfix="\n"):
def dict_to_markdown(d, prefix=MARKDOWN_TITLE_PREFIX, kv_sep="\n", postfix="\n"):
markdown_str = ""
for key, value in d.items():
markdown_str += f"{prefix}{key}{kv_sep}{value}{postfix}"

34
metagpt/actions/design_api.py
View file

@ -26,6 +26,7 @@ from metagpt.const import DATA_API_DESIGN_FILE_REPO, SEQ_FLOW_FILE_REPO
from metagpt.logs import logger
from metagpt.schema import Document, Documents, Message
from metagpt.utils.mermaid import mermaid_to_file
from metagpt.utils.report import DocsReporter
NEW_REQ_TEMPLATE = """
### Legacy Content
@ -70,31 +71,34 @@ class WriteDesign(Action):
return ActionOutput(content=changed_files.model_dump_json(), instruct_content=changed_files)
async def _new_system_design(self, context):
node = await DESIGN_API_NODE.fill(context=context, llm=self.llm)
node = await DESIGN_API_NODE.fill(context=context, llm=self.llm, schema=self.prompt_schema)
return node
async def _merge(self, prd_doc, system_design_doc):
context = NEW_REQ_TEMPLATE.format(old_design=system_design_doc.content, context=prd_doc.content)
node = await REFINED_DESIGN_NODE.fill(context=context, llm=self.llm)
node = await REFINED_DESIGN_NODE.fill(context=context, llm=self.llm, schema=self.prompt_schema)
system_design_doc.content = node.instruct_content.model_dump_json()
return system_design_doc
async def _update_system_design(self, filename) -> Document:
prd = await self.repo.docs.prd.get(filename)
old_system_design_doc = await self.repo.docs.system_design.get(filename)
if not old_system_design_doc:
system_design = await self._new_system_design(context=prd.content)
doc = await self.repo.docs.system_design.save(
filename=filename,
content=system_design.instruct_content.model_dump_json(),
dependencies={prd.root_relative_path},
)
else:
doc = await self._merge(prd_doc=prd, system_design_doc=old_system_design_doc)
await self.repo.docs.system_design.save_doc(doc=doc, dependencies={prd.root_relative_path})
await self._save_data_api_design(doc)
await self._save_seq_flow(doc)
await self.repo.resources.system_design.save_pdf(doc=doc)
async with DocsReporter(enable_llm_stream=True) as reporter:
await reporter.async_report({"type": "design"}, "meta")
if not old_system_design_doc:
system_design = await self._new_system_design(context=prd.content)
doc = await self.repo.docs.system_design.save(
filename=filename,
content=system_design.instruct_content.model_dump_json(),
dependencies={prd.root_relative_path},
)
else:
doc = await self._merge(prd_doc=prd, system_design_doc=old_system_design_doc)
await self.repo.docs.system_design.save_doc(doc=doc, dependencies={prd.root_relative_path})
await self._save_data_api_design(doc)
await self._save_seq_flow(doc)
md = await self.repo.resources.system_design.save_pdf(doc=doc)
await reporter.async_report(self.repo.workdir / md.root_relative_path, "path")
return doc
async def _save_data_api_design(self, design_doc):

96
metagpt/actions/di/execute_nb_code.py
View file

@ -13,9 +13,10 @@ from typing import Literal, Tuple
import nbformat
from nbclient import NotebookClient
from nbclient.exceptions import CellTimeoutError, DeadKernelError
from nbclient.exceptions import CellExecutionComplete, CellTimeoutError, DeadKernelError
from nbclient.util import ensure_async
from nbformat import NotebookNode
from nbformat.v4 import new_code_cell, new_markdown_cell, new_output
from nbformat.v4 import new_code_cell, new_markdown_cell, new_output, output_from_msg
from rich.box import MINIMAL
from rich.console import Console, Group
from rich.live import Live
@ -25,32 +26,64 @@ from rich.syntax import Syntax
from metagpt.actions import Action
from metagpt.const import DEFAULT_WORKSPACE_ROOT
from metagpt.logs import ToolLogItem, log_tool_output, logger
from metagpt.logs import logger
from metagpt.utils.report import NotebookReporter
INSTALL_KEEPLEN = 500
class RealtimeOutputNotebookClient(NotebookClient):
"""Realtime output of Notebook execution."""
def __init__(self, *args, notebook_reporter=None, **kwargs) -> None:
super().__init__(*args, **kwargs)
self.notebook_reporter = notebook_reporter or NotebookReporter()
async def _async_poll_output_msg(self, parent_msg_id: str, cell: NotebookNode, cell_index: int) -> None:
"""Implement a feature to enable sending messages."""
assert self.kc is not None
while True:
msg = await ensure_async(self.kc.iopub_channel.get_msg(timeout=None))
await self._send_msg(msg)
if msg["parent_header"].get("msg_id") == parent_msg_id:
try:
# Will raise CellExecutionComplete when completed
self.process_message(msg, cell, cell_index)
except CellExecutionComplete:
return
async def _send_msg(self, msg: dict):
msg_type = msg.get("header", {}).get("msg_type")
if msg_type not in ["stream", "error", "execute_result"]:
return
await self.notebook_reporter.async_report(output_from_msg(msg), "content")
class ExecuteNbCode(Action):
"""execute notebook code block, return result to llm, and display it."""
nb: NotebookNode
nb_client: NotebookClient
nb_client: NotebookClient = None
console: Console
interaction: str
timeout: int = 600
def __init__(
self,
nb=nbformat.v4.new_notebook(),
timeout=600,
):
def __init__(self, nb=nbformat.v4.new_notebook(), timeout=600):
super().__init__(
nb=nb,
nb_client=NotebookClient(nb, timeout=timeout, resources={"metadata": {"path": DEFAULT_WORKSPACE_ROOT}}),
timeout=timeout,
console=Console(),
interaction=("ipython" if self.is_ipython() else "terminal"),
)
self.reporter = NotebookReporter()
self.nb_client = RealtimeOutputNotebookClient(
nb,
timeout=timeout,
resources={"metadata": {"path": DEFAULT_WORKSPACE_ROOT}},
notebook_reporter=self.reporter,
)
async def build(self):
if self.nb_client.kc is None or not await self.nb_client.kc.is_alive():
@ -175,6 +208,8 @@ class ExecuteNbCode(Action):
"""set timeout for run code.
returns the success or failure of the cell execution, and an optional error message.
"""
await self.reporter.async_report(cell, "content")
try:
await self.nb_client.async_execute_cell(cell, cell_index)
return self.parse_outputs(self.nb.cells[-1].outputs)
@ -196,35 +231,36 @@ class ExecuteNbCode(Action):
"""
self._display(code, language)
if language == "python":
# add code to the notebook
self.add_code_cell(code=code)
async with self.reporter:
if language == "python":
# add code to the notebook
self.add_code_cell(code=code)
# build code executor
await self.build()
# build code executor
await self.build()
# run code
cell_index = len(self.nb.cells) - 1
success, outputs = await self.run_cell(self.nb.cells[-1], cell_index)
# run code
cell_index = len(self.nb.cells) - 1
success, outputs = await self.run_cell(self.nb.cells[-1], cell_index)
if "!pip" in code:
success = False
outputs = outputs[-INSTALL_KEEPLEN:]
if "!pip" in code:
success = False
outputs = outputs[-INSTALL_KEEPLEN:]
elif language == "markdown":
# add markdown content to markdown cell in a notebook.
self.add_markdown_cell(code)
# return True, beacuse there is no execution failure for markdown cell.
outputs, success = code, True
else:
raise ValueError(f"Only support for language: python, markdown, but got {language}, ")
file_path = DEFAULT_WORKSPACE_ROOT / "code.ipynb"
nbformat.write(self.nb, file_path)
log_tool_output(ToolLogItem(name="file_path", value=file_path), tool_name="ExecuteNbCode")
await self.reporter.async_report(file_path, "path")
return outputs, success
elif language == "markdown":
# add markdown content to markdown cell in a notebook.
self.add_markdown_cell(code)
# return True, beacuse there is no execution failure for markdown cell.
return code, True
else:
raise ValueError(f"Only support for language: python, markdown, but got {language}, ")
def remove_escape_and_color_codes(input_str: str):
# 使用正则表达式去除jupyter notebook输出结果中的转义字符和颜色代码

28
metagpt/actions/project_management.py
View file

@ -19,6 +19,7 @@ from metagpt.actions.project_management_an import PM_NODE, REFINED_PM_NODE
from metagpt.const import PACKAGE_REQUIREMENTS_FILENAME
from metagpt.logs import logger
from metagpt.schema import Document, Documents
from metagpt.utils.report import DocsReporter
NEW_REQ_TEMPLATE = """
### Legacy Content
@ -59,18 +60,21 @@ class WriteTasks(Action):
async def _update_tasks(self, filename):
system_design_doc = await self.repo.docs.system_design.get(filename)
task_doc = await self.repo.docs.task.get(filename)
if task_doc:
task_doc = await self._merge(system_design_doc=system_design_doc, task_doc=task_doc)
await self.repo.docs.task.save_doc(doc=task_doc, dependencies={system_design_doc.root_relative_path})
else:
rsp = await self._run_new_tasks(context=system_design_doc.content)
task_doc = await self.repo.docs.task.save(
filename=filename,
content=rsp.instruct_content.model_dump_json(),
dependencies={system_design_doc.root_relative_path},
)
await self._update_requirements(task_doc)
await self.repo.resources.api_spec_and_task.save_pdf(doc=task_doc)
async with DocsReporter(enable_llm_stream=True) as reporter:
await reporter.async_report({"type": "task"}, "meta")
if task_doc:
task_doc = await self._merge(system_design_doc=system_design_doc, task_doc=task_doc)
await self.repo.docs.task.save_doc(doc=task_doc, dependencies={system_design_doc.root_relative_path})
else:
rsp = await self._run_new_tasks(context=system_design_doc.content)
task_doc = await self.repo.docs.task.save(
filename=filename,
content=rsp.instruct_content.model_dump_json(),
dependencies={system_design_doc.root_relative_path},
)
await self._update_requirements(task_doc)
md = await self.repo.resources.api_spec_and_task.save_pdf(doc=task_doc)
await reporter.async_report(self.repo.workdir / md.root_relative_path, "path")
return task_doc
async def _run_new_tasks(self, context):

21
metagpt/actions/write_code.py
View file

@ -28,6 +28,7 @@ from metagpt.logs import logger
from metagpt.schema import CodingContext, Document, RunCodeResult
from metagpt.utils.common import CodeParser
from metagpt.utils.project_repo import ProjectRepo
from metagpt.utils.report import EditorReporter
PROMPT_TEMPLATE = """
NOTICE
@ -65,6 +66,11 @@ ATTENTION: Use '##' to SPLIT SECTIONS, not '#'. Output format carefully referenc
## {filename}
...
```
## Code: {filename}
```javascript
// {filename}
...
```
# Instruction: Based on the context, follow "Format example", write code.
@ -139,12 +145,15 @@ class WriteCode(Action):
summary_log=summary_doc.content if summary_doc else "",
)
logger.info(f"Writing {coding_context.filename}..")
code = await self.write_code(prompt)
if not coding_context.code_doc:
# avoid root_path pydantic ValidationError if use WriteCode alone
root_path = self.context.src_workspace if self.context.src_workspace else ""
coding_context.code_doc = Document(filename=coding_context.filename, root_path=str(root_path))
coding_context.code_doc.content = code
async with EditorReporter(enable_llm_stream=True) as reporter:
await reporter.async_report({"filename": coding_context.filename}, "meta")
code = await self.write_code(prompt)
if not coding_context.code_doc:
# avoid root_path pydantic ValidationError if use WriteCode alone
root_path = self.context.src_workspace if self.context.src_workspace else ""
coding_context.code_doc = Document(filename=coding_context.filename, root_path=str(root_path))
coding_context.code_doc.content = code
await reporter.async_report(self.repo.workdir / coding_context.code_doc.root_relative_path, "path")
return coding_context
@staticmethod

5
metagpt/actions/write_code_plan_and_change_an.py
View file

@ -184,6 +184,11 @@ Role: You are a professional engineer; The main goal is to complete incremental
## {filename}
...
```
## Code: {filename}
```javascript
// {filename}
...
```
# Instruction: Based on the context, follow "Format example", write or rewrite code.
## Write/Rewrite Code: Only write one file {filename}, write or rewrite complete code using triple quotes based on the following attentions and context.

9
metagpt/actions/write_code_review.py
View file

@ -110,11 +110,16 @@ LGTM
REWRITE_CODE_TEMPLATE = """
# Instruction: rewrite code based on the Code Review and Actions
## Rewrite Code: CodeBlock. If it still has some bugs, rewrite {filename} with triple quotes. Do your utmost to optimize THIS SINGLE FILE. Return all completed codes and prohibit the return of unfinished codes.
```Code
## Rewrite Code: CodeBlock. If it still has some bugs, rewrite {filename} using a Markdown code block, with the filename docstring preceding the code block. Do your utmost to optimize THIS SINGLE FILE. Return all completed codes and prohibit the return of unfinished codes.
```python
## {filename}
...
```
or
```javascript
// {filename}
...
```
"""

39
metagpt/actions/write_prd.py
View file

@ -37,6 +37,7 @@ from metagpt.schema import BugFixContext, Document, Documents, Message
from metagpt.utils.common import CodeParser
from metagpt.utils.file_repository import FileRepository
from metagpt.utils.mermaid import mermaid_to_file
from metagpt.utils.report import DocsReporter
CONTEXT_TEMPLATE = """
### Project Name
@ -102,17 +103,22 @@ class WritePRD(Action):
async def _handle_new_requirement(self, req: Document) -> ActionOutput:
"""handle new requirement"""
project_name = self.project_name
context = CONTEXT_TEMPLATE.format(requirements=req, project_name=project_name)
exclude = [PROJECT_NAME.key] if project_name else []
node = await WRITE_PRD_NODE.fill(context=context, llm=self.llm, exclude=exclude) # schema=schema
await self._rename_workspace(node)
new_prd_doc = await self.repo.docs.prd.save(
filename=FileRepository.new_filename() + ".json", content=node.instruct_content.model_dump_json()
)
await self._save_competitive_analysis(new_prd_doc)
await self.repo.resources.prd.save_pdf(doc=new_prd_doc)
return Documents.from_iterable(documents=[new_prd_doc]).to_action_output()
async with DocsReporter(enable_llm_stream=True) as reporter:
await reporter.async_report({"type": "prd"}, "meta")
project_name = self.project_name
context = CONTEXT_TEMPLATE.format(requirements=req, project_name=project_name)
exclude = [PROJECT_NAME.key] if project_name else []
node = await WRITE_PRD_NODE.fill(
context=context, llm=self.llm, exclude=exclude, schema=self.prompt_schema
) # schema=schema
await self._rename_workspace(node)
new_prd_doc = await self.repo.docs.prd.save(
filename=FileRepository.new_filename() + ".json", content=node.instruct_content.model_dump_json()
)
await self._save_competitive_analysis(new_prd_doc)
md = await self.repo.resources.prd.save_pdf(doc=new_prd_doc)
await reporter.async_report(self.repo.workdir / md.root_relative_path, "path")
return Documents.from_iterable(documents=[new_prd_doc]).to_action_output()
async def _handle_requirement_update(self, req: Document, related_docs: list[Document]) -> ActionOutput:
# ... requirement update logic ...
@ -146,10 +152,13 @@ class WritePRD(Action):
return related_doc
async def _update_prd(self, req: Document, prd_doc: Document) -> Document:
new_prd_doc: Document = await self._merge(req, prd_doc)
await self.repo.docs.prd.save_doc(doc=new_prd_doc)
await self._save_competitive_analysis(new_prd_doc)
await self.repo.resources.prd.save_pdf(doc=new_prd_doc)
async with DocsReporter(enable_llm_stream=True) as reporter:
await reporter.async_report({"type": "prd"}, "meta")
new_prd_doc: Document = await self._merge(req, prd_doc)
await self.repo.docs.prd.save_doc(doc=new_prd_doc)
await self._save_competitive_analysis(new_prd_doc)
md = await self.repo.resources.prd.save_pdf(doc=new_prd_doc)
await reporter.async_report(self.repo.workdir / md.root_relative_path, "path")
return new_prd_doc
async def _save_competitive_analysis(self, prd_doc: Document):

2
metagpt/actions/write_prd_an.py
View file

@ -165,7 +165,7 @@ ANYTHING_UNCLEAR = ActionNode(
key="Anything UNCLEAR",
expected_type=str,
instruction="Mention any aspects of the project that are unclear and try to clarify them.",
example="",
example="Currently, all aspects of the project are clear.",
)
ISSUE_TYPE = ActionNode(

2
metagpt/configs/role_custom_config.py
View file

@ -14,6 +14,6 @@ class RoleCustomConfig(YamlModel):
role: role's className or role's role_id
To be expanded
"""
role: str = ""
llm: LLMConfig

6
metagpt/const.py
View file

@ -140,5 +140,11 @@ LLM_API_TIMEOUT = 300
# Assistant alias
ASSISTANT_ALIAS = "response"
# Markdown
MARKDOWN_TITLE_PREFIX = "## "
# Reporter
METAGPT_REPORTER_DEFAULT_URL = os.environ.get("METAGPT_REPORTER_URL", "")
# Metadata defines
AGENT = "agent"

38
metagpt/logs.py
View file

@ -8,8 +8,10 @@
from __future__ import annotations
import asyncio
import inspect
import sys
from contextvars import ContextVar
from datetime import datetime
from functools import partial
from typing import Any
@ -19,6 +21,8 @@ from pydantic import BaseModel, Field
from metagpt.const import METAGPT_ROOT
LLM_STREAM_QUEUE: ContextVar[asyncio.Queue] = ContextVar("llm-stream")
class ToolLogItem(BaseModel):
type_: str = Field(alias="type", default="str", description="Data type of `value` field.")
@ -47,6 +51,20 @@ logger = define_log_level()
def log_llm_stream(msg):
"""
Logs a message to the LLM stream.
Args:
msg: The message to be logged.
Notes:
If the LLM_STREAM_QUEUE has not been set (e.g., if `create_llm_stream_queue` has not been called),
the message will not be added to the LLM stream queue.
"""
queue = get_llm_stream_queue()
if queue:
queue.put_nowait(msg)
_llm_stream_log(msg)
@ -102,4 +120,24 @@ async def _tool_output_log_async(*args, **kwargs):
pass
def create_llm_stream_queue():
"""Creates a new LLM stream queue and sets it in the context variable.
Returns:
The newly created asyncio.Queue instance.
"""
queue = asyncio.Queue()
LLM_STREAM_QUEUE.set(queue)
return queue
def get_llm_stream_queue():
"""Retrieves the current LLM stream queue from the context variable.
Returns:
The asyncio.Queue instance if set, otherwise None.
"""
return LLM_STREAM_QUEUE.get(None)
_get_human_input = input # get human input from console by default

3
metagpt/roles/role.py
View file

@ -443,6 +443,9 @@ class Role(SerializationMixin, ContextMixin, BaseModel):
"""If the role belongs to env, then the role's messages will be broadcast to env"""
if not msg:
return
if all(to in {any_to_str(self), self.name} for to in msg.send_to): # Message to myself
self.put_message(msg)
return
if not self.rc.env:
# If env does not exist, do not publish the message
return

12
metagpt/schema.py
View file

@ -47,10 +47,11 @@ from metagpt.const import (
SYSTEM_DESIGN_FILE_REPO,
TASK_FILE_REPO,
)
from metagpt.logs import ToolLogItem, log_tool_output, logger
from metagpt.logs import logger
from metagpt.repo_parser import DotClassInfo
from metagpt.utils.common import CodeParser, any_to_str, any_to_str_set, import_class
from metagpt.utils.exceptions import handle_exception
from metagpt.utils.report import TaskReporter
from metagpt.utils.serialize import (
actionoutout_schema_to_mapping,
actionoutput_mapping_to_str,
@ -578,14 +579,7 @@ class Plan(BaseModel):
current_task_id = task.task_id
break
self.current_task_id = current_task_id
log_tool_output(
[
ToolLogItem(type="object", name="tasks", value=self.tasks),
ToolLogItem(type="object", name="current_task_id", value=self.current_task_id),
],
tool_name="Plan",
)
TaskReporter().report({"tasks": self.tasks, "current_task_id": current_task_id})
@property
def current_task(self) -> Task:

3
metagpt/software_company.py
View file

@ -8,7 +8,6 @@ import typer
from metagpt.const import CONFIG_ROOT
from metagpt.utils.common import any_to_str
from metagpt.utils.project_repo import ProjectRepo
app = typer.Typer(add_completion=False, pretty_exceptions_show_locals=False)
@ -26,7 +25,7 @@ def generate_repo(
reqa_file="",
max_auto_summarize_code=0,
recover_path=None,
) -> ProjectRepo:
):
"""Run the startup logic. Can be called from CLI or other Python scripts."""
from metagpt.config2 import config
from metagpt.context import Context

31
metagpt/tools/libs/browser.py
View file

@ -1,8 +1,8 @@
from playwright.async_api import async_playwright
from metagpt.const import DEFAULT_WORKSPACE_ROOT
from metagpt.logs import ToolLogItem, log_tool_output_async
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.report import BrowserReporter
@register_tool()
@ -20,6 +20,7 @@ class Browser:
self.pages = {}
self.current_page_url = None
self.current_page = None
self.reporter = BrowserReporter()
async def start(self):
"""Starts Playwright and launches a browser"""
@ -34,21 +35,19 @@ class Browser:
async def open_new_page(self, url: str):
"""open a new page in the browser and view the page"""
page = await self.browser.new_page()
await page.goto(url)
self.pages[url] = page
await self._set_current_page(page, url)
await log_tool_output_async(
ToolLogItem(type="object", name="open_new_page", value=self.current_page), tool_name="Browser"
)
async with self.reporter as reporter:
page = await self.browser.new_page()
await reporter.async_report(url, "url")
await page.goto(url)
self.pages[url] = page
await self._set_current_page(page, url)
await reporter.async_report(page, "page")
async def switch_page(self, url: str):
"""switch to an opened page in the browser and view the page"""
if url in self.pages:
await self._set_current_page(self.pages[url], url)
await log_tool_output_async(
ToolLogItem(type="object", name="switch_page", value=self.current_page), tool_name="Browser"
)
await self.reporter.async_report(self.current_page, "page")
else:
print(f"Page not found: {url}")
@ -110,9 +109,8 @@ class Browser:
index = len(search_results) - 1
element = search_results[index]["element_obj"]
await element.scroll_into_view_if_needed()
await log_tool_output_async(
ToolLogItem(type="object", name="scroll_page", value=self.current_page), tool_name="Browser"
)
await self.reporter.async_report(self.current_page, "page")
print(f"Successfully scrolled to the {index}-th search result")
print(await self._view())
@ -152,9 +150,8 @@ class Browser:
async def scroll_current_page(self, offset: int = 500):
"""scroll the current page by offset pixels, negative value means scrolling up, will print out observed content after scrolling"""
await self.current_page.evaluate(f"window.scrollBy(0, {offset})")
await log_tool_output_async(
ToolLogItem(type="object", name="scroll_page", value=self.current_page), tool_name="Browser"
)
await self.reporter.async_report(self.current_page, "page")
print(f"Scrolled current page by {offset} pixels.")
print(await self._view())

4
metagpt/tools/libs/deployer.py
View file

@ -1,5 +1,5 @@
from metagpt.logs import ToolLogItem, log_tool_output
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.report import ServerReporter
# An un-implemented tool reserved for deploying a local service to public
@ -8,4 +8,4 @@ class Deployer:
"""Deploy a local service to public. Used only for final deployment, you should NOT use it for development and testing."""
def deploy_to_public(self, local_url: str):
log_tool_output(ToolLogItem(name="local_url", value=local_url), tool_name="Deployer")
ServerReporter().report(local_url, "local_url")

18
metagpt/tools/libs/file_manager.py
View file

@ -4,8 +4,8 @@ import subprocess
from pydantic import BaseModel, Field
from metagpt.logs import ToolLogItem, log_tool_output
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.report import EditorReporter
class FileBlock(BaseModel):
@ -23,17 +23,20 @@ class FileBlock(BaseModel):
class FileManager:
"""A tool for reading, understanding, writing, and editing files"""
def __init__(self) -> None:
self.resource = EditorReporter()
def write(self, path: str, content: str):
"""Write the whole content to a file."""
with open(path, "w") as f:
f.write(content)
log_tool_output(ToolLogItem(name="write_file_path", value=path), tool_name="FileManager")
self.resource.report(path, "path")
def read(self, path: str) -> str:
"""Read the whole content of a file."""
with open(path, "r") as f:
self.resource.report(path, "path")
return f.read()
log_tool_output(ToolLogItem(name="read_file_path", value=path), tool_name="FileManager")
def search_content(self, symbol: str, root_path: str = "", window: int = 20) -> FileBlock:
"""
@ -78,10 +81,7 @@ class FileManager:
symbol=symbol,
symbol_line=i + 1,
)
log_tool_output(
ToolLogItem(type="object", name="file_block_searched", value=result),
tool_name="FileManager",
)
self.resource.report(result.file_path, "path")
return result
return None
@ -124,9 +124,7 @@ class FileManager:
block_start_line=start_line,
block_end_line=-1 if end_line < start_line else start_line + new_block_content.count("\n"),
)
log_tool_output(
ToolLogItem(type="object", name="file_block_written", value=new_file_block), tool_name="FileManager"
)
self.resource.report(new_file_block.file_path, "path")
return f"Content written successfully to {file_path}"

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

@ -3,13 +3,13 @@
from __future__ import annotations
from pathlib import Path
from typing import Optional
from typing import Optional, Union
from github.Issue import Issue
from github.PullRequest import PullRequest
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.git_repository import GitRepository
from metagpt.utils.git_repository import GitBranch, GitRepository
# @register_tool(tags=["git"])
@ -18,41 +18,33 @@ async def git_clone(url: str, output_dir: str | Path = None) -> Path:
Clones a Git repository from the given URL.
Args:
url (str): The URL of the Git repository to clone.
output_dir (str or Path, optional): The directory where the repository will be cloned.
If not provided, the repository will be cloned into the current working directory.
url (Union[str, Path]): The URL or local path of the Git repository to clone.
output_dir (Union[str, Path], optional): The directory where the repository should be cloned.
If None, the repository will be cloned into the current working directory. Defaults to None.
Returns:
Path: The path to the cloned repository.
Raises:
ValueError: If the specified Git root is invalid.
Example:
>>> # git clone to /TO/PATH
>>> url = 'https://github.com/geekan/MetaGPT.git'
>>> output_dir = "/TO/PATH"
>>> repo_dir = await git_clone(url=url, output_dir=output_dir)
>>> print(repo_dir)
/TO/PATH/MetaGPT
>>> # git clone to default directory.
>>> url = 'https://github.com/geekan/MetaGPT.git'
>>> repo_dir = await git_clone(url)
>>> print(repo_dir)
/WORK_SPACE/downloads/MetaGPT
>>> url = "https://github.com/iorisa/snake-game.git"
>>> local_path = await git_clone(url=url)
>>> print(local_path)
/local/path/to/snake-game
"""
repo = await GitRepository.clone_from(url, output_dir)
repo = await GitRepository.clone_from(url=url, output_dir=output_dir)
return repo.workdir
@register_tool(
tags=["software development", "git", "Checks out the specific commit/branch/tag of the local Git repository."]
)
async def git_checkout(repo_dir: str | Path, commit_id: str):
"""
Checks out a specific commit in a Git repository.
Args:
repo_dir (str or Path): The directory containing the Git repository.
commit_id (str): The ID of the commit to check out.
commit_id (str): The ID of the commit or the name of branch/tag to check out.
Raises:
ValueError: If the specified Git root is invalid.
@ -69,145 +61,171 @@ async def git_checkout(repo_dir: str | Path, commit_id: str):
await repo.checkout(commit_id)
@register_tool(tags=["git"])
async def create_pull_request(
@register_tool(tags=["software development", "git", "Commit the changes and push to remote git repository."])
async def git_push(
local_path: Union[str, Path],
access_token: str,
comments: str = "Commit",
new_branch: str = "",
) -> GitBranch:
"""
Pushes changes from a local Git repository to its remote counterpart.
Args:
local_path (Union[str, Path]): The path to the local Git repository.
access_token (str): The access token for authentication. Use `get_env` to get access token.
comments (str, optional): The commit message to use. Defaults to "Commit".
new_branch (str, optional): The name of the new branch to create and push changes to.
If not provided, changes will be pushed to the current branch. Defaults to "".
Returns:
GitBranch: The branch to which the changes were pushed.
Raises:
ValueError: If the provided local_path does not point to a valid Git repository.
Example:
>>> url = "https://github.com/iorisa/snake-game.git"
>>> local_path = await git_clone(url=url)
>>> from metagpt.tools.libs import get_env
>>> access_token = await get_env(key="access_token", app_name="github") # Read access token from enviroment variables.
>>> comments = "Archive"
>>> new_branch = "feature/new"
>>> branch = await git_push(local_path=local_path, access_token=access_token, comments=comments, new_branch=new_branch)
>>> base = branch.base
>>> head = branch.head
>>> repo_name = branch.repo_name
>>> print(f"base branch:'{base}', head branch:'{head}', repo_name:'{repo_name}'")
base branch:'master', head branch:'feature/new', repo_name:'iorisa/snake-game'
"""
if not GitRepository.is_git_dir(local_path):
raise ValueError("Invalid local git repository")
repo = GitRepository(local_path=local_path, auto_init=False)
branch = await repo.push(new_branch=new_branch, comments=comments, access_token=access_token)
return branch
@register_tool(tags=["software development", "git", "create a git pull/merge request"])
async def git_create_pull(
base: str,
head: str,
base_repo_name: str,
access_token: str,
head_repo_name: Optional[str] = None,
title: Optional[str] = None,
body: Optional[str] = None,
issue: Optional[Issue] = None,
) -> PullRequest:
"""
Creates a pull request in a Git repository.
Creates a pull request on a Git repository.
Args:
access_token (str): The access token for authentication.
base (str): The name of the base branch of the pull request (e.g., 'main', 'master').
head (str): The name of the head branch of the pull request (e.g., 'feature-branch').
base (str): The base branch of the pull request.
head (str): The head branch of the pull request.
base_repo_name (str): The full repository name (user/repo) where the pull request will be created.
access_token (str): The access token for authentication. Use `get_env` to get access token.
head_repo_name (Optional[str], optional): The full repository name (user/repo) where the pull request will merge from. Defaults to None.
title (Optional[str]): The title of the pull request.
body (Optional[str]): The body of the pull request.
title (Optional[str], optional): The title of the pull request. Defaults to None.
body (Optional[str], optional): The body of the pull request. Defaults to None.
issue (Optional[Issue], optional): The related issue of the pull request. Defaults to None.
Example:
>>> # push and create pull
>>> url = "https://github.com/iorisa/snake-game.git"
>>> local_path = await git_clone(url=url)
>>> from metagpt.tools.libs import get_env
>>> access_token = await get_env(key="access_token", app_name="github")
>>> comments = "Archive"
>>> new_branch = "feature/new"
>>> branch = await git_push(local_path=local_path, access_token=access_token, comments=comments, new_branch=new_branch)
>>> base = branch.base
>>> head = branch.head
>>> repo_name = branch.repo_name
>>> print(f"base branch:'{base}', head branch:'{head}', repo_name:'{repo_name}'")
base branch:'master', head branch:'feature/new', repo_name:'iorisa/snake-game'
>>> title = "feat: modify http lib",
>>> body = "Change HTTP library used to send requests"
>>> pr = await git_create_pull(
>>> base_repo_name=repo_name,
>>> base=base,
>>> head=head,
>>> title=title,
>>> body=body,
>>> access_token=access_token,
>>> )
>>> print(pr)
PullRequest("feat: modify http lib")
>>> # create pull request
>>> base_repo_name = "geekan/MetaGPT"
>>> head_repo_name = "ioris/MetaGPT"
>>> base = "master"
>>> head = "feature/http"
>>> title = "feat: modify http lib",
>>> body = "Change HTTP library used to send requests"
>>> from metagpt.tools.libs import get_env
>>> access_token = await get_env(key="access_token", app_name="github")
>>> pr = await git_create_pull(
>>> base_repo_name=base_repo_name,
>>> head_repo_name=head_repo_name,
>>> base=base,
>>> head=head,
>>> title=title,
>>> body=body,
>>> access_token=access_token,
>>> )
>>> print(pr)
PullRequest("feat: modify http lib")
Returns:
PullRequest: The created pull request object.
Raises:
ValueError: If `access_token` is invalid. Visit: "https://github.com/settings/tokens"
Any exceptions that might occur during the pull request creation process.
Note:
This function is intended to be used in an asynchronous context (with `await`).
Example:
>>> # Merge Request
>>> repo_name = "user/repo" # "user/repo" for example: "https://github.com/user/repo.git"
>>> base = "master" # branch that merge to
>>> head = "feature/new_feature" # branch that merge from
>>> title = "Implement new feature"
>>> body = "This pull request adds functionality X, Y, and Z."
>>> pull_request = await create_pull_request(
repo_name=repo_name,
base=base,
head=head,
title=title,
body=body,
access_token=get_env("git_access_token")
)
>>> print(pull_request)
PullRequest(title="Implement new feature", number=26)
>>> # Pull Request
>>> base_repo_name = "user1/repo1" # for example: "user1/repo1" from "https://github.com/user1/repo1.git"
>>> head_repo_name = "user2/repo2" # for example: "user2/repo2" from "https://github.com/user2/repo2.git"
>>> base = "master" # branch that merge to
>>> head = "feature/new_feature" # branch that merge from
>>> title = "Implement new feature"
>>> body = "This pull request adds functionality X, Y, and Z."
>>> pull_request = await create_pull_request(
base_repo_name=base_repo_name,
head_repo_name=head_repo_name,
base=base,
head=head,
title=title,
body=body,
access_token=get_env("git_access_token")
)
>>> print(pull_request)
PullRequest(title="Implement new feature", number=26)
PullRequest: The created pull request.
"""
return await GitRepository.create_pull(
base_repo_name=base_repo_name,
head_repo_name=head_repo_name,
base=base,
head=head,
base_repo_name=base_repo_name,
head_repo_name=head_repo_name,
title=title,
body=body,
issue=issue,
access_token=access_token,
)
@register_tool(tags=["git"])
async def create_issue(
access_token: str,
@register_tool(tags=["software development", "create a git issue"])
async def git_create_issue(
repo_name: str,
title: str,
access_token: str,
body: Optional[str] = None,
assignee: Optional[str] = None,
labels: Optional[list[str]] = None,
) -> Issue:
"""
Creates an issue in the specified repository.
Creates an issue on a Git repository.
Args:
access_token (str): The access token for authentication.
Visit `https://github.com/settings/tokens` to obtain a personal access token.
For more authentication options, visit: `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html`
repo_name (str): The full repository name (user/repo) where the issue will be created.
repo_name (str): The name of the repository.
title (str): The title of the issue.
access_token (str): The access token for authentication. Use `get_env` to get access token.
body (Optional[str], optional): The body of the issue. Defaults to None.
assignee (Optional[str], optional): The username of the assignee for the issue. Defaults to None.
labels (Optional[list[str]], optional): A list of label names to associate with the issue. Defaults to None.
Returns:
Issue: The created issue object.
Example:
>>> # Create an issue with title and body
>>> repo_name = "username/repository"
>>> title = "Bug Report"
>>> body = "I found a bug in the application."
>>> issue = await create_issue(
repo_name=repo_name,
title=title,
body=body,
access_token=get_env("git_access_token")
)
>>> repo_name = "geekan/MetaGPT"
>>> title = "This is a new issue"
>>> from metagpt.tools.libs import get_env
>>> access_token = await get_env(key="access_token", app_name="github")
>>> body = "This is the issue body."
>>> issue = await git_create_issue(
>>> repo_name=repo_name,
>>> title=title,
>>> access_token=access_token,
>>> body=body,
>>> )
>>> print(issue)
Issue(title="Bug Report", number=26)
Issue("This is a new issue")
>>> # Create an issue with title, body, assignee, and labels
>>> repo_name = "username/repository"
>>> title = "Bug Report"
>>> body = "I found a bug in the application."
>>> assignee = "john_doe"
>>> labels = ["enhancement", "help wanted"]
>>> issue = await create_issue(
repo_name=repo_name,
title=title,
body=body,
assignee=assigee,
labels=labels,
access_token=get_env("git_access_token")
)
>>> print(issue)
Issue(title="Bug Report", number=26)
Returns:
Issue: The created issue.
"""
return await GitRepository.create_issue(
repo_name=repo_name, title=title, body=body, assignee=assignee, labels=labels, access_token=access_token
)
return await GitRepository.create_issue(repo_name=repo_name, title=title, body=body, access_token=access_token)

11
metagpt/tools/libs/shell.py
View file

@ -49,12 +49,5 @@ async def shell_execute(
"""
cwd = str(cwd) if cwd else None
shell = True if isinstance(command, str) else False
process = subprocess.Popen(command, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, shell=shell)
try:
# Wait for the process to complete, with a timeout
stdout, stderr = process.communicate(timeout=timeout)
return stdout.decode("utf-8"), stderr.decode("utf-8"), process.returncode
except subprocess.TimeoutExpired:
process.kill() # Kill the process if it times out
stdout, stderr = process.communicate()
raise ValueError(f"{stdout.decode('utf-8')}\n{stderr.decode('utf-8')}")
result = subprocess.run(command, cwd=cwd, capture_output=True, text=True, env=env, timeout=timeout, shell=shell)
return result.stdout, result.stderr, result.returncode

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

@ -6,10 +6,8 @@ from pathlib import Path
from metagpt.const import ASSISTANT_ALIAS
from metagpt.logs import ToolLogItem, log_tool_output
from metagpt.tools.tool_registry import register_tool
@register_tool(tags=["software development", "import git repo"])
async def import_git_repo(url: str) -> Path:
"""
Imports a project from a Git website and formats it to MetaGPT project format to enable incremental appending requirements.

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

@ -2,8 +2,8 @@ import subprocess
import threading
from queue import Queue
from metagpt.logs import TOOL_LOG_END_MARKER, ToolLogItem, log_tool_output
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.report import END_MARKER_VALUE, TerminalReporter
@register_tool()
@ -31,6 +31,7 @@ class Terminal:
executable="/bin/bash",
)
self.stdout_queue = Queue()
self.observer = TerminalReporter()
self._check_state()
@ -65,7 +66,7 @@ class Terminal:
# Send the command
self.process.stdin.write(cmd + self.command_terminator)
self.process.stdin.write(
f'echo "{TOOL_LOG_END_MARKER.value}"' + self.command_terminator # write EOF
f'echo "{END_MARKER_VALUE}"' + self.command_terminator # write EOF
) # Unique marker to signal command end
self.process.stdin.flush()
if daemon:
@ -99,28 +100,26 @@ class Terminal:
return self.run_command(cmd, daemon=daemon)
def _read_and_process_output(self, cmd):
cmd_output = []
log_tool_output(
output=ToolLogItem(name="cmd", value=cmd + self.command_terminator), tool_name="Terminal"
) # log the command
with self.observer as observer:
cmd_output = []
observer.report(cmd + self.command_terminator, "cmd")
# report the command
# Read the output until the unique marker is found
while True:
line = self.process.stdout.readline()
ix = line.rfind(TOOL_LOG_END_MARKER.value)
if ix >= 0:
line = line[0:ix]
if line:
log_tool_output(
output=ToolLogItem(name="output", value=line), tool_name="Terminal"
) # log stdout in real-time
cmd_output.append(line)
log_tool_output(TOOL_LOG_END_MARKER)
break
# log stdout in real-time
log_tool_output(output=ToolLogItem(name="output", value=line), tool_name="Terminal")
cmd_output.append(line)
self.stdout_queue.put(line)
# Read the output until the unique marker is found
while True:
line = self.process.stdout.readline()
ix = line.rfind(END_MARKER_VALUE)
if ix >= 0:
line = line[0:ix]
if line:
observer.report(line, "output")
# report stdout in real-time
cmd_output.append(line)
break
# log stdout in real-time
observer.report(line, "output")
cmd_output.append(line)
self.stdout_queue.put(line)
return "".join(cmd_output)

4
metagpt/utils/common.py
View file

@ -37,7 +37,7 @@ from PIL import Image
from pydantic_core import to_jsonable_python
from tenacity import RetryCallState, RetryError, _utils
from metagpt.const import MESSAGE_ROUTE_TO_ALL
from metagpt.const import MARKDOWN_TITLE_PREFIX, MESSAGE_ROUTE_TO_ALL
from metagpt.logs import logger
from metagpt.utils.exceptions import handle_exception
@ -65,7 +65,7 @@ class OutputParser:
@classmethod
def parse_blocks(cls, text: str):
# 首先根据"##"将文本分割成不同的block
blocks = text.split("##")
blocks = text.split(MARKDOWN_TITLE_PREFIX)
# 创建一个字典，用于存储每个block的标题和内容
block_dict = {}

6
metagpt/utils/file_repository.py
View file

@ -198,8 +198,9 @@ class FileRepository:
:type dependencies: List[str], optional
"""
await self.save(filename=doc.filename, content=doc.content, dependencies=dependencies)
doc = await self.save(filename=doc.filename, content=doc.content, dependencies=dependencies)
logger.debug(f"File Saved: {str(doc.filename)}")
return doc
async def save_pdf(self, doc: Document, with_suffix: str = ".md", dependencies: List[str] = None):
"""Save a Document instance as a PDF file.
@ -216,8 +217,9 @@ class FileRepository:
"""
m = json.loads(doc.content)
filename = Path(doc.filename).with_suffix(with_suffix) if with_suffix is not None else Path(doc.filename)
await self.save(filename=str(filename), content=json_to_markdown(m), dependencies=dependencies)
doc = await self.save(filename=str(filename), content=json_to_markdown(m), dependencies=dependencies)
logger.debug(f"File Saved: {str(filename)}")
return doc
async def delete(self, filename: Path | str):
"""Delete a file from the file repository.

107
metagpt/utils/git_repository.py
View file

@ -13,11 +13,12 @@ import shutil
import uuid
from enum import Enum
from pathlib import Path
from subprocess import TimeoutExpired
from typing import Dict, List, Optional
from git.repo import Repo
from git.repo.fun import is_git_dir
from github import Auth, Github
from github import Auth, BadCredentialsException, Github
from github.GithubObject import NotSet
from github.Issue import Issue
from github.Label import Label
@ -25,6 +26,7 @@ from github.Milestone import Milestone
from github.NamedUser import NamedUser
from github.PullRequest import PullRequest
from gitignore_parser import parse_gitignore
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_random_exponential
from metagpt.logs import logger
@ -49,6 +51,12 @@ class RateLimitError(Exception):
super().__init__(self.message)
class GitBranch(BaseModel):
head: str
base: str
repo_name: str
class GitRepository:
"""A class representing a Git repository.
@ -177,6 +185,52 @@ class GitRepository:
return None
return Path(self._repository.working_dir)
@property
def current_branch(self) -> str:
"""
Returns the name of the current active branch.
Returns:
str: The name of the current active branch.
"""
return self._repository.active_branch.name
@property
def remote_url(self) -> str:
try:
return self._repository.remotes.origin.url
except AttributeError:
return ""
@property
def repo_name(self) -> str:
if self.remote_url:
# This assumes a standard HTTPS or SSH format URL
# HTTPS format example: https://github.com/username/repo_name.git
# SSH format example: git@github.com:username/repo_name.git
if self.remote_url.startswith("https://"):
return self.remote_url.split("/", maxsplit=3)[-1].replace(".git", "")
elif self.remote_url.startswith("git@"):
return self.remote_url.split(":")[-1].replace(".git", "")
return ""
def new_branch(self, branch_name: str) -> str:
"""
Creates a new branch with the given name.
Args:
branch_name (str): The name of the new branch to create.
Returns:
str: The name of the newly created branch.
If the provided branch_name is empty, returns the name of the current active branch.
"""
if not branch_name:
return self.current_branch
new_branch = self._repository.create_head(branch_name)
new_branch.checkout()
return new_branch.name
def archive(self, comments="Archive"):
"""Archive the current state of the Git repository.
@ -186,6 +240,57 @@ class GitRepository:
self.add_change(self.changed_files)
self.commit(comments)
async def push(
self, new_branch: str, comments="Archive", access_token: Optional[str] = None, auth: Optional[Auth] = None
) -> GitBranch:
"""
Pushes changes to the remote repository.
Args:
new_branch (str): The name of the new branch to be pushed.
comments (str, optional): Comments to be associated with the push. Defaults to "Archive".
access_token (str, optional): Access token for authentication. Defaults to None. Visit `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html`, `https://github.com/PyGithub/PyGithub/blob/main/doc/examples/Authentication.rst`.
auth (Auth, optional): Optional authentication object. Defaults to None.
Returns:
GitBranch: The pushed branch object.
Raises:
ValueError: If neither `auth` nor `access_token` is provided.
BadCredentialsException: If authentication fails due to bad credentials or timeout.
Note:
This function assumes that `self.current_branch`, `self.new_branch()`, `self.archive()`,
`ctx.config.proxy`, `ctx.config`, `self.remote_url`, `shell_execute()`, and `logger` are
defined and accessible within the scope of this function.
"""
if not auth and not access_token:
raise ValueError('`access_token` is invalid. Visit: "https://github.com/settings/tokens"')
from metagpt.context import Context
base = self.current_branch
head = base if not new_branch else self.new_branch(new_branch)
self.archive(comments)
ctx = Context()
env = ctx.new_environ()
proxy = ["-c", f"http.proxy={ctx.config.proxy}"] if ctx.config.proxy else []
token = access_token or auth.token
remote_url = f"https://{token}@" + self.remote_url.removeprefix("https://")
command = ["git"] + proxy + ["push", remote_url]
logger.info(" ".join(command).replace(token, "<TOKEN>"))
try:
stdout, stderr, return_code = await shell_execute(
command=command, cwd=str(self.workdir), env=env, timeout=15
)
except TimeoutExpired as e:
info = str(e).replace(token, "<TOKEN>")
raise BadCredentialsException(status=401, message=info)
info = f"{stdout}\n{stderr}\nexit: {return_code}\n"
info = info.replace(token, "<TOKEN>")
logger.info(info)
return GitBranch(base=base, head=head, repo_name=self.repo_name)
def new_file_repository(self, relative_path: Path | str = ".") -> FileRepository:
"""Create a new instance of FileRepository associated with this Git repository.

305
metagpt/utils/report.py Normal file
View file

@ -0,0 +1,305 @@
import asyncio
import os
import typing
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Literal, Optional, Union
from urllib.parse import unquote, urlparse, urlunparse
from uuid import UUID, uuid4
from aiohttp import ClientSession, UnixConnector
from playwright.async_api import Page as AsyncPage
from playwright.sync_api import Page as SyncPage
from pydantic import BaseModel, Field, PrivateAttr
from metagpt.const import METAGPT_REPORTER_DEFAULT_URL
from metagpt.logs import create_llm_stream_queue, get_llm_stream_queue
if typing.TYPE_CHECKING:
from metagpt.roles.role import Role
try:
import requests_unixsocket as requests
except ImportError:
import requests
from contextvars import ContextVar
CURRENT_ROLE: ContextVar["Role"] = ContextVar("role")
class BlockType(str, Enum):
"""Enumeration for different types of blocks."""
TERMINAL = "Terminal"
TASK = "Task"
BROWSER = "Browser"
BROWSER_RT = "Browser-RT"
EDITOR = "Editor"
GALLERY = "Gallery"
NOTEBOOK = "Notebook"
DOCS = "Docs"
END_MARKER_NAME = "end_marker"
END_MARKER_VALUE = "\x18\x19\x1B\x18"
class ResourceReporter(BaseModel):
"""Base class for resource reporting."""
block: BlockType = Field(description="The type of block that is reporting the resource")
uuid: UUID = Field(default_factory=uuid4, description="The unique identifier for the resource")
is_chunk: bool = Field(False, description="Indicates whether the report is a chunk of a stream")
enable_llm_stream: bool = Field(False, description="Indicates whether to connect to an LLM stream for reporting")
callback_url: str = Field(METAGPT_REPORTER_DEFAULT_URL, description="The URL to which the report should be sent")
_llm_task: Optional[asyncio.Task] = PrivateAttr(None)
def report(self, value: Any, name: str):
"""Synchronously report resource observation data.
Args:
value: The data to report.
name: The type name of the data.
"""
return self._report(value, name)
async def async_report(self, value: Any, name: str):
"""Asynchronously report resource observation data.
Args:
value: The data to report.
name: The type name of the data.
"""
return await self._async_report(value, name)
@classmethod
def set_report_fn(cls, fn: Callable):
"""Set the synchronous report function.
Args:
fn: A callable function used for synchronous reporting. For example:
>>> def _report(self, value: Any, name: str):
... print(value, name)
"""
cls._report = fn
@classmethod
def set_async_report_fn(cls, fn: Callable):
"""Set the asynchronous report function.
Args:
fn: A callable function used for asynchronous reporting. For example:
```python
>>> async def _report(self, value: Any, name: str):
... print(value, name)
```
"""
cls._async_report = fn
def _report(self, value: Any, name: str):
if not self.callback_url:
return
data = self._format_data(value, name)
resp = requests.post(self.callback_url, json=data)
resp.raise_for_status()
return resp.text
async def _async_report(self, value: Any, name: str):
if not self.callback_url:
return
data = self._format_data(value, name)
url = self.callback_url
_result = urlparse(url)
sessiion_kwargs = {}
if _result.scheme.endswith("+unix"):
parsed_list = list(_result)
parsed_list[0] = parsed_list[0][:-5]
parsed_list[1] = "fake.org"
url = urlunparse(parsed_list)
sessiion_kwargs["connector"] = UnixConnector(path=unquote(_result.netloc))
async with ClientSession(**sessiion_kwargs) as client:
async with client.post(url, json=data) as resp:
resp.raise_for_status()
return await resp.text()
def _format_data(self, value, name):
data = self.model_dump(mode="json", exclude=("callback_url", "llm_stream"))
data["value"] = str(value) if isinstance(value, Path) else value
data["name"] = name
role = CURRENT_ROLE.get(None)
if role:
role_name = role.name
else:
role_name = os.environ.get("METAGPT_ROLE")
data["role"] = role_name
return data
def __enter__(self):
"""Enter the synchronous streaming callback context."""
self.is_chunk = True
return self
def __exit__(self, *args, **kwargs):
"""Exit the synchronous streaming callback context."""
self.report(None, END_MARKER_NAME)
self.is_chunk = False
async def __aenter__(self):
"""Enter the asynchronous streaming callback context."""
self.is_chunk = True
if self.enable_llm_stream:
queue = create_llm_stream_queue()
self._llm_task = asyncio.create_task(self._llm_stream_report(queue))
return self
async def __aexit__(self, *args, **kwargs):
"""Exit the asynchronous streaming callback context."""
self.is_chunk = False
if self.enable_llm_stream:
self._llm_task.cancel()
self._llm_task = None
await self.async_report(None, END_MARKER_NAME)
async def _llm_stream_report(self, queue: asyncio.Queue):
while self.is_chunk:
await self.async_report(await queue.get(), "content")
async def wait_llm_stream_report(self):
"""Wait for the LLM stream report to complete."""
queue = get_llm_stream_queue()
while self._llm_task:
if queue.empty():
break
await asyncio.sleep(0.01)
class TerminalReporter(ResourceReporter):
"""Terminal output callback for streaming reporting of command and output.
The terminal has state, and an agent can open multiple terminals and input different commands into them.
To correctly display these states, each terminal should have its own unique ID, so in practice, each terminal
should instantiate its own TerminalReporter object.
"""
block: Literal[BlockType.TERMINAL] = BlockType.TERMINAL
def report(self, value: str, name: Literal["cmd", "output"]):
"""Report terminal command or output synchronously."""
return super().report(value, name)
async def async_report(self, value: str, name: Literal["cmd", "output"]):
"""Report terminal command or output asynchronously."""
return await super().async_report(value, name)
class BrowserReporter(ResourceReporter):
"""Browser output callback for streaming reporting of requested URL and page content.
The browser has state, so in practice, each browser should instantiate its own BrowserReporter object.
"""
block: Literal[BlockType.BROWSER] = BlockType.BROWSER
def report(self, value: Union[str, SyncPage], name: Literal["url", "page"]):
"""Report browser URL or page content synchronously."""
if name == "page":
value = {"page_url": value.url, "title": value.title(), "screenshot": str(value.screenshot())}
return super().report(value, name)
async def async_report(self, value: Union[str, AsyncPage], name: Literal["url", "page"]):
"""Report browser URL or page content asynchronously."""
if name == "page":
value = {"page_url": value.url, "title": await value.title(), "screenshot": str(await value.screenshot())}
return await super().async_report(value, name)
class ServerReporter(ResourceReporter):
"""Callback for server deployment reporting."""
block: Literal[BlockType.BROWSER_RT] = BlockType.BROWSER_RT
def report(self, value: str, name: Literal["local_url"] = "local_url"):
"""Report server deployment synchronously."""
return super().report(value, name)
async def async_report(self, value: str, name: Literal["local_url"] = "local_url"):
"""Report server deployment asynchronously."""
return await super().async_report(value, name)
class ObjectReporter(ResourceReporter):
"""Callback for reporting complete object resources."""
def report(self, value: dict, name: Literal["object"] = "object"):
"""Report object resource synchronously."""
return super().report(value, name)
async def async_report(self, value: dict, name: Literal["object"] = "object"):
"""Report object resource asynchronously."""
return await super().async_report(value, name)
class TaskReporter(ObjectReporter):
"""Reporter for object resources to Task Block."""
block: Literal[BlockType.TASK] = BlockType.TASK
class FileReporter(ResourceReporter):
"""File resource callback for reporting complete file paths.
There are two scenarios: if the file needs to be output in its entirety at once, use non-streaming callback;
if the file can be partially output for display first, use streaming callback.
"""
def report(self, value: Union[Path, dict, Any], name: Literal["path", "meta", "content"] = "path"):
"""Report file resource synchronously."""
return super().report(value, name)
async def async_report(self, value: Path, name: Literal["path", "meta", "content"] = "path"):
"""Report file resource asynchronously."""
return await super().async_report(value, name)
class NotebookReporter(FileReporter):
"""Equivalent to FileReporter(block=BlockType.NOTEBOOK)."""
block: Literal[BlockType.NOTEBOOK] = BlockType.NOTEBOOK
class DocsReporter(FileReporter):
"""Equivalent to FileReporter(block=BlockType.DOCS)."""
block: Literal[BlockType.DOCS] = BlockType.DOCS
class EditorReporter(FileReporter):
"""Equivalent to FileReporter(block=BlockType.Editor)."""
block: Literal[BlockType.EDITOR] = BlockType.EDITOR
class GalleryReporter(FileReporter):
"""Image resource callback for reporting complete file paths.
Since images need to be complete before display, each callback is a complete file path. However, the Gallery
needs to display the type of image and prompt, so if there is meta information, it should be reported in a
streaming manner.
"""
block: Literal[BlockType.GALLERY] = BlockType.GALLERY
def report(self, value: Union[dict, Path], name: Literal["meta", "path"] = "path"):
"""Report image resource synchronously."""
return super().report(value, name)
async def async_report(self, value: Union[dict, Path], name: Literal["meta", "path"] = "path"):
"""Report image resource asynchronously."""
return await super().async_report(value, name)

3
requirements.txt
View file

@ -70,4 +70,5 @@ qianfan==0.3.2
dashscope==1.14.1
rank-bm25==0.2.2 # for tool recommendation
gymnasium==0.29.1
pylint~=3.0.3
pylint~=3.0.3
pygithub~=2.3

16
tests/conftest.py
View file

@ -247,14 +247,16 @@ def search_engine_mocker(aiohttp_mocker, curl_cffi_mocker, httplib2_mocker, sear
@pytest.fixture
def http_server():
async def handler(request):
return aiohttp.web.Response(
text="""<!DOCTYPE html><html lang="en"><head><meta charset="UTF-8">
<title>MetaGPT</title></head><body><h1>MetaGPT</h1></body></html>""",
content_type="text/html",
)
async def start(handler=None):
if handler is None:
async def handler(request):
return aiohttp.web.Response(
text="""<!DOCTYPE html><html lang="en"><head><meta charset="UTF-8">
<title>MetaGPT</title></head><body><h1>MetaGPT</h1></body></html>""",
content_type="text/html",
)
async def start():
server = aiohttp.web.Server(handler)
runner = aiohttp.web.ServerRunner(server)
await runner.setup()

182
tests/metagpt/test_reporter.py Normal file
View file

@ -0,0 +1,182 @@
import ast
from contextlib import asynccontextmanager
import aiohttp.web
import pytest
from metagpt.logs import log_llm_stream
from metagpt.utils.report import (
END_MARKER_NAME,
BlockType,
BrowserReporter,
DocsReporter,
EditorReporter,
NotebookReporter,
ServerReporter,
TaskReporter,
TerminalReporter,
)
class MockFileLLM:
def __init__(self, data: str):
self.data = data
async def aask(self, *args, **kwargs) -> str:
for i in self.data.splitlines(keepends=True):
log_llm_stream(i)
log_llm_stream("\n")
return self.data
@asynccontextmanager
async def callback_server(http_server):
callback_data = []
async def handler(request):
callback_data.append(await request.json())
return aiohttp.web.json_response({})
server, url = await http_server(handler)
yield url, callback_data
await server.stop()
@pytest.mark.asyncio
async def test_terminal_report(http_server):
async with callback_server(http_server) as (url, callback_data):
async with TerminalReporter(callback_url=url) as reporter:
await reporter.async_report("ls -a", "cmd")
await reporter.async_report("main.py\n", "output")
await reporter.async_report("setup.py\n", "output")
assert all(BlockType.TERMINAL is BlockType(i["block"]) for i in callback_data)
assert all(i["uuid"] == callback_data[0]["uuid"] for i in callback_data[1:])
assert "".join(i["value"] for i in callback_data if i["name"] != END_MARKER_NAME) == "ls -amain.py\nsetup.py\n"
@pytest.mark.asyncio
async def test_browser_report(http_server):
img = b"\x89PNG\r\n\x1a\n\x00\x00"
web_url = "https://docs.deepwisdom.ai"
class AsyncPage:
async def screenshot(self):
return img
async with callback_server(http_server) as (url, callback_data):
async with BrowserReporter(callback_url=url) as reporter:
await reporter.async_report(web_url, "url")
await reporter.async_report(AsyncPage(), "page")
assert all(BlockType.BROWSER is BlockType(i["block"]) for i in callback_data)
assert all(i["uuid"] == callback_data[0]["uuid"] for i in callback_data[1:])
assert len(callback_data) == 3
assert callback_data[-1]["name"] == END_MARKER_NAME
assert callback_data[0]["name"] == "url"
assert callback_data[0]["value"] == web_url
assert callback_data[1]["name"] == "page"
assert ast.literal_eval(callback_data[1]["value"]) == img
@pytest.mark.asyncio
async def test_server_reporter(http_server):
local_url = "http://127.0.0.1:8080/index.html"
async with callback_server(http_server) as (url, callback_data):
reporter = ServerReporter(callback_url=url)
await reporter.async_report(local_url)
assert all(BlockType.BROWSER_RT is BlockType(i["block"]) for i in callback_data)
assert len(callback_data) == 1
assert callback_data[0]["name"] == "local_url"
assert callback_data[0]["value"] == local_url
assert not callback_data[0]["is_chunk"]
@pytest.mark.asyncio
async def test_task_reporter(http_server):
task = {"current_task_id": "", "tasks": []}
async with callback_server(http_server) as (url, callback_data):
reporter = TaskReporter(callback_url=url)
await reporter.async_report(task)
assert all(BlockType.TASK is BlockType(i["block"]) for i in callback_data)
assert len(callback_data) == 1
assert callback_data[0]["name"] == "object"
assert callback_data[0]["value"] == task
@pytest.mark.asyncio
async def test_notebook_reporter(http_server):
code = {
"cell_type": "code",
"execution_count": None,
"id": "e1841c44",
"metadata": {},
"outputs": [],
"source": ["\n", "import time\n", "print('will sleep 1s.')\n", "time.sleep(1)\n", "print('end.')\n", ""],
}
output1 = {"name": "stdout", "output_type": "stream", "text": ["will sleep 1s.\n"]}
output2 = {"name": "stdout", "output_type": "stream", "text": ["will sleep 1s.\n"]}
code_path = "/data/main.ipynb"
async with callback_server(http_server) as (url, callback_data):
async with NotebookReporter(callback_url=url) as reporter:
await reporter.async_report(code, "content")
await reporter.async_report(output1, "content")
await reporter.async_report(output2, "content")
await reporter.async_report(code_path, "path")
assert all(BlockType.NOTEBOOK is BlockType(i["block"]) for i in callback_data)
assert len(callback_data) == 5
assert callback_data[-1]["name"] == END_MARKER_NAME
assert callback_data[-2]["name"] == "path"
assert callback_data[-2]["value"] == code_path
assert all(i["uuid"] == callback_data[0]["uuid"] for i in callback_data[1:])
assert [i["value"] for i in callback_data if i["name"] == "content"] == [code, output1, output2]
@pytest.mark.asyncio
@pytest.mark.parametrize(
("data", "file_path", "meta", "block", "report_cls"),
(
(
"## Language\n\nen_us\n\n## Programming Language\n\nPython\n\n## Original Requirements\n\nCreate a 2048 gam...",
"/data/prd.md",
{"type": "write_prd"},
BlockType.DOCS,
DocsReporter,
),
(
"#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n\nprint('Hello World')\n",
"/data/main.py",
{"type": "write_code"},
BlockType.EDITOR,
EditorReporter,
),
),
ids=["test_docs_reporter", "test_editor_reporter"],
)
async def test_llm_stream_reporter(data, file_path, meta, block, report_cls, http_server):
async with callback_server(http_server) as (url, callback_data):
async with report_cls(callback_url=url, enable_llm_stream=True) as reporter:
await reporter.async_report(meta, "meta")
await MockFileLLM(data).aask("")
await reporter.wait_llm_stream_report()
await reporter.async_report(file_path, "path")
assert callback_data
assert all(block is BlockType(i["block"]) for i in callback_data)
assert all(i["uuid"] == callback_data[0]["uuid"] for i in callback_data[1:])
chunks, names = [], set()
for i in callback_data:
name = i["name"]
names.add(name)
if name == "meta":
assert i["value"] == meta
elif name == "path":
assert i["value"] == file_path
elif name == END_MARKER_NAME:
pass
elif name == "content":
chunks.append(i["value"])
else:
raise ValueError
assert "".join(chunks[:-1]) == data
assert names == {"meta", "path", "content", END_MARKER_NAME}

61
tests/metagpt/tools/libs/test_git.py
View file

@ -3,12 +3,17 @@
from __future__ import annotations
import os
import uuid
import pytest
from github import Auth, Github
from pydantic import BaseModel
from metagpt.context import Context
from metagpt.roles.di.data_interpreter import DataInterpreter
from metagpt.schema import UserMessage
from metagpt.tools.libs.git import git_checkout, git_clone
from metagpt.utils.common import awrite
from metagpt.utils.git_repository import GitRepository
@ -17,7 +22,7 @@ class SWEBenchItem(BaseModel):
repo: str
def get_env(key):
async def get_env(key: str, app_name: str = ""):
return os.environ.get(key)
@ -37,8 +42,9 @@ async def test_git(url: str, commit_id: str):
@pytest.mark.skip
def test_login():
auth = Auth.Login(get_env("GITHUB_USER"), get_env("GITHUB_PWD"))
@pytest.mark.asyncio
async def test_login():
auth = Auth.Login(await get_env("GITHUB_USER"), await get_env("GITHUB_PWD"))
g = Github(auth=auth)
repo = g.get_repo("geekan/MetaGPT")
topics = repo.get_topics()
@ -55,7 +61,7 @@ async def test_new_issue():
repo_name="iorisa/MetaGPT",
title="This is a new issue",
body="This is the issue body",
access_token=get_env("GITHUB_PERSONAL_ACCESS_TOKEN"),
access_token=await get_env(key="access_token", app_name="github"),
)
print(issue)
assert issue.number
@ -74,20 +80,21 @@ async def test_new_pr():
>>> - [x] Send 'POST' request with/without body
"""
pr = await GitRepository.create_pull(
repo_name="iorisa/MetaGPT",
base_repo_name="iorisa/MetaGPT",
base="send18",
head="fixbug/gbk",
title="Test pr",
body=body,
access_token=get_env("GITHUB_PERSONAL_ACCESS_TOKEN"),
access_token=await get_env(key="access_token", app_name="github"),
)
print(pr)
assert pr
@pytest.mark.skip
def test_auth():
access_token = get_env("GITHUB_PERSONAL_ACCESS_TOKEN")
@pytest.mark.asyncio
async def test_auth():
access_token = await get_env(key="access_token", app_name="github")
auth = Auth.Token(access_token)
g = Github(auth=auth)
u = g.get_user()
@ -98,5 +105,43 @@ def test_auth():
pass
@pytest.mark.skip
@pytest.mark.asyncio
async def test_github(context):
repo = await GitRepository.clone_from(url="https://github.com/iorisa/snake-game.git")
content = uuid.uuid4().hex
await awrite(filename=repo.workdir / "README.md", data=content)
branch = await repo.push(
new_branch=f"feature/{content[0:8]}", access_token=await get_env(key="access_token", app_name="github")
)
pr = await GitRepository.create_pull(
base=branch.base,
head=branch.head,
base_repo_name=branch.repo_name,
title=f"new pull {content[0:8]}",
access_token=await get_env(key="access_token", app_name="github"),
)
assert pr
@pytest.mark.asyncio
@pytest.mark.parametrize(
"content",
[
# "create a new issue to github repo 'iorisa/snake-game' :'The snake did not grow longer after eating'",
"Resolve the issue #1 'Snake not growing longer after eating' in the GitHub repository https://github.com/iorisa/snake-game.git', and create a new pull request about the issue"
],
)
async def test_git_create_issue(content: str):
context = Context()
di = DataInterpreter(context=context, tools=["<all>"])
prerequisite = "from metagpt.tools.libs import get_env"
await di.execute_code.run(code=prerequisite, language="python")
di.put_message(UserMessage(content=content))
while not di.is_idle:
await di.run()
if __name__ == "__main__":
pytest.main([__file__, "-s"])

50
tests/metagpt/tools/libs/test_software_development.py
View file

@ -1,59 +1,17 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from typing import Dict
import pytest
from metagpt.tools.libs import (
fix_bug,
git_archive,
run_qa_test,
write_codes,
write_design,
write_prd,
write_project_plan,
)
from metagpt.tools.libs.software_development import import_git_repo
@pytest.mark.asyncio
async def test_software_team():
path = await write_prd("snake game")
assert path
path = await write_design(path)
assert path
path = await write_project_plan(path)
assert path
path = await write_codes(path)
assert path
path = await run_qa_test(path)
assert path
issue = """
pygame 2.0.1 (SDL 2.0.14, Python 3.9.17)
Hello from the pygame community. https://www.pygame.org/contribute.html
Traceback (most recent call last):
File "/Users/ix/github/bak/MetaGPT/workspace/snake_game/snake_game/main.py", line 10, in <module>
main()
File "/Users/ix/github/bak/MetaGPT/workspace/snake_game/snake_game/main.py", line 7, in main
game.start_game()
File "/Users/ix/github/bak/MetaGPT/workspace/snake_game/snake_game/game.py", line 81, in start_game
x
NameError: name 'x' is not defined
"""
path = await fix_bug(path, issue)
assert path
new_path = await write_prd("snake game with moving enemy", path)
assert new_path == path
git_log = await git_archive(new_path)
assert git_log
async def get_env_description() -> Dict[str, str]:
return {'await get_env(key="access_token", app_name="github")': "get the access token for github authentication."}
@pytest.mark.skip
@pytest.mark.asyncio
async def test_import_repo():
url = "https://github.com/spec-first/connexion.git"