Clase BaseTool

Herencia de BaseTool para casos avanzados

Aunque el decorador @tool es la opción preferida para la mayoría de herramientas en LangChain, 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 que va más allá de las capacidades del decorador.

Cuándo usar BaseTool en lugar de @tool

La herencia de BaseTool se vuelve necesaria en situaciones donde requerimos funcionalidades avanzadas 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 y definir las propiedades requeridas:

from langchain_core.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
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.historial = []  # Estado interno
    
    def _run(self, operacion: str, precision: int = 2) -> str:
        """Implementación síncrona de la herramienta"""
        try:
            resultado = eval(operacion)  # En producción usar ast.literal_eval
            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"""
        # Para operaciones simples, delegamos a la versión síncrona
        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. Esto es especialmente útil para herramientas que necesitan recordar información previa:

from datetime import datetime
from typing import Dict, List

class GestorSesionInput(BaseModel):
    accion: str = Field(description="Acción a realizar: '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
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.datos_sesion: Dict[str, str] = {}
        self.timestamp_creacion = datetime.now()
    
    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 para obtener"
            
            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 f"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 args_schema personalizado

BaseTool permite implementar validaciones complejas que van más allá de los tipos básicos, utilizando las capacidades completas de Pydantic:

from pydantic import BaseModel, Field, validator
from typing import Literal

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"
    )
    
    @validator('endpoint')
    def validar_endpoint(cls, v):
        if not v.startswith(('http://', 'https://')):
            raise ValueError('El endpoint debe comenzar con http:// o https://')
        return v
    
    @validator('parametros')
    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
    
    def __init__(self, timeout: int = 30, **kwargs):
        super().__init__(**kwargs)
        self.timeout = timeout
        self.contador_requests = 0
    
    def _run(self, endpoint: str, metodo: str = "GET", 
             parametros: Optional[Dict[str, str]] = None) -> str:
        self.contador_requests += 1
        
        # Simulación de consulta HTTP
        resultado = {
            'endpoint': endpoint,
            'metodo': metodo,
            'parametros': parametros or {},
            'request_numero': self.contador_requests,
            'status': 'success'
        }
        
        return f"Consulta realizada: {resultado}"
    
    async def _arun(self, endpoint: str, metodo: str = "GET", 
                    parametros: Optional[Dict[str, str]] = None) -> str:
        # En una implementación real, aquí usaríamos aiohttp
        return self._run(endpoint, metodo, parametros)

Integración con LCEL y modelos de chat

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

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

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

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

# Crear una cadena LCEL
cadena = llm_con_herramientas | (lambda x: x.tool_calls[0] if x.tool_calls else x)

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

print(f"Respuesta del modelo: {respuesta}")

La herencia de BaseTool proporciona el control granular necesario para casos de uso avanzados, manteniendo la compatibilidad completa con el ecosistema de LangChain y permitiendo implementar funcionalidades que serían imposibles con el decorador @tool estándar.

Manejo de errores y respuestas personalizadas

El manejo robusto de errores es fundamental cuando trabajamos con herramientas que interactúan con recursos externos o realizan operaciones que pueden fallar. BaseTool proporciona mecanismos específicos para capturar, procesar y responder a errores de manera elegante, permitiendo que nuestras aplicaciones mantengan su funcionalidad incluso cuando las herramientas encuentran problemas.

Implementación de handle_tool_error

El método handle_tool_error nos permite definir cómo debe comportarse nuestra herramienta cuando ocurre una excepción. Este enfoque es más sofisticado que simplemente capturar errores con try-catch, ya que permite personalizar completamente la respuesta:

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
import requests
import time
from typing import Type, Optional

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
    
    def __init__(self, max_reintentos: int = 3, **kwargs):
        super().__init__(**kwargs)
        self.max_reintentos = max_reintentos
        self.reintentos_realizados = 0
    
    def _run(self, url: str, timeout: int = 10) -> str:
        """Implementación con manejo de errores personalizado"""
        for intento in range(self.max_reintentos):
            try:
                response = requests.get(url, timeout=timeout)
                response.raise_for_status()
                
                # Resetear contador en caso de éxito
                self.reintentos_realizados = 0
                
                return f"Consulta exitosa a {url}. Status: {response.status_code}"
                
            except requests.exceptions.Timeout:
                if intento < self.max_reintentos - 1:
                    time.sleep(2 ** intento)  # Backoff exponencial
                    continue
                raise Exception(f"Timeout después de {self.max_reintentos} intentos")
                
            except requests.exceptions.ConnectionError:
                if intento < self.max_reintentos - 1:
                    time.sleep(1)
                    continue
                raise Exception(f"Error de conexión persistente a {url}")
                
            except requests.exceptions.HTTPError as e:
                # No reintentar errores HTTP 4xx
                if 400 <= e.response.status_code < 500:
                    raise Exception(f"Error del cliente: {e.response.status_code}")
                # Reintentar errores 5xx
                if intento < self.max_reintentos - 1:
                    time.sleep(2)
                    continue
                raise Exception(f"Error del servidor persistente: {e.response.status_code}")
    
    async def _arun(self, url: str, timeout: int = 10) -> str:
        # En producción usaríamos aiohttp para requests asíncronos
        return self._run(url, timeout)

Respuestas estructuradas para diferentes tipos de error

Una estrategia avanzada es categorizar los errores y proporcionar respuestas específicas que ayuden tanto al modelo como al usuario final a entender qué ocurrió:

from enum import Enum
from dataclasses import dataclass
import json

class TipoError(Enum):
    CONEXION = "conexion"
    TIMEOUT = "timeout"
    AUTORIZACION = "autorizacion"
    RECURSO_NO_ENCONTRADO = "recurso_no_encontrado"
    SERVIDOR = "servidor"
    VALIDACION = "validacion"

@dataclass
class RespuestaError:
    tipo: TipoError
    mensaje: str
    sugerencia: str
    reintentable: bool

class ProcesadorArchivosInput(BaseModel):
    ruta_archivo: str = Field(description="Ruta del archivo a procesar")
    operacion: str = Field(description="Operación: 'leer', 'escribir', 'eliminar'")
    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) -> RespuestaError:
        """Categoriza errores y genera respuestas estructuradas"""
        error_str = str(error).lower()
        
        if "permission denied" in error_str:
            return RespuestaError(
                tipo=TipoError.AUTORIZACION,
                mensaje="Sin permisos para acceder al archivo",
                sugerencia="Verificar permisos del archivo o ejecutar con privilegios adecuados",
                reintentable=False
            )
        
        elif "no such file" in error_str:
            return RespuestaError(
                tipo=TipoError.RECURSO_NO_ENCONTRADO,
                mensaje="El archivo especificado no existe",
                sugerencia="Verificar la ruta del archivo o crear el archivo primero",
                reintentable=False
            )
        
        elif "disk full" in error_str or "no space" in error_str:
            return RespuestaError(
                tipo=TipoError.SERVIDOR,
                mensaje="Espacio insuficiente en disco",
                sugerencia="Liberar espacio en disco antes de continuar",
                reintentable=True
            )
        
        else:
            return RespuestaError(
                tipo=TipoError.SERVIDOR,
                mensaje=f"Error inesperado: {str(error)}",
                sugerencia="Revisar logs del sistema para más detalles",
                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"Archivo leído exitosamente. Contenido: {contenido_leido[:100]}..."
            
            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"Contenido escrito exitosamente en {ruta_archivo}"
            
            elif operacion == "eliminar":
                import os
                os.remove(ruta_archivo)
                return f"Archivo {ruta_archivo} eliminado exitosamente"
            
            else:
                raise ValueError(f"Operación '{operacion}' no soportada")
                
        except Exception as e:
            respuesta_error = self._categorizar_error(e)
            
            # Formato JSON para respuestas estructuradas
            error_info = {
                "error": True,
                "tipo": respuesta_error.tipo.value,
                "mensaje": respuesta_error.mensaje,
                "sugerencia": respuesta_error.sugerencia,
                "reintentable": respuesta_error.reintentable,
                "operacion_fallida": operacion,
                "archivo": ruta_archivo
            }
            
            return json.dumps(error_info, ensure_ascii=False, indent=2)
    
    async def _arun(self, ruta_archivo: str, operacion: str, 
                    contenido: Optional[str] = None) -> str:
        return self._run(ruta_archivo, operacion, contenido)

Herramientas con recuperación automática y estado de error

Para casos más complejos, podemos implementar herramientas que mantengan estado sobre errores y apliquen estrategias de recuperación automática:

from datetime import datetime, timedelta
from typing import Dict, List

class EstadoError:
    def __init__(self):
        self.errores_recientes: List[Dict] = []
        self.ultimo_exito: Optional[datetime] = None
        self.en_modo_degradado: bool = False

class ServicioExternoInput(BaseModel):
    endpoint: str = Field(description="Endpoint del servicio")
    datos: Optional[Dict] = Field(default=None, description="Datos a enviar")

class ServicioExterno(BaseTool):
    name: str = "servicio_externo"
    description: str = "Interactúa con servicios externos con recuperación automática"
    args_schema: Type[BaseModel] = ServicioExternoInput
    handle_tool_error: bool = True
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.estado_error = EstadoError()
        self.umbral_modo_degradado = 3  # Errores consecutivos
    
    def _evaluar_salud_servicio(self) -> bool:
        """Evalúa si el servicio está en condiciones de operar"""
        ahora = datetime.now()
        errores_recientes = [
            e for e in self.estado_error.errores_recientes
            if ahora - e['timestamp'] < timedelta(minutes=5)
        ]
        
        return len(errores_recientes) < self.umbral_modo_degradado
    
    def _registrar_error(self, error: Exception, endpoint: str):
        """Registra un error en el historial"""
        self.estado_error.errores_recientes.append({
            'timestamp': datetime.now(),
            'error': str(error),
            'endpoint': endpoint,
            'tipo': type(error).__name__
        })
        
        # Mantener solo errores de los últimos 10 minutos
        limite_tiempo = datetime.now() - timedelta(minutes=10)
        self.estado_error.errores_recientes = [
            e for e in self.estado_error.errores_recientes
            if e['timestamp'] > limite_tiempo
        ]
    
    def _modo_degradado_respuesta(self, endpoint: str) -> str:
        """Respuesta cuando el servicio está en modo degradado"""
        ultimo_error = self.estado_error.errores_recientes[-1] if self.estado_error.errores_recientes else None
        
        respuesta = {
            "error": True,
            "modo": "degradado",
            "mensaje": f"Servicio {endpoint} temporalmente no disponible",
            "ultimo_error": ultimo_error['error'] if ultimo_error else "Desconocido",
            "tiempo_ultimo_error": ultimo_error['timestamp'].isoformat() if ultimo_error else None,
            "sugerencia": "El servicio se reactivará automáticamente cuando esté disponible"
        }
        
        return json.dumps(respuesta, ensure_ascii=False, indent=2)
    
    def _run(self, endpoint: str, datos: Optional[Dict] = None) -> str:
        # Verificar estado del servicio
        if not self._evaluar_salud_servicio():
            self.estado_error.en_modo_degradado = True
            return self._modo_degradado_respuesta(endpoint)
        
        try:
            # Simulación de llamada a servicio externo
            if "error" in endpoint.lower():
                raise ConnectionError("Servicio no disponible")
            
            # Éxito - actualizar estado
            self.estado_error.ultimo_exito = datetime.now()
            self.estado_error.en_modo_degradado = False
            
            resultado = {
                "exito": True,
                "endpoint": endpoint,
                "datos_enviados": datos or {},
                "timestamp": datetime.now().isoformat(),
                "respuesta": "Operación completada exitosamente"
            }
            
            return json.dumps(resultado, ensure_ascii=False, indent=2)
            
        except Exception as e:
            self._registrar_error(e, endpoint)
            
            # Respuesta de error detallada
            respuesta_error = {
                "error": True,
                "endpoint": endpoint,
                "mensaje": str(e),
                "timestamp": datetime.now().isoformat(),
                "errores_recientes": len(self.estado_error.errores_recientes),
                "proximo_intento_sugerido": (datetime.now() + timedelta(minutes=1)).isoformat()
            }
            
            return json.dumps(respuesta_error, ensure_ascii=False, indent=2)
    
    async def _arun(self, endpoint: str, datos: Optional[Dict] = None) -> str:
        return self._run(endpoint, datos)

Integración con modelos de chat y manejo de errores

Las herramientas con manejo avanzado de errores se integran naturalmente con los modelos de chat, proporcionando información rica que permite al modelo tomar decisiones informadas:

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# Crear instancias de herramientas con manejo de errores
consulta_web = ConsultaWeb(max_reintentos=2)
procesador = ProcesadorArchivos()
servicio = ServicioExterno()

# Configurar modelo con herramientas
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_con_herramientas = llm.bind_tools([consulta_web, procesador, servicio])

# Mensaje del sistema que instruye sobre manejo de errores
sistema = SystemMessage(content="""
Eres un asistente que usa herramientas con manejo robusto de errores.
Cuando una herramienta devuelva un error:
1. Analiza el tipo de error y si es reintentable
2. Sugiere alternativas cuando sea posible
3. Explica al usuario qué ocurrió de manera clara
""")

# Ejemplo de uso con manejo de errores
respuesta = llm_con_herramientas.invoke([
    sistema,
    HumanMessage(content="Intenta leer el archivo /ruta/inexistente.txt")
])

print(f"Respuesta con manejo de errores: {respuesta}")

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