Project: An Agent with Dynamic Model Routing

Project overview

In the seven previous capsules you learned to customize agents without rewriting them: individual hooks with @before_model and @after_model, powerful interceptors with @wrap_model_call and @wrap_tool_call, dynamic models to pick the model at runtime, dynamic tools and dynamic prompts to adapt the agent to its context, and the AgentMiddleware class to package it all into reusable modules. You saw each concept on its own. Now you're going to combine everything into a real system: an agent that automatically picks the model based on the complexity of the question.

The core idea is simple but powerful: not every question needs the same model. "What is the capital of France?" doesn't need GPT-4.1 — a cheap model like GPT-4.1-mini answers it perfectly for a fraction of the cost. But "Analyze the implications of a microservices architecture vs a monolith for a startup with 10 engineers" does need deep reasoning. A smart agent should make that distinction on its own.

In this project you'll build exactly that: a classification middleware that analyzes the complexity of each question, a routing system that picks the right model, a logging middleware that records every decision, and dynamic tools that enable advanced capabilities only when the question calls for them. The result is an agent that optimizes cost and quality at the same time — exactly what you need in production.

The system runs as an interactive terminal chat where you can ask simple and complex questions and watch, in real time, which model was chosen, why, and which tools are available.

---

Project goal

Build an agent with dynamic model routing that automatically picks between a cheap model and a powerful one based on the complexity of the question, with full logging and dynamic tools.

By the end of this project:

• 🔧 You'll know how to implement a complexity classifier as middleware
• 🔧 You'll build a routing system that selects models at runtime
• 🔧 You'll add logging middleware that documents every decision
• 🔧 You'll implement dynamic tools by complexity level
• 🔧 You'll compose several middleware into a production-ready agent
• 🔧 You'll have a working system that ties the whole module together

---

Technical specs

Tech stack

Component	Version	Purpose
Python	3.11+	Runtime
LangChain	v1.2+	LLM framework + middleware
LangGraph	v1.0+	create_agent
langchain-openai	latest	Model provider
python-dotenv	latest	Environment variables

Initial setup

pip install langchain langgraph langchain-openai python-dotenv

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

# .env
OPENAI_API_KEY=sk-...

Project structure

dynamic-routing-agent/
├── .env                          # API key
├── dynamic_routing_agent.py      # Main code (everything in one file)
└── requirements.txt              # Dependencies

# requirements.txt
langchain>=0.3.0
langgraph>=0.3.0
langchain-openai>=0.3.0
python-dotenv>=1.0.0

---

Step 1: Define the tools

The agent needs two levels of tools: basic ones (always available) and advanced ones (only for complex questions).

Basic tools: search and calculator

Available for every kind of question. They're fast, low-cost operations.

from langchain_core.tools import tool

KNOWLEDGE_BASE = {
    "python": {
        "title": "Python Programming Language",
        "content": (
            "Python is a high-level, interpreted language with dynamic typing. "
            "Created by Guido van Rossum in 1991. Used in AI/ML, data science, web, "
            "automation. Ecosystem: PyTorch, TensorFlow, Django, FastAPI. "
            "Community: >8M developers. Current version: 3.12."
        ),
        "source": "https://python.org",
    },
    "rust": {
        "title": "Rust Programming Language",
        "content": (
            "Rust is a systems language with memory safety and no garbage collector. "
            "Created by Mozilla in 2010. Unique ownership system. Performance comparable to C/C++. "
            "Used in: operating systems, WebAssembly, CLIs, networking. "
            "Adopted by the Linux kernel, Android, Windows."
        ),
        "source": "https://rust-lang.org",
    },
    "microservices": {
        "title": "Microservices Architecture",
        "content": (
            "Microservices is an architectural style that structures an application as a "
            "collection of small, autonomous, independently deployable services. "
            "Upsides: independent scaling, independent deploys, heterogeneous technology. "
            "Downsides: network complexity, eventual consistency, operational overhead. "
            "Alternative: a modular monolith for teams under 20 people."
        ),
        "source": "https://microservices.io",
    },
    "langchain": {
        "title": "LangChain Framework",
        "content": (
            "LangChain is the most widely adopted open-source framework for LLM applications. "
            "v1.2+ includes: init_chat_model, create_agent, the middleware system. "
            "Ecosystem: LangGraph (orchestration), LangSmith (observability), Deep Agents. "
            "Used by: AI startups, enterprises, research labs."
        ),
        "source": "https://langchain.com",
    },
}


@tool
def search(query: str) -> str:
    """Search for general information about a topic. Useful for factual questions."""
    query_lower = query.lower()
    results = []
    for key, data in KNOWLEDGE_BASE.items():
        if key in query_lower or any(word in data["content"].lower() for word in query_lower.split()):
            results.append(f"📄 {data['title']}: {data['content']} (Source: {data['source']})")

    if results:
        return f"Found {len(results)} result(s):\n" + "\n\n".join(results)
    return f"No specific results for '{query}'. Available topics: {', '.join(KNOWLEDGE_BASE.keys())}"


