"""工具基础接口定义""" import time from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional, Union from pydantic import BaseModel, Field from enum import Enum from app.models.tool_model import ToolType, ToolStatus class ParameterType(str, Enum): """参数类型枚举""" STRING = "string" INTEGER = "integer" NUMBER = "number" BOOLEAN = "boolean" ARRAY = "array" OBJECT = "object" class ToolParameter(BaseModel): """工具参数定义""" name: str = Field(..., description="参数名称") type: ParameterType = Field(..., description="参数类型") description: str = Field("", description="参数描述") required: bool = Field(False, description="是否必需") default: Any = Field(None, description="默认值") enum: Optional[List[Any]] = Field(None, description="枚举值") minimum: Optional[Union[int, float]] = Field(None, description="最小值") maximum: Optional[Union[int, float]] = Field(None, description="最大值") pattern: Optional[str] = Field(None, description="正则表达式模式") class Config: use_enum_values = True class ToolResult(BaseModel): """工具执行结果""" success: bool = Field(..., description="执行是否成功") data: Any = Field(None, description="返回数据") error: Optional[str] = Field(None, description="错误信息") error_code: Optional[str] = Field(None, description="错误代码") execution_time: float = Field(..., description="执行时间（秒）") token_usage: Optional[Dict[str, int]] = Field(None, description="Token使用情况") metadata: Dict[str, Any] = Field(default_factory=dict, description="额外元数据") @classmethod def success_result( cls, data: Any, execution_time: float, token_usage: Optional[Dict[str, int]] = None, metadata: Optional[Dict[str, Any]] = None ) -> "ToolResult": """创建成功结果""" return cls( success=True, data=data, execution_time=execution_time, token_usage=token_usage, metadata=metadata or {} ) @classmethod def error_result( cls, error: str, execution_time: float, error_code: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None ) -> "ToolResult": """创建错误结果""" return cls( success=False, error=error, error_code=error_code, execution_time=execution_time, metadata=metadata or {} ) class ToolInfo(BaseModel): """工具信息""" id: str = Field(..., description="工具ID") name: str = Field(..., description="工具名称") description: str = Field(..., description="工具描述") tool_type: ToolType = Field(..., description="工具类型") version: str = Field("1.0.0", description="工具版本") parameters: List[ToolParameter] = Field(default_factory=list, description="工具参数") status: ToolStatus = Field(ToolStatus.ACTIVE, description="工具状态") tags: List[str] = Field(default_factory=list, description="工具标签") tenant_id: Optional[str] = Field(None, description="租户ID") class Config: use_enum_values = True class BaseTool(ABC): """所有工具的基础抽象类""" def __init__(self, tool_id: str, config: Dict[str, Any]): """初始化工具 Args: tool_id: 工具ID config: 工具配置 """ self.tool_id = tool_id self.config = config self._status = ToolStatus.ACTIVE @property @abstractmethod def name(self) -> str: """工具名称""" pass @property @abstractmethod def description(self) -> str: """工具描述""" pass @property @abstractmethod def tool_type(self) -> ToolType: """工具类型""" pass @property def version(self) -> str: """工具版本""" return self.config.get("version", "1.0.0") @property def status(self) -> ToolStatus: """工具状态""" return self._status @status.setter def status(self, value: ToolStatus): """设置工具状态""" self._status = value @property @abstractmethod def parameters(self) -> List[ToolParameter]: """工具参数定义""" pass @property def tags(self) -> List[str]: """工具标签""" return self.config.get("tags", []) def get_info(self) -> ToolInfo: """获取工具信息""" return ToolInfo( id=self.tool_id, name=self.name, description=self.description, tool_type=self.tool_type, version=self.version, parameters=self.parameters, status=self.status, tags=self.tags, tenant_id=self.config.get("tenant_id") ) def validate_parameters(self, parameters: Dict[str, Any]) -> Dict[str, str]: """验证参数 Args: parameters: 输入参数 Returns: 验证错误字典，空字典表示验证通过 """ errors = {} param_definitions = {p.name: p for p in self.parameters} # 检查必需参数 for param_def in self.parameters: if param_def.required and param_def.name not in parameters: errors[param_def.name] = f"Required parameter '{param_def.name}' is missing" # 检查参数类型和约束 for param_name, param_value in parameters.items(): if param_name not in param_definitions: continue param_def = param_definitions[param_name] # 类型检查 if not self._validate_parameter_type(param_value, param_def): errors[param_name] = f"Parameter '{param_name}' has invalid type, expected {param_def.type}" # 约束检查 constraint_error = self._validate_parameter_constraints(param_value, param_def) if constraint_error: errors[param_name] = constraint_error return errors def _validate_parameter_type(self, value: Any, param_def: ToolParameter) -> bool: """验证参数类型""" if value is None: return not param_def.required type_mapping = { ParameterType.STRING: str, ParameterType.INTEGER: int, ParameterType.NUMBER: (int, float), ParameterType.BOOLEAN: bool, ParameterType.ARRAY: list, ParameterType.OBJECT: dict } expected_type = type_mapping.get(param_def.type) if expected_type: return isinstance(value, expected_type) return True def _validate_parameter_constraints(self, value: Any, param_def: ToolParameter) -> Optional[str]: """验证参数约束""" if value is None: return None # 枚举值检查 if param_def.enum and value not in param_def.enum: return f"Value must be one of {param_def.enum}" # 数值范围检查 if param_def.type in [ParameterType.INTEGER, ParameterType.NUMBER]: if param_def.minimum is not None and value < param_def.minimum: return f"Value must be >= {param_def.minimum}" if param_def.maximum is not None and value > param_def.maximum: return f"Value must be <= {param_def.maximum}" # 字符串模式检查 if param_def.type == ParameterType.STRING and param_def.pattern: import re if not re.match(param_def.pattern, str(value)): return f"Value must match pattern: {param_def.pattern}" return None @abstractmethod async def execute(self, **kwargs) -> ToolResult: """执行工具 Args: **kwargs: 工具参数 Returns: 执行结果 """ pass async def safe_execute(self, **kwargs) -> ToolResult: """安全执行工具（包含参数验证和异常处理） Args: **kwargs: 工具参数 Returns: 执行结果 """ start_time = time.time() try: # 参数验证 validation_errors = self.validate_parameters(kwargs) if validation_errors: execution_time = time.time() - start_time error_msg = "; ".join([f"{k}: {v}" for k, v in validation_errors.items()]) return ToolResult.error_result( error=f"Parameter validation failed: {error_msg}", error_code="VALIDATION_ERROR", execution_time=execution_time ) # 执行工具 result = await self.execute(**kwargs) return result except Exception as e: execution_time = time.time() - start_time return ToolResult.error_result( error=str(e), error_code="EXECUTION_ERROR", execution_time=execution_time ) def to_langchain_tool(self): """转换为Langchain工具格式""" from .langchain_adapter import LangchainAdapter return LangchainAdapter.convert_tool(self) def __repr__(self): return f"<{self.__class__.__name__}(id={self.tool_id}, name={self.name})>"