api/core/tools/entities/tool_entities.py

from enum import Enum
from typing import Any, Optional, Union, cast

from pydantic import BaseModel, Field

from core.tools.entities.common_entities import I18nObject


class ToolProviderType(Enum):
    """
        Enum class for tool provider
    """
    BUILT_IN = "built-in"
    DATASET_RETRIEVAL = "dataset-retrieval"
    APP_BASED = "app-based"
    API_BASED = "api-based"

    @classmethod
    def value_of(cls, value: str) -> 'ToolProviderType':
        """
        Get value of given mode.

        :param value: mode value
        :return: mode
        """
        for mode in cls:
            if mode.value == value:
                return mode
        raise ValueError(f'invalid mode value {value}')
    
class ApiProviderSchemaType(Enum):
    """
    Enum class for api provider schema type.
    """
    OPENAPI = "openapi"
    SWAGGER = "swagger"
    OPENAI_PLUGIN = "openai_plugin"
    OPENAI_ACTIONS = "openai_actions"

    @classmethod
    def value_of(cls, value: str) -> 'ApiProviderSchemaType':
        """
        Get value of given mode.

        :param value: mode value
        :return: mode
        """
        for mode in cls:
            if mode.value == value:
                return mode
        raise ValueError(f'invalid mode value {value}')
    
class ApiProviderAuthType(Enum):
    """
    Enum class for api provider auth type.
    """
    NONE = "none"
    API_KEY = "api_key"

    @classmethod
    def value_of(cls, value: str) -> 'ApiProviderAuthType':
        """
        Get value of given mode.

        :param value: mode value
        :return: mode
        """
        for mode in cls:
            if mode.value == value:
                return mode
        raise ValueError(f'invalid mode value {value}')

class ToolInvokeMessage(BaseModel):
    class MessageType(Enum):
        TEXT = "text"
        IMAGE = "image"
        LINK = "link"
        BLOB = "blob"
        IMAGE_LINK = "image_link"

    type: MessageType = MessageType.TEXT
    """
        plain text, image url or link url
    """
    message: Union[str, bytes] = None
    meta: dict[str, Any] = None
    save_as: str = ''

class ToolInvokeMessageBinary(BaseModel):
    mimetype: str = Field(..., description="The mimetype of the binary")
    url: str = Field(..., description="The url of the binary")
    save_as: str = ''

class ToolParameterOption(BaseModel):
    value: str = Field(..., description="The value of the option")
    label: I18nObject = Field(..., description="The label of the option")

class ToolParameter(BaseModel):
    class ToolParameterType(Enum):
        STRING = "string"
        NUMBER = "number"
        BOOLEAN = "boolean"
        SELECT = "select"
        SECRET_INPUT = "secret-input"

    class ToolParameterForm(Enum):
        SCHEMA = "schema" # should be set while adding tool
        FORM = "form"     # should be set before invoking tool
        LLM = "llm"       # will be set by LLM

    name: str = Field(..., description="The name of the parameter")
    label: I18nObject = Field(..., description="The label presented to the user")
    human_description: I18nObject = Field(..., description="The description presented to the user")
    type: ToolParameterType = Field(..., description="The type of the parameter")
    form: ToolParameterForm = Field(..., description="The form of the parameter, schema/form/llm")
    llm_description: Optional[str] = None
    required: Optional[bool] = False
    default: Optional[Union[int, str]] = None
    min: Optional[Union[float, int]] = None
    max: Optional[Union[float, int]] = None
    options: Optional[list[ToolParameterOption]] = None

    @classmethod
    def get_simple_instance(cls, 
                       name: str, llm_description: str, type: ToolParameterType, 
                       required: bool, options: Optional[list[str]] = None) -> 'ToolParameter':
        """
            get a simple tool parameter

            :param name: the name of the parameter
            :param llm_description: the description presented to the LLM
            :param type: the type of the parameter
            :param required: if the parameter is required
            :param options: the options of the parameter
        """
        # convert options to ToolParameterOption
        if options:
            options = [ToolParameterOption(value=option, label=I18nObject(en_US=option, zh_Hans=option)) for option in options]
        return cls(
            name=name,
            label=I18nObject(en_US='', zh_Hans=''),
            human_description=I18nObject(en_US='', zh_Hans=''),
            type=type,
            form=cls.ToolParameterForm.LLM,
            llm_description=llm_description,
            required=required,
            options=options,
        )

class ToolProviderIdentity(BaseModel):
    author: str = Field(..., description="The author of the tool")
    name: str = Field(..., description="The name of the tool")
    description: I18nObject = Field(..., description="The description of the tool")
    icon: str = Field(..., description="The icon of the tool")
    label: I18nObject = Field(..., description="The label of the tool")

class ToolDescription(BaseModel):
    human: I18nObject = Field(..., description="The description presented to the user")
    llm: str = Field(..., description="The description presented to the LLM")

class ToolIdentity(BaseModel):
    author: str = Field(..., description="The author of the tool")
    name: str = Field(..., description="The name of the tool")
    label: I18nObject = Field(..., description="The label of the tool")
    provider: str = Field(..., description="The provider of the tool")
    icon: Optional[str] = None

class ToolCredentialsOption(BaseModel):
    value: str = Field(..., description="The value of the option")
    label: I18nObject = Field(..., description="The label of the option")

