Module 1: Models and Providers

Project: Multi-Provider Chat with Fallback

Next →

Project overview

In the previous seven capsules you learned to initialize models with init_chat_model, configure parameters, execute with invoke(), stream() and batch(), get typed responses with with_structured_output, process images with multimodal models, and control costs with rate limiting and caching. You saw each concept on its own, in isolated examples. Now you're going to combine all of it into a real system.

In this project you build a multi-provider chat with automatic fallback. The system tries to answer using OpenAI. If OpenAI fails — an API error, a rate limit, or an invalid API key — it automatically tries Anthropic. If Anthropic fails too, it tries Google. All of this happens transparently: the user watches their answer arrive token by token without knowing which provider generated it.

On top of that, every response carries structured metadata: which provider answered, which model was used, how long it took in milliseconds, and how many tokens it consumed. This metadata uses a Pydantic model — exactly like you learned in the Structured Output capsule — but applied to operational data instead of LLM responses.

This fallback pattern isn't academic. Services like AWS, Stripe and any critical API implement provider fallback as standard practice. When you finish this project, you'll have a working system you can adapt for any application that needs resilience against LLM provider failures.

Project goal

Build an interactive terminal chat that uses multiple LLM providers with automatic fallback, progressive streaming, and structured metadata for every response.

When you complete this project:

  • 🔧 You'll know how to integrate multiple providers (init_chat_model) into a single system with fallback
  • 🔧 You'll implement streaming with per-provider error handling
  • 🔧 You'll capture operational metadata (provider, latency, tokens) using Pydantic models
  • 🔧 You'll have a working terminal chat that demonstrates real resilience

Technical specs

Tech stack

ComponentVersionPurpose
Python3.11+Runtime
LangChainv1.2+LLM framework
langchain-openailatestOpenAI provider
langchain-anthropiclatestAnthropic provider
langchain-google-genailatestGoogle provider
pydanticv2+Metadata model
python-dotenvlatestEnvironment variables

Initial setup

Before you start, make sure you have the dependencies installed and the API keys configured:

# Install dependencies
pip install langchain langchain-openai langchain-anthropic langchain-google-genai python-dotenv pydantic

Create a .env file at the root of your project:

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

You don't need all three API keys for the project to work. The fallback system is designed precisely to handle unavailable providers. With at least one valid API key, the chat runs.

Project structure

multi-provider-chat/
├── .env                  # API keys
├── chat.py               # Main code (everything in one file)
└── requirements.txt      # Dependencies

# 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

For this mini-project, all the code goes in a single chat.py file. You don't need a complex structure — the goal is to integrate concepts, not to design an architecture.

Step 1: Configure the providers

The first step is to define the providers and write a function that tries to initialize each one. If a provider has no API key configured or hits some other problem, it gets marked as unavailable but the program doesn't stop.

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():
    """Initializes every available provider.
    Returns a dict with the providers that could be created.
    """
    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
# Expected output (with all 3 API keys configured):
  ✅ OpenAI GPT-4.1 Mini
  ✅ Anthropic Claude Sonnet 4
  ✅ Google Gemini 2.0 Flash

# Expected output (without an Anthropic API key):
  ✅ OpenAI GPT-4.1 Mini
  ❌ Anthropic Claude Sonnet 4: Did not find anthropic_api_key...
  ✅ Google Gemini 2.0 Flash

Notice that init_chat_model can fail while creating the model if it doesn't find the matching API key — it depends on the provider. Some providers validate the key at initialization, others on the first call. The try/except handles both cases.

The PROVIDERS list defines the priority order for the fallback. OpenAI is tried first, Anthropic second, Google third. You can change the order based on your preferences or costs.

Step 2: Implement the fallback

The manual approach with try/except

The manual approach gives you full control: you can know exactly which provider answered, measure latency, and capture tokens. Each provider is tried in order until one responds successfully.

import time

def invoke_with_fallback(models, messages):
    """Tries invoke() with each provider in order.
    Returns (response, provider_id) from the first provider that works.
    """
    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} failed → trying the next one...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(
        f"All providers failed:\n{error_detail}"
    )

Let's test the fallback:

from langchain_core.messages import HumanMessage, SystemMessage

models = init_providers()

messages = [
    SystemMessage(content="Answer in one short sentence."),
    HumanMessage(content="What is Python?"),
]

response, provider_id, latency = invoke_with_fallback(models, messages)
print(f"\nProvider: {provider_id}")
print(f"Response: {response.content}")
print(f"Latency: {latency:.0f}ms")
# Expected output (OpenAI available):
Provider: openai
Response: Python is an interpreted, high-level, general-purpose programming language.
Latency: 823ms

# Expected output (OpenAI fails → fallback):
  ⚠️  Fallback: openai failed → trying the next one...
