Dynamic Models: Smart Selection

Capsule overview

In the previous capsules you learned to intercept model calls with @wrap_model_call and to customize tools with @wrap_tool_call. But so far, you've always used a single model for everything. That's like driving a freight truck to the corner store for bread: it works, but it's wasteful.

One of the most powerful patterns in the middleware system is model routing: the agent automatically picks which model to use based on the complexity of each request. Simple questions like "What time is it?" go to a fast, cheap model (gpt-4.1-mini). Complex questions like "Analyze the differences between 5 AI frameworks" go to a powerful one (gpt-4.1 or claude-sonnet). This lets you cut costs by up to 80% without giving up quality on the answers that actually need it.

The key is @wrap_model_call: you already know it intercepts the model call — now you'll use it to swap the model itself based on how the question gets classified. Combined with init_chat_model, you can pre-initialize several models and switch between them transparently.

The concept: routing models by complexity

In a production system, not all questions are equal:

Using gpt-4.1 for everything works, but if 70% of your requests are simple, you're paying 10x more than you need to on those. Model routing automates that decision.

The routing architecture

User sends a question
       │
       ▼
┌─────────────────────┐
│  classify_complexity │  ← Heuristics: length, keywords, tools
│  "simple" / "complex"│
└──────────┬──────────┘
           │
     ┌─────┴─────┐
     │           │
     ▼           ▼
 gpt-4.1-mini  gpt-4.1
 (fast,        (powerful,
  cheap)        precise)
     │           │
     └─────┬─────┘
           │
           ▼
      Answer to the user

You don't need a model to do the classifying: simple text heuristics are enough for most cases.

Classifying complexity: simple heuristics

Before you can implement the routing, you need a function that classifies each request. Let's start with text-based heuristics.

Classifying by length and keywords

from dotenv import load_dotenv
load_dotenv()

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

def classify_complexity(messages) -> str:
    """Classify the request's complexity as 'simple' or 'complex'."""
    last_message = messages[-1].content if messages else ""

    complex_keywords = [
        "analyze", "compare", "research", "explain in detail",
        "multi-step", "evaluate", "design", "architecture",
        "pros and cons", "trade-offs", "in depth"
    ]

    is_long = len(last_message) > 200
    has_complex_keywords = any(kw in last_message.lower() for kw in complex_keywords)

    if is_long or has_complex_keywords:
        return "complex"
    return "simple"

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Result: {query} is a key concept in technology."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])

from langchain_core.messages import HumanMessage

simple_msgs = [HumanMessage(content="What is Python?")]
complex_msgs = [HumanMessage(content="Analyze in detail the differences between Python and Rust for distributed backend systems, taking into account performance, memory safety, and ecosystem.")]

print(f"'What is Python?' → {classify_complexity(simple_msgs)}")
print(f"'Analyze in detail...' → {classify_complexity(complex_msgs)}")
# Expected output:
# 'What is Python?' → simple
# 'Analyze in detail...' → complex

Multi-level classification

For more granular routing, you can use three levels instead of two:

from dotenv import load_dotenv
load_dotenv()

from langchain_core.messages import HumanMessage

def classify_complexity_v2(messages) -> str:
    """Classify complexity as 'trivial', 'moderate', or 'complex'."""
    last_message = messages[-1].content if messages else ""
    text = last_message.lower()

    trivial_patterns = ["hi", "thanks", "ok", "yes", "no", "bye"]
    complex_keywords = [
        "analyze", "compare", "research", "explain in detail",
        "design", "architecture", "evaluate", "in depth"
    ]

    if any(text.strip() == p for p in trivial_patterns):
        return "trivial"
    if len(last_message) > 200 or any(kw in text for kw in complex_keywords):
        return "complex"
    return "moderate"

test_cases = [
    "Hi",
    "What is Python?",
    "Analyze the differences between microservices and monoliths, weighing scalability, maintenance, deployment, and operating costs of each architecture.",
]

for msg_text in test_cases:
    msgs = [HumanMessage(content=msg_text)]
    level = classify_complexity_v2(msgs)
    print(f"[{level:>8}] {msg_text[:60]}...")
# Expected output:
# [ trivial] Hi...
# [moderate] What is Python?...
# [ complex] Analyze the differences between microservices and monoliths...

Implementing routing with @wrap_model_call

This is where it all comes together. You use @wrap_model_call to intercept the model call and swap out the model based on the complexity classification.

Basic routing: two models

from dotenv import load_dotenv
load_dotenv()

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

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

def classify_complexity(messages) -> str:
    """Classify the request's complexity."""
    last_message = messages[-1].content if messages else ""
    complex_keywords = ["analyze", "compare", "research", "explain in detail", "multi-step"]
    is_complex = (
        len(last_message) > 200 or
        any(kw in last_message.lower() for kw in complex_keywords)
    )
    return "complex" if is_complex else "simple"

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Result: {query} — relevant information found."

def wrap_model_call(messages, config, call_next):
    """Pick the model based on the question's complexity."""
    complexity = classify_complexity(messages)

    if complexity == "simple":
        print(f"[ROUTING] → gpt-4.1-mini (simple)")
        response = cheap_model.invoke(messages)
    else:
        print(f"[ROUTING] → gpt-4.1 (complex)")
        response = powerful_model.invoke(messages)

    return response

