Clase BaseTool

Herencia de BaseTool para casos avanzados

Aunque el decorador @tool es la opción preferida para la mayoría de herramientas, existen escenarios específicos donde heredar directamente de BaseTool proporciona mayor control y flexibilidad. Esta aproximación es especialmente útil cuando necesitamos implementar herramientas con estado interno, validaciones complejas o lógica de procesamiento avanzada.

Cuándo usar BaseTool en lugar de @tool

La herencia de BaseTool se vuelve necesaria en situaciones donde requerimos funcionalidades que el decorador no puede proporcionar:

• Herramientas con estado interno: Cuando necesitamos mantener información entre llamadas
• Validación compleja de argumentos: Más allá de los tipos básicos de Pydantic
• Lógica de inicialización personalizada: Configuración específica durante la creación de la herramienta
• Métodos auxiliares privados: Funcionalidad interna que no debe ser expuesta

Estructura básica de una herramienta personalizada

Para crear una herramienta heredando de BaseTool, necesitamos implementar los métodos esenciales:

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type, Optional

class CalculadoraAvanzadaInput(BaseModel):
    operacion: str = Field(description="Operación matemática a realizar")
    precision: int = Field(default=2, description="Decimales de precisión")

class CalculadoraAvanzada(BaseTool):
    name: str = "calculadora_avanzada"
    description: str = "Realiza cálculos matemáticos con precisión configurable"
    args_schema: Type[BaseModel] = CalculadoraAvanzadaInput
    
    # Estado interno
    historial: list = []
    
    def _run(self, operacion: str, precision: int = 2) -> str:
        """Implementación síncrona de la herramienta"""
        try:
            resultado = eval(operacion)
            resultado_formateado = round(resultado, precision)
            
            # Mantener historial (estado interno)
            self.historial.append({
                'operacion': operacion,
                'resultado': resultado_formateado
            })
            
            return f"Resultado: {resultado_formateado}"
        except Exception as e:
            return f"Error en el cálculo: {str(e)}"
    
    async def _arun(self, operacion: str, precision: int = 2) -> str:
        """Implementación asíncrona de la herramienta"""
        return self._run(operacion, precision)

Implementación de herramientas con estado persistente

Una de las ventajas principales de BaseTool es la capacidad de mantener estado entre llamadas:

from datetime import datetime
from typing import Dict

class GestorSesionInput(BaseModel):
    accion: str = Field(description="Acción: 'guardar', 'obtener', 'listar'")
    clave: Optional[str] = Field(default=None, description="Clave para guardar/obtener")
    valor: Optional[str] = Field(default=None, description="Valor a guardar")

class GestorSesion(BaseTool):
    name: str = "gestor_sesion"
    description: str = "Gestiona datos de sesión durante la conversación"
    args_schema: Type[BaseModel] = GestorSesionInput
    
    # Estado interno
    datos_sesion: Dict[str, str] = {}
    
    def _run(self, accion: str, clave: Optional[str] = None, 
             valor: Optional[str] = None) -> str:
        if accion == "guardar":
            if not clave or not valor:
                return "Error: Se requieren clave y valor para guardar"
            self.datos_sesion[clave] = valor
            return f"Guardado: {clave} = {valor}"
        
        elif accion == "obtener":
            if not clave:
                return "Error: Se requiere una clave"
            valor_obtenido = self.datos_sesion.get(clave, "No encontrado")
            return f"Valor de '{clave}': {valor_obtenido}"
        
        elif accion == "listar":
            if not self.datos_sesion:
                return "No hay datos en la sesión"
            items = [f"{k}: {v}" for k, v in self.datos_sesion.items()]
            return "Datos de sesión:\n" + "\n".join(items)
        
        return f"Acción '{accion}' no reconocida"
    
    async def _arun(self, accion: str, clave: Optional[str] = None, 
                    valor: Optional[str] = None) -> str:
        return self._run(accion, clave, valor)

Validación avanzada con Pydantic v2

BaseTool permite implementar validaciones complejas utilizando las capacidades de Pydantic v2:

from pydantic import BaseModel, Field, field_validator
from typing import Literal, Dict

class ConsultaAPIInput(BaseModel):
    endpoint: str = Field(description="Endpoint de la API a consultar")
    metodo: Literal["GET", "POST", "PUT", "DELETE"] = Field(
        default="GET", description="Método HTTP"
    )
    parametros: Optional[Dict[str, str]] = Field(
        default=None, description="Parámetros de la consulta"
    )
    
    @field_validator('endpoint')
    @classmethod
    def validar_endpoint(cls, v):
        if not v.startswith(('http://', 'https://')):
            raise ValueError('El endpoint debe comenzar con http:// o https://')
        return v
    
    @field_validator('parametros')
    @classmethod
    def validar_parametros(cls, v):
        if v and len(v) > 10:
            raise ValueError('Máximo 10 parámetros permitidos')
        return v

class ConsultorAPI(BaseTool):
    name: str = "consultor_api"
    description: str = "Realiza consultas HTTP con validación avanzada"
    args_schema: Type[BaseModel] = ConsultaAPIInput
    
    # Estado interno
    contador_requests: int = 0
    
    def _run(self, endpoint: str, metodo: str = "GET", 
             parametros: Optional[Dict[str, str]] = None) -> str:
        self.contador_requests += 1
        
        resultado = {
            'endpoint': endpoint,
            'metodo': metodo,
            'parametros': parametros or {},
            'request_numero': self.contador_requests
        }
        
        return f"Consulta realizada: {resultado}"
    
    async def _arun(self, endpoint: str, metodo: str = "GET", 
                    parametros: Optional[Dict[str, str]] = None) -> str:
        return self._run(endpoint, metodo, parametros)