Provider: anthropic
Response: Python is a versatile programming language, known for its clear and readable syntax.
Latency: 1205ms

The with_fallbacks() approach (built-in)

LangChain ships a with_fallbacks() method that chains models automatically. It's more concise but gives you less control over the 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("What is LangChain?")
print(response.content)
# Expected output: LangChain is an open-source framework for building
# applications with language models...

with_fallbacks() is perfect when all you need is resilience without tracking. But for this project we use the manual approach because we want to capture which provider answered and how long it took — information with_fallbacks() doesn't expose directly.

Step 3: Add streaming

Streaming is what makes a chat feel responsive. Instead of waiting 2-5 seconds for the full response to arrive, the user watches the tokens appear progressively — exactly like in ChatGPT or Claude.

The stream_with_fallback function applies the same fallback logic but using stream() instead of invoke():

import time

def stream_with_fallback(models, messages):
    """Tries stream() with each provider in order.
    Prints tokens progressively and returns the full text + 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} failed → trying the next one...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(
        f"All providers failed:\n{error_detail}"
    )
# Expected output (the tokens appear one by one):
🤖 [OpenAI GPT-4.1 Mini]: A REST API is an interface that lets
applications communicate with each other over the HTTP protocol...
📊 openai | 1842ms | 28→47 tokens

Two important details about the streaming:

  1. flush=True is critical in the print(). Without it, Python buffers the output and the tokens don't show up progressively — you'd see them all at once at the end, defeating the whole purpose of streaming.

  2. usage_metadata in chunks — not every provider sends a token count during streaming. OpenAI typically includes it in the last chunk; other providers may not include it at all. That's why we initialize the counters at 0 and only update them if the data is there.

Step 4: Structured output for the metadata

Every chat response generates operational data: which provider handled it, how long it took, how many tokens it consumed. Instead of carrying that data around as loose variables, we wrap it in a Pydantic model — exactly like you learned in capsule 05.

Define the metadata model

from pydantic import BaseModel, Field


class ChatMetadata(BaseModel):
    """Operational metadata for each chat response."""

    provider: str = Field(
        description="Identifier of the provider that answered"
    )
    model: str = Field(
        description="Full model identifier"
    )
    latency_ms: float = Field(
        description="Total response time in milliseconds"
    )
    input_tokens: int = Field(
        description="Tokens consumed by the prompt"
    )
    output_tokens: int = Field(
        description="Tokens generated in the response"
    )

    def summary(self) -> str:
        """Human-readable summary of the 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)"
        )

Wire the metadata into the streaming

Now we update stream_with_fallback so it returns a ChatMetadata object instead of loose values:

import time
from langchain_core.messages import AIMessage


def stream_with_fallback(models, messages):
    """Streaming with fallback. Returns (text, 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} failed → trying the next one...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(f"All providers failed:\n{error_detail}")

Now every response comes with a typed ChatMetadata object attached. You can reach for metadata.provider, metadata.latency_ms, or call metadata.summary() for a readable digest. No strings to parse, no dictionaries with keys you have to guess.

Step 5: Interactive chat loop

The last step ties everything together in a conversation loop. The chat keeps a message history (so the model has context for the conversation), applies automatic fallback, shows the streaming, and prints the metadata after each response.

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage


SYSTEM_PROMPT = """You are a technical assistant specialized in programming.
Answer concisely and directly. If they ask you about something outside
programming, answer briefly and steer back to technical topics."""


def chat():
    """Main loop of the multi-provider chat."""
    print("=" * 55)
    print("  Multi-Provider Chat with Fallback")
    print("  Type 'exit' to quit")
    print("  Type 'status' to see the active providers")
    print("  Type 'stats' to see the session statistics")
    print("=" * 55)

    print("\nInitializing providers...")
    models = init_providers()

    if not models:
        print("\n❌ No providers available. Check your API keys in .env")
        return

    print(f"\n{len(models)} provider(s) available. Ready to chat!\n")

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

    while True:
        try:
            user_input = input("You: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\nSee you next time!")
            break

        if not user_input:
            continue

        if user_input.lower() in ("exit", "quit", "bye"):
            print("\nSee you next time!")
            break

        if user_input.lower() == "status":
            print("\nActive providers:")
            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--- Session summary ---")
        print_session_stats(session_metadata)


def print_session_stats(metadata_list):
    """Prints the accumulated statistics for the session."""
    if not metadata_list:
        print("\nNo statistics yet.\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  📊 Total messages: {len(metadata_list)}")
    print(f"  📊 Total tokens: {total_tokens}")
    print(f"  📊 Average latency: {avg_latency:.0f}ms")
    print(f"  📊 Providers used:")
    for provider, count in provider_counts.items():
        print(f"     - {provider}: {count} response(s)")
    print()
# Expected output:
=======================================================
  Multi-Provider Chat with Fallback
  Type 'exit' to quit
  ...
=======================================================

Initializing providers...
  ✅ OpenAI GPT-4.1 Mini
  ✅ Anthropic Claude Sonnet 4
  ✅ Google Gemini 2.0 Flash

3 provider(s) available. Ready to chat!

You: What is a decorator in Python?

🤖 [OpenAI GPT-4.1 Mini]: A decorator is a function that takes another
function as an argument and extends its behavior without modifying it
directly. You apply it with the @decorator syntax above the
function's definition.
📊 openai (openai:gpt-4.1-mini) | 1203ms | 45↑ 38↓ (83 total)

You: exit

See you next time!

--- Session summary ---
  📊 Total messages: 1
  📊 Total tokens: 83
  📊 Average latency: 1203ms
  📊 Providers used:
     - openai: 1 response(s)

Complete code

This is the complete chat.py file. Copy it, configure your .env, and run it with python chat.py:

"""
Multi-Provider Chat with Fallback
Module 1 — LangChain & LangGraph: From Chains to Agents

