mirror of
https://github.com/hwchase17/langchain.git
synced 2025-09-08 14:31:55 +00:00
community[minor]: Connery Tool and Toolkit (#14506)
## Summary This PR implements the "Connery Action Tool" and "Connery Toolkit". Using them, you can integrate Connery actions into your LangChain agents and chains. Connery is an open-source plugin infrastructure for AI. With Connery, you can easily create a custom plugin with a set of actions and seamlessly integrate them into your LangChain agents and chains. Connery will handle the rest: runtime, authorization, secret management, access management, audit logs, and other vital features. Additionally, Connery and our community offer a wide range of ready-to-use open-source plugins for your convenience. Learn more about Connery: - GitHub: https://github.com/connery-io/connery-platform - Documentation: https://docs.connery.io - Twitter: https://twitter.com/connery_io ## TODOs - [x] API wrapper - [x] Integration tests - [x] Connery Action Tool - [x] Docs - [x] Example - [x] Integration tests - [x] Connery Toolkit - [x] Docs - [x] Example - [x] Formatting (`make format`) - [x] Linting (`make lint`) - [x] Testing (`make test`)
This commit is contained in:
committed by
GitHub
parent
8457c31c04
commit
32c5be8b73
163
libs/community/langchain_community/tools/connery/tool.py
Normal file
163
libs/community/langchain_community/tools/connery/tool.py
Normal file
@@ -0,0 +1,163 @@
|
||||
import asyncio
|
||||
from functools import partial
|
||||
from typing import Any, Dict, List, Optional, Type
|
||||
|
||||
from langchain_core.callbacks.manager import (
|
||||
AsyncCallbackManagerForToolRun,
|
||||
CallbackManagerForToolRun,
|
||||
)
|
||||
from langchain_core.pydantic_v1 import BaseModel, Field, create_model, root_validator
|
||||
from langchain_core.tools import BaseTool
|
||||
|
||||
from langchain_community.tools.connery.models import Action, Parameter
|
||||
|
||||
|
||||
class ConneryAction(BaseTool):
|
||||
"""
|
||||
A LangChain Tool wrapping a Connery Action.
|
||||
"""
|
||||
|
||||
name: str
|
||||
description: str
|
||||
args_schema: Type[BaseModel]
|
||||
|
||||
action: Action
|
||||
connery_service: Any
|
||||
|
||||
def _run(
|
||||
self,
|
||||
run_manager: Optional[CallbackManagerForToolRun] = None,
|
||||
**kwargs: Dict[str, str],
|
||||
) -> Dict[str, str]:
|
||||
"""
|
||||
Runs the Connery Action with the provided input.
|
||||
Parameters:
|
||||
kwargs (Dict[str, str]): The input dictionary expected by the action.
|
||||
Returns:
|
||||
Dict[str, str]: The output of the action.
|
||||
"""
|
||||
|
||||
return self.connery_service.run_action(self.action.id, kwargs)
|
||||
|
||||
async def _arun(
|
||||
self,
|
||||
run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
|
||||
**kwargs: Dict[str, str],
|
||||
) -> Dict[str, str]:
|
||||
"""
|
||||
Runs the Connery Action asynchronously with the provided input.
|
||||
Parameters:
|
||||
kwargs (Dict[str, str]): The input dictionary expected by the action.
|
||||
Returns:
|
||||
Dict[str, str]: The output of the action.
|
||||
"""
|
||||
|
||||
func = partial(self._run, **kwargs)
|
||||
return await asyncio.get_event_loop().run_in_executor(None, func)
|
||||
|
||||
def get_schema_json(self) -> str:
|
||||
"""
|
||||
Returns the JSON representation of the Connery Action Tool schema.
|
||||
This is useful for debugging.
|
||||
Returns:
|
||||
str: The JSON representation of the Connery Action Tool schema.
|
||||
"""
|
||||
|
||||
return self.args_schema.schema_json(indent=2)
|
||||
|
||||
@root_validator()
|
||||
def validate_attributes(cls, values: dict) -> dict:
|
||||
"""
|
||||
Validate the attributes of the ConneryAction class.
|
||||
Parameters:
|
||||
values (dict): The arguments to validate.
|
||||
Returns:
|
||||
dict: The validated arguments.
|
||||
"""
|
||||
|
||||
# Import ConneryService here and check if it is an instance
|
||||
# of ConneryService to avoid circular imports
|
||||
from .service import ConneryService
|
||||
|
||||
if not isinstance(values.get("connery_service"), ConneryService):
|
||||
raise ValueError(
|
||||
"The attribute 'connery_service' must be an instance of ConneryService."
|
||||
)
|
||||
|
||||
if not values.get("name"):
|
||||
raise ValueError("The attribute 'name' must be set.")
|
||||
if not values.get("description"):
|
||||
raise ValueError("The attribute 'description' must be set.")
|
||||
if not values.get("args_schema"):
|
||||
raise ValueError("The attribute 'args_schema' must be set.")
|
||||
if not values.get("action"):
|
||||
raise ValueError("The attribute 'action' must be set.")
|
||||
if not values.get("connery_service"):
|
||||
raise ValueError("The attribute 'connery_service' must be set.")
|
||||
|
||||
return values
|
||||
|
||||
@classmethod
|
||||
def create_instance(cls, action: Action, connery_service: Any) -> "ConneryAction":
|
||||
"""
|
||||
Creates a Connery Action Tool from a Connery Action.
|
||||
Parameters:
|
||||
action (Action): The Connery Action to wrap in a Connery Action Tool.
|
||||
connery_service (ConneryService): The Connery Service
|
||||
to run the Connery Action. We use Any here to avoid circular imports.
|
||||
Returns:
|
||||
ConneryAction: The Connery Action Tool.
|
||||
"""
|
||||
|
||||
# Import ConneryService here and check if it is an instance
|
||||
# of ConneryService to avoid circular imports
|
||||
from .service import ConneryService
|
||||
|
||||
if not isinstance(connery_service, ConneryService):
|
||||
raise ValueError(
|
||||
"The connery_service must be an instance of ConneryService."
|
||||
)
|
||||
|
||||
input_schema = cls._create_input_schema(action.inputParameters)
|
||||
description = action.title + (
|
||||
": " + action.description if action.description else ""
|
||||
)
|
||||
|
||||
instance = cls(
|
||||
name=action.id,
|
||||
description=description,
|
||||
args_schema=input_schema,
|
||||
action=action,
|
||||
connery_service=connery_service,
|
||||
)
|
||||
|
||||
return instance
|
||||
|
||||
@classmethod
|
||||
def _create_input_schema(cls, inputParameters: List[Parameter]) -> Type[BaseModel]:
|
||||
"""
|
||||
Creates an input schema for a Connery Action Tool
|
||||
based on the input parameters of the Connery Action.
|
||||
Parameters:
|
||||
inputParameters: List of input parameters of the Connery Action.
|
||||
Returns:
|
||||
Type[BaseModel]: The input schema for the Connery Action Tool.
|
||||
"""
|
||||
|
||||
dynamic_input_fields: Dict[str, Any] = {}
|
||||
|
||||
for param in inputParameters:
|
||||
default = ... if param.validation and param.validation.required else None
|
||||
title = param.title
|
||||
description = param.title + (
|
||||
": " + param.description if param.description else ""
|
||||
)
|
||||
type = param.type
|
||||
|
||||
dynamic_input_fields[param.key] = (
|
||||
str,
|
||||
Field(default, title=title, description=description, type=type),
|
||||
)
|
||||
|
||||
InputModel = create_model("InputSchema", **dynamic_input_fields)
|
||||
return InputModel
|
Reference in New Issue
Block a user