from enum import Enum from typing import Any, Optional, Union, cast from pydantic import BaseModel, Field, field_validator from core.tools.entities.common_entities import I18nObject class ToolLabelEnum(Enum): SEARCH = 'search' IMAGE = 'image' VIDEOS = 'videos' WEATHER = 'weather' FINANCE = 'finance' DESIGN = 'design' TRAVEL = 'travel' SOCIAL = 'social' NEWS = 'news' MEDICAL = 'medical' PRODUCTIVITY = 'productivity' EDUCATION = 'education' BUSINESS = 'business' ENTERTAINMENT = 'entertainment' UTILITIES = 'utilities' OTHER = 'other' class ToolProviderType(Enum): """ Enum class for tool provider """ BUILT_IN = "builtin" WORKFLOW = "workflow" API = "api" APP = "app" DATASET_RETRIEVAL = "dataset-retrieval" @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" FILE_VAR = "file_var" 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 = '' file_var: Optional[dict[str, Any]] = None class ToolParameterOption(BaseModel): value: str = Field(..., description="The value of the option") label: I18nObject = Field(..., description="The label of the option") @field_validator('value', mode='before') @classmethod def transform_id_to_str(cls, value) -> str: if not isinstance(value, str): return str(value) else: return value class ToolParameter(BaseModel): class ToolParameterType(str, Enum): STRING = "string" NUMBER = "number" BOOLEAN = "boolean" SELECT = "select" SECRET_INPUT = "secret-input" FILE = "file" 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") tags: Optional[list[ToolLabelEnum]] = Field(default=[], description="The tags 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.model_dump() 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 WorkflowToolParameterConfiguration(BaseModel): """ Workflow tool configuration """ name: str = Field(..., description="The name of the parameter") description: str = Field(..., description="The description of the parameter") form: ToolParameter.ToolParameterForm = Field(..., description="The form of the parameter") 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, } class ToolLabel(BaseModel): """ Tool label """ name: str = Field(..., description="The name of the tool") label: I18nObject = Field(..., description="The label of the tool") icon: str = Field(..., description="The icon of the tool") class ToolInvokeFrom(Enum): """ Enum class for tool invoke """ WORKFLOW = "workflow" AGENT = "agent"