@tool
def calculator(expression: str) -> str:
    """Evaluate math expressions. Useful for numeric calculations."""
    try:
        result = eval(expression)
        return f"Result: {expression} = {result}"
    except Exception as e:
        return f"Error evaluating '{expression}': {e}"

Advanced tools: deep_research and data_analysis

Only available for complex questions. They simulate costlier, deeper operations.

@tool
def deep_research(topic: str, aspects: list[str]) -> str:
    """Deep research on a topic across several aspects.
    Only used for complex questions that require detailed analysis."""
    results = []
    for aspect in aspects:
        topic_lower = topic.lower()
        if topic_lower in KNOWLEDGE_BASE:
            data = KNOWLEDGE_BASE[topic_lower]
            results.append(f"🔬 {aspect}: Based on {data['title']} — {data['content']}")
        else:
            results.append(f"🔬 {aspect}: General analysis of {topic} (no specific data)")

    return (
        f"Deep research on '{topic}':\n"
        f"Aspects analyzed: {len(aspects)}\n\n"
        + "\n\n".join(results)
    )


@tool
def data_analysis(data_description: str, analysis_type: str) -> str:
    """Analyze data with a specific type of analysis.
    Only used for complex questions that require data processing."""
    analyses = {
        "comparison": f"📊 Comparison: {data_description} — 3 comparison dimensions identified: performance, cost, and ease of adoption.",
        "tradeoff": f"📊 Trade-offs: {data_description} — Upsides vs downsides analyzed across 4 categories: technical, organizational, financial, and temporal.",
        "recommendation": f"📊 Recommendation: {data_description} — Based on the analysis, the recommendation depends on context: team, budget, and timeline.",
        "trend": f"📊 Trend: {data_description} — Trend analysis shows sustained growth over the last 3 years.",
    }
    return analyses.get(
        analysis_type.lower(),
        f"📊 Analysis ({analysis_type}): {data_description} — Analysis completed with general results."
    )

---

Step 2: Build the complexity classifier middleware

This middleware analyzes the user's question and classifies it as "simple" or "complex". The classification rests on heuristic signals: length, the presence of analytical keywords, and the structure of the question.

import re

COMPLEXITY_KEYWORDS = [
    "analyze", "compare", "evaluate", "explain in detail",
    "implications", "trade-off", "tradeoff", "advantages and disadvantages",
    "architecture", "design", "strategy", "in depth",
    "vs", "versus", "differences between", "pros and cons",
    "recommendation", "recommend", "when to use", "best option",
    "impact", "consequences", "long term",
]


def classify_complexity(text: str) -> dict:
    """Classify how complex a question is."""
    text_lower = text.lower()
    word_count = len(text.split())

    keyword_matches = [kw for kw in COMPLEXITY_KEYWORDS if kw in text_lower]
    has_multiple_questions = text.count("?") > 1
    has_connectors = any(c in text_lower for c in [" and ", " but ", " however", " moreover"])

    score = 0
    reasons = []

    if word_count > 20:
        score += 1
        reasons.append(f"long question ({word_count} words)")
    if keyword_matches:
        score += len(keyword_matches)
        reasons.append(f"analytical keywords: {', '.join(keyword_matches[:3])}")
    if has_multiple_questions:
        score += 1
        reasons.append("multiple questions")
    if has_connectors:
        score += 1
        reasons.append("complex connectors")

    complexity = "complex" if score >= 2 else "simple"

    return {
        "complexity": complexity,
        "score": score,
        "reasons": reasons,
        "word_count": word_count,
    }

---

Step 3: Implement the model routing middleware

This is the heart of the project. It uses the classifier to decide which model to invoke: gpt-4.1-mini (cheap, fast) for simple questions, gpt-4.1 (powerful, expensive) for complex ones.

from langchain.agents import AgentMiddleware
from langchain.chat_models import init_chat_model


class ModelRoutingMiddleware(AgentMiddleware):
    """Pick the model based on the complexity of the question."""

    def __init__(
        self,
        simple_model: str = "openai:gpt-4.1-mini",
        complex_model: str = "openai:gpt-4.1",
    ):
        self._simple = init_chat_model(simple_model, temperature=0)
        self._complex = init_chat_model(complex_model, temperature=0.2)
        self.simple_model_name = simple_model
        self.complex_model_name = complex_model
        self.last_classification = None

    def wrap_model_call(self, messages, config, call_next):
        user_messages = [
            m for m in messages
            if hasattr(m, "type") and m.type == "human"
        ]

        if user_messages:
            last_user_msg = user_messages[-1].content
            classification = classify_complexity(last_user_msg)
        else:
            classification = {"complexity": "simple", "score": 0, "reasons": ["no user message"], "word_count": 0}

        self.last_classification = classification

        if classification["complexity"] == "complex":
            selected = self._complex
            model_name = self.complex_model_name
        else:
            selected = self._simple
            model_name = self.simple_model_name

        print(f"\n  🧠 Model selected: {model_name}")
        print(f"     Complexity: {classification['complexity']} (score: {classification['score']})")
        if classification["reasons"]:
            print(f"     Reasons: {', '.join(classification['reasons'])}")

        tools_on_messages = []
        for m in messages:
            if hasattr(m, "tool_calls") and m.tool_calls:
                tools_on_messages = m.tool_calls
                break

        bound_tools = config.get("__bound_tools__", [])
        if bound_tools:
            selected = selected.bind_tools(bound_tools)

        return selected.invoke(messages)

