add support for langchain tools as functions

thehunmonkgroup · thehunmonkgroup · commit 8168a0a44935 · 2023-06-25T20:09:45.000-04:00
diff --git a/README.md b/README.md
@@ -606,6 +606,26 @@ A preset can be edited by using the `/preset-edit` command:
 Note the special `return_on_function_call` and `return_on_function_response` metadata attributes, which can be used to
 control the return value, useful when using the `ApiBackend`module, or via [workflows](#workflows)
 
+#### Support for Langchain tools
+
+[Langchain](https://docs.langchain.com) has many useful [tools](https://python.langchain.com/docs/modules/agents/tools/) that
+can be used in function calls.
+
+To use a Langchain tool as function:
+
+1. Find the name of the tool class, e.g. `MoveFileTool` or `ShellTool`.
+2. Prefix that class name with `Langchain-`
+3. Add it to the `functions` list for the preset:
+   ```yaml
+   metadata:
+     # Usual preset metadata.
+   model_customizations:
+     # Other attributes.
+     model_kwargs:
+       functions:
+         - Langchain-ShellTool
+   ```
+
 
 ### Flask API (experimental)
 
diff --git a/lwe/backends/api/backend.py b/lwe/backends/api/backend.py
@@ -164,8 +164,12 @@ def init_function_cache(self):
                     self.function_cache.append(function)
 
     def function_cache_add(self, function_name):
-        if function_name not in self.function_manager.functions:
-            return False
+        if self.function_manager.is_langchain_tool(function_name):
+            if not self.function_manager.get_langchain_tool(function_name):
+                return False
+        else:
+            if function_name not in self.function_manager.functions:
+                return False
         if function_name not in self.function_cache:
             self.function_cache.append(function_name)
         return True
diff --git a/lwe/core/function_manager.py b/lwe/core/function_manager.py
@@ -4,10 +4,14 @@
 
 from pathlib import Path
 
+import langchain.tools
+
 from lwe.core.config import Config
 from lwe.core.logger import Logger
 import lwe.core.util as util
 
+LANGCHAIN_TOOL_PREFIX = "Langchain-"
+
 class FunctionManager():
     """
     Manage functions.
@@ -58,6 +62,45 @@ def load_function(self, function_name):
             return True, function_filepath, message
         return False, None, f"Function {function_name} not found"
 
+    def is_langchain_tool(self, function_name):
+        self.log.debug(f"Checking for Langchain tool: {function_name}")
+        return function_name.lower().startswith(LANGCHAIN_TOOL_PREFIX.lower())
+
+    def get_langchain_tool(self, function_name):
+        self.log.debug(f"Loading Langchain tool: {function_name}")
+        tool_name = util.remove_prefix(function_name, LANGCHAIN_TOOL_PREFIX)
+        tool = getattr(langchain.tools, tool_name)
+        try:
+            tool_instance = tool()
+            return tool_instance
+        except Exception as e:
+            self.log.warning(f"Could not load Langchaine tool: {function_name}: {str(e)}")
+            return None
+
+    def get_langchain_tool_spec(self, function_name):
+        self.log.debug(f"Loading tool spec for Langchain tool: {function_name}")
+        tool_instance = self.get_langchain_tool(function_name)
+        if not tool_instance:
+            raise RuntimeError(f"Langchain tool {function_name} not found")
+        spec = langchain.tools.format_tool_to_openai_function(tool_instance)
+        spec['name'] = function_name
+        return spec
+
+    def run_langchain_tool(self, function_name, input_data):
+        self.log.debug(f"Running langchaing tool: {function_name} with data: {input_data}")
+        tool_instance = self.get_langchain_tool(function_name)
+        if not tool_instance:
+            raise RuntimeError(f"Langchain tool {function_name} not found")
+        try:
+            result = tool_instance.run(input_data)
+        except Exception as e:
+            message = f"Error: Exception occurred while running langchain tool {function_name}: {str(e)}"
+            self.log.error(message)
+            return False, None, message
+        message = f"Langchain tool {function_name} executed successfully, output data: {result}"
+        self.log.info(message)
+        return True, result, message
+
     def load_functions(self):
         self.log.debug("Loading functions from dirs: %s" % ", ".join(self.all_function_dirs))
         self.functions = {}
@@ -99,6 +142,8 @@ def setup_function_instance(self, function_name, function_path):
 
     def get_function_config(self, function_name):
         self.log.debug(f"Getting config for function: {function_name}")
+        if self.is_langchain_tool(function_name):
+            return self.get_langchain_tool_spec(function_name)
         try:
             _success, function_path, user_message = self.load_function(function_name)
             function_instance = self.setup_function_instance(function_name, function_path)
@@ -111,6 +156,8 @@ def get_function_config(self, function_name):
     def run_function(self, function_name, input_data):
         if isinstance(input_data, str):
             input_data = json.loads(input_data)
+        if self.is_langchain_tool(function_name):
+            return self.run_langchain_tool(function_name, input_data)
         self.log.debug(f"Running function: {function_name} with data: {input_data}")
         success, function_path, user_message = self.load_function(function_name)
         if not success:
