Proyecto: Chat Multi-Proveedor con Fallback

Descripción del proyecto

En las siete cápsulas anteriores aprendiste a inicializar modelos con init_chat_model, configurar parámetros, ejecutar con invoke(), stream() y batch(), obtener respuestas tipadas con with_structured_output, procesar imágenes con modelos multimodales, y controlar costos con rate limiting y caching. Cada concepto lo viste por separado, con ejemplos aislados. Ahora vas a combinar todo en un sistema real.

En este proyecto construyes un chat multi-proveedor con fallback automático. El sistema intenta responder usando OpenAI. Si OpenAI falla — por un error de API, por exceder rate limits, o porque la API key no es válida — automáticamente intenta con Anthropic. Si Anthropic también falla, prueba con Google. Todo esto ocurre de forma transparente: el usuario ve su respuesta llegar token por token sin saber qué proveedor la generó.

Además, cada respuesta incluye metadata estructurada: qué proveedor respondió, qué modelo se usó, cuánto tardó en milisegundos, y cuántos tokens consumió. Esta metadata usa un modelo Pydantic — exactamente como aprendiste en la cápsula de Structured Output — pero aplicado a datos operacionales en vez de respuestas del LLM.

Este patrón de fallback no es académico. Servicios como AWS, Stripe y cualquier API crítica implementan fallback entre proveedores como práctica estándar. Al terminar este proyecto, tendrás un sistema funcional que puedes adaptar para cualquier aplicación que necesite resiliencia frente a fallos de proveedores de LLMs.

---

Objetivo del proyecto

Construir un chat interactivo en terminal que use múltiples proveedores de LLMs con fallback automático, streaming progresivo, y metadata estructurada por cada respuesta.

Al completar este proyecto:

• 🔧 Sabrás integrar múltiples proveedores (init_chat_model) en un solo sistema con fallback
• 🔧 Implementarás streaming con manejo de errores por proveedor
• 🔧 Capturarás metadata operacional (proveedor, latencia, tokens) usando modelos Pydantic
• 🔧 Tendrás un chat funcional en terminal que demuestra resiliencia real

---

Especificaciones técnicas

Stack tecnológico

Componente	Versión	Propósito
Python	3.11+	Runtime
LangChain	v1.2+	Framework de LLMs
langchain-openai	latest	Proveedor OpenAI
langchain-anthropic	latest	Proveedor Anthropic
langchain-google-genai	latest	Proveedor Google
pydantic	v2+	Modelo de metadata
python-dotenv	latest	Variables de entorno

Setup inicial

Antes de empezar, asegúrate de tener las dependencias instaladas y las API keys configuradas:

# Instalar dependencias
pip install langchain langchain-openai langchain-anthropic langchain-google-genai python-dotenv pydantic

Crea un archivo .env en la raíz de tu proyecto:

# .env
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=AI...

No necesitas las tres API keys para que el proyecto funcione. El sistema de fallback está diseñado precisamente para manejar proveedores no disponibles. Con al menos una API key válida, el chat funciona.

Estructura del proyecto

chat-multi-proveedor/
├── .env                  # API keys
├── chat.py               # Código principal (todo en un archivo)
└── requirements.txt      # Dependencias

# requirements.txt
langchain>=0.3.0
langchain-openai>=0.3.0
langchain-anthropic>=0.3.0
langchain-google-genai>=2.0.0
python-dotenv>=1.0.0
pydantic>=2.0.0

Para este mini-proyecto, todo el código va en un solo archivo chat.py. No necesitas estructura compleja — el objetivo es integrar conceptos, no diseñar arquitectura.

---

Paso 1: Configurar proveedores

El primer paso es definir los proveedores y crear una función que intente inicializar cada uno. Si un proveedor no tiene API key configurada o tiene otro problema, lo marca como no disponible pero no detiene el programa.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PROVIDERS = [
    {
        "id": "openai",
        "model_id": "openai:gpt-4.1-mini",
        "display_name": "OpenAI GPT-4.1 Mini",
    },
    {
        "id": "anthropic",
        "model_id": "anthropic:claude-sonnet-4-20250514",
        "display_name": "Anthropic Claude Sonnet 4",
    },
    {
        "id": "google",
        "model_id": "google_genai:gemini-2.0-flash",
        "display_name": "Google Gemini 2.0 Flash",
    },
]


def init_providers():
    """Inicializa todos los proveedores disponibles.
    Retorna un dict con los proveedores que se pudieron crear.
    """
    models = {}
    for provider in PROVIDERS:
        try:
            model = init_chat_model(
                provider["model_id"],
                temperature=0.7,
                max_tokens=1024,
            )
            models[provider["id"]] = {
                "model": model,
                "display_name": provider["display_name"],
                "model_id": provider["model_id"],
            }
            print(f"  ✅ {provider['display_name']}")
        except Exception as e:
            print(f"  ❌ {provider['display_name']}: {e}")
    return models

# Output esperado (con 3 API keys configuradas):
  ✅ OpenAI GPT-4.1 Mini
  ✅ Anthropic Claude Sonnet 4
  ✅ Google Gemini 2.0 Flash

# Output esperado (sin API key de Anthropic):
  ✅ OpenAI GPT-4.1 Mini
  ❌ Anthropic Claude Sonnet 4: Did not find anthropic_api_key...
  ✅ Google Gemini 2.0 Flash

Nota que init_chat_model puede fallar al crear el modelo si no encuentra la API key correspondiente — depende del proveedor. Algunos proveedores validan la key al inicializar, otros al hacer la primera llamada. El try/except maneja ambos casos.

La lista PROVIDERS define el orden de prioridad para el fallback. OpenAI se intenta primero, Anthropic segundo, Google tercero. Puedes cambiar el orden según tus preferencias o costos.

---

Paso 2: Implementar fallback

Enfoque manual con try/except

El enfoque manual te da control total: puedes saber exactamente qué proveedor respondió, medir latencia, y capturar tokens. Cada proveedor se intenta en orden hasta que uno responde exitosamente.

import time

def invoke_with_fallback(models, messages):
    """Intenta invoke() con cada proveedor en orden.
    Retorna (response, provider_id) del primer proveedor que funcione.
    """
    errors = []

    for provider_id, provider_info in models.items():
        try:
            start = time.time()
            response = provider_info["model"].invoke(messages)
            latency_ms = (time.time() - start) * 1000

            return response, provider_id, latency_ms

        except Exception as e:
            errors.append(f"{provider_id}: {e}")
            print(f"  ⚠️  Fallback: {provider_id} falló → intentando siguiente...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(
        f"Todos los proveedores fallaron:\n{error_detail}"
    )

Probemos el fallback:

from langchain_core.messages import HumanMessage, SystemMessage

models = init_providers()

messages = [
    SystemMessage(content="Responde en una frase corta."),
    HumanMessage(content="¿Qué es Python?"),
]

response, provider_id, latency = invoke_with_fallback(models, messages)
print(f"\nProveedor: {provider_id}")
print(f"Respuesta: {response.content}")
print(f"Latencia: {latency:.0f}ms")

# Output esperado (OpenAI disponible):
Proveedor: openai
Respuesta: Python es un lenguaje de programación interpretado, de alto nivel y propósito general.
Latencia: 823ms

# Output esperado (OpenAI falla → fallback):
  ⚠️  Fallback: openai falló → intentando siguiente...
Proveedor: anthropic
Respuesta: Python es un lenguaje de programación versátil, conocido por su sintaxis clara y legible.
Latencia: 1205ms

Enfoque con with_fallbacks() (built-in)

LangChain incluye un método with_fallbacks() que encadena modelos automáticamente. Es más conciso pero te da menos control sobre la metadata:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

openai_model = init_chat_model("openai:gpt-4.1-mini", temperature=0.7)
anthropic_model = init_chat_model("anthropic:claude-sonnet-4-20250514", temperature=0.7)
google_model = init_chat_model("google_genai:gemini-2.0-flash", temperature=0.7)

model_with_fallback = openai_model.with_fallbacks(
    [anthropic_model, google_model]
)

response = model_with_fallback.invoke("¿Qué es LangChain?")
print(response.content)
# Output esperado: LangChain es un framework open-source para construir
# aplicaciones con modelos de lenguaje...

with_fallbacks() es perfecto cuando solo necesitas resiliencia sin tracking. Pero para este proyecto usamos el enfoque manual porque queremos capturar qué proveedor respondió y cuánto tardó — información que with_fallbacks() no expone directamente.

---

Paso 3: Agregar streaming

El streaming es lo que hace que un chat se sienta responsivo. En vez de esperar 2-5 segundos a que llegue la respuesta completa, el usuario ve los tokens aparecer progresivamente — exactamente como en ChatGPT o Claude.

La función stream_with_fallback aplica la misma lógica de fallback pero usando stream() en vez de invoke():

import time

def stream_with_fallback(models, messages):
    """Intenta stream() con cada proveedor en orden.
    Imprime tokens progresivamente y retorna el texto completo + metadata.
    """
    errors = []

    for provider_id, provider_info in models.items():
        try:
            start = time.time()
            full_response = ""
            input_tokens = 0
            output_tokens = 0

            print(f"\n🤖 [{provider_info['display_name']}]: ", end="", flush=True)

            for chunk in provider_info["model"].stream(messages):
                if chunk.content:
                    print(chunk.content, end="", flush=True)
                    full_response += chunk.content

                if hasattr(chunk, "usage_metadata") and chunk.usage_metadata:
                    input_tokens = chunk.usage_metadata.get("input_tokens", input_tokens)
                    output_tokens = chunk.usage_metadata.get("output_tokens", output_tokens)

            print()
            latency_ms = (time.time() - start) * 1000

            return full_response, provider_id, latency_ms, input_tokens, output_tokens

        except Exception as e:
            errors.append(f"{provider_id}: {e}")
            print(f"\n  ⚠️  Fallback: {provider_id} falló → intentando siguiente...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(
        f"Todos los proveedores fallaron:\n{error_detail}"
    )

# Output esperado (los tokens aparecen uno a uno):
🤖 [OpenAI GPT-4.1 Mini]: Una API REST es una interfaz que permite a
aplicaciones comunicarse entre sí usando el protocolo HTTP...
📊 openai | 1842ms | 28→47 tokens

Dos detalles importantes sobre el streaming:

1. flush=True es crítico en el print(). Sin él, Python almacena el output en un buffer y los tokens no se muestran progresivamente — los verías todos de golpe al final, derrotando el propósito del streaming.

2. usage_metadata en chunks — no todos los proveedores envían conteo de tokens durante el streaming. OpenAI típicamente los incluye en el último chunk; otros proveedores pueden no incluirlos. Por eso inicializamos los contadores en 0 y los actualizamos solo si están disponibles.

---

Paso 4: Structured output para metadata

Cada respuesta del chat genera datos operacionales: qué proveedor la manejó, cuánto tardó, cuántos tokens consumió. En vez de manejar estos datos como variables sueltas, los encapsulamos en un modelo Pydantic — exactamente como aprendiste en la cápsula 05.

Definir el modelo de metadata

from pydantic import BaseModel, Field


class ChatMetadata(BaseModel):
    """Metadata operacional de cada respuesta del chat."""

    provider: str = Field(
        description="Identificador del proveedor que respondió"
    )
    model: str = Field(
        description="Identificador completo del modelo"
    )
    latency_ms: float = Field(
        description="Tiempo total de respuesta en milisegundos"
    )
    input_tokens: int = Field(
        description="Tokens consumidos en el prompt"
    )
    output_tokens: int = Field(
        description="Tokens generados en la respuesta"
    )

    def summary(self) -> str:
        """Resumen legible de la metadata."""
        total = self.input_tokens + self.output_tokens
        return (
            f"📊 {self.provider} ({self.model}) | "
            f"{self.latency_ms:.0f}ms | "
            f"{self.input_tokens}↑ {self.output_tokens}↓ ({total} total)"
        )

Integrar metadata en el streaming

Ahora actualizamos stream_with_fallback para que retorne un objeto ChatMetadata en vez de valores sueltos:

import time
from langchain_core.messages import AIMessage