agent = create_agent(
    cheap_model,
    [search],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "What is Python?")]})
print(f"\nAnswer: {result['messages'][-1].content[:100]}")
# Expected output:
# [ROUTING] → gpt-4.1-mini (simple)
# [ROUTING] → gpt-4.1-mini (simple)
# 
# Answer: Python is a high-level, interpreted, general-purpose programming language...

Notice that [ROUTING] can show up several times: once for the first model call (which decides whether to call tools), and again when the agent produces the final answer.

Routing with cost logging

from dotenv import load_dotenv
load_dotenv()

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

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

COST_PER_CALL = {
    "gpt-4.1-mini": 0.0004,
    "gpt-4.1": 0.005,
}

routing_log = []

def classify_complexity(messages) -> str:
    last_message = messages[-1].content if messages else ""
    complex_keywords = ["analyze", "compare", "research", "explain in detail"]
    if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
        return "complex"
    return "simple"

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Result: {query} is a modern technology concept."

def wrap_model_call(messages, config, call_next):
    complexity = classify_complexity(messages)

    if complexity == "simple":
        model_name = "gpt-4.1-mini"
        response = cheap_model.invoke(messages)
    else:
        model_name = "gpt-4.1"
        response = powerful_model.invoke(messages)

    cost = COST_PER_CALL[model_name]
    routing_log.append({
        "model": model_name,
        "complexity": complexity,
        "cost": cost,
    })
    print(f"[ROUTING] {model_name} | complexity={complexity} | cost=${cost}")

    return response

agent = create_agent(
    cheap_model,
    [search],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "What is Docker?")]})
print(f"\nAnswer: {result['messages'][-1].content[:100]}")
print(f"\n--- Routing summary ---")
total_cost = sum(entry["cost"] for entry in routing_log)
print(f"Total calls: {len(routing_log)}")
print(f"Total cost: ${total_cost:.4f}")
print(f"Cost if everything were gpt-4.1: ${len(routing_log) * COST_PER_CALL['gpt-4.1']:.4f}")
print(f"Savings: ${(len(routing_log) * COST_PER_CALL['gpt-4.1']) - total_cost:.4f}")
# Expected output:
# [ROUTING] gpt-4.1-mini | complexity=simple | cost=$0.0004
# [ROUTING] gpt-4.1-mini | complexity=simple | cost=$0.0004
# 
# Answer: Docker is a container platform that lets you package applications with all their...
# 
# --- Routing summary ---
# Total calls: 2
# Total cost: $0.0008
# Cost if everything were gpt-4.1: $0.0100
# Savings: $0.0092

init_chat_model for dynamic selection

init_chat_model takes a string in the form "provider:model_name", which lets you initialize models from different providers behind the same interface.

Initializing several providers

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

models = {
    "cheap": init_chat_model("openai:gpt-4.1-mini"),
    "balanced": init_chat_model("openai:gpt-4.1"),
    "powerful": init_chat_model("anthropic:claude-sonnet-4-20250514"),
}

for name, model in models.items():
    response = model.invoke([("user", "Say 'hello' in one word.")])
    print(f"[{name:>9}] {response.content}")
# Expected output:
# [    cheap] Hello
# [ balanced] Hello
# [ powerful] Hello

Routing with dynamic init_chat_model

Instead of pre-initializing every model, you can create the model on demand:

from dotenv import load_dotenv
load_dotenv()

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

MODEL_REGISTRY = {
    "simple": "openai:gpt-4.1-mini",
    "complex": "openai:gpt-4.1",
}

model_cache = {}

def get_model(complexity: str):
    """Get or create a model for a given complexity."""
    if complexity not in model_cache:
        model_id = MODEL_REGISTRY[complexity]
        model_cache[complexity] = init_chat_model(model_id)
        print(f"[CACHE] Model '{model_id}' initialized and cached")
    return model_cache[complexity]

def classify_complexity(messages) -> str:
    last_message = messages[-1].content if messages else ""
    complex_keywords = ["analyze", "compare", "research", "explain in detail"]
    if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
        return "complex"
    return "simple"

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Result: {query} is relevant in the current context."

def wrap_model_call(messages, config, call_next):
    complexity = classify_complexity(messages)
    model = get_model(complexity)
    model_id = MODEL_REGISTRY[complexity]
    print(f"[ROUTING] → {model_id} ({complexity})")
    return model.invoke(messages)

base_model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(base_model, [search], wrap_model_call=wrap_model_call)

result = agent.invoke({"messages": [("user", "What is Kubernetes?")]})
print(f"\nAnswer: {result['messages'][-1].content[:100]}")
# Expected output:
# [CACHE] Model 'openai:gpt-4.1-mini' initialized and cached
# [ROUTING] → openai:gpt-4.1-mini (simple)
# [ROUTING] → openai:gpt-4.1-mini (simple)
# 
# Answer: Kubernetes is an open-source container orchestration platform that automates...

The cache keeps you from re-initializing models on every call — they get built once and reused.

Cost optimization: tracking savings

In production, you need real metrics to justify the routing. This pattern tracks the accumulated savings.

from dotenv import load_dotenv
load_dotenv()

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

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

