Compare commits

...

2 Commits

Author SHA1 Message Date
Harrison Chase
450be8d15f cr 2023-02-09 23:50:27 -08:00
dkowitz
46354d4f64 Added Tools for PowerShell and WSL (#947)
I added utilities for PowerShell and WSL. Both tools are based on the
bash utility. I also updated 'load_tools.py' with the new tools.
2023-02-09 23:49:59 -08:00
3 changed files with 81 additions and 9 deletions

View File

@@ -13,7 +13,9 @@ from langchain.requests import RequestsWrapper
from langchain.serpapi import SerpAPIWrapper
from langchain.utilities.bash import BashProcess
from langchain.utilities.google_search import GoogleSearchAPIWrapper
from langchain.utilities.powershell import PowerShellProcess
from langchain.utilities.wolfram_alpha import WolframAlphaAPIWrapper
from langchain.utilities.wsl import WSLProcess
def _get_python_repl() -> Tool:
@@ -40,10 +42,28 @@ def _get_terminal() -> Tool:
)
def _get_wsl() -> Tool:
return Tool(
"WSL",
WSLProcess().run,
"Executes bash commands in a linux system running on Windows Subsystem for Linux (WSL). Input should be valid bash commands, and the output will be any output from running that command.",
)
def _get_powershell() -> Tool:
return Tool(
"PowerShell",
PowerShellProcess().run,
"Executes commands in PowerShell. Input should be valid commands, and the output will be any output from running that command.",
)
_BASE_TOOLS = {
"python_repl": _get_python_repl,
"requests": _get_requests,
"terminal": _get_terminal,
"wsl": _get_wsl,
"powershell": _get_powershell,
}
@@ -65,10 +85,9 @@ def _get_pal_colored_objects(llm: BaseLLM) -> Tool:
def _get_llm_math(llm: BaseLLM) -> Tool:
return Tool(
name="Calculator",
description="Useful for when you need to answer questions about math.",
func=LLMMathChain(llm=llm, callback_manager=llm.callback_manager).run,
coroutine=LLMMathChain(llm=llm, callback_manager=llm.callback_manager).arun,
"Calculator",
LLMMathChain(llm=llm).run,
"Useful for when you need to answer questions about math.",
)
@@ -133,10 +152,9 @@ def _get_google_search(**kwargs: Any) -> Tool:
def _get_serpapi(**kwargs: Any) -> Tool:
return Tool(
name="Search",
description="A search engine. Useful for when you need to answer questions about current events. Input should be a search query.",
func=SerpAPIWrapper(**kwargs).run,
coroutine=SerpAPIWrapper(**kwargs).arun,
"Search",
SerpAPIWrapper(**kwargs).run,
"A search engine. Useful for when you need to answer questions about current events. Input should be a search query.",
)
@@ -147,7 +165,7 @@ _EXTRA_LLM_TOOLS = {
_EXTRA_OPTIONAL_TOOLS = {
"wolfram-alpha": (_get_wolfram_alpha, ["wolfram_alpha_appid"]),
"google-search": (_get_google_search, ["google_api_key", "google_cse_id"]),
"serpapi": (_get_serpapi, ["serpapi_api_key", "aiosession"]),
"serpapi": (_get_serpapi, ["serpapi_api_key"]),
}

View File

@@ -0,0 +1,28 @@
"""Wrapper around subprocess to run commands."""
import subprocess
from typing import List, Union
class PowerShellProcess:
"""Executes PowerShell commands and returns the output."""
def __init__(self, strip_newlines: bool = False):
"""Initialize with stripping newlines."""
self.strip_newlines = strip_newlines
def run(self, commands: Union[str, List[str]]) -> str:
"""Run commands and return final output."""
if isinstance(commands, str):
commands = [commands]
commands = ";".join(commands)
try:
output = subprocess.check_output(
["powershell.exe", "-Command", commands]
).decode()
except subprocess.CalledProcessError as error:
return str(error)
if self.strip_newlines:
output = output.strip()
return output

View File

@@ -0,0 +1,26 @@
"""Wrapper around subprocess to run commands."""
import subprocess
from typing import List, Union
class WSLProcess:
"""Executes bash commands on Windows Subsystem for Linux (WSL) and returns the output."""
def __init__(self, strip_newlines: bool = False):
"""Initialize with stripping newlines."""
self.strip_newlines = strip_newlines
def run(self, commands: Union[str, List[str]]) -> str:
"""Run commands and return final output."""
if isinstance(commands, str):
commands = [commands]
commands = ";".join(commands)
commands = "wsl.exe " + commands
try:
output = subprocess.check_output(commands, shell=True).decode()
except subprocess.CalledProcessError as error:
return str(error)
if self.strip_newlines:
output = output.strip()
return output