class ToolProviderCredentials(BaseModel):
    class CredentialsType(Enum):
        SECRET_INPUT = "secret-input"
        TEXT_INPUT = "text-input"
        SELECT = "select"
        BOOLEAN = "boolean"

        @classmethod
        def value_of(cls, value: str) -> "ToolProviderCredentials.CredentialsType":
            """
            Get value of given mode.

            :param value: mode value
            :return: mode
            """
            for mode in cls:
                if mode.value == value:
                    return mode
            raise ValueError(f'invalid mode value {value}')
        
        @staticmethod
        def default(value: str) -> str:
            return ""

    name: str = Field(..., description="The name of the credentials")
    type: CredentialsType = Field(..., description="The type of the credentials")
    required: bool = False
    default: Optional[Union[int, str]] = None
    options: Optional[list[ToolCredentialsOption]] = None
    label: Optional[I18nObject] = None
    help: Optional[I18nObject] = None
    url: Optional[str] = None
    placeholder: Optional[I18nObject] = None

    def to_dict(self) -> dict:
        return {
            'name': self.name,
            'type': self.type.value,
            'required': self.required,
            'default': self.default,
            'options': self.options,
            'help': self.help.to_dict() if self.help else None,
            'label': self.label.to_dict(),
            'url': self.url,
            'placeholder': self.placeholder.to_dict() if self.placeholder else None,
        }

class ToolRuntimeVariableType(Enum):
    TEXT = "text"
    IMAGE = "image"

class ToolRuntimeVariable(BaseModel):
    type: ToolRuntimeVariableType = Field(..., description="The type of the variable")
    name: str = Field(..., description="The name of the variable")
    position: int = Field(..., description="The position of the variable")
    tool_name: str = Field(..., description="The name of the tool")

class ToolRuntimeTextVariable(ToolRuntimeVariable):
    value: str = Field(..., description="The value of the variable")

class ToolRuntimeImageVariable(ToolRuntimeVariable):
    value: str = Field(..., description="The path of the image")

class ToolRuntimeVariablePool(BaseModel):
    conversation_id: str = Field(..., description="The conversation id")
    user_id: str = Field(..., description="The user id")
    tenant_id: str = Field(..., description="The tenant id of assistant")

    pool: list[ToolRuntimeVariable] = Field(..., description="The pool of variables")

    def __init__(self, **data: Any):
        pool = data.get('pool', [])
        # convert pool into correct type
        for index, variable in enumerate(pool):
            if variable['type'] == ToolRuntimeVariableType.TEXT.value:
                pool[index] = ToolRuntimeTextVariable(**variable)
            elif variable['type'] == ToolRuntimeVariableType.IMAGE.value:
                pool[index] = ToolRuntimeImageVariable(**variable)
        super().__init__(**data)

    def dict(self) -> dict:
        return {
            'conversation_id': self.conversation_id,
            'user_id': self.user_id,
            'tenant_id': self.tenant_id,
            'pool': [variable.dict() for variable in self.pool],
        }
    
    def set_text(self, tool_name: str, name: str, value: str) -> None:
        """
            set a text variable
        """
        for variable in self.pool:
            if variable.name == name:
                if variable.type == ToolRuntimeVariableType.TEXT:
                    variable = cast(ToolRuntimeTextVariable, variable)
                    variable.value = value
                    return
                
        variable = ToolRuntimeTextVariable(
            type=ToolRuntimeVariableType.TEXT,
            name=name,
            position=len(self.pool),
            tool_name=tool_name,
            value=value,
        )

        self.pool.append(variable)

    def set_file(self, tool_name: str, value: str, name: str = None) -> None:
        """
            set an image variable

            :param tool_name: the name of the tool
            :param value: the id of the file
        """
        # check how many image variables are there
        image_variable_count = 0
        for variable in self.pool:
            if variable.type == ToolRuntimeVariableType.IMAGE:
                image_variable_count += 1

        if name is None:
            name = f"file_{image_variable_count}"

        for variable in self.pool:
            if variable.name == name:
                if variable.type == ToolRuntimeVariableType.IMAGE:
                    variable = cast(ToolRuntimeImageVariable, variable)
                    variable.value = value
                    return
                
        variable = ToolRuntimeImageVariable(
            type=ToolRuntimeVariableType.IMAGE,
            name=name,
            position=len(self.pool),
            tool_name=tool_name,
            value=value,
        )

        self.pool.append(variable)

class ModelToolPropertyKey(Enum):
    IMAGE_PARAMETER_NAME = "image_parameter_name"

class ModelToolConfiguration(BaseModel):
    """
    Model tool configuration
    """
    type: str = Field(..., description="The type of the model tool")
    model: str = Field(..., description="The model")
    label: I18nObject = Field(..., description="The label of the model tool")
    properties: dict[ModelToolPropertyKey, Any] = Field(..., description="The properties of the model tool")

class ModelToolProviderConfiguration(BaseModel):
    """
    Model tool provider configuration
    """
    provider: str = Field(..., description="The provider of the model tool")
    models: list[ModelToolConfiguration] = Field(..., description="The models of the model tool")
    label: I18nObject = Field(..., description="The label of the model tool")

class ToolInvokeMeta(BaseModel):
    """
    Tool invoke meta
    """
    time_cost: float = Field(..., description="The time cost of the tool invoke")
    error: Optional[str] = None
    tool_config: Optional[dict] = None

    @classmethod
    def empty(cls) -> 'ToolInvokeMeta':
        """
        Get an empty instance of ToolInvokeMeta
        """
        return cls(time_cost=0.0, error=None, tool_config={})
    
    @classmethod
    def error_instance(cls, error: str) -> 'ToolInvokeMeta':
        """
        Get an instance of ToolInvokeMeta with error
        """
        return cls(time_cost=0.0, error=error, tool_config={})
    
    def to_dict(self) -> dict:
        return {
            'time_cost': self.time_cost,
            'error': self.error,
            'tool_config': self.tool_config,
        }