COST_PER_1K_TOKENS = {
    "gpt-4.1-mini": {"input": 0.0004, "output": 0.0016},
    "gpt-4.1": {"input": 0.002, "output": 0.008},
}

class CostTracker:
    def __init__(self):
        self.calls = []

    def log(self, model_name: str, complexity: str):
        baseline_cost = COST_PER_1K_TOKENS["gpt-4.1"]["input"]
        actual_cost = COST_PER_1K_TOKENS[model_name]["input"]
        self.calls.append({
            "model": model_name,
            "complexity": complexity,
            "actual_cost": actual_cost,
            "baseline_cost": baseline_cost,
        })

    def summary(self) -> dict:
        total_actual = sum(c["actual_cost"] for c in self.calls)
        total_baseline = sum(c["baseline_cost"] for c in self.calls)
        savings = total_baseline - total_actual
        pct = (savings / total_baseline * 100) if total_baseline > 0 else 0
        return {
            "total_calls": len(self.calls),
            "simple_calls": sum(1 for c in self.calls if c["complexity"] == "simple"),
            "complex_calls": sum(1 for c in self.calls if c["complexity"] == "complex"),
            "actual_cost": total_actual,
            "baseline_cost": total_baseline,
            "savings": savings,
            "savings_pct": pct,
        }

tracker = CostTracker()

def classify_complexity(messages) -> str:
    last_message = messages[-1].content if messages else ""
    complex_keywords = ["analyze", "compare", "research", "explain in detail"]
    if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
        return "complex"
    return "simple"

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Result: {query} is a relevant topic."

def wrap_model_call(messages, config, call_next):
    complexity = classify_complexity(messages)
    if complexity == "simple":
        model_name = "gpt-4.1-mini"
        response = cheap_model.invoke(messages)
    else:
        model_name = "gpt-4.1"
        response = powerful_model.invoke(messages)
    tracker.log(model_name, complexity)
    return response

agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)

questions = [
    "What is Python?",
    "Hi",
    "Analyze the differences between REST and GraphQL in detail",
    "What is 2+2?",
    "Compare the pros and cons of microservices vs monoliths",
]

for q in questions:
    print(f"\n--- Question: {q[:50]}... ---")
    result = agent.invoke({"messages": [("user", q)]})
    print(f"Answer: {result['messages'][-1].content[:80]}...")

s = tracker.summary()
print(f"\n{'='*50}")
print(f"COST SUMMARY")
print(f"{'='*50}")
print(f"Total model calls: {s['total_calls']}")
print(f"  → Simple (gpt-4.1-mini): {s['simple_calls']}")
print(f"  → Complex (gpt-4.1):     {s['complex_calls']}")
print(f"Actual cost:   ${s['actual_cost']:.4f}/1K tokens")
print(f"Baseline cost: ${s['baseline_cost']:.4f}/1K tokens")
print(f"Savings:       ${s['savings']:.4f} ({s['savings_pct']:.1f}%)")
# Expected output:
# --- Question: What is Python?... ---
# Answer: Python is a high-level, interpreted, general-purpose programming language...
# 
# --- Question: Hi... ---
# Answer: Hi there! How can I help you?...
# 
# --- Question: Analyze the differences between REST and GraphQL i... ---
# Answer: REST and GraphQL are two different paradigms for designing APIs. REST follows a...
# 
# --- Question: What is 2+2?... ---
# Answer: 2 + 2 = 4...
# 
# --- Question: Compare the pros and cons of microservices vs mono... ---
# Answer: Microservices and monoliths represent two different architectural approaches...
# 
# ==================================================
# COST SUMMARY
# ==================================================
# Total model calls: 10
# → Simple (gpt-4.1-mini): 6
# → Complex (gpt-4.1):     4
# Actual cost:   $0.0104/1K tokens
# Baseline cost: $0.0200/1K tokens
# Savings:       $0.0096 (48.0%)

Latency optimization: a fast model for real-time

Routing doesn't just save money — it also cuts latency. Smaller models respond faster, which improves the user experience in interactive interfaces.

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool

fast_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

def classify_complexity(messages) -> str:
    last_message = messages[-1].content if messages else ""
    complex_keywords = ["analyze", "compare", "research", "explain in detail"]
    if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
        return "complex"
    return "simple"

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Result for '{query}': relevant data found."

latency_log = []

def wrap_model_call(messages, config, call_next):
    complexity = classify_complexity(messages)
    start = time.time()

    if complexity == "simple":
        model_name = "gpt-4.1-mini"
        response = fast_model.invoke(messages)
    else:
        model_name = "gpt-4.1"
        response = powerful_model.invoke(messages)

    elapsed_ms = (time.time() - start) * 1000
    latency_log.append({"model": model_name, "latency_ms": elapsed_ms})
    print(f"[LATENCY] {model_name}: {elapsed_ms:.0f}ms")
    return response

agent = create_agent(fast_model, [search], wrap_model_call=wrap_model_call)

result = agent.invoke({"messages": [("user", "What is Docker?")]})
print(f"\nAnswer: {result['messages'][-1].content[:80]}")

if latency_log:
    avg_latency = sum(e["latency_ms"] for e in latency_log) / len(latency_log)
    print(f"\nAverage latency: {avg_latency:.0f}ms across {len(latency_log)} calls")