---

Step 4: Add the logging middleware

This middleware records every operation the agent performs, with details about the model used, the tools executed, and performance metrics.

import time
from datetime import datetime


class LoggingMiddleware(AgentMiddleware):
    """Log every operation in detail."""

    def __init__(self):
        self.log_entries: list[dict] = []
        self._call_count = 0

    def before_model(self, messages, config):
        self._call_count += 1
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "model_call",
            "call_number": self._call_count,
            "message_count": len(messages),
        }
        self.log_entries.append(entry)
        print(f"  📋 [{entry['timestamp']}] Model call #{self._call_count} ({len(messages)} msgs)")

    def after_model(self, response, config):
        has_tool_calls = bool(getattr(response, "tool_calls", []))
        content_len = len(response.content) if response.content else 0
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "model_response",
            "has_tool_calls": has_tool_calls,
            "content_length": content_len,
        }
        self.log_entries.append(entry)
        if has_tool_calls:
            tool_names = [tc["name"] for tc in response.tool_calls]
            print(f"  📋 Model decided to call tools: {', '.join(tool_names)}")
        else:
            preview = response.content[:80] if response.content else "(empty)"
            print(f"  📋 Model answered: {preview}...")

    def wrap_tool_call(self, tool_call, config, call_next):
        start = time.time()
        print(f"  🔧 Running: {tool_call['name']}({str(tool_call['args'])[:60]})")
        result = call_next(tool_call, config)
        elapsed = time.time() - start
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "tool_execution",
            "tool": tool_call["name"],
            "elapsed_ms": round(elapsed * 1000),
        }
        self.log_entries.append(entry)
        preview = str(result)[:80]
        print(f"  📥 Result ({elapsed * 1000:.0f}ms): {preview}")
        return result

    def get_summary(self) -> dict:
        model_calls = sum(1 for e in self.log_entries if e["type"] == "model_call")
        tool_execs = [e for e in self.log_entries if e["type"] == "tool_execution"]
        tools_used = {}
        for e in tool_execs:
            tools_used[e["tool"]] = tools_used.get(e["tool"], 0) + 1
        return {
            "model_calls": model_calls,
            "tool_executions": len(tool_execs),
            "tools_used": tools_used,
            "total_events": len(self.log_entries),
        }

    def reset(self):
        self.log_entries.clear()
        self._call_count = 0

---

Step 5: Implement dynamic tools by complexity

The dynamic tools middleware filters the available tools by the complexity classification. Simple questions only reach search and calculator. Complex questions reach everything: search, calculator, deep_research and data_analysis.

class DynamicToolsMiddleware(AgentMiddleware):
    """Enable the advanced tools only for complex questions."""

    def __init__(self):
        self.basic_tools = [search, calculator]
        self.advanced_tools = [search, calculator, deep_research, data_analysis]

    def get_tools(self, state):
        messages = state.get("messages", [])
        user_messages = [
            m for m in messages
            if hasattr(m, "type") and m.type == "human"
        ]

        if user_messages:
            last_msg = user_messages[-1].content
            classification = classify_complexity(last_msg)
        else:
            classification = {"complexity": "simple"}

        if classification["complexity"] == "complex":
            tool_names = [t.name for t in self.advanced_tools]
            print(f"  🔧 Tools enabled (complex): {', '.join(tool_names)}")
            return self.advanced_tools

        tool_names = [t.name for t in self.basic_tools]
        print(f"  🔧 Tools enabled (simple): {', '.join(tool_names)}")
        return self.basic_tools

---

Step 6: Compose it all into an agent

Now we combine the three middleware into the final agent. Order matters: logging goes first (outer layer, captures everything), then routing (decides the model), and dynamic tools last (filters the tools).

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent

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

logging_mw = LoggingMiddleware()
routing_mw = ModelRoutingMiddleware(
    simple_model="openai:gpt-4.1-mini",
    complex_model="openai:gpt-4.1",
)
tools_mw = DynamicToolsMiddleware()

SYSTEM_PROMPT = """You are an intelligent assistant with access to search and analysis tools.

Instructions:
- For simple factual questions, answer concisely
- For complex analytical questions, use the deep research and data analysis tools
- Always cite your sources when you use the search tools
- If you don't have enough information, say so honestly"""

agent = create_agent(
    model,
    tools=[search, calculator, deep_research, data_analysis],
    prompt=SYSTEM_PROMPT,
    middleware=[logging_mw, routing_mw, tools_mw],
)

---

Step 7: An interactive chat loop with visible routing

The interactive loop shows, in real time, which model was selected, which tools were enabled, and the agent's full reasoning process.

def display_routing_decision(routing_mw, question):
    """Show the routing decision before running."""
    classification = classify_complexity(question)
    print(f"\n{'─' * 60}")
    print(f"  📨 Question: {question}")
    print(f"{'─' * 60}")
    print(f"  📊 Classification:")
    print(f"     Complexity: {classification['complexity'].upper()}")
    print(f"     Score: {classification['score']}")
    if classification["reasons"]:
        for reason in classification["reasons"]:
            print(f"     → {reason}")
    print(f"{'─' * 60}")


