Error Handling y Troubleshooting

Descripción de la cápsula

Las tools fallan. Las APIs externas se caen, los modelos inventan nombres de tools que no existen, los argumentos llegan con el tipo incorrecto, y los timeouts ocurren justo cuando más necesitas la respuesta. Si tu sistema de tool calling no maneja estos fallos, una sola tool rota puede colapsar toda la aplicación.

En esta cápsula aprenderás a construir tool calling robusto que sobrevive a errores reales. Verás los cinco tipos de errores más comunes en tool calling, cómo validar argumentos antes de ejecutar una tool, cómo envolver ejecuciones en try/except con retry logic y backoff exponencial, y cómo reportar errores de forma estructurada con ToolException. También aprenderás a debuggear tool calls inspeccionando los mensajes que el modelo genera — una habilidad que usarás constantemente cuando algo no funcione como esperas.

La diferencia entre un prototipo y un sistema de producción no es la funcionalidad: es cómo maneja los fallos. Después de esta cápsula, tus tools manejarán errores como lo haría un servicio profesional.

Los 5 errores más comunes en tool calling

Antes de aprender a manejar errores, necesitas reconocerlos. Estos son los cinco que encontrarás con más frecuencia:

1. Tool not found (el modelo alucina un nombre)

El modelo decide llamar una tool que no existe. Esto ocurre cuando el modelo "inventa" un nombre de tool basándose en su entrenamiento general, en vez de limitarse a las tools que le vinculaste con bind_tools.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Obtiene el clima actual de una ciudad."""
    return f"Soleado, 25°C en {city}"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([get_weather])

response = model_with_tools.invoke("Busca información sobre Python en Google")
print(response.tool_calls)
# Output posible (el modelo puede intentar llamar una tool inexistente):
# [{'name': 'google_search', 'args': {'query': 'Python'}, 'id': 'call_abc123'}]
# ↑ 'google_search' NO existe en nuestras tools

Por qué ocurre: El modelo sabe que existen tools de búsqueda por su entrenamiento. Aunque solo le vinculaste get_weather, a veces genera un tool call hacia una tool que "cree" que debería existir. Esto es más frecuente con modelos menos capaces o cuando el prompt no es específico.

2. Argumentos inválidos (tipos incorrectos)

El modelo pasa argumentos que no coinciden con el schema de la tool. Por ejemplo, envía un string donde se espera un número, o un solo valor donde se espera una lista.

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):
    operation: str = Field(description="Operación: 'add', 'subtract', 'multiply', 'divide'")
    a: float = Field(description="Primer número")
    b: float = Field(description="Segundo número")

@tool(args_schema=CalculatorInput)
def calculator(operation: str, a: float, b: float) -> str:
    """Realiza operaciones matemáticas básicas."""
    operations = {
        "add": a + b,
        "subtract": a - b,
        "multiply": a * b,
        "divide": a / b if b != 0 else "Error: división por cero",
    }
    if operation not in operations:
        return f"Operación '{operation}' no soportada"
    return str(operations[operation])

# El modelo podría pasar:
# {'operation': 'sum', 'a': 'diez', 'b': 5}
# 'sum' no es una operación válida, y 'diez' no es un float

3. Timeout en la ejecución (API externa lenta)

La tool llama a una API externa que tarda demasiado o no responde. Sin timeout, tu aplicación se queda esperando indefinidamente.

4. Formato de retorno inesperado

La tool retorna un tipo de dato que el flujo no esperaba — por ejemplo, None en vez de un string, o un diccionario cuando se esperaba texto plano. El modelo necesita recibir un string para generar su respuesta final.

5. Rate limiting en APIs externas

La API externa que tu tool consulta bloquea las solicitudes por exceder el límite de velocidad. A diferencia del rate limiting del modelo (que viste en la cápsula 07 del Módulo 1), este ocurre en el servicio que tu tool llama.

Validar argumentos antes de ejecutar

La primera línea de defensa es verificar que los argumentos son válidos antes de ejecutar la lógica de la tool. Esto evita errores innecesarios y da mensajes claros al modelo para que se autocorrija.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool

VALID_TICKERS = {"AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"}

@tool
def get_stock_price(ticker: str) -> str:
    """Obtiene el precio actual de una acción. Tickers válidos: AAPL, GOOGL, MSFT, AMZN, TSLA."""
    ticker = ticker.upper().strip()

    if not ticker:
        return "Error: ticker vacío. Proporciona un símbolo como 'AAPL'."

    if ticker not in VALID_TICKERS:
        return f"Error: ticker '{ticker}' no reconocido. Tickers válidos: {', '.join(sorted(VALID_TICKERS))}"

    prices = {"AAPL": 185.50, "GOOGL": 142.30, "MSFT": 420.80, "AMZN": 195.20, "TSLA": 248.60}
    return f"{ticker}: ${prices[ticker]:.2f} USD"

result = get_stock_price.invoke({"ticker": "AAPL"})
print(result)
# Output esperado: AAPL: $185.50 USD

result = get_stock_price.invoke({"ticker": "INVALID"})
print(result)
# Output esperado: Error: ticker 'INVALID' no reconocido. Tickers válidos: AAPL, AMZN, GOOGL, MSFT, TSLA

Punto clave: Cuando retornas un mensaje de error descriptivo (en vez de lanzar una excepción), el modelo recibe ese mensaje como ToolMessage y puede autocorregirse — por ejemplo, pidiendo al usuario que clarifique el ticker.

Try/except para tool execution

No importa cuánto valides los argumentos: la ejecución puede fallar por razones externas (red, API caída, datos corruptos). Envolver la ejecución en try/except es obligatorio para cualquier tool que interactúe con servicios externos.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool
import json

@tool
def fetch_user_data(user_id: str) -> str:
    """Obtiene datos de un usuario por su ID."""
    try:
        if not user_id.isdigit():
            return f"Error: user_id debe ser numérico, recibí '{user_id}'"

        users = {
            "1": {"name": "Ana García", "email": "ana@example.com", "role": "admin"},
            "2": {"name": "Carlos López", "email": "carlos@example.com", "role": "user"},
        }

        if user_id not in users:
            return f"Error: usuario con ID '{user_id}' no encontrado"

        return json.dumps(users[user_id], ensure_ascii=False)

    except Exception as e:
        return f"Error inesperado al buscar usuario: {type(e).__name__}: {e}"

print(fetch_user_data.invoke({"user_id": "1"}))
# Output esperado: {"name": "Ana García", "email": "ana@example.com", "role": "admin"}

print(fetch_user_data.invoke({"user_id": "999"}))
# Output esperado: Error: usuario con ID '999' no encontrado

print(fetch_user_data.invoke({"user_id": "abc"}))
# Output esperado: Error: user_id debe ser numérico, recibí 'abc'

Patrón: tool wrapper defensivo

Cuando tienes muchas tools, puedes crear un wrapper que aplique try/except automáticamente:

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool

def safe_tool_execute(tool_fn, args: dict) -> str:
    """Ejecuta una tool con manejo de errores estándar."""
    try:
        result = tool_fn.invoke(args)
        if result is None:
            return "La herramienta no retornó resultado."
        return str(result)
    except TypeError as e:
        return f"Error de argumentos: {e}"
    except TimeoutError:
        return "Error: la operación tardó demasiado. Intenta de nuevo."
    except Exception as e:
        return f"Error ejecutando {tool_fn.name}: {type(e).__name__}: {e}"

@tool
def divide(a: float, b: float) -> str:
    """Divide dos números."""
    return str(a / b)

print(safe_tool_execute(divide, {"a": 10, "b": 3}))
# Output esperado: 3.3333333333333335

print(safe_tool_execute(divide, {"a": 10, "b": 0}))
# Output esperado: Error ejecutando divide: ZeroDivisionError: float division by zero

Retry logic con backoff exponencial

Los errores transitorios — timeouts de red, rate limits temporales, errores 503 del servidor — se resuelven solos si esperas un momento y reintentes. Pero reintentar inmediatamente suele empeorar las cosas (más carga en un servidor ya sobrecargado). El patrón estándar es backoff exponencial: esperar 1s, luego 2s, luego 4s.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool
import time
import random

def execute_with_retry(tool_fn, args: dict, max_retries: int = 3, base_delay: float = 1.0) -> str:
    """Ejecuta un tool con retry y backoff exponencial."""
    for attempt in range(max_retries):
        try:
            result = tool_fn.invoke(args)
            return result
        except Exception as e:
            is_last_attempt = attempt == max_retries - 1
            if is_last_attempt:
                return f"Error después de {max_retries} intentos: {type(e).__name__}: {e}"

            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            print(f"  ⚠️ Intento {attempt + 1} falló: {e}. Reintentando en {delay:.1f}s...")
            time.sleep(delay)

    return "Error: se agotaron los reintentos"

@tool
def unreliable_api(query: str) -> str:
    """Simula una API que falla intermitentemente."""
    if random.random() < 0.7:
        raise ConnectionError("Servidor no disponible")
    return f"Resultado para: {query}"

result = execute_with_retry(unreliable_api, {"query": "test"})
print(f"Resultado: {result}")
# Output esperado (varía por aleatoriedad):
#   ⚠️ Intento 1 falló: Servidor no disponible. Reintentando en 1.3s...
#   ⚠️ Intento 2 falló: Servidor no disponible. Reintentando en 2.2s...
# Resultado: Resultado para: test

¿Por qué agregar jitter (ruido aleatorio)?

El random.uniform(0, 0.5) añade un pequeño delay aleatorio a cada retry. Sin jitter, si 100 clientes fallan al mismo tiempo y todos reintentan en exactamente 1s, el servidor recibe 100 solicitudes simultáneas de nuevo. Con jitter, las solicitudes se distribuyen en una ventana temporal, dando al servidor tiempo para recuperarse.

Retry selectivo: solo errores transitorios

No todos los errores merecen retry. Un error de "ticker no encontrado" no se va a resolver reintentando. Solo deberías reintentar errores transitorios:

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool, ToolException
import time
import random

TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError)

def execute_with_smart_retry(tool_fn, args: dict, max_retries: int = 3) -> str:
    """Retry solo para errores transitorios. Errores permanentes fallan inmediatamente."""
    for attempt in range(max_retries):
        try:
            return tool_fn.invoke(args)
        except TRANSIENT_ERRORS as e:
            if attempt == max_retries - 1:
                return f"Error transitorio persistente después de {max_retries} intentos: {e}"
            delay = (2 ** attempt) + random.uniform(0, 0.5)
            print(f"  ⚠️ Error transitorio (intento {attempt + 1}): {e}. Retry en {delay:.1f}s...")
            time.sleep(delay)
        except ToolException as e:
            return f"Error de tool (no reintentable): {e}"
        except Exception as e:
            return f"Error permanente: {type(e).__name__}: {e}"

    return "Error: se agotaron los reintentos"

Graceful degradation: qué hacer cuando una tool falla permanentemente

A veces la tool simplemente no va a funcionar — la API está caída desde hace horas, la key expiró, o el servicio dejó de existir. En estos casos necesitas una estrategia de degradación graceful: dar una respuesta útil al usuario aunque la tool no funcione.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool

@tool
def get_weather_with_fallback(city: str) -> str:
    """Obtiene el clima actual de una ciudad."""
    try:
        raise ConnectionError("API del clima no disponible")
    except ConnectionError:
        return (
            f"⚠️ No pude obtener el clima en tiempo real para {city}. "
            f"El servicio de clima no está disponible en este momento. "
            f"Puedes consultar directamente en https://weather.com"
        )

result = get_weather_with_fallback.invoke({"city": "Madrid"})
print(result)
# Output esperado:
# ⚠️ No pude obtener el clima en tiempo real para Madrid. El servicio de clima
# no está disponible en este momento. Puedes consultar directamente en
# https://weather.com

Comparación: fail-fast vs retry vs fallback

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool
import time
import random

@tool
def robust_weather(city: str) -> str:
    """Obtiene el clima con estrategia completa de error handling."""
    if not city or len(city) < 2:
        return f"Error: ciudad inválida '{city}'. Proporciona un nombre de ciudad válido."

    weather_data = {"Madrid": "22°C, soleado", "London": "14°C, nublado", "Tokyo": "28°C, húmedo"}

    max_retries = 3
    for attempt in range(max_retries):
        try:
            if random.random() < 0.3:
                raise ConnectionError("Timeout de red")

            if city in weather_data:
                return f"Clima en {city}: {weather_data[city]}"
            else:
                return f"No tengo datos de clima para {city}. Ciudades disponibles: {', '.join(weather_data.keys())}"

        except ConnectionError as e:
            if attempt < max_retries - 1:
                time.sleep(0.1 * (2 ** attempt))
                continue
            return f"⚠️ No pude conectar con el servicio de clima después de {max_retries} intentos. Intenta más tarde."

result = robust_weather.invoke({"city": "Madrid"})
print(result)
# Output esperado: Clima en Madrid: 22°C, soleado

ToolException para error reporting estructurado

LangChain provee ToolException como una forma estructurada de reportar errores en tools. Cuando una tool lanza ToolException, LangChain la captura y la convierte en un ToolMessage con el error — permitiendo que el modelo vea el error y decida qué hacer (reintentar, pedir clarificación, o usar otra tool).

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool, ToolException

@tool(handle_tool_error=True)
def get_stock_price(ticker: str) -> str:
    """Obtiene el precio actual de una acción."""
    prices = {"AAPL": 185.50, "GOOGL": 142.30, "MSFT": 420.80}

    ticker = ticker.upper().strip()
    if ticker not in prices:
        raise ToolException(
            f"Ticker '{ticker}' no encontrado. "
            f"Tickers disponibles: {', '.join(sorted(prices.keys()))}. "
            f"Verifica el símbolo e intenta de nuevo."
        )

    return f"{ticker}: ${prices[ticker]:.2f} USD"

print(get_stock_price.invoke({"ticker": "AAPL"}))
# Output esperado: AAPL: $185.50 USD

print(get_stock_price.invoke({"ticker": "INVALID"}))
# Output esperado: Ticker 'INVALID' no encontrado. Tickers disponibles: AAPL, GOOGL, MSFT.
# Verifica el símbolo e intenta de nuevo.

handle_tool_error: controlar cómo se reporta el error

El parámetro handle_tool_error en @tool controla qué pasa cuando la tool lanza ToolException:

from langchain_core.tools import tool, ToolException

# Opción 1: handle_tool_error=True → retorna str(exception) como ToolMessage
@tool(handle_tool_error=True)
def tool_v1(x: str) -> str:
    """Demo tool."""
    raise ToolException("Algo falló")

# Opción 2: handle_tool_error="mensaje custom" → retorna ese mensaje
@tool(handle_tool_error="La herramienta no está disponible. Intenta reformular tu pregunta.")
def tool_v2(x: str) -> str:
    """Demo tool."""
    raise ToolException("Error interno")

# Opción 3: handle_tool_error=callable → procesa el error con una función
def custom_handler(error: ToolException) -> str:
    return f"⚠️ Error controlado: {error}. Por favor intenta con otros parámetros."

@tool(handle_tool_error=custom_handler)
def tool_v3(x: str) -> str:
    """Demo tool."""
    raise ToolException("Dato no disponible")

print(tool_v1.invoke({"x": "test"}))
# Output esperado: Algo falló

print(tool_v2.invoke({"x": "test"}))
# Output esperado: La herramienta no está disponible. Intenta reformular tu pregunta.

print(tool_v3.invoke({"x": "test"}))
# Output esperado: ⚠️ Error controlado: Dato no disponible. Por favor intenta con otros parámetros.

ToolException vs return de string de error

¿Cuándo usar ToolException y cuándo simplemente retornar un string con el error?

Regla práctica: Usa return "Error: ..." durante desarrollo. Migra a ToolException cuando necesites monitoreo y logging diferenciado entre resultados exitosos y errores.

Debugging de tool calls: inspeccionar mensajes

Cuando algo no funciona — el modelo no llama la tool que esperas, pasa argumentos incorrectos, o el resultado no se integra bien — necesitas inspeccionar los mensajes. Los tool calls viven dentro del AIMessage que genera el modelo.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def calculate(expression: str) -> str:
    """Evalúa una expresión matemática."""
    try:
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return f"Error: expresión contiene caracteres no permitidos"
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([calculate])

response = model_with_tools.invoke("¿Cuánto es 15% de 240?")

print("=== Inspección del AIMessage ===")
print(f"Content: '{response.content}'")
print(f"Tool calls: {response.tool_calls}")
print(f"Número de tool calls: {len(response.tool_calls)}")

if response.tool_calls:
    tc = response.tool_calls[0]
    print(f"\n=== Detalle del tool call ===")
    print(f"  Name: {tc['name']}")
    print(f"  Args: {tc['args']}")
    print(f"  ID:   {tc['id']}")

# Output esperado:
# === Inspección del AIMessage ===
# Content: ''
# Tool calls: [{'name': 'calculate', 'args': {'expression': '240 * 0.15'}, 'id': 'call_abc123'}]
# Número de tool calls: 1
#
# === Detalle del tool call ===
#   Name: calculate
#   Args: {'expression': '240 * 0.15'}
#   ID:   call_abc123

Inspeccionar el flujo completo

Para debuggear el loop completo (model → tool → model), imprime cada mensaje de la conversación:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def calculate(expression: str) -> str:
    """Evalúa una expresión matemática."""
    allowed_chars = set("0123456789+-*/.() ")
    if not all(c in allowed_chars for c in expression):
        return f"Error: expresión contiene caracteres no permitidos"
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([calculate])

messages = [HumanMessage(content="¿Cuánto es 15% de 240?")]

response = model_with_tools.invoke(messages)
messages.append(response)

if response.tool_calls:
    for tc in response.tool_calls:
        result = calculate.invoke(tc["args"])
        messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))

    final = model_with_tools.invoke(messages)
    messages.append(final)

print("=== Flujo completo de mensajes ===")
for i, msg in enumerate(messages):
    msg_type = type(msg).__name__
    if hasattr(msg, "tool_calls") and msg.tool_calls:
        tools = [tc["name"] for tc in msg.tool_calls]
        print(f"  [{i}] {msg_type}: tool_calls={tools}")
    elif hasattr(msg, "tool_call_id"):
        print(f"  [{i}] {msg_type}: '{msg.content}' (id={msg.tool_call_id})")
    else:
        print(f"  [{i}] {msg_type}: '{msg.content[:80]}'")

# Output esperado:
# === Flujo completo de mensajes ===
#   [0] HumanMessage: '¿Cuánto es 15% de 240?'
#   [1] AIMessage: tool_calls=['calculate']
#   [2] ToolMessage: '36.0' (id=call_abc123)
#   [3] AIMessage: 'El 15% de 240 es 36.'

Tool execution loop con error handling integrado

El tool execution loop que aprendiste en la cápsula 04 asumía que las tools siempre funcionan. Aquí está la versión robusta que maneja errores en cada paso:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool, ToolException
from langchain_core.messages import HumanMessage, ToolMessage
import time
import random

@tool
def get_weather(city: str) -> str:
    """Obtiene el clima actual de una ciudad."""
    weather = {"Madrid": "22°C, soleado", "London": "14°C, nublado", "Tokyo": "28°C, húmedo"}
    city_normalized = city.strip().title()
    if city_normalized not in weather:
        raise ToolException(f"Ciudad '{city}' no encontrada. Disponibles: {', '.join(weather.keys())}")
    return f"Clima en {city_normalized}: {weather[city_normalized]}"

@tool
def calculate(expression: str) -> str:
    """Evalúa una expresión matemática simple."""
    allowed_chars = set("0123456789+-*/.() ")
    if not all(c in allowed_chars for c in expression):
        raise ToolException(f"Expresión inválida: caracteres no permitidos en '{expression}'")
    try:
        return str(eval(expression))
    except Exception as e:
        raise ToolException(f"No pude evaluar '{expression}': {e}")

tools = [get_weather, calculate]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

def execute_tool_call(tool_call: dict, max_retries: int = 2) -> str:
    """Ejecuta un tool call individual con validación y retry."""
    tool_name = tool_call["name"]
    tool_args = tool_call["args"]
    tool_id = tool_call["id"]

    if tool_name not in tools_by_name:
        return f"Error: herramienta '{tool_name}' no existe. Herramientas disponibles: {', '.join(tools_by_name.keys())}"

    selected_tool = tools_by_name[tool_name]

    for attempt in range(max_retries):
        try:
            result = selected_tool.invoke(tool_args)
            return result
        except ToolException as e:
            return str(e)
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(0.5 * (2 ** attempt))
                continue
            return f"Error después de {max_retries} intentos en '{tool_name}': {e}"

    return f"Error inesperado en '{tool_name}'"

def run_tool_loop(question: str, max_iterations: int = 5) -> str:
    """Ejecuta el loop completo model→tool→model con error handling."""
    messages = [HumanMessage(content=question)]

    for iteration in range(max_iterations):
        response = model_with_tools.invoke(messages)
        messages.append(response)

        if not response.tool_calls:
            return response.content

        print(f"  Iteración {iteration + 1}: {len(response.tool_calls)} tool call(s)")

        for tc in response.tool_calls:
            result = execute_tool_call(tc)
            print(f"    → {tc['name']}({tc['args']}) = {result[:60]}")
            messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))

    return "Se alcanzó el límite de iteraciones. El modelo no generó respuesta final."

answer = run_tool_loop("¿Cuánto es 100 * 1.16 y qué clima hace en Madrid?")
print(f"\nRespuesta: {answer}")
# Output esperado:
#   Iteración 1: 2 tool call(s)
#     → calculate({'expression': '100 * 1.16'}) = 116.0
#     → get_weather({'city': 'Madrid'}) = Clima en Madrid: 22°C, soleado
#
# Respuesta: 100 × 1.16 = 116.0. En Madrid hace 22°C y está soleado.

Conexión con el proyecto

En el proyecto de este módulo (Cápsula 08), construirás un asistente con herramientas externas que integra todo lo que aprendiste sobre error handling:

• Validación de argumentos: Cada tool del asistente (weather, calculator, web search) valida sus inputs antes de ejecutar
• Try/except defensivo: Todas las tools envuelven su lógica en try/except para que un fallo no colapse el sistema
• Graceful degradation: Cuando una tool falla, el asistente informa al usuario y sigue funcionando con las demás tools
• Tool execution loop robusto: El loop que implementarás maneja tools inexistentes, errores de ejecución, y límites de iteraciones

Los patrones de esta cápsula son la base para que tu asistente sea confiable, no solo funcional.

Troubleshooting

Problema 1: El modelo llama una tool que no existe

KeyError: 'web_search'

Causa: El modelo genera un tool_call con un nombre que no está en tu diccionario de tools. Esto pasa cuando el modelo "alucina" un nombre de tool.

Solución: Siempre valida que el nombre existe antes de ejecutar:

if tool_name not in tools_by_name:
    error_msg = f"Tool '{tool_name}' no existe. Disponibles: {list(tools_by_name.keys())}"
    messages.append(ToolMessage(content=error_msg, tool_call_id=tool_call["id"]))

Al retornar el error como ToolMessage, el modelo lo ve y puede autocorregirse en la siguiente iteración.

Problema 2: ToolException no se captura y crashea el programa

langchain_core.tools.base.ToolException: Ticker 'XYZ' no encontrado

Causa: Lanzas ToolException pero no configuraste handle_tool_error=True en el decorador @tool, ni la capturas en tu try/except.

Solución: Agrega handle_tool_error=True al decorador o captura ToolException explícitamente:

@tool(handle_tool_error=True)
def my_tool(x: str) -> str:
    """Mi tool."""
    raise ToolException("Error controlado")

# O captura en tu loop:
try:
    result = tool_fn.invoke(args)
except ToolException as e:
    result = str(e)

Problema 3: El retry loop no espera entre intentos

Causa: Usas time.sleep(0) o el delay no escala. Verifica que el cálculo de backoff es correcto: delay = base * (2 ** attempt) da 1s, 2s, 4s para base=1.

Solución:

delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
print(f"Esperando {delay:.1f}s antes de reintentar...")
time.sleep(delay)

Problema 4: ToolMessage sin tool_call_id causa error

ValueError: ToolMessage must have a tool_call_id

Causa: Al crear un ToolMessage, no pasaste el tool_call_id que vino en el tool_call original.

Solución: Siempre pasa el ID del tool call:

for tc in response.tool_calls:
    result = execute_tool(tc)
    messages.append(ToolMessage(
        content=result,
        tool_call_id=tc["id"]    # ← obligatorio
    ))

Problema 5: El modelo entra en loop infinito de tool calls

Causa: El modelo sigue llamando tools sin generar una respuesta final. Esto pasa cuando la respuesta de la tool no satisface al modelo y sigue intentando.

Solución: Limita las iteraciones del loop y agrega un mensaje de corte:

def run_tool_loop(question, max_iterations=5):
    for iteration in range(max_iterations):
        response = model_with_tools.invoke(messages)
        if not response.tool_calls:
            return response.content
        # ... ejecutar tools ...
    return "Límite de iteraciones alcanzado."

Problema 6: Los errores de tool no llegan al modelo

Causa: Capturas la excepción pero no la envías como ToolMessage. El modelo espera una respuesta para cada tool call; si no la recibe, el flujo se rompe.

Solución: Siempre retorna un ToolMessage por cada tool call, incluso si fue un error:

try:
    result = tool_fn.invoke(args)
except Exception as e:
    result = f"Error: {e}"

messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))

Problema 7: handle_tool_error con función custom no funciona

Causa: Tu función custom no tiene la firma correcta. Debe recibir ToolException y retornar str.

Solución:

def my_handler(error: ToolException) -> str:
    return f"Error manejado: {error}"

@tool(handle_tool_error=my_handler)
def my_tool(x: str) -> str:
    """Mi tool."""
    raise ToolException("algo falló")

Ejercicios

Ejercicio 1: Validación defensiva de argumentos (Básico)

Crea una tool convert_temperature que convierta entre Celsius y Fahrenheit. Debe validar que la escala sea "C" o "F", que la temperatura sea un número razonable (-100 a 1000), y retornar mensajes de error claros.

from langchain_core.tools import tool

@tool
def convert_temperature(value: float, from_scale: str) -> str:
    """Convierte temperatura entre Celsius y Fahrenheit."""
    from_scale = from_scale.upper().strip()

    if from_scale not in ("C", "F"):
        return f"Error: escala '{from_scale}' no válida. Usa 'C' (Celsius) o 'F' (Fahrenheit)."

    if not -100 <= value <= 1000:
        return f"Error: temperatura {value} fuera de rango razonable (-100 a 1000)."

    if from_scale == "C":
        result = (value * 9/5) + 32
        return f"{value}°C = {result:.1f}°F"
    else:
        result = (value - 32) * 5/9
        return f"{value}°F = {result:.1f}°C"

print(convert_temperature.invoke({"value": 100, "from_scale": "C"}))
print(convert_temperature.invoke({"value": 72, "from_scale": "F"}))
print(convert_temperature.invoke({"value": 25, "from_scale": "K"}))
print(convert_temperature.invoke({"value": 5000, "from_scale": "C"}))
# Output esperado:
# 100°C = 212.0°F
# 72°F = 22.2°C
# Error: escala 'K' no válida. Usa 'C' (Celsius) o 'F' (Fahrenheit).
# Error: temperatura 5000 fuera de rango razonable (-100 a 1000).

Ejercicio 2: Tool con ToolException y handler custom (Básico)

Crea una tool lookup_country que busque información de un país por nombre. Usa ToolException para errores y un handler custom que formatee el error de forma amigable.

from langchain_core.tools import tool, ToolException

def friendly_error_handler(error: ToolException) -> str:
    return f"🔍 No pude completar la búsqueda: {error}. Intenta con otro país."

@tool(handle_tool_error=friendly_error_handler)
def lookup_country(country: str) -> str:
    """Busca información básica de un país."""
    countries = {
        "México": {"capital": "Ciudad de México", "población": "130M", "idioma": "Español"},
        "España": {"capital": "Madrid", "población": "47M", "idioma": "Español"},
        "Japón": {"capital": "Tokio", "población": "125M", "idioma": "Japonés"},
    }

    country_normalized = country.strip().title()
    if country_normalized not in countries:
        available = ", ".join(sorted(countries.keys()))
        raise ToolException(f"País '{country}' no encontrado. Disponibles: {available}")

    info = countries[country_normalized]
    return f"{country_normalized} — Capital: {info['capital']}, Población: {info['población']}, Idioma: {info['idioma']}"

print(lookup_country.invoke({"country": "México"}))
print(lookup_country.invoke({"country": "Francia"}))
# Output esperado:
# México — Capital: Ciudad de México, Población: 130M, Idioma: Español
# 🔍 No pude completar la búsqueda: País 'Francia' no encontrado. Disponibles: España, Japón, México. Intenta con otro país.

Ejercicio 3: Retry con backoff para API inestable (Medio)

Crea una función resilient_invoke que ejecute cualquier tool con retry y backoff exponencial. Debe aceptar max_retries y base_delay, imprimir cada intento, y distinguir entre errores transitorios (retry) y errores permanentes (fail-fast).

from langchain_core.tools import tool, ToolException
import time
import random

TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError)

def resilient_invoke(tool_fn, args: dict, max_retries: int = 3, base_delay: float = 1.0) -> dict:
    """Ejecuta una tool con retry inteligente.
    Retorna dict con resultado, intentos, y tiempo total.
    """
    start = time.time()

    for attempt in range(max_retries):
        try:
            result = tool_fn.invoke(args)
            return {
                "success": True,
                "result": result,
                "attempts": attempt + 1,
                "elapsed_ms": (time.time() - start) * 1000,
            }
        except TRANSIENT_ERRORS as e:
            if attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 0.3)
                print(f"  ⚠️ Intento {attempt + 1}/{max_retries}: {type(e).__name__}. Retry en {delay:.1f}s...")
                time.sleep(delay)
            else:
                return {
                    "success": False,
                    "result": f"Error transitorio persistente: {e}",
                    "attempts": max_retries,
                    "elapsed_ms": (time.time() - start) * 1000,
                }
        except ToolException as e:
            return {
                "success": False,
                "result": f"Error de tool: {e}",
                "attempts": attempt + 1,
                "elapsed_ms": (time.time() - start) * 1000,
            }
        except Exception as e:
            return {
                "success": False,
                "result": f"Error permanente: {type(e).__name__}: {e}",
                "attempts": attempt + 1,
                "elapsed_ms": (time.time() - start) * 1000,
            }

call_count = 0

@tool
def flaky_api(query: str) -> str:
    """Simula una API que falla las primeras 2 veces."""
    global call_count
    call_count += 1
    if call_count <= 2:
        raise ConnectionError(f"Timeout (intento interno #{call_count})")
    return f"Resultado exitoso para: {query}"

call_count = 0
result = resilient_invoke(flaky_api, {"query": "test"}, max_retries=3, base_delay=0.1)
print(f"\nResultado: {result}")
# Output esperado:
#   ⚠️ Intento 1/3: ConnectionError. Retry en 0.1s...
#   ⚠️ Intento 2/3: ConnectionError. Retry en 0.3s...
#
# Resultado: {'success': True, 'result': 'Resultado exitoso para: test', 'attempts': 3, 'elapsed_ms': 452.3}

Ejercicio 4: Debug inspector para tool calls (Medio)

Crea una función debug_tool_loop que ejecute el loop completo model→tool→model e imprima un reporte detallado de cada paso: tipo de mensaje, contenido, tool calls, y tiempos.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
import time

@tool
def multiply(a: float, b: float) -> str:
    """Multiplica dos números."""
    return str(a * b)

@tool
def add(a: float, b: float) -> str:
    """Suma dos números."""
    return str(a + b)

tools = [multiply, add]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

def debug_tool_loop(question: str, max_iterations: int = 5) -> str:
    """Tool loop con debug detallado de cada paso."""
    messages = [HumanMessage(content=question)]
    print(f"{'=' * 60}")
    print(f"DEBUG TOOL LOOP")
    print(f"Pregunta: {question}")
    print(f"Tools disponibles: {list(tools_by_name.keys())}")
    print(f"{'=' * 60}")

    for iteration in range(max_iterations):
        print(f"\n--- Iteración {iteration + 1} ---")

        start = time.time()
        response = model_with_tools.invoke(messages)
        model_time = (time.time() - start) * 1000

        print(f"  [MODEL] ({model_time:.0f}ms)")
        print(f"    Content: '{response.content[:100]}'" if response.content else "    Content: (vacío)")
        print(f"    Tool calls: {len(response.tool_calls)}")

        messages.append(response)

        if not response.tool_calls:
            print(f"\n{'=' * 60}")
            print(f"RESULTADO FINAL: {response.content[:200]}")
            print(f"{'=' * 60}")
            return response.content

        for i, tc in enumerate(response.tool_calls):
            print(f"\n  [TOOL CALL {i + 1}]")
            print(f"    Name: {tc['name']}")
            print(f"    Args: {tc['args']}")
            print(f"    ID:   {tc['id']}")

            start = time.time()
            if tc["name"] in tools_by_name:
                try:
                    result = tools_by_name[tc["name"]].invoke(tc["args"])
                except Exception as e:
                    result = f"Error: {e}"
            else:
                result = f"Error: tool '{tc['name']}' no existe"
            tool_time = (time.time() - start) * 1000

            print(f"    Result: {result}")
            print(f"    Time: {tool_time:.0f}ms")

            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

    return "Límite de iteraciones alcanzado"

result = debug_tool_loop("¿Cuánto es 7 * 8 + 3 * 4?")
# Output esperado:
# ============================================================
# DEBUG TOOL LOOP
# Pregunta: ¿Cuánto es 7 * 8 + 3 * 4?
# Tools disponibles: ['multiply', 'add']
# ============================================================
#
# --- Iteración 1 ---
#   [MODEL] (823ms)
#     Content: (vacío)
#     Tool calls: 2
#
#   [TOOL CALL 1]
#     Name: multiply
#     Args: {'a': 7.0, 'b': 8.0}
#     ID:   call_abc123
#     Result: 56.0
#     Time: 0ms
#
#   [TOOL CALL 2]
#     Name: multiply
#     Args: {'a': 3.0, 'b': 4.0}
#     ID:   call_def456
#     Result: 12.0
#     Time: 0ms
#
# --- Iteración 2 ---
#   [MODEL] (645ms)
#     Content: (vacío)
#     Tool calls: 1
#
#   [TOOL CALL 1]
#     Name: add
#     Args: {'a': 56.0, 'b': 12.0}
#     ID:   call_ghi789
#     Result: 68.0
#     Time: 0ms
#
# --- Iteración 3 ---
#   [MODEL] (412ms)
#     Content: '7 × 8 + 3 × 4 = 68'
#     Tool calls: 0
#
# ============================================================
# RESULTADO FINAL: 7 × 8 + 3 × 4 = 68
# ============================================================

Ejercicio 5: Sistema de tools con fallback entre fuentes (Difícil)

Crea una tool smart_search que busque información en tres "fuentes" (simuladas). Si la primera fuente falla, intenta la segunda. Si la segunda falla, intenta la tercera. Retorna el resultado de la primera fuente que funcione, junto con metadata de qué fuente respondió.

from langchain_core.tools import tool
import random

def source_api_primary(query: str) -> str:
    """Fuente principal — falla 50% del tiempo."""
    if random.random() < 0.5:
        raise ConnectionError("API primaria no disponible")
    return f"[Fuente primaria] Resultado para '{query}': información detallada y actualizada."

def source_api_secondary(query: str) -> str:
    """Fuente secundaria — falla 30% del tiempo."""
    if random.random() < 0.3:
        raise ConnectionError("API secundaria no disponible")
    return f"[Fuente secundaria] Resultado para '{query}': información general."

def source_cache(query: str) -> str:
    """Cache local — nunca falla pero datos pueden ser antiguos."""
    return f"[Cache local] Resultado para '{query}': datos de hace 24 horas."

@tool
def smart_search(query: str) -> str:
    """Busca información usando múltiples fuentes con fallback automático."""
    sources = [
        ("API Primaria", source_api_primary),
        ("API Secundaria", source_api_secondary),
        ("Cache Local", source_cache),
    ]

    errors = []

    for source_name, source_fn in sources:
        try:
            result = source_fn(query)
            if errors:
                fallback_info = f" (después de fallos en: {', '.join(errors)})"
            else:
                fallback_info = ""
            return f"{result}{fallback_info}"
        except Exception as e:
            errors.append(source_name)
            continue

    return f"Error: todas las fuentes fallaron — {', '.join(errors)}"

random.seed(42)
for i in range(5):
    result = smart_search.invoke({"query": f"Python tutorial"})
    print(f"Búsqueda {i + 1}: {result}")
    print()
# Output esperado (varía por seed):
# Búsqueda 1: [Fuente primaria] Resultado para 'Python tutorial': información detallada y actualizada.
#
# Búsqueda 2: [Fuente secundaria] Resultado para 'Python tutorial': información general. (después de fallos en: API Primaria)
#
# Búsqueda 3: [Cache local] Resultado para 'Python tutorial': datos de hace 24 horas. (después de fallos en: API Primaria, API Secundaria)
# ...

Ejercicio 6: Tool execution loop completo con error handling (Difícil)

Implementa una función safe_agent_loop que ejecute el loop model→tool→model con: validación de tool names, try/except por tool call, retry para errores transitorios, límite de iteraciones, y un resumen de ejecución al final (tools llamadas, errores, tiempo total).

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool, ToolException
from langchain_core.messages import HumanMessage, ToolMessage
import time
import random

@tool
def get_price(product: str) -> str:
    """Obtiene el precio de un producto."""
    prices = {"laptop": 999.99, "mouse": 29.99, "keyboard": 79.99, "monitor": 349.99}
    product = product.lower().strip()
    if product not in prices:
        raise ToolException(f"Producto '{product}' no encontrado. Disponibles: {', '.join(prices.keys())}")
    return f"{product}: ${prices[product]}"

@tool
def calculate_discount(price: float, percent: float) -> str:
    """Calcula el precio con descuento."""
    if percent < 0 or percent > 100:
        raise ToolException(f"Porcentaje {percent} inválido. Debe estar entre 0 y 100.")
    discount = price * (percent / 100)
    final = price - discount
    return f"Precio original: ${price:.2f}, Descuento: ${discount:.2f} ({percent}%), Precio final: ${final:.2f}"

tools = [get_price, calculate_discount]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

def safe_agent_loop(question: str, max_iterations: int = 5, max_retries: int = 2) -> dict:
    """Loop completo con error handling, retry, y métricas."""
    messages = [HumanMessage(content=question)]
    execution_log = []
    start_time = time.time()

    for iteration in range(max_iterations):
        response = model_with_tools.invoke(messages)
        messages.append(response)

        if not response.tool_calls:
            elapsed = (time.time() - start_time) * 1000
            return {
                "answer": response.content,
                "iterations": iteration + 1,
                "tool_calls": len(execution_log),
                "errors": sum(1 for log in execution_log if not log["success"]),
                "elapsed_ms": elapsed,
                "log": execution_log,
            }

        for tc in response.tool_calls:
            log_entry = {"tool": tc["name"], "args": tc["args"], "success": False, "result": ""}

            if tc["name"] not in tools_by_name:
                log_entry["result"] = f"Tool '{tc['name']}' no existe"
                execution_log.append(log_entry)
                messages.append(ToolMessage(
                    content=f"Error: herramienta '{tc['name']}' no disponible. Herramientas válidas: {', '.join(tools_by_name.keys())}",
                    tool_call_id=tc["id"],
                ))
                continue

            result = None
            for attempt in range(max_retries):
                try:
                    result = tools_by_name[tc["name"]].invoke(tc["args"])
                    log_entry["success"] = True
                    log_entry["result"] = result
                    break
                except ToolException as e:
                    log_entry["result"] = str(e)
                    result = str(e)
                    break
                except Exception as e:
                    if attempt < max_retries - 1:
                        time.sleep(0.1 * (2 ** attempt))
                    else:
                        log_entry["result"] = f"Error tras {max_retries} intentos: {e}"
                        result = log_entry["result"]

            execution_log.append(log_entry)
            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

    elapsed = (time.time() - start_time) * 1000
    return {
        "answer": "Límite de iteraciones alcanzado.",
        "iterations": max_iterations,
        "tool_calls": len(execution_log),
        "errors": sum(1 for log in execution_log if not log["success"]),
        "elapsed_ms": elapsed,
        "log": execution_log,
    }

result = safe_agent_loop("¿Cuánto cuesta un laptop con 15% de descuento?")

print(f"Respuesta: {result['answer']}")
print(f"Iteraciones: {result['iterations']}")
print(f"Tool calls: {result['tool_calls']}")
print(f"Errores: {result['errors']}")
print(f"Tiempo: {result['elapsed_ms']:.0f}ms")
print(f"\nLog de ejecución:")
for log in result["log"]:
    status = "✅" if log["success"] else "❌"
    print(f"  {status} {log['tool']}({log['args']}) → {log['result'][:60]}")
# Output esperado:
# Respuesta: Un laptop cuesta $999.99. Con un 15% de descuento, el precio final es $849.99.
# Iteraciones: 3
# Tool calls: 2
# Errores: 0
# Tiempo: 2845ms
#
# Log de ejecución:
#   ✅ get_price({'product': 'laptop'}) → laptop: $999.99
#   ✅ calculate_discount({'price': 999.99, 'percent': 15.0}) → Precio original: $999.99, Descuento: $150.00 (15.0%), P

Resumen

En esta cápsula aprendiste:

• Los 5 errores más comunes en tool calling: tool not found, argumentos inválidos, timeout, formato inesperado, rate limiting
• Validar argumentos antes de ejecutar evita errores innecesarios y da mensajes claros al modelo
• Try/except envuelve toda ejecución que interactúe con servicios externos
• Retry con backoff exponencial maneja errores transitorios: delay = base * (2 ** attempt)
• Jitter (ruido aleatorio) distribuye retries para no sobrecargar servidores
• Graceful degradation da respuestas útiles cuando una tool falla permanentemente
• ToolException reporta errores de forma estructurada; handle_tool_error controla cómo se presentan
• Inspeccionar mensajes (tool_calls, content, ToolMessage) es la forma de debuggear flujos de tool calling
• Solo reintentar errores transitorios (red, timeout); errores de negocio deben fallar inmediatamente

Próxima cápsula: Proyecto — construirás un asistente con herramientas externas que integra todo lo aprendido: tools con @tool, bind_tools, el tool execution loop, parallel calls, y el error handling robusto de esta cápsula.

Recursos adicionales

1. LangChain Tool Error Handling — Guía oficial de manejo de errores en tools
2. ToolException API Reference — Referencia de la clase ToolException
3. LangChain Tools How-To — Índice de guías how-to para tools
4. Exponential Backoff and Jitter (AWS) — Artículo clásico de AWS sobre backoff strategies
5. LangChain Tool Calling — Conceptos de tool calling en LangChain
6. Python Exception Handling Best Practices — Referencia oficial de manejo de excepciones en Python

Módulo 2 — LangChain & LangGraph: From Chains to Agents