# Expected output:
# [LATENCY] gpt-4.1-mini: 450ms
# [LATENCY] gpt-4.1-mini: 380ms
# 
# Answer: Docker is a container platform that lets you package applications with
# 
# Average latency: 415ms across 2 calls

Multi-criteria routing: complexity + budget + latency

In real systems, the routing decision doesn't hinge on complexity alone. You can factor in several things: remaining budget, latency requirements, and the type of task.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from dataclasses import dataclass

@dataclass
class RoutingConfig:
    budget_remaining: float = 1.0
    max_latency_ms: float = 5000
    prefer_quality: bool = False

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

MODEL_COSTS = {
    "gpt-4.1-mini": 0.0004,
    "gpt-4.1": 0.005,
}

routing_config = RoutingConfig(budget_remaining=0.05, max_latency_ms=3000)

def classify_complexity(messages) -> str:
    last_message = messages[-1].content if messages else ""
    complex_keywords = ["analyze", "compare", "research", "explain in detail"]
    if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
        return "complex"
    return "simple"

def select_model(complexity: str, config: RoutingConfig) -> tuple:
    """Pick a model weighing several criteria."""
    if config.budget_remaining < MODEL_COSTS["gpt-4.1"]:
        return cheap_model, "gpt-4.1-mini", "not enough budget for gpt-4.1"

    if config.max_latency_ms < 1000:
        return cheap_model, "gpt-4.1-mini", "latency requirement < 1s"

    if complexity == "complex" or config.prefer_quality:
        return powerful_model, "gpt-4.1", "high complexity or a preference for quality"

    return cheap_model, "gpt-4.1-mini", "simple request, optimizing for cost"

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Result: {query} is a relevant engineering concept."

def wrap_model_call(messages, config, call_next):
    complexity = classify_complexity(messages)
    model, model_name, reason = select_model(complexity, routing_config)

    cost = MODEL_COSTS[model_name]
    routing_config.budget_remaining -= cost

    print(f"[ROUTING] {model_name} | reason: {reason}")
    print(f"[BUDGET]  remaining: ${routing_config.budget_remaining:.4f}")

    return model.invoke(messages)

agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)

result = agent.invoke({
    "messages": [("user", "What is Kubernetes?")]
})
print(f"\nAnswer: {result['messages'][-1].content[:100]}")
# Expected output:
# [ROUTING] gpt-4.1-mini | reason: simple request, optimizing for cost
# [BUDGET]  remaining: $0.0496
# [ROUTING] gpt-4.1-mini | reason: simple request, optimizing for cost
# [BUDGET]  remaining: $0.0492
# 
# Answer: Kubernetes is an open-source container orchestration platform that makes deploying...

Fallback patterns: escalating on quality

An advanced pattern: send the request to the cheap model first, judge the quality of the answer, and if it isn't good enough, re-send it to the powerful one. That gets you the best of both worlds.

Simple fallback: by response length

from dotenv import load_dotenv
load_dotenv()

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

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

MIN_RESPONSE_LENGTH = 50

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Result: {query} has plenty of practical applications."

def wrap_model_call(messages, config, call_next):
    print(f"[FALLBACK] Trying gpt-4.1-mini...")
    response = cheap_model.invoke(messages)

    if len(response.content) < MIN_RESPONSE_LENGTH:
        print(f"[FALLBACK] Short answer ({len(response.content)} chars), escalating to gpt-4.1...")
        response = powerful_model.invoke(messages)
        print(f"[FALLBACK] gpt-4.1 replied ({len(response.content)} chars)")
    else:
        print(f"[FALLBACK] gpt-4.1-mini was enough ({len(response.content)} chars)")

    return response

agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)

result = agent.invoke({"messages": [("user", "What is LangChain?")]})
print(f"\nAnswer: {result['messages'][-1].content[:120]}")
# Expected output:
# [FALLBACK] Trying gpt-4.1-mini...
# [FALLBACK] gpt-4.1-mini was enough (187 chars)
# [FALLBACK] Trying gpt-4.1-mini...
# [FALLBACK] gpt-4.1-mini was enough (203 chars)
# 
# Answer: LangChain is an open-source framework designed to make it easier to build applications that use language mod...

Fallback with content checking

from dotenv import load_dotenv
load_dotenv()

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

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

def quality_check(response_text: str) -> bool:
    """Check whether the answer clears a minimum quality bar."""
    if len(response_text) < 30:
        return False
    low_quality_indicators = [
        "i don't know", "i don't have information", "i can't",
        "as a language model", "no lo sé"
    ]
    if any(indicator in response_text.lower() for indicator in low_quality_indicators):
        return False
    return True

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Result: {query} is an important technical concept."

fallback_stats = {"direct": 0, "escalated": 0}

def wrap_model_call(messages, config, call_next):
    response = cheap_model.invoke(messages)

    if quality_check(response.content):
        fallback_stats["direct"] += 1
        print(f"[QUALITY] ✅ gpt-4.1-mini passed the quality check")
        return response

    fallback_stats["escalated"] += 1
    print(f"[QUALITY] ⚠️ Escalating to gpt-4.1")
    return powerful_model.invoke(messages)

agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)

