Merge branch 'di_mgx' into 'mgx_ops'

allow di to use terminal tool, add tool log fn, expand truncate len

See merge request pub/MetaGPT!9

examples/di/use_github_repo.py

@@ -0,0 +1,18 @@
+import asyncio
+
+from metagpt.roles.di.data_interpreter import DataInterpreter
+
+USE_GOT_REPO_REQ = """
+This is a link to the GOT github repo: https://github.com/spcl/graph-of-thoughts.git.
+Clone it, read the README to understand the usage, install it, and finally run the quick start example.
+**Note the config for LLM is at `config/config_got.json`, use this path directly.** Don't write all codes in one response, each time, just write code for one step.
+"""
+
+
+async def main():
+    di = DataInterpreter(tools=["Terminal"])
+    await di.run(USE_GOT_REPO_REQ)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

metagpt/actions/di/execute_nb_code.py

@@ -106,7 +106,7 @@ class ExecuteNbCode(Action):
         else:
             cell["outputs"].append(new_output(output_type="stream", name="stdout", text=str(output)))
 
-    def parse_outputs(self, outputs: list[str], keep_len: int = 2000) -> Tuple[bool, str]:
+    def parse_outputs(self, outputs: list[str], keep_len: int = 5000) -> Tuple[bool, str]:
         """Parses the outputs received from notebook execution."""
         assert isinstance(outputs, list)
         parsed_output, is_success = [], True

metagpt/logs.py

