Source code for langchain_community.tools.shell.tool

import logging
import platform
import warnings
from typing import Any, List, Optional, Type, Union

from langchain_core.callbacks import (
    CallbackManagerForToolRun,
)
from langchain_core.pydantic_v1 import BaseModel, Field, root_validator
from langchain_core.tools import BaseTool

logger = logging.getLogger(__name__)


class ShellInput(BaseModel):
    """Bash Shell工具的命令。"""

    commands: Union[str, List[str]] = Field(
        ...,
        description="需要运行的shell命令列表。 Deserialized using json.loads",
    )
    """需要运行的shell命令列表。"""

    @root_validator
    def _validate_commands(cls, values: dict) -> dict:
        """验证命令。"""
        # TODO: Add real validators
        commands = values.get("commands")
        if not isinstance(commands, list):
            values["commands"] = [commands]
        # Warn that the bash tool is not safe
        warnings.warn(
            "The shell tool has no safeguards by default. Use at your own risk."
        )
        return values


def _get_default_bash_process() -> Any:
    """获取默认的bash进程。"""
    try:
        from langchain_experimental.llm_bash.bash import BashProcess
    except ImportError:
        raise ImportError(
            "BashProcess has been moved to langchain experimental."
            "To use this tool, install langchain-experimental "
            "with `pip install langchain-experimental`."
        )
    return BashProcess(return_err_output=True)


def _get_platform() -> str:
    """获取平台。"""
    system = platform.system()
    if system == "Darwin":
        return "MacOS"
    return system


class ShellTool(BaseTool):
    """用于运行shell命令的工具。"""

    process: Any = Field(default_factory=_get_default_bash_process)
    """运行命令的Bash进程。"""

    name: str = "terminal"
    """工具的名称。"""

    description: str = f"Run shell commands on this {_get_platform()} machine."
    """工具的描述。"""

    args_schema: Type[BaseModel] = ShellInput
    """输入参数的模式。"""

    ask_human_input: bool = False
    """如果为True，则在bash shell中执行语言模型生成的命令之前，会提示用户确认（y/n）。"""

    def _run(
        self,
        commands: Union[str, List[str]],
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        """运行命令并返回最终输出。"""

        print(f"Executing command:\n {commands}")  # noqa: T201

        try:
            if self.ask_human_input:
                user_input = input("Proceed with command execution? (y/n): ").lower()
                if user_input == "y":
                    return self.process.run(commands)
                else:
                    logger.info("Invalid input. User aborted command execution.")
                    return None  # type: ignore[return-value]
            else:
                return self.process.run(commands)

        except Exception as e:
            logger.error(f"Error during command execution: {e}")
            return None  # type: ignore[return-value]