result = agent.invoke({"messages": [("user", "Explain what Docker is")]})
print(f"\nAnswer: {result['messages'][-1].content[:100]}")
print(f"\nStats: {fallback_stats}")
# Expected output:
# [QUALITY] ✅ gpt-4.1-mini passed the quality check
# [QUALITY] ✅ gpt-4.1-mini passed the quality check
# 
# Answer: Docker is a containerization platform that lets you package applications along with all...
# 
# Stats: {'direct': 2, 'escalated': 0}

Comparison: static model vs dynamic routing

When is dynamic routing worth the effort versus just using a single model?

When to use which

• ✅ Static model — Prototypes, internal apps, budget isn't a concern
• ✅ Dynamic routing — High-volume production with a mix of simple and complex requests
• ⚠️ Don't use dynamic routing if your volume is low (<100 requests/day) — the complexity isn't worth it
• ❌ Never use only the cheap model for everything if quality matters on the complex questions

A side-by-side example

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

COST_PER_CALL = {"gpt-4.1-mini": 0.0004, "gpt-4.1": 0.005}

request_profile = {
    "simple": 70,
    "complex": 30,
}

cost_static_powerful = 100 * COST_PER_CALL["gpt-4.1"]
cost_static_cheap = 100 * COST_PER_CALL["gpt-4.1-mini"]
cost_dynamic = (
    request_profile["simple"] * COST_PER_CALL["gpt-4.1-mini"] +
    request_profile["complex"] * COST_PER_CALL["gpt-4.1"]
)

print("Scenario: 100 requests (70% simple, 30% complex)")
print(f"  gpt-4.1 only:      ${cost_static_powerful:.4f}")
print(f"  gpt-4.1-mini only: ${cost_static_cheap:.4f}  (poor quality on the complex ones)")
print(f"  Dynamic routing:   ${cost_dynamic:.4f}  (high quality where it matters)")
print(f"\nSavings vs static gpt-4.1: {((cost_static_powerful - cost_dynamic) / cost_static_powerful * 100):.1f}%")
# Expected output:
# Scenario: 100 requests (70% simple, 30% complex)
#   gpt-4.1 only:      $0.5000
#   gpt-4.1-mini only: $0.0400  (poor quality on the complex ones)
#   Dynamic routing:   $0.1780  (high quality where it matters)
# 
# Savings vs static gpt-4.1: 64.4%

Connection to the project

In the module project (Capsule 08), you'll build a complete agent with dynamic model routing: the agent automatically picks between the cheap and the powerful model based on complexity, combined with logging middleware to monitor every routing decision. That will integrate with dynamic tools (Capsule 06) and the AgentMiddleware class (Capsule 07) to create a composed middleware system.

Troubleshooting

Problem 1: Everything gets classified as "simple"

Cause: Your complexity keywords don't match the user's actual messages, or you're reading the wrong message. Fix: Check which message you're analyzing — inside the agent's loop, messages includes the whole history:

def classify_complexity(messages):
    last_message = messages[-1].content if messages else ""
    print(f"[DEBUG] Analyzing: '{last_message[:80]}'")
    # ... rest of classification

Problem 2: wrap_model_call doesn't intercept every call

Cause: Some of the agent's internal calls may bypass the wrapper if it isn't wired up correctly. Fix: Make sure you're passing wrap_model_call as a create_agent parameter:

agent = create_agent(
    model,
    tools,
    wrap_model_call=wrap_model_call,  # Not as a decorator — as a parameter
)

Problem 3: The model cache grows without bound

Cause: If you call init_chat_model inside the wrapper without a cache, you build a brand-new model on every call. Fix: Pre-initialize the models outside the wrapper, or use a cache dictionary:

model_cache = {}
def get_model(model_id):
    if model_id not in model_cache:
        model_cache[model_id] = init_chat_model(model_id)
    return model_cache[model_id]

Problem 4: The fallback always fires (answers always come out "short")

Cause: Your quality threshold is too high, or you're measuring before the model has finished the answer. Fix: Tune the thresholds and check that response.content actually holds the complete answer:

print(f"[DEBUG] Answer length: {len(response.content)}")
print(f"[DEBUG] Content: {response.content[:200]}")

Problem 5: The costs don't match what you expected

Cause: Each agent invocation can make several model calls (one to decide on tools, another for the final answer), and the wrapper runs on every one of them. Fix: Keep in mind that a single agent.invoke() can trigger 2+ calls into the wrapper:

def wrap_model_call(messages, config, call_next):
    # This function runs N times per invoke(), not just once
    print(f"[DEBUG] Call #{len(routing_log) + 1}")
    # ...

Exercises

Exercise 1: Basic routing with two models (Easy)

Build an agent with a wrap_model_call that sends questions under 50 characters to gpt-4.1-mini and longer ones to gpt-4.1. Print which model gets used on each call.

from dotenv import load_dotenv
load_dotenv()

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

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

@tool
def search(query: str) -> str:
    """Search for information about a topic."""
    return f"Info: {query} is a topic of interest."

def wrap_model_call(messages, config, call_next):
    last_msg = messages[-1].content if messages else ""
    if len(last_msg) < 50:
        print(f"[SHORT] → gpt-4.1-mini ({len(last_msg)} chars)")
        return cheap_model.invoke(messages)
    else:
        print(f"[LONG] → gpt-4.1 ({len(last_msg)} chars)")
        return powerful_model.invoke(messages)

agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)

result = agent.invoke({"messages": [("user", "What is Python?")]})
print(f"Answer: {result['messages'][-1].content[:80]}")
# Expected output:
# [SHORT] → gpt-4.1-mini (15 chars)
# [SHORT] → gpt-4.1-mini (15 chars)
# Answer: Python is a high-level, interpreted, general-purpose programming language...

Exercise 2: CostTracker with a summary (Easy)

Write a CostTracker class that records each model call with its estimated cost. At the end, print a summary with the total spend and the savings versus always using the expensive model.

from dotenv import load_dotenv
load_dotenv()

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

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

COSTS = {"gpt-4.1-mini": 0.0004, "gpt-4.1": 0.005}

class CostTracker:
    def __init__(self):
        self.entries = []

    def log(self, model_name: str):
        self.entries.append({"model": model_name, "cost": COSTS[model_name]})

    def summary(self):
        total = sum(e["cost"] for e in self.entries)
        baseline = len(self.entries) * COSTS["gpt-4.1"]
        saving = baseline - total
        print(f"Calls: {len(self.entries)}")
        print(f"Actual cost:   ${total:.4f}")
        print(f"Baseline cost: ${baseline:.4f}")
        print(f"Savings:       ${saving:.4f} ({saving/baseline*100:.1f}%)")

tracker = CostTracker()

def classify_complexity(messages) -> str:
    last_msg = messages[-1].content if messages else ""
    if len(last_msg) > 100 or "analyze" in last_msg.lower():
        return "complex"
    return "simple"

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Result: {query}."

def wrap_model_call(messages, config, call_next):
    complexity = classify_complexity(messages)
    if complexity == "simple":
        tracker.log("gpt-4.1-mini")
        return cheap_model.invoke(messages)
    else:
        tracker.log("gpt-4.1")
        return powerful_model.invoke(messages)

agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)

for q in ["Hi", "What is Python?", "Analyze REST vs GraphQL"]:
    agent.invoke({"messages": [("user", q)]})

print("\n--- Summary ---")
tracker.summary()
# Expected output:
# --- Summary ---
# Calls: 6
# Actual cost:   $0.0116
# Baseline cost: $0.0300
# Savings:       $0.0184 (61.3%)

Exercise 3: Multi-level classification (Medium)

Implement three-level classification (trivial, moderate, complex) and give each level a different model: gpt-4.1-mini for trivial, gpt-4.1-mini at a higher temperature for moderate, and gpt-4.1 for complex.

from dotenv import load_dotenv
load_dotenv()

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

trivial_model = init_chat_model("openai:gpt-4.1-mini", temperature=0)
moderate_model = init_chat_model("openai:gpt-4.1-mini", temperature=0.7)
complex_model = init_chat_model("openai:gpt-4.1")

def classify_three_levels(messages) -> str:
    last_msg = messages[-1].content if messages else ""
    text = last_msg.lower().strip()

    trivial_patterns = ["hi", "thanks", "ok", "yes", "no", "goodbye", "bye"]
    complex_keywords = ["analyze", "compare", "research", "design", "explain in detail"]

    if text in trivial_patterns or len(text) < 10:
        return "trivial"
    if len(last_msg) > 200 or any(kw in text for kw in complex_keywords):
        return "complex"
    return "moderate"

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Result: {query} is relevant."

routing_history = []

def wrap_model_call(messages, config, call_next):
    level = classify_three_levels(messages)
    models = {
        "trivial": (trivial_model, "gpt-4.1-mini (t=0)"),
        "moderate": (moderate_model, "gpt-4.1-mini (t=0.7)"),
        "complex": (complex_model, "gpt-4.1"),
    }
    model, name = models[level]
    routing_history.append({"level": level, "model": name})
    print(f"[{level:>8}] → {name}")
    return model.invoke(messages)

agent = create_agent(trivial_model, [search], wrap_model_call=wrap_model_call)

for q in ["Hi", "What is Python?", "Analyze microservices vs monoliths"]:
    print(f"\nQuestion: {q}")
    result = agent.invoke({"messages": [("user", q)]})
    print(f"Answer: {result['messages'][-1].content[:60]}...")

print(f"\n--- Routing history ---")
for entry in routing_history:
    print(f"  [{entry['level']:>8}] {entry['model']}")
# Expected output:
# Question: Hi
# [ trivial] → gpt-4.1-mini (t=0)
# Answer: Hi there! How can I help you?...
# 
# Question: What is Python?
# [moderate] → gpt-4.1-mini (t=0.7)
# [moderate] → gpt-4.1-mini (t=0.7)
# Answer: Python is a high-level, interpreted programming languag...
# 
# Question: Analyze microservices vs monoliths
# [ complex] → gpt-4.1
# [ complex] → gpt-4.1
# Answer: Microservices and monoliths are two architectural appro...
# 
# --- Routing history ---
#   [ trivial] gpt-4.1-mini (t=0)
#   [moderate] gpt-4.1-mini (t=0.7)
#   [moderate] gpt-4.1-mini (t=0.7)
#   [ complex] gpt-4.1
#   [ complex] gpt-4.1

Exercise 4: Fallback with a quality check (Medium)