Integración con modelos de chat

Las herramientas personalizadas creadas con BaseTool se integran perfectamente con los modelos de chat de LangChain:

from langchain_openai import ChatOpenAI
from langchain.messages import HumanMessage

# Crear instancias de las herramientas
calculadora = CalculadoraAvanzada()
gestor = GestorSesion()
consultor = ConsultorAPI()

# Configurar el modelo con las herramientas
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_con_herramientas = llm.bind_tools([calculadora, gestor, consultor])

# Ejemplo de uso
respuesta = llm_con_herramientas.invoke([
    HumanMessage(content="Calcula 15.7 * 8.3 con 3 decimales de precisión")
])

print(f"Respuesta: {respuesta}")

Manejo de errores y respuestas personalizadas

El manejo robusto de errores es fundamental cuando trabajamos con herramientas que interactúan con recursos externos. BaseTool proporciona mecanismos específicos para capturar, procesar y responder a errores de manera elegante.

Implementación de handle_tool_error

El atributo handle_tool_error nos permite definir cómo debe comportarse nuestra herramienta cuando ocurre una excepción:

import time

class ConsultaWebInput(BaseModel):
    url: str = Field(description="URL a consultar")
    timeout: int = Field(default=10, description="Timeout en segundos")

class ConsultaWeb(BaseTool):
    name: str = "consulta_web"
    description: str = "Consulta páginas web con manejo de errores robusto"
    args_schema: Type[BaseModel] = ConsultaWebInput
    handle_tool_error: bool = True
    
    # Configuración
    max_reintentos: int = 3
    
    def _run(self, url: str, timeout: int = 10) -> str:
        """Implementación con manejo de errores personalizado"""
        for intento in range(self.max_reintentos):
            try:
                # Simulación de petición HTTP
                if "error" in url.lower():
                    raise ConnectionError("Servicio no disponible")
                
                return f"Consulta exitosa a {url}"
                
            except ConnectionError:
                if intento < self.max_reintentos - 1:
                    time.sleep(2 ** intento)  # Backoff exponencial
                    continue
                raise Exception(f"Error persistente después de {self.max_reintentos} intentos")
    
    async def _arun(self, url: str, timeout: int = 10) -> str:
        return self._run(url, timeout)

Respuestas estructuradas para diferentes tipos de error

Una estrategia avanzada es categorizar los errores y proporcionar respuestas específicas:

from enum import Enum
import json

class TipoError(Enum):
    CONEXION = "conexion"
    AUTORIZACION = "autorizacion"
    VALIDACION = "validacion"

class ProcesadorArchivosInput(BaseModel):
    ruta_archivo: str = Field(description="Ruta del archivo a procesar")
    operacion: str = Field(description="Operación: 'leer', 'escribir'")
    contenido: Optional[str] = Field(default=None, description="Contenido para escribir")

class ProcesadorArchivos(BaseTool):
    name: str = "procesador_archivos"
    description: str = "Procesa archivos con manejo detallado de errores"
    args_schema: Type[BaseModel] = ProcesadorArchivosInput
    handle_tool_error: bool = True
    
    def _categorizar_error(self, error: Exception) -> dict:
        """Categoriza errores y genera respuestas estructuradas"""
        error_str = str(error).lower()
        
        if "permission" in error_str:
            return {
                "tipo": TipoError.AUTORIZACION.value,
                "mensaje": "Sin permisos para acceder al archivo",
                "reintentable": False
            }
        elif "no such file" in error_str:
            return {
                "tipo": TipoError.VALIDACION.value,
                "mensaje": "El archivo no existe",
                "reintentable": False
            }
        return {
            "tipo": TipoError.CONEXION.value,
            "mensaje": str(error),
            "reintentable": True
        }
    
    def _run(self, ruta_archivo: str, operacion: str, 
             contenido: Optional[str] = None) -> str:
        try:
            if operacion == "leer":
                with open(ruta_archivo, 'r', encoding='utf-8') as f:
                    contenido_leido = f.read()
                return f"Contenido: {contenido_leido[:200]}..."
            
            elif operacion == "escribir":
                if not contenido:
                    raise ValueError("Se requiere contenido para escribir")
                with open(ruta_archivo, 'w', encoding='utf-8') as f:
                    f.write(contenido)
                return f"Escrito exitosamente en {ruta_archivo}"
            
            raise ValueError(f"Operación '{operacion}' no soportada")
                
        except Exception as e:
            error_info = self._categorizar_error(e)
            return json.dumps({"error": True, **error_info}, ensure_ascii=False)
    
    async def _arun(self, ruta_archivo: str, operacion: str, 
                    contenido: Optional[str] = None) -> str:
        return self._run(ruta_archivo, operacion, contenido)

El manejo personalizado de errores en BaseTool transforma las herramientas de simples ejecutores de funciones a componentes inteligentes capaces de diagnosticar problemas y proporcionar información valiosa tanto para el modelo como para el usuario final.