Source code for langchain_community.tools.connery.tool

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):
    """康纳动作工具。"""

    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]:
        """运行Connery Action并提供输入。
参数：
    kwargs（Dict[str, str]）：操作所需的输入字典。
返回：
    Dict[str, str]：操作的输出。
"""

        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]:
        """使用提供的输入异步运行Connery操作。
参数：
    kwargs（Dict[str，str]）：操作所需的输入字典。
返回：
    Dict[str，str]：操作的输出。
"""

        func = partial(self._run, **kwargs)
        return await asyncio.get_event_loop().run_in_executor(None, func)

    def get_schema_json(self) -> str:
        """返回Connery Action Tool模式的JSON表示。
这对于调试很有用。
返回：
    str：Connery Action Tool模式的JSON表示。
"""

        return self.args_schema.schema_json(indent=2)

    @root_validator()
    def validate_attributes(cls, values: dict) -> dict:
        """验证 ConneryAction 类的属性。
参数：
    values（dict）：要验证的参数。
返回：
    dict：验证后的参数。
"""

        # 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":
        """创建一个Connery操作工具，从一个Connery操作中创建。
参数：
    action（Action）：要包装在Connery操作工具中的Connery操作。
    connery_service（ConneryService）：Connery服务，用于运行Connery操作。我们在这里使用Any来避免循环导入。
返回：
    ConneryAction：Connery操作工具。
"""

        # 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]:
        """根据Connery Action的输入参数，为Connery Action工具创建一个输入模式。
参数：
    inputParameters：Connery Action的输入参数列表。
返回：
    Type[BaseModel]：Connery Action工具的输入模式。
"""

        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