def chat_loop():
    """Interactive loop with visible dynamic routing."""
    print("=" * 60)
    print("  🤖 Agent with Dynamic Model Routing")
    print("  Ask anything — the agent picks the model.")
    print("  Commands: 'stats' (metrics), 'exit' (quit)")
    print("=" * 60)

    while True:
        try:
            question = input("\n💬 Your question: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\nGoodbye!")
            break

        if not question:
            continue

        if question.lower() in ("exit", "quit", "bye"):
            print("\nGoodbye!")
            break

        if question.lower() == "stats":
            summary = logging_mw.get_summary()
            print(f"\n📊 Accumulated metrics:")
            print(f"   Model calls: {summary['model_calls']}")
            print(f"   Tool executions: {summary['tool_executions']}")
            print(f"   Tools used: {summary['tools_used']}")
            print(f"   Total events: {summary['total_events']}")
            continue

        display_routing_decision(routing_mw, question)

        try:
            result = agent.invoke(
                {"messages": [("user", question)]}
            )

            print(f"\n{'─' * 60}")
            print(f"  💡 Answer:")
            print(f"{'─' * 60}")
            print(f"\n{result['messages'][-1].content}")
            print(f"\n{'─' * 60}")

        except Exception as e:
            print(f"\n❌ Error: {e}")
            print("   Try another question.")


if __name__ == "__main__":
    chat_loop()

---

The complete code

This is the full dynamic_routing_agent.py file. Copy it and run it directly.

"""
Agent with Dynamic Model Routing
Module 4 — LangChain & LangGraph: From Chains to Agents

Automatically picks the model based on the complexity of the question.
"""

from dotenv import load_dotenv
load_dotenv()

import re
import time
from datetime import datetime

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


# =============================================================================
# KNOWLEDGE BASE
# =============================================================================

KNOWLEDGE_BASE = {
    "python": {
        "title": "Python Programming Language",
        "content": (
            "Python is a high-level, interpreted language with dynamic typing. "
            "Created by Guido van Rossum in 1991. Used in AI/ML, data science, web, "
            "automation. Ecosystem: PyTorch, TensorFlow, Django, FastAPI. "
            "Community: >8M developers. Current version: 3.12."
        ),
        "source": "https://python.org",
    },
    "rust": {
        "title": "Rust Programming Language",
        "content": (
            "Rust is a systems language with memory safety and no garbage collector. "
            "Created by Mozilla in 2010. Unique ownership system. Performance comparable to C/C++. "
            "Used in: operating systems, WebAssembly, CLIs, networking. "
            "Adopted by the Linux kernel, Android, Windows."
        ),
        "source": "https://rust-lang.org",
    },
    "microservices": {
        "title": "Microservices Architecture",
        "content": (
            "Microservices is an architectural style that structures an application as a "
            "collection of small, autonomous, independently deployable services. "
            "Upsides: independent scaling, independent deploys, heterogeneous technology. "
            "Downsides: network complexity, eventual consistency, operational overhead. "
            "Alternative: a modular monolith for teams under 20 people."
        ),
        "source": "https://microservices.io",
    },
    "langchain": {
        "title": "LangChain Framework",
        "content": (
            "LangChain is the most widely adopted open-source framework for LLM applications. "
            "v1.2+ includes: init_chat_model, create_agent, the middleware system. "
            "Ecosystem: LangGraph (orchestration), LangSmith (observability), Deep Agents. "
            "Used by: AI startups, enterprises, research labs."
        ),
        "source": "https://langchain.com",
    },
    "typescript": {
        "title": "TypeScript Language",
        "content": (
            "TypeScript is a typed superset of JavaScript developed by Microsoft. "
            "It adds optional static types, interfaces and enums. It compiles to JavaScript. "
            "Adopted by: Angular, React (with TS), Vue 3, Next.js, Deno. "
            "Market share: >78% of new JavaScript projects use TypeScript."
        ),
        "source": "https://typescriptlang.org",
    },
}


# =============================================================================
# TOOLS — BASIC
# =============================================================================

@tool
def search(query: str) -> str:
    """Search for general information about a topic. Useful for factual questions."""
    query_lower = query.lower()
    results = []
    for key, data in KNOWLEDGE_BASE.items():
        if key in query_lower or any(
            word in data["content"].lower() for word in query_lower.split() if len(word) > 3
        ):
            results.append(
                f"📄 {data['title']}: {data['content']} (Source: {data['source']})"
            )

    if results:
        return f"Found {len(results)} result(s):\n" + "\n\n".join(results)
    return (
        f"No specific results for '{query}'. "
        f"Available topics: {', '.join(KNOWLEDGE_BASE.keys())}"
    )


@tool
def calculator(expression: str) -> str:
    """Evaluate math expressions. Useful for numeric calculations."""
    try:
        result = eval(expression)
        return f"Result: {expression} = {result}"
    except Exception as e:
        return f"Error evaluating '{expression}': {e}"


# =============================================================================
# TOOLS — ADVANCED (complex questions only)
# =============================================================================