def stream_with_fallback(models, messages):
    """Streaming con fallback. Retorna (texto, AIMessage, ChatMetadata)."""
    errors = []

    for provider_id, provider_info in models.items():
        try:
            start = time.time()
            full_response = ""
            input_tokens = 0
            output_tokens = 0

            print(f"\n🤖 [{provider_info['display_name']}]: ", end="", flush=True)

            for chunk in provider_info["model"].stream(messages):
                if chunk.content:
                    print(chunk.content, end="", flush=True)
                    full_response += chunk.content

                if hasattr(chunk, "usage_metadata") and chunk.usage_metadata:
                    input_tokens = chunk.usage_metadata.get("input_tokens", input_tokens)
                    output_tokens = chunk.usage_metadata.get("output_tokens", output_tokens)

            print()
            latency_ms = (time.time() - start) * 1000

            metadata = ChatMetadata(
                provider=provider_id,
                model=provider_info["model_id"],
                latency_ms=round(latency_ms, 2),
                input_tokens=input_tokens,
                output_tokens=output_tokens,
            )

            ai_message = AIMessage(content=full_response)

            return full_response, ai_message, metadata

        except Exception as e:
            errors.append(f"{provider_id}: {e}")
            print(f"\n  ⚠️  Fallback: {provider_id} falló → intentando siguiente...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(f"Todos los proveedores fallaron:\n{error_detail}")

Ahora cada respuesta viene acompañada de un objeto ChatMetadata tipado. Puedes acceder a metadata.provider, metadata.latency_ms, o llamar metadata.summary() para un resumen legible. No hay strings que parsear ni diccionarios con keys que adivinar.

---

Paso 5: Chat loop interactivo

El último paso une todo en un loop de conversación. El chat mantiene historial de mensajes (para que el modelo tenga contexto de la conversación), aplica fallback automático, muestra streaming, e imprime la metadata después de cada respuesta.

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage


SYSTEM_PROMPT = """Eres un asistente técnico especializado en programación.
Responde de forma concisa y directa. Si te preguntan algo fuera de
programación, responde brevemente y redirige a temas técnicos."""


def chat():
    """Loop principal del chat multi-proveedor."""
    print("=" * 55)
    print("  Chat Multi-Proveedor con Fallback")
    print("  Escribe 'salir' para terminar")
    print("  Escribe 'status' para ver proveedores activos")
    print("  Escribe 'stats' para ver estadísticas de la sesión")
    print("=" * 55)

    print("\nInicializando proveedores...")
    models = init_providers()

    if not models:
        print("\n❌ No hay proveedores disponibles. Verifica tus API keys en .env")
        return

    print(f"\n{len(models)} proveedor(es) disponible(s). ¡Listo para chatear!\n")

    history = [SystemMessage(content=SYSTEM_PROMPT)]
    session_metadata = []

    while True:
        try:
            user_input = input("Tú: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\n¡Hasta luego!")
            break

        if not user_input:
            continue

        if user_input.lower() in ("salir", "exit", "quit"):
            print("\n¡Hasta luego!")
            break

        if user_input.lower() == "status":
            print("\nProveedores activos:")
            for pid, pinfo in models.items():
                print(f"  ✅ {pinfo['display_name']} ({pid})")
            print()
            continue

        if user_input.lower() == "stats":
            print_session_stats(session_metadata)
            continue

        history.append(HumanMessage(content=user_input))

        try:
            text, ai_message, metadata = stream_with_fallback(models, history)
            history.append(ai_message)
            session_metadata.append(metadata)
            print(metadata.summary())

        except RuntimeError as e:
            print(f"\n❌ {e}")
            history.pop()

    if session_metadata:
        print("\n--- Resumen de sesión ---")
        print_session_stats(session_metadata)


def print_session_stats(metadata_list):
    """Imprime estadísticas acumuladas de la sesión."""
    if not metadata_list:
        print("\nNo hay estadísticas todavía.\n")
        return

    total_tokens = sum(m.input_tokens + m.output_tokens for m in metadata_list)
    avg_latency = sum(m.latency_ms for m in metadata_list) / len(metadata_list)

    provider_counts = {}
    for m in metadata_list:
        provider_counts[m.provider] = provider_counts.get(m.provider, 0) + 1

    print(f"\n  📊 Mensajes totales: {len(metadata_list)}")
    print(f"  📊 Tokens totales: {total_tokens}")
    print(f"  📊 Latencia promedio: {avg_latency:.0f}ms")
    print(f"  📊 Proveedores usados:")
    for provider, count in provider_counts.items():
        print(f"     - {provider}: {count} respuesta(s)")
    print()

# Output esperado:
=======================================================
  Chat Multi-Proveedor con Fallback
  Escribe 'salir' para terminar
  ...
=======================================================

Inicializando proveedores...
  ✅ OpenAI GPT-4.1 Mini
  ✅ Anthropic Claude Sonnet 4
  ✅ Google Gemini 2.0 Flash

3 proveedor(es) disponible(s). ¡Listo para chatear!

Tú: ¿Qué es un decorador en Python?

🤖 [OpenAI GPT-4.1 Mini]: Un decorador es una función que recibe otra
función como argumento y extiende su comportamiento sin modificarla
directamente. Se aplica con la sintaxis @decorador encima de la
definición de la función.
📊 openai (openai:gpt-4.1-mini) | 1203ms | 45↑ 38↓ (83 total)

Tú: salir

¡Hasta luego!

--- Resumen de sesión ---
  📊 Mensajes totales: 1
  📊 Tokens totales: 83
  📊 Latencia promedio: 1203ms
  📊 Proveedores usados:
     - openai: 1 respuesta(s)

---

Código completo

Este es el archivo chat.py completo. Cópialo, configura tu .env, y ejecútalo con python chat.py:

"""
Chat Multi-Proveedor con Fallback
Módulo 1 — LangChain & LangGraph: From Chains to Agents

Requiere: pip install langchain langchain-openai langchain-anthropic
          langchain-google-genai python-dotenv pydantic
"""

import time

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from pydantic import BaseModel, Field


# --- Configuración de proveedores ---

PROVIDERS = [
    {
        "id": "openai",
        "model_id": "openai:gpt-4.1-mini",
        "display_name": "OpenAI GPT-4.1 Mini",
    },
    {
        "id": "anthropic",
        "model_id": "anthropic:claude-sonnet-4-20250514",
        "display_name": "Anthropic Claude Sonnet 4",
    },
    {
        "id": "google",
        "model_id": "google_genai:gemini-2.0-flash",
        "display_name": "Google Gemini 2.0 Flash",
    },
]

SYSTEM_PROMPT = """Eres un asistente técnico especializado en programación.
Responde de forma concisa y directa. Si te preguntan algo fuera de
programación, responde brevemente y redirige a temas técnicos."""


# --- Modelo de metadata ---

class ChatMetadata(BaseModel):
    """Metadata operacional de cada respuesta del chat."""

    provider: str = Field(description="Identificador del proveedor que respondió")
    model: str = Field(description="Identificador completo del modelo")
    latency_ms: float = Field(description="Tiempo total de respuesta en milisegundos")
    input_tokens: int = Field(description="Tokens consumidos en el prompt")
    output_tokens: int = Field(description="Tokens generados en la respuesta")

    def summary(self) -> str:
        total = self.input_tokens + self.output_tokens
        return (
            f"📊 {self.provider} ({self.model}) | "
            f"{self.latency_ms:.0f}ms | "
            f"{self.input_tokens}↑ {self.output_tokens}↓ ({total} total)"
        )


# --- Inicialización ---

def init_providers():
    """Inicializa todos los proveedores disponibles."""
    models = {}
    for provider in PROVIDERS:
        try:
            model = init_chat_model(
                provider["model_id"],
                temperature=0.7,
                max_tokens=1024,
            )
            models[provider["id"]] = {
                "model": model,
                "display_name": provider["display_name"],
                "model_id": provider["model_id"],
            }
            print(f"  ✅ {provider['display_name']}")
        except Exception as e:
            print(f"  ❌ {provider['display_name']}: {e}")
    return models


# --- Fallback con invoke ---

def invoke_with_fallback(models, messages):
    """invoke() con fallback. Retorna (response, ChatMetadata)."""
    errors = []

    for provider_id, provider_info in models.items():
        try:
            start = time.time()
            response = provider_info["model"].invoke(messages)
            latency_ms = (time.time() - start) * 1000

            usage = response.usage_metadata or {}

            metadata = ChatMetadata(
                provider=provider_id,
                model=provider_info["model_id"],
                latency_ms=round(latency_ms, 2),
                input_tokens=usage.get("input_tokens", 0),
                output_tokens=usage.get("output_tokens", 0),
            )

            return response, metadata

        except Exception as e:
            errors.append(f"{provider_id}: {e}")
            print(f"  ⚠️  Fallback: {provider_id} falló → intentando siguiente...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(f"Todos los proveedores fallaron:\n{error_detail}")


# --- Fallback con streaming ---

def stream_with_fallback(models, messages):
    """stream() con fallback. Retorna (texto, AIMessage, ChatMetadata)."""
    errors = []

    for provider_id, provider_info in models.items():
        try:
            start = time.time()
            full_response = ""
            input_tokens = 0
            output_tokens = 0

            print(f"\n🤖 [{provider_info['display_name']}]: ", end="", flush=True)

            for chunk in provider_info["model"].stream(messages):
                if chunk.content:
                    print(chunk.content, end="", flush=True)
                    full_response += chunk.content

                if hasattr(chunk, "usage_metadata") and chunk.usage_metadata:
                    input_tokens = chunk.usage_metadata.get(
                        "input_tokens", input_tokens
                    )
                    output_tokens = chunk.usage_metadata.get(
                        "output_tokens", output_tokens
                    )

            print()
            latency_ms = (time.time() - start) * 1000

            metadata = ChatMetadata(
                provider=provider_id,
                model=provider_info["model_id"],
                latency_ms=round(latency_ms, 2),
                input_tokens=input_tokens,
                output_tokens=output_tokens,
            )

            ai_message = AIMessage(content=full_response)
            return full_response, ai_message, metadata

        except Exception as e:
            errors.append(f"{provider_id}: {e}")
            print(f"\n  ⚠️  Fallback: {provider_id} falló → intentando siguiente...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(f"Todos los proveedores fallaron:\n{error_detail}")


# --- Estadísticas de sesión ---

def print_session_stats(metadata_list):
    """Imprime estadísticas acumuladas de la sesión."""
    if not metadata_list:
        print("\nNo hay estadísticas todavía.\n")
        return

    total_tokens = sum(m.input_tokens + m.output_tokens for m in metadata_list)
    avg_latency = sum(m.latency_ms for m in metadata_list) / len(metadata_list)

    provider_counts = {}
    for m in metadata_list:
        provider_counts[m.provider] = provider_counts.get(m.provider, 0) + 1

    print(f"\n  📊 Mensajes totales: {len(metadata_list)}")
    print(f"  📊 Tokens totales: {total_tokens}")
    print(f"  📊 Latencia promedio: {avg_latency:.0f}ms")
    print(f"  📊 Proveedores usados:")
    for provider, count in provider_counts.items():
        print(f"     - {provider}: {count} respuesta(s)")
    print()


# --- Chat loop ---

def chat():
    """Loop principal del chat multi-proveedor."""
    print("=" * 55)
    print("  Chat Multi-Proveedor con Fallback")
    print("  Escribe 'salir' para terminar")
    print("  Escribe 'status' para ver proveedores activos")
    print("  Escribe 'stats' para ver estadísticas de la sesión")
    print("=" * 55)

    print("\nInicializando proveedores...")
    models = init_providers()

    if not models:
        print("\n❌ No hay proveedores disponibles. Verifica tus API keys en .env")
        return

    print(f"\n{len(models)} proveedor(es) disponible(s). ¡Listo para chatear!\n")

    history = [SystemMessage(content=SYSTEM_PROMPT)]
    session_metadata = []

    while True:
        try:
            user_input = input("Tú: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\n¡Hasta luego!")
            break

        if not user_input:
            continue

        if user_input.lower() in ("salir", "exit", "quit"):
            print("\n¡Hasta luego!")
            break

        if user_input.lower() == "status":
            print("\nProveedores activos:")
            for pid, pinfo in models.items():
                print(f"  ✅ {pinfo['display_name']} ({pid})")
            print()
            continue

        if user_input.lower() == "stats":
            print_session_stats(session_metadata)
            continue

        history.append(HumanMessage(content=user_input))

        try:
            text, ai_message, metadata = stream_with_fallback(models, history)
            history.append(ai_message)
            session_metadata.append(metadata)
            print(metadata.summary())

        except RuntimeError as e:
            print(f"\n❌ {e}")
            history.pop()

    if session_metadata:
        print("\n--- Resumen de sesión ---")
        print_session_stats(session_metadata)


if __name__ == "__main__":
    chat()

Ejecútalo:

python chat.py

---

Criterios de éxito

Tu proyecto está completo cuando cumples los cuatro criterios:

• ✅ El chat funciona con al menos 2 proveedores — puedes cambiar entre ellos configurando/desconfigurando API keys
• ✅ El fallback se activa correctamente — al invalidar la API key del primer proveedor, el chat responde automáticamente con el siguiente
• ✅ El streaming muestra tokens progresivamente — ves los tokens aparecer uno a uno en la terminal, no la respuesta completa de golpe
• ✅ La metadata incluye proveedor y tokens — después de cada respuesta ves el resumen con proveedor, latencia y conteo de tokens

---

Cómo probar el fallback

El fallback solo se activa cuando un proveedor falla. Para provocar un fallo controlado, invalida temporalmente la API key del primer proveedor.

Método 1: API key inválida en .env

# .env — modifica temporalmente
OPENAI_API_KEY=sk-invalida-12345    # ← key inválida
ANTHROPIC_API_KEY=sk-ant-...         # ← key válida
GOOGLE_API_KEY=AI...                 # ← key válida

Ejecuta el chat y verás:

Inicializando proveedores...
  ✅ OpenAI GPT-4.1 Mini
  ✅ Anthropic Claude Sonnet 4
  ✅ Google Gemini 2.0 Flash

Tú: Hola

  ⚠️  Fallback: openai falló → intentando siguiente...

🤖 [Anthropic Claude Sonnet 4]: ¡Hola! Soy un asistente técnico
especializado en programación. ¿En qué puedo ayudarte?
📊 anthropic (anthropic:claude-sonnet-4-20250514) | 1456ms | 32↑ 24↓ (56 total)

OpenAI falla al intentar autenticar, y el sistema automáticamente usa Anthropic.

Método 2: Desde código

Para testear sin modificar .env, inserta un proveedor ficticio al inicio de la lista:

PROVIDERS = [
    {
        "id": "ficticio",
        "model_id": "openai:modelo-que-no-existe",
        "display_name": "Proveedor Ficticio",
    },
    {
        "id": "openai",
        "model_id": "openai:gpt-4.1-mini",
        "display_name": "OpenAI GPT-4.1 Mini",
    },
]

El proveedor ficticio fallará siempre, y el sistema usará OpenAI como fallback.

---

Errores comunes

1. ModuleNotFoundError: No module named 'langchain_openai'

Causa: No instalaste los paquetes de proveedor.

# Solución: instalar los paquetes faltantes
pip install langchain-openai langchain-anthropic langchain-google-genai

Cada proveedor tiene su propio paquete. langchain base no los incluye.

2. AuthenticationError: Incorrect API key

Causa: La API key en .env es inválida o expiró. Verifica que el formato es correcto (sin comillas): OPENAI_API_KEY=sk-proj-abc123.... Un error común es poner la key de un proveedor en la variable de otro.

3. Los tokens no aparecen progresivamente (se ven todos de golpe)

Causa: Falta flush=True en el print().

# Incorrecto — Python acumula output en buffer
print(chunk.content, end="")

# Correcto — forzar flush del buffer
print(chunk.content, end="", flush=True)

Sin flush=True, Python espera a que se acumule suficiente texto en el buffer interno antes de escribir a la terminal. Con streaming queremos que cada token aparezca inmediatamente.

4. RateLimitError: Rate limit exceeded

Causa: Estás haciendo demasiadas llamadas demasiado rápido.

from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=1,
    check_every_n_seconds=0.1,
    max_bucket_size=10,
)

model = init_chat_model(
    "openai:gpt-4.1-mini",
    rate_limiter=rate_limiter,
)

Agrega un InMemoryRateLimiter a los modelos que están dando este error. Como aprendiste en la cápsula 07, el rate limiter controla la velocidad de las llamadas automáticamente.

5. usage_metadata es None

Causa: No todos los proveedores/modelos reportan tokens en streaming.

# Solución: siempre verificar antes de acceder
if hasattr(chunk, "usage_metadata") and chunk.usage_metadata:
    input_tokens = chunk.usage_metadata.get("input_tokens", 0)

Algunos proveedores solo reportan tokens en el último chunk, otros no los reportan en streaming. Tu código ya maneja esto con el patrón de inicializar en 0 y actualizar si los datos están disponibles.

6. TimeoutError o conexión lenta

Causa: La API del proveedor no responde a tiempo.

model = init_chat_model(
    "openai:gpt-4.1-mini",
    timeout=30,        # segundos máximos de espera
    max_retries=2,     # reintentos automáticos
)

Configura timeout y max_retries en init_chat_model. El timeout evita que tu aplicación se quede colgada esperando indefinidamente. Los retries manejan errores transitorios de red.

7. El historial crece demasiado y empiezo a recibir errores de tokens

Causa: Estás enviando toda la conversación como contexto. Cada mensaje anterior consume tokens de input.

MAX_HISTORY = 20

if len(history) > MAX_HISTORY:
    system = history[0]
    history = [system] + history[-(MAX_HISTORY - 1):]

Mantén una ventana de los últimos N mensajes. Siempre conserva el SystemMessage (primer elemento) y recorta los mensajes más antiguos.

8. load_dotenv() no encuentra el archivo .env

Causa: El archivo .env no está en el directorio desde donde ejecutas el script. Verifica con pwd que estás en el directorio correcto. Si el .env está en otra ubicación, pasa la ruta: load_dotenv("/ruta/completa/al/.env").

---

Ideas para extender

Si terminaste el proyecto y quieres ir más allá:

• 🚀 Agregar Ollama como fallback local — init_chat_model("ollama:llama3.2") como último recurso cuando todos los proveedores cloud fallan
• 🚀 Selección manual de proveedor — Comando usar openai para forzar un proveedor específico
• 🚀 Log de costos estimados — Calcular costo por respuesta usando precios por token de cada proveedor
• 🚀 Rate limiting por proveedor — InMemoryRateLimiter con límites diferenciados por proveedor
• 🚀 Modo de comparación — Enviar el mismo prompt a todos los proveedores y comparar respuestas

---

Conexión con el siguiente módulo

En este módulo aprendiste a trabajar con modelos: inicializarlos, configurarlos, ejecutarlos de múltiples formas, y combinarlos con fallback. Pero los modelos solos tienen una limitación fundamental — solo pueden generar texto. No pueden buscar en internet, consultar una base de datos, ni ejecutar código.

En el Módulo 2: Tools y Tool Calling, aprenderás a darle herramientas a los modelos. Crearás funciones Python que el modelo puede "llamar" cuando necesita información externa. El mismo chat multi-proveedor que construiste aquí podría extenderse con tools para buscar clima, hacer cálculos, o consultar APIs — transformando un chatbot en un asistente que realmente hace cosas.

---

Recursos para el proyecto

1. init_chat_model API Reference — Documentación completa de la función universal para inicializar modelos
2. LangChain Fallbacks — Guía oficial del método with_fallbacks() para encadenar modelos
3. Streaming en LangChain — Patterns de streaming con stream() y astream()
4. Pydantic v2 Documentation — Referencia de modelos, Field, y validación
5. Token Usage Tracking — Cómo monitorear consumo de tokens con usage_metadata
6. OpenAI API Error Codes — Referencia de errores y cómo manejarlos (aplica al diseño de fallback)

---

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