Implement a fallback that tries gpt-4.1-mini first. If the answer contains phrases like "I don't know" or "I don't have information", escalate automatically to gpt-4.1.

from dotenv import load_dotenv
load_dotenv()

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

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

LOW_QUALITY_PHRASES = [
    "i don't know", "i don't have information", "i can't answer",
    "as a language model", "i don't have access",
]

def is_low_quality(text: str) -> bool:
    return any(phrase in text.lower() for phrase in LOW_QUALITY_PHRASES)

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Result: {query} has production applications."

fallback_log = {"accepted": 0, "escalated": 0}

def wrap_model_call(messages, config, call_next):
    response = cheap_model.invoke(messages)

    if is_low_quality(response.content):
        fallback_log["escalated"] += 1
        print(f"[FALLBACK] ⚠️ Low quality detected, escalating to gpt-4.1")
        return powerful_model.invoke(messages)

    fallback_log["accepted"] += 1
    print(f"[FALLBACK] ✅ gpt-4.1-mini accepted")
    return response

agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)

result = agent.invoke({"messages": [("user", "What is FastAPI?")]})
print(f"\nAnswer: {result['messages'][-1].content[:80]}")
print(f"\nStats: {fallback_log}")
# Expected output:
# [FALLBACK] ✅ gpt-4.1-mini accepted
# [FALLBACK] ✅ gpt-4.1-mini accepted
# 
# Answer: FastAPI is a modern, fast framework for building APIs with Python, based on...
# 
# Stats: {'accepted': 2, 'escalated': 0}

Exercise 5: Routing on a limited budget (Hard)

Build a routing system that respects a maximum budget. If the budget is about to run out, force the cheap model regardless of complexity. Simulate 5 questions and show the remaining budget after each one.

from dotenv import load_dotenv
load_dotenv()

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

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

COSTS = {"gpt-4.1-mini": 0.0004, "gpt-4.1": 0.005}

class BudgetRouter:
    def __init__(self, budget: float):
        self.budget = budget
        self.initial_budget = budget
        self.history = []

    def select_and_charge(self, complexity: str) -> tuple:
        if self.budget < COSTS["gpt-4.1"]:
            model_name = "gpt-4.1-mini"
            reason = "forced by budget"
        elif complexity == "complex":
            model_name = "gpt-4.1"
            reason = "high complexity"
        else:
            model_name = "gpt-4.1-mini"
            reason = "simple request"

        cost = COSTS[model_name]
        self.budget -= cost
        self.history.append({
            "model": model_name, "cost": cost,
            "budget_after": self.budget, "reason": reason,
        })
        return (powerful_model if model_name == "gpt-4.1" else cheap_model), model_name, reason

router = BudgetRouter(budget=0.015)

def classify_complexity(messages) -> str:
    last_msg = messages[-1].content if messages else ""
    complex_keywords = ["analyze", "compare", "research"]
    if any(kw in last_msg.lower() for kw in complex_keywords):
        return "complex"
    return "simple"

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Result: {query}."

def wrap_model_call(messages, config, call_next):
    complexity = classify_complexity(messages)
    model, name, reason = router.select_and_charge(complexity)
    print(f"  [{name}] reason={reason} | remaining=${router.budget:.4f}")
    return model.invoke(messages)

agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)

questions = [
    "What is Python?",
    "Analyze the pros and cons of Docker",
    "Hi",
    "Compare Kubernetes vs Docker Swarm",
    "What time is it?",
]

for i, q in enumerate(questions, 1):
    print(f"\n[Q{i}] {q}")
    result = agent.invoke({"messages": [("user", q)]})
    print(f"  Answer: {result['messages'][-1].content[:60]}...")

print(f"\n{'='*50}")
print(f"Starting budget: ${router.initial_budget:.4f}")
print(f"Final budget:    ${router.budget:.4f}")
print(f"Spent:           ${router.initial_budget - router.budget:.4f}")
# Expected output:
# [Q1] What is Python?
#   [gpt-4.1-mini] reason=simple request | remaining=$0.0146
#   [gpt-4.1-mini] reason=simple request | remaining=$0.0142
#   Answer: Python is a high-level, interpreted programming language...
# 
# [Q2] Analyze the pros and cons of Docker
#   [gpt-4.1] reason=high complexity | remaining=$0.0092
#   [gpt-4.1] reason=high complexity | remaining=$0.0042
#   Answer: Docker has several pros and cons worth weighing up...
# 
# [Q3] Hi
#   [gpt-4.1-mini] reason=simple request | remaining=$0.0038
#   Answer: Hi there! How can I help you?...
# 
# [Q4] Compare Kubernetes vs Docker Swarm
#   [gpt-4.1-mini] reason=forced by budget | remaining=$0.0034
#   [gpt-4.1-mini] reason=forced by budget | remaining=$0.0030
#   Answer: Kubernetes and Docker Swarm are orchestration platforms...
# 
# [Q5] What time is it?
#   [gpt-4.1-mini] reason=simple request | remaining=$0.0026
#   Answer: I don't have access to the current time, but you can check...
# 
# ==================================================
# Starting budget: $0.0150
# Final budget:    $0.0026
# Spent:           $0.0124

Exercise 6: Real-time routing dashboard (Hard)

Build a complete system that combines complexity routing, cost tracking, and latency tracking. At the end, print a dashboard with all the metrics.

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool

cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")

COSTS = {"gpt-4.1-mini": 0.0004, "gpt-4.1": 0.005}

class RoutingDashboard:
    def __init__(self):
        self.entries = []

    def log(self, model: str, complexity: str, latency_ms: float):
        self.entries.append({
            "model": model, "complexity": complexity,
            "latency_ms": latency_ms, "cost": COSTS[model],
        })

    def print_dashboard(self):
        if not self.entries:
            print("No data.")
            return

        total_cost = sum(e["cost"] for e in self.entries)
        baseline_cost = len(self.entries) * COSTS["gpt-4.1"]
        avg_latency = sum(e["latency_ms"] for e in self.entries) / len(self.entries)
        simple_count = sum(1 for e in self.entries if e["complexity"] == "simple")
        complex_count = sum(1 for e in self.entries if e["complexity"] == "complex")
        simple_lat = [e["latency_ms"] for e in self.entries if e["complexity"] == "simple"]
        complex_lat = [e["latency_ms"] for e in self.entries if e["complexity"] == "complex"]

        print(f"\n{'='*55}")
        print(f"  ROUTING DASHBOARD")
        print(f"{'='*55}")
        print(f"  Total calls:        {len(self.entries)}")
        print(f"  Simple:             {simple_count}")
        print(f"  Complex:            {complex_count}")
        print(f"{'─'*55}")
        print(f"  Total cost:         ${total_cost:.4f}")
        print(f"  Baseline cost:      ${baseline_cost:.4f}")
        print(f"  Savings:            ${baseline_cost - total_cost:.4f} ({(baseline_cost - total_cost)/baseline_cost*100:.1f}%)")
        print(f"{'─'*55}")
        print(f"  Average latency:    {avg_latency:.0f}ms")
        if simple_lat:
            print(f"  Simple latency:     {sum(simple_lat)/len(simple_lat):.0f}ms")
        if complex_lat:
            print(f"  Complex latency:    {sum(complex_lat)/len(complex_lat):.0f}ms")
        print(f"{'='*55}")

dashboard = RoutingDashboard()

def classify_complexity(messages) -> str:
    last_msg = messages[-1].content if messages else ""
    complex_keywords = ["analyze", "compare", "research", "explain in detail"]
    if len(last_msg) > 150 or any(kw in last_msg.lower() for kw in complex_keywords):
        return "complex"
    return "simple"

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Result: {query} is an important concept."

def wrap_model_call(messages, config, call_next):
    complexity = classify_complexity(messages)
    start = time.time()

    if complexity == "simple":
        model_name = "gpt-4.1-mini"
        response = cheap_model.invoke(messages)
    else:
        model_name = "gpt-4.1"
        response = powerful_model.invoke(messages)

    latency_ms = (time.time() - start) * 1000
    dashboard.log(model_name, complexity, latency_ms)
    return response

agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)

questions = [
    "What is Docker?",
    "Analyze in detail the advantages of microservices over monoliths",
    "Hi",
    "What is Python?",
    "Compare React, Vue, and Angular for enterprise applications",
]

for q in questions:
    print(f"→ {q[:50]}...")
    agent.invoke({"messages": [("user", q)]})

dashboard.print_dashboard()
# Expected output:
# → What is Docker?...
# → Analyze in detail the advantages of microservices...
# → Hi...
# → What is Python?...
# → Compare React, Vue, and Angular for enterprise app...
# 
# =======================================================
#   ROUTING DASHBOARD
# =======================================================
#   Total calls:        10
#   Simple:             6
#   Complex:            4
# ───────────────────────────────────────────────────────
#   Total cost:         $0.0224
#   Baseline cost:      $0.0500
#   Savings:            $0.0276 (55.2%)
# ───────────────────────────────────────────────────────
#   Average latency:    520ms
#   Simple latency:     380ms
#   Complex latency:    730ms
# =======================================================

Summary

In this capsule you learned:

• Model routing automatically picks the best model for each request based on its complexity
• classify_complexity() uses simple heuristics (length, keywords) to classify requests without needing an extra model
• @wrap_model_call is the perfect interception point for routing — you already knew how to intercept the call, now you swap the model itself
• init_chat_model lets you initialize models from several providers ("openai:gpt-4.1-mini", "anthropic:claude-sonnet") behind the same interface
• Cost tracking measures the real savings from routing versus a static model — typically 40-80%
• Latency tracking confirms that cheaper models are meaningfully faster
• Multi-criteria routing weighs complexity + budget + latency for better decisions
• Fallback patterns try the cheap model first and escalate when the quality falls short

Next capsule: Dynamic Tools and Dynamic Prompts — how to filter the available tools by user permissions and generate dynamic prompts from context.

Further reading

1. init_chat_model API Reference — Reference for init_chat_model and the supported providers
2. How to add custom middleware to agents — The official guide to middleware in agents
3. Model Routing Patterns — Model routing patterns in LangChain
4. OpenAI Pricing — Current pricing for OpenAI models
5. Anthropic Pricing — Pricing for Claude models
6. LangSmith Cost Tracking — Monitoring costs in production with LangSmith

Module 4 — LangChain & LangGraph: From Chains to Agents