Requires: 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


# --- Provider configuration ---

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 = """You are a technical assistant specialized in programming.
Answer concisely and directly. If they ask you about something outside
programming, answer briefly and steer back to technical topics."""


# --- Metadata model ---

class ChatMetadata(BaseModel):
    """Operational metadata for each chat response."""

    provider: str = Field(description="Identifier of the provider that answered")
    model: str = Field(description="Full model identifier")
    latency_ms: float = Field(description="Total response time in milliseconds")
    input_tokens: int = Field(description="Tokens consumed by the prompt")
    output_tokens: int = Field(description="Tokens generated in the response")

    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)"
        )


# --- Initialization ---

def init_providers():
    """Initializes every available provider."""
    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 with invoke ---

def invoke_with_fallback(models, messages):
    """invoke() with fallback. Returns (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} failed → trying the next one...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(f"All providers failed:\n{error_detail}")


# --- Fallback with streaming ---

def stream_with_fallback(models, messages):
    """stream() with fallback. Returns (text, 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} failed → trying the next one...")
            continue

    error_detail = "\n".join(errors)
    raise RuntimeError(f"All providers failed:\n{error_detail}")


# --- Session statistics ---

def print_session_stats(metadata_list):
    """Prints the accumulated statistics for the session."""
    if not metadata_list:
        print("\nNo statistics yet.\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  📊 Total messages: {len(metadata_list)}")
    print(f"  📊 Total tokens: {total_tokens}")
    print(f"  📊 Average latency: {avg_latency:.0f}ms")
    print(f"  📊 Providers used:")
    for provider, count in provider_counts.items():
        print(f"     - {provider}: {count} response(s)")
    print()


# --- Chat loop ---

def chat():
    """Main loop of the multi-provider chat."""
    print("=" * 55)
    print("  Multi-Provider Chat with Fallback")
    print("  Type 'exit' to quit")
    print("  Type 'status' to see the active providers")
    print("  Type 'stats' to see the session statistics")
    print("=" * 55)

    print("\nInitializing providers...")
    models = init_providers()

    if not models:
        print("\n❌ No providers available. Check your API keys in .env")
        return

    print(f"\n{len(models)} provider(s) available. Ready to chat!\n")

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

    while True:
        try:
            user_input = input("You: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\nSee you next time!")
            break

        if not user_input:
            continue

        if user_input.lower() in ("exit", "quit", "bye"):
            print("\nSee you next time!")
            break

        if user_input.lower() == "status":
            print("\nActive providers:")
            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--- Session summary ---")
        print_session_stats(session_metadata)


if __name__ == "__main__":
    chat()

Run it:

python chat.py

Success criteria

Your project is complete when you meet all four criteria:

  • ✅ The chat works with at least 2 providers — you can switch between them by configuring/unconfiguring API keys
  • ✅ The fallback kicks in correctly — when you invalidate the first provider's API key, the chat automatically answers with the next one
  • ✅ The streaming shows tokens progressively — you see the tokens appear one by one in the terminal, not the whole response at once
  • ✅ The metadata includes provider and tokens — after each response you see the summary with provider, latency and token count

How to test the fallback

The fallback only kicks in when a provider fails. To force a controlled failure, temporarily invalidate the first provider's API key.

Method 1: an invalid API key in .env

# .env — modify temporarily
OPENAI_API_KEY=sk-invalid-12345      # ← invalid key
ANTHROPIC_API_KEY=sk-ant-...         # ← valid key
GOOGLE_API_KEY=AI...                 # ← valid key

Run the chat and you'll see:

Initializing providers...
  ✅ OpenAI GPT-4.1 Mini
  ✅ Anthropic Claude Sonnet 4
  ✅ Google Gemini 2.0 Flash

You: Hi

  ⚠️  Fallback: openai failed → trying the next one...

