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
Guarda tu progreso
Inicia sesión para no perder tu progreso y accede a miles de tutoriales, ejercicios prácticos y nuestro asistente de IA.
Más de 25.000 desarrolladores ya confían en CertiDevs
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.
Aprendizajes de esta lección
- Comprender cuándo es necesario heredar de BaseTool en lugar de usar el decorador @tool.
- Aprender a implementar herramientas personalizadas con estado interno y lógica avanzada.
- Aplicar validaciones complejas mediante schemas personalizados con Pydantic.
- Implementar manejo robusto y personalizado de errores en herramientas.
- Integrar herramientas basadas en BaseTool con modelos de chat y LCEL en LangChain.
Completa LangChain y certifícate
Únete a nuestra plataforma y accede a miles de tutoriales, ejercicios prácticos, proyectos reales y nuestro asistente de IA personalizado para acelerar tu aprendizaje.
Asistente IA
Resuelve dudas al instante
Ejercicios
Practica con proyectos reales
Certificados
Valida tus conocimientos
Más de 25.000 desarrolladores ya se han certificado con CertiDevs