@@ -6,15 +6,29 @@
 @File    : logs.py
 """
 
+from __future__ import annotations
+
 import sys
 from datetime import datetime
 from functools import partial
 
 from loguru import logger as _logger
+from pydantic import BaseModel, Field
 
 from metagpt.const import METAGPT_ROOT
 
 
+class ToolLogItem(BaseModel):
+    type_: str = Field(alias="type", default="str", description="Data type of `value` field.")
+    name: str
+    value: str
+
+
+TOOL_LOG_END_MARKER = ToolLogItem(
+    type="str", name="end_marker", value="#END#"
+)  # A special log item to suggest the end of a stream log
+
+
 def define_log_level(print_level="INFO", logfile_level="DEBUG", name: str = None):
     """Adjust the log level to above level"""
     current_date = datetime.now()
@@ -34,9 +48,22 @@ def log_llm_stream(msg):
     _llm_stream_log(msg)
 
 
+def log_tool_output(output: ToolLogItem | list[ToolLogItem], tool_name: str = ""):
+    """interface for logging tool output, can be set to log tool output in different ways to different places with set_tool_output_logfunc"""
+    _tool_output_log(output=output, tool_name=tool_name)
+
+
 def set_llm_stream_logfunc(func):
     global _llm_stream_log
     _llm_stream_log = func
 
 
+def set_tool_output_logfunc(func):
+    global _tool_output_log
+    _tool_output_log = func
+
+
 _llm_stream_log = partial(print, end="")
+
+
+_tool_output_log = lambda output, tool_name: print(output)

metagpt/prompts/di/write_analysis_code.py

@@ -1,4 +1,8 @@
-INTERPRETER_SYSTEM_MSG = """As a data scientist, you need to help user to achieve their goal step by step in a continuous Jupyter notebook. Since it is a notebook environment, don't use asyncio.run. Instead, use await if you need to call an async function."""
+INTERPRETER_SYSTEM_MSG = """
+As a data scientist, you need to help user to achieve their goal step by step in a continuous Jupyter notebook.
+Since it is a notebook environment, don't use asyncio.run. Instead, use await if you need to call an async function.
+If you want to use shell command such as git clone, pip install packages, navigate folders, read file, etc., use Terminal tool if available before trying ! in notebook block.
+"""
 
 STRUCTUAL_PROMPT = """
 # User Requirement

metagpt/schema.py

@@ -45,7 +45,7 @@ from metagpt.const import (
     SYSTEM_DESIGN_FILE_REPO,
     TASK_FILE_REPO,
 )
-from metagpt.logs import logger
+from metagpt.logs import ToolLogItem, log_tool_output, logger
 from metagpt.repo_parser import DotClassInfo
 from metagpt.utils.common import any_to_str, any_to_str_set, import_class
 from metagpt.utils.exceptions import handle_exception
@@ -433,6 +433,13 @@ class Plan(BaseModel):
             final_tasks = self.tasks[:prefix_length] + new_tasks[prefix_length:]
             self.tasks = final_tasks
 
+        log_tool_output(
+            ToolLogItem(
+                name="output", value="\n\n".join([f"Task {task.task_id}: {task.instruction}" for task in self.tasks])
+            ),
+            tool_name="Plan",
+        )
+
         # Update current_task_id to the first unfinished task in the merged list
         self._update_current_task()
 

metagpt/tools/libs/__init__.py

@@ -11,6 +11,7 @@ from metagpt.tools.libs import (
     gpt_v_generator,
     web_scraping,
     email_login,
+    terminal,
 )
 
 _ = (
@@ -20,4 +21,5 @@ _ = (
     gpt_v_generator,
     web_scraping,
     email_login,
+    terminal,
 )  # Avoid pre-commit error

metagpt/tools/libs/terminal.py

@@ -0,0 +1,66 @@
+import subprocess
+
+from metagpt.logs import TOOL_LOG_END_MARKER, ToolLogItem, log_tool_output
+from metagpt.tools.tool_registry import register_tool
+
+
+@register_tool()
+class Terminal:
+    """A tool for running terminal commands. Don't initialize a new instance of this class if one already exists."""
+
+    def __init__(self):
+        self.shell_command = ["bash"]  # FIXME: should consider windows support later
+        self.command_terminator = "\n"
+
+        # Start a persistent shell process
+        self.process = subprocess.Popen(
+            self.shell_command,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            bufsize=1,  # Line buffered
+        )
+
+    def run_command(self, cmd: str) -> str:
+        """
+        Run a command in the terminal and return the output.
+        When the command is being executed, stream the output to the terminal.
+        Maintains state across commands, such as current directory.
+
+        Args:
+            cmd (str): The command to run in the terminal.
+
+        Returns:
+            str: The output of the terminal command.
+        """
+        cmd_output = []
+
+        # 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
+        )  # Unique marker to signal command end
+        self.process.stdin.flush()
+        log_tool_output(
+            output=ToolLogItem(name="cmd", value=cmd + self.command_terminator), tool_name="Terminal"
+        )  # log the command
+
+        # Read the output until the unique marker is found
+        while True:
+            line = self.process.stdout.readline()
+            if line.strip() == TOOL_LOG_END_MARKER.value:
+                log_tool_output(TOOL_LOG_END_MARKER)
+                break
+            log_tool_output(
+                output=ToolLogItem(name="output", value=line), tool_name="Terminal"
+            )  # log stdout in real-time
+            cmd_output.append(line)
+
+        return "".join(cmd_output)
+
+    def close(self):
+        """Close the persistent shell process."""
+        self.process.stdin.close()
+        self.process.terminate()
+        self.process.wait()

tests/metagpt/tools/libs/test_terminal.py

@@ -0,0 +1,15 @@
+from metagpt.const import DATA_PATH, METAGPT_ROOT
+from metagpt.tools.libs.terminal import Terminal
+
+
+def test_terminal():
+    terminal = Terminal()
+
+    terminal.run_command(f"cd {METAGPT_ROOT}")
+    output = terminal.run_command("pwd")
+    assert output.strip() == str(METAGPT_ROOT)
+
+    # pwd now should be METAGPT_ROOT, cd data should land in DATA_PATH
+    terminal.run_command("cd data")
+    output = terminal.run_command("pwd")
+    assert output.strip() == str(DATA_PATH)