🤖 [Anthropic Claude Sonnet 4]: Hi! I'm a technical assistant
specialized in programming. How can I help you?
📊 anthropic (anthropic:claude-sonnet-4-20250514) | 1456ms | 32↑ 24↓ (56 total)

OpenAI fails when it tries to authenticate, and the system automatically uses Anthropic.

Method 2: from the code

To test without touching .env, insert a fake provider at the top of the list:

PROVIDERS = [
    {
        "id": "fake",
        "model_id": "openai:model-that-does-not-exist",
        "display_name": "Fake Provider",
    },
    {
        "id": "openai",
        "model_id": "openai:gpt-4.1-mini",
        "display_name": "OpenAI GPT-4.1 Mini",
    },
]

The fake provider will always fail, and the system will use OpenAI as the fallback.

Common errors

1. ModuleNotFoundError: No module named 'langchain_openai'

Cause: You didn't install the provider packages.

# Fix: install the missing packages
pip install langchain-openai langchain-anthropic langchain-google-genai

Each provider has its own package. Base langchain doesn't include them.

2. AuthenticationError: Incorrect API key

Cause: The API key in .env is invalid or expired. Check that the format is correct (no quotes): OPENAI_API_KEY=sk-proj-abc123.... A common mistake is putting one provider's key in another provider's variable.

3. The tokens don't appear progressively (they all show up at once)

Cause: flush=True is missing from the print().

# Wrong — Python buffers the output
print(chunk.content, end="")

# Right — force a flush of the buffer
print(chunk.content, end="", flush=True)

Without flush=True, Python waits until enough text piles up in the internal buffer before writing to the terminal. With streaming we want each token to appear immediately.

4. RateLimitError: Rate limit exceeded

Cause: You're making too many calls too fast.

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,
)

Add an InMemoryRateLimiter to the models that are throwing this error. As you learned in capsule 07, the rate limiter controls the pace of your calls automatically.

5. usage_metadata is None

Cause: Not every provider/model reports tokens while streaming.

# Fix: always check before accessing
if hasattr(chunk, "usage_metadata") and chunk.usage_metadata:
    input_tokens = chunk.usage_metadata.get("input_tokens", 0)

Some providers only report tokens in the last chunk, others don't report them during streaming at all. Your code already handles this with the pattern of initializing at 0 and updating if the data is available.

6. TimeoutError or a slow connection

Cause: The provider's API isn't responding in time.

model = init_chat_model(
    "openai:gpt-4.1-mini",
    timeout=30,        # max seconds to wait
    max_retries=2,     # automatic retries
)

Configure timeout and max_retries in init_chat_model. The timeout keeps your application from hanging indefinitely. The retries handle transient network errors.

7. The history grows too large and I start getting token errors

Cause: You're sending the whole conversation as context. Every previous message consumes input tokens.

MAX_HISTORY = 20

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

Keep a window of the last N messages. Always preserve the SystemMessage (the first element) and trim the oldest messages.

8. load_dotenv() can't find the .env file

Cause: The .env file isn't in the directory you're running the script from. Check with pwd that you're in the right directory. If the .env lives somewhere else, pass the path: load_dotenv("/full/path/to/.env").

Ideas to extend it

If you finished the project and want to go further:

  • 🚀 Add Ollama as a local fallback — init_chat_model("ollama:llama3.2") as the last resort when every cloud provider fails
  • 🚀 Manual provider selection — a use openai command to force a specific provider
  • 🚀 Estimated cost log — compute the cost per response using each provider's per-token pricing
  • 🚀 Per-provider rate limiting — InMemoryRateLimiter with different limits for each provider
  • 🚀 Comparison mode — send the same prompt to every provider and compare the answers

Connection with the next module

In this module you learned to work with models: initialize them, configure them, run them in several ways, and combine them with fallback. But models on their own have one fundamental limitation — all they can do is generate text. They can't search the internet, query a database, or run code.

In Module 2: Tools and Tool Calling, you'll learn to give models tools. You'll write Python functions the model can "call" when it needs external information. The same multi-provider chat you built here could be extended with tools to look up the weather, do calculations, or query APIs — turning a chatbot into an assistant that actually does things.

Project resources

  1. init_chat_model API Reference — Full documentation of the universal function for initializing models
  2. LangChain Fallbacks — Official guide to the with_fallbacks() method for chaining models
  3. Streaming in LangChain — Streaming patterns with stream() and astream()
  4. Pydantic v2 Documentation — Reference for models, Field, and validation
  5. Token Usage Tracking — How to monitor token consumption with usage_metadata
  6. OpenAI API Error Codes — Reference for errors and how to handle them (relevant to fallback design)

Module 1 — LangChain & LangGraph: From Chains to Agents

← PreviousLocal Models, Caching and Rate LimitingNext →Introduction: why models need tools