@tool
def deep_research(topic: str, aspects: list[str]) -> str:
    """Deep research on a topic across several aspects.
    Only used for complex questions that require detailed analysis."""
    results = []
    for aspect in aspects:
        topic_lower = topic.lower()
        if topic_lower in KNOWLEDGE_BASE:
            data = KNOWLEDGE_BASE[topic_lower]
            results.append(f"🔬 {aspect}: Based on {data['title']} — {data['content']}")
        else:
            results.append(
                f"🔬 {aspect}: General analysis of '{topic}' for the aspect '{aspect}' "
                f"(no specific data in the knowledge base)"
            )

    return (
        f"Deep research on '{topic}':\n"
        f"Aspects analyzed: {len(aspects)}\n\n"
        + "\n\n".join(results)
    )


@tool
def data_analysis(data_description: str, analysis_type: str) -> str:
    """Analyze data with a specific type of analysis.
    Only used for complex questions that require data processing."""
    analyses = {
        "comparison": (
            f"📊 Comparison: {data_description} — 3 dimensions identified: "
            f"performance, cost, and ease of adoption."
        ),
        "tradeoff": (
            f"📊 Trade-offs: {data_description} — Upsides vs downsides across 4 categories: "
            f"technical, organizational, financial, and temporal."
        ),
        "recommendation": (
            f"📊 Recommendation: {data_description} — The recommendation depends on context: "
            f"team size, budget, and project timeline."
        ),
        "trend": (
            f"📊 Trend: {data_description} — Trend analysis shows sustained growth "
            f"over the last 3 years, with recent acceleration."
        ),
    }
    return analyses.get(
        analysis_type.lower(),
        f"📊 Analysis ({analysis_type}): {data_description} — Completed with general results.",
    )


# =============================================================================
# COMPLEXITY CLASSIFIER
# =============================================================================

COMPLEXITY_KEYWORDS = [
    "analyze", "compare", "evaluate", "explain in detail",
    "implications", "trade-off", "tradeoff", "advantages and disadvantages",
    "architecture", "design", "strategy", "in depth",
    "vs", "versus", "differences between", "pros and cons",
    "recommendation", "recommend", "when to use", "best option",
    "impact", "consequences", "long term",
]


def classify_complexity(text: str) -> dict:
    """Classify how complex a question is."""
    text_lower = text.lower()
    word_count = len(text.split())

    keyword_matches = [kw for kw in COMPLEXITY_KEYWORDS if kw in text_lower]
    has_multiple_questions = text.count("?") > 1
    has_connectors = any(
        c in text_lower for c in [" and ", " but ", " however", " moreover"]
    )

    score = 0
    reasons = []

    if word_count > 20:
        score += 1
        reasons.append(f"long question ({word_count} words)")
    if keyword_matches:
        score += len(keyword_matches)
        reasons.append(f"analytical keywords: {', '.join(keyword_matches[:3])}")
    if has_multiple_questions:
        score += 1
        reasons.append("multiple questions")
    if has_connectors:
        score += 1
        reasons.append("complex connectors")

    complexity = "complex" if score >= 2 else "simple"

    return {
        "complexity": complexity,
        "score": score,
        "reasons": reasons,
        "word_count": word_count,
    }


# =============================================================================
# MIDDLEWARE: Model Routing
# =============================================================================

class ModelRoutingMiddleware(AgentMiddleware):
    """Pick the model based on the complexity of the question."""

    def __init__(
        self,
        simple_model: str = "openai:gpt-4.1-mini",
        complex_model: str = "openai:gpt-4.1",
    ):
        self._simple = init_chat_model(simple_model, temperature=0)
        self._complex = init_chat_model(complex_model, temperature=0.2)
        self.simple_model_name = simple_model
        self.complex_model_name = complex_model
        self.last_classification = None
        self.routing_history: list[dict] = []

    def wrap_model_call(self, messages, config, call_next):
        user_messages = [
            m for m in messages
            if hasattr(m, "type") and m.type == "human"
        ]

        if user_messages:
            last_user_msg = user_messages[-1].content
            classification = classify_complexity(last_user_msg)
        else:
            classification = {
                "complexity": "simple",
                "score": 0,
                "reasons": ["no user message"],
                "word_count": 0,
            }

        self.last_classification = classification

        if classification["complexity"] == "complex":
            selected = self._complex
            model_name = self.complex_model_name
        else:
            selected = self._simple
            model_name = self.simple_model_name

        self.routing_history.append({
            "model": model_name,
            "complexity": classification["complexity"],
            "score": classification["score"],
            "reasons": classification["reasons"],
        })

        print(f"\n  🧠 Model selected: {model_name}")
        print(f"     Complexity: {classification['complexity']} (score: {classification['score']})")
        if classification["reasons"]:
            print(f"     Reasons: {', '.join(classification['reasons'])}")

        bound_tools = config.get("__bound_tools__", [])
        if bound_tools:
            selected = selected.bind_tools(bound_tools)

        return selected.invoke(messages)


# =============================================================================
# MIDDLEWARE: Logging
# =============================================================================

class LoggingMiddleware(AgentMiddleware):
    """Log every operation in detail."""

    def __init__(self):
        self.log_entries: list[dict] = []
        self._call_count = 0

    def before_model(self, messages, config):
        self._call_count += 1
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "model_call",
            "call_number": self._call_count,
            "message_count": len(messages),
        }
        self.log_entries.append(entry)
        print(
            f"  📋 [{entry['timestamp']}] Model call #{self._call_count} "
            f"({len(messages)} msgs)"
        )

    def after_model(self, response, config):
        has_tool_calls = bool(getattr(response, "tool_calls", []))
        content_len = len(response.content) if response.content else 0
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "model_response",
            "has_tool_calls": has_tool_calls,
            "content_length": content_len,
        }
        self.log_entries.append(entry)
        if has_tool_calls:
            tool_names = [tc["name"] for tc in response.tool_calls]
            print(f"  📋 Model decided to call tools: {', '.join(tool_names)}")
        else:
            preview = response.content[:80] if response.content else "(empty)"
            print(f"  📋 Model answered: {preview}...")

    def wrap_tool_call(self, tool_call, config, call_next):
        start = time.time()
        print(f"  🔧 Running: {tool_call['name']}({str(tool_call['args'])[:60]})")

        result = call_next(tool_call, config)

        elapsed = time.time() - start
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "tool_execution",
            "tool": tool_call["name"],
            "elapsed_ms": round(elapsed * 1000),
        }
        self.log_entries.append(entry)
        preview = str(result)[:80]
        print(f"  📥 Result ({elapsed * 1000:.0f}ms): {preview}")
        return result

    def get_summary(self) -> dict:
        model_calls = sum(1 for e in self.log_entries if e["type"] == "model_call")
        tool_execs = [e for e in self.log_entries if e["type"] == "tool_execution"]
        tools_used: dict[str, int] = {}
        for e in tool_execs:
            tools_used[e["tool"]] = tools_used.get(e["tool"], 0) + 1
        return {
            "model_calls": model_calls,
            "tool_executions": len(tool_execs),
            "tools_used": tools_used,
            "total_events": len(self.log_entries),
        }

    def reset(self):
        self.log_entries.clear()
        self._call_count = 0


# =============================================================================
# MIDDLEWARE: Dynamic Tools
# =============================================================================

class DynamicToolsMiddleware(AgentMiddleware):
    """Enable the advanced tools only for complex questions."""

    def __init__(self):
        self.basic_tools = [search, calculator]
        self.advanced_tools = [search, calculator, deep_research, data_analysis]

    def get_tools(self, state):
        messages = state.get("messages", [])
        user_messages = [
            m for m in messages
            if hasattr(m, "type") and m.type == "human"
        ]

        if user_messages:
            last_msg = user_messages[-1].content
            classification = classify_complexity(last_msg)
        else:
            classification = {"complexity": "simple"}

        if classification["complexity"] == "complex":
            tool_names = [t.name for t in self.advanced_tools]
            print(f"  🔧 Tools enabled (complex): {', '.join(tool_names)}")
            return self.advanced_tools

        tool_names = [t.name for t in self.basic_tools]
        print(f"  🔧 Tools enabled (simple): {', '.join(tool_names)}")
        return self.basic_tools


# =============================================================================
# AGENT SETUP
# =============================================================================

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

logging_mw = LoggingMiddleware()
routing_mw = ModelRoutingMiddleware(
    simple_model="openai:gpt-4.1-mini",
    complex_model="openai:gpt-4.1",
)
tools_mw = DynamicToolsMiddleware()

SYSTEM_PROMPT = """You are an intelligent assistant with access to search and analysis tools.

Instructions:
- For simple factual questions, answer concisely
- For complex analytical questions, use the deep research and data analysis tools
- Always cite your sources when you use the search tools
- If you don't have enough information, say so honestly"""

agent = create_agent(
    model,
    tools=[search, calculator, deep_research, data_analysis],
    prompt=SYSTEM_PROMPT,
    middleware=[logging_mw, routing_mw, tools_mw],
)


# =============================================================================
# INTERACTIVE CHAT LOOP
# =============================================================================

def display_routing_decision(question):
    """Show the complexity classification."""
    classification = classify_complexity(question)
    print(f"\n{'─' * 60}")
    print(f"  📨 Question: {question}")
    print(f"{'─' * 60}")
    print(f"  📊 Up-front classification:")
    print(f"     Complexity: {classification['complexity'].upper()}")
    print(f"     Score: {classification['score']}")
    if classification["reasons"]:
        for reason in classification["reasons"]:
            print(f"     → {reason}")
    print(f"{'─' * 60}")


def chat_loop():
    """Interactive loop with visible dynamic routing."""
    print("=" * 60)
    print("  🤖 Agent with Dynamic Model Routing")
    print("  Ask anything — the agent picks the model.")
    print("  Commands: 'stats' (metrics), 'exit' (quit)")
    print("=" * 60)

    while True:
        try:
            question = input("\n💬 Your question: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\nGoodbye!")
            break

        if not question:
            continue

        if question.lower() in ("exit", "quit", "bye"):
            summary = logging_mw.get_summary()
            print(f"\n📊 Session summary:")
            print(f"   Model calls: {summary['model_calls']}")
            print(f"   Tools executed: {summary['tool_executions']}")
            print(f"   Tools used: {summary['tools_used']}")
            print(f"\n🧠 Routing history:")
            for i, r in enumerate(routing_mw.routing_history, 1):
                print(f"   {i}. {r['model']} ({r['complexity']}, score={r['score']})")
            print("\nGoodbye!")
            break

        if question.lower() == "stats":
            summary = logging_mw.get_summary()
            print(f"\n📊 Accumulated metrics:")
            print(f"   Model calls: {summary['model_calls']}")
            print(f"   Tool executions: {summary['tool_executions']}")
            print(f"   Tools used: {summary['tools_used']}")
            print(f"   Total events: {summary['total_events']}")
            print(f"\n🧠 Routing history:")
            for i, r in enumerate(routing_mw.routing_history, 1):
                print(f"   {i}. {r['model']} ({r['complexity']}, score={r['score']})")
            continue

        display_routing_decision(question)

        try:
            result = agent.invoke(
                {"messages": [("user", question)]}
            )

            print(f"\n{'─' * 60}")
            print(f"  💡 Answer:")
            print(f"{'─' * 60}")
            print(f"\n{result['messages'][-1].content}")
            print(f"\n{'─' * 60}")

        except Exception as e:
            print(f"\n❌ Error: {e}")
            print("   Try another question.")


if __name__ == "__main__":
    chat_loop()

Run it:

python dynamic_routing_agent.py

---

Success criteria

Your project is complete when it meets all five criteria:

• ✅ Simple questions use the cheap model — "What is Python?" uses gpt-4.1-mini, visible in the logs
• ✅ Complex questions use the powerful model — "Compare microservices vs monolith and recommend one for a startup" uses gpt-4.1, visible in the logs
• ✅ The logs show the selected model and the reason — every question shows the classification, the score, the reasons and the chosen model
• ✅ Advanced tools only available for complex requests — simple questions only reach search and calculator; complex questions also reach deep_research and data_analysis
• ✅ The system works as an interactive chat — you can ask several questions and see the accumulated metrics with stats

---

How to test it

Test 1: A simple question (should use gpt-4.1-mini)

💬 Your question: What is Python?

────────────────────────────────────────────────────────────
  📨 Question: What is Python?
────────────────────────────────────────────────────────────
  📊 Up-front classification:
     Complexity: SIMPLE
     Score: 0
────────────────────────────────────────────────────────────
  🔧 Tools enabled (simple): search, calculator
  📋 [15:30:01] Model call #1 (2 msgs)

  🧠 Model selected: openai:gpt-4.1-mini
     Complexity: simple (score: 0)
  📋 Model decided to call tools: search
  🔧 Running: search({'query': 'Python'})
  📥 Result (1ms): Found 1 result(s): 📄 Python Programming Language...
  📋 [15:30:02] Model call #2 (4 msgs)

  🧠 Model selected: openai:gpt-4.1-mini
     Complexity: simple (score: 0)
  📋 Model answered: Python is a high-level, interpreted programming language...

────────────────────────────────────────────────────────────
  💡 Answer:
────────────────────────────────────────────────────────────

Python is a high-level, interpreted programming language
with dynamic typing...

Test 2: A complex question (should use gpt-4.1)

💬 Your question: Compare the advantages and disadvantages of microservices vs monolith and recommend when to use each one

────────────────────────────────────────────────────────────
  📨 Question: Compare the advantages and disadvantages of...
────────────────────────────────────────────────────────────
  📊 Up-front classification:
     Complexity: COMPLEX
     Score: 4
     → long question (16 words)
     → analytical keywords: compare, advantages and disadvantages, recommend
────────────────────────────────────────────────────────────
  🔧 Tools enabled (complex): search, calculator, deep_research, data_analysis
  📋 [15:31:00] Model call #3 (2 msgs)

  🧠 Model selected: openai:gpt-4.1
     Complexity: complex (score: 4)
     Reasons: long question, analytical keywords: compare, advantages and disadvantages, recommend
  📋 Model decided to call tools: deep_research, data_analysis
  🔧 Running: deep_research({'topic': 'microservices', 'aspects': [...]})
  📥 Result (1ms): Deep research on 'microservices'...
  🔧 Running: data_analysis({'data_description': '...', 'analysis_type': 'tradeoff'})
  📥 Result (0ms): 📊 Trade-offs: ...
  📋 [15:31:03] Model call #4 (6 msgs)

  🧠 Model selected: openai:gpt-4.1
     Complexity: complex (score: 4)
  📋 Model answered: [Detailed analysis with recommendations]...

────────────────────────────────────────────────────────────
  💡 Answer:
────────────────────────────────────────────────────────────

## Microservices vs Monolith

### Advantages of Microservices:
- Independent scaling...
...

Test 3: Check the metrics

💬 Your question: stats

📊 Accumulated metrics:
   Model calls: 4
   Tool executions: 3
   Tools used: {'search': 1, 'deep_research': 1, 'data_analysis': 1}
   Total events: 10

🧠 Routing history:
   1. openai:gpt-4.1-mini (simple, score=0)
   2. openai:gpt-4.1-mini (simple, score=0)
   3. openai:gpt-4.1 (complex, score=4)
   4. openai:gpt-4.1 (complex, score=4)

Test 4: A borderline question

💬 Your question: What are the advantages of Rust?

────────────────────────────────────────────────────────────
  📊 Up-front classification:
     Complexity: SIMPLE
     Score: 0
────────────────────────────────────────────────────────────

(uses gpt-4.1-mini because "advantages" on its own isn't enough — it needs "advantages and disadvantages")

---

Common errors

1. ModuleNotFoundError: No module named 'langchain'

Cause: You didn't install the dependencies.

pip install langchain langgraph langchain-openai python-dotenv

2. The classifier marks everything as "simple"

Cause: The complexity keywords don't match the way you phrase your questions. The classifier looks for exact matches.

Fix: Add more keywords to COMPLEXITY_KEYWORDS that reflect your vocabulary. You can also tune the score threshold (currently >= 2 for complex).

3. AuthenticationError with the powerful model

Cause: Your API key doesn't have access to gpt-4.1. Some OpenAI accounts only have access to the mini models.

Fix: Switch the complex model to one you do have available:

routing_mw = ModelRoutingMiddleware(
    simple_model="openai:gpt-4.1-mini",
    complex_model="openai:gpt-4.1-mini",  # same model for testing
)

4. The routing middleware picks the wrong model on intermediate calls

Cause: After the first model call (the one that decides to call tools), the second call no longer has the user's message as the last message — it has the tool results. The classifier sees those technical messages and classifies them as "simple".

Fix: The middleware looks for the last message with type == "human", not simply the last message. Check that the user_messages logic filters by type correctly.

5. The advanced tools show up on simple questions

Cause: DynamicToolsMiddleware.get_tools() and the router's classification can diverge if the message state changes between the two evaluations.

Fix: Both middleware use classify_complexity() with the same logic, so they should agree. If you see inconsistencies, check that the last user message is the same in both cases.

6. RateLimitError from too many calls

Cause: Every question triggers 2-4 model calls (classification + tool calling + answer). Complex questions trigger more because they use advanced tools that return more context.

Fix: Add a RateLimitMiddleware (you saw it in capsule 07) to the middleware list, or wait between questions.

7. The agent doesn't use the advanced tools even though the question is complex

Cause: The system prompt isn't directive enough about using deep_research and data_analysis. The model may decide to answer straight away.

Fix: Make the system prompt more explicit: "For complex questions that require analysis, ALWAYS use deep_research and/or data_analysis before answering."

8. TypeError: 'NoneType' object is not subscriptable on result["messages"]

Cause: The agent returned None, usually because an exception was silently swallowed inside the middleware.

Fix: Check that every wrap_model_call and wrap_tool_call returns a value. Forgetting the return is the most common error of all.

---

Ideas to extend it

If you finished the project and want to go further:

• An LLM-based classifier — Replace the keyword heuristic with a fast call to gpt-4.1-mini that classifies the complexity. It costs more but it's far more accurate on ambiguous cases
• Three model tiers — Add a middle tier: gpt-4.1-mini for simple, gpt-4.1-mini with a higher temperature for medium, and gpt-4.1 for complex. Tune the classifier with three thresholds
• A cost budget — Add a BudgetMiddleware that tracks the estimated cost (input tokens × price per token) and falls back to the cheap model when a daily budget is exceeded
• More providers — Add Anthropic as an option: gpt-4.1-mini for simple, claude-sonnet for complex. Requires pip install langchain-anthropic and ANTHROPIC_API_KEY
• A feedback loop — Let the user say "that answer wasn't good enough" and have the agent re-run with the powerful model. Implement it as a special command in the chat loop
• Persistence — Add MemorySaver so the agent remembers previous conversations and uses the history to improve its complexity classification

---

Connecting to the next module

In this project you used LangChain's whole middleware system to build an intelligent agent that optimizes cost and quality. But everything runs inside create_agent — a linear loop of model → tools → model → tools → answer.

What happens when you need the agent to take completely different paths depending on the situation? When one node processes data while another searches for information in parallel? When the flow has to loop for a retry only on error? When it has to pause and ask a human for approval?

In Module 5: Introduction to LangGraph, you'll learn to build workflows as graphs: nodes that are functions, edges that define the flow, and conditional edges that create dynamic branching. LangGraph gives you full control over orchestration — cycles, branching, subgraphs and state persistence — which goes far beyond what middleware can do. The create_agent you used in this module is, in fact, a LangGraph graph under the hood. In Module 5, you'll learn to build those graphs yourself.

---

Project resources

1. LangChain Agents Middleware — The official guide to the middleware system
2. create_agent API Reference — Reference with the middleware parameters
3. LangChain init_chat_model — Multi-provider initialization for routing
4. OpenAI Model Pricing — Model pricing, so you can see what routing actually saves
5. LangGraph Agents Conceptual Guide — How create_agent works internally as a graph
6. Middleware Pattern in Software — The design pattern behind the middleware system

---

Module 4 — LangChain & LangGraph: From Chains to Agents