Learn how to build a production-grade semantic router that classifies user intent and dispatches queries to specialized handlers. Covers embedding-based and LLM-based classification, hybrid routing with confidence thresholds, and wiring everything to real downstream chains, indexes, and agents.
Imagine you've built a customer-facing AI assistant for a mid-sized SaaS company. Users ask it everything: "How do I reset my password?", "What's my invoice total for Q3?", "Can you write me a SQL query to pull churn data?", and "I'm furious that my account got suspended without warning." Each of those questions demands a completely different response strategy — one needs a documentation lookup, one needs a live database query, one needs a code-generation chain, and one needs a carefully tuned empathetic tone with escalation logic baked in. If you route all four to the same generic chain, you'll get mediocre results on all of them.
This is the core problem that semantic routing solves. Rather than running every query through one monolithic pipeline, you build a dispatcher that classifies user intent and sends each query to the handler best equipped to serve it — a specialized RAG index, a tool-calling agent, a fine-tuned prompt chain, a human escalation path, or some combination. The result is a system that's simultaneously faster, cheaper, more accurate, and more maintainable than the all-in-one alternative.
By the end of this lesson, you'll have built a production-grade semantic router from scratch. We'll cover both embedding-based and LLM-based classification strategies, show you how to compose them into a hybrid system, and wire everything up to specialized downstream handlers. We'll also cover failure modes, fallback behavior, and how to monitor routing decisions in production.
What you'll learn:
You should be comfortable with:
You don't need to have used LangChain or any specific orchestration framework — we'll build the core logic from first principles and show where frameworks can help.
The naive approach to routing is keyword matching: if the message contains "invoice" or "billing," send it to the billing handler; if it contains "error" or "bug," send it to support. This works until it doesn't — which is approximately day three of production traffic.
Users don't phrase things the way you expect. "I keep getting charged twice" doesn't contain the word "billing." "My dashboard is totally broken" might be a UI bug or a permission issue. "Can you help me understand my options?" could be pre-sales, support, or a cancellation attempt. Keyword rules become an unmaintainable spider web of regexes and special cases.
Semantic routing sidesteps this by working at the meaning level rather than the token level. Instead of matching words, you're matching intent — the underlying communicative goal of the message. Two messages that share zero vocabulary ("I need to export my data" vs. "How do I get a copy of everything I've stored?") can have identical intent and should be routed identically.
There are two primary mechanisms for achieving this:
Embedding similarity routing encodes the incoming query as a vector and compares it against a set of reference vectors (one or more per route). The route whose reference vectors are most similar wins. This is fast and cheap — it requires no LLM call — but it's a blunt instrument for nuanced or ambiguous queries.
LLM-based classification sends the query to a language model with a structured classification prompt. This is slower and more expensive but handles ambiguity, context-dependency, and multi-intent queries much more gracefully.
In production, you almost always want both: use embedding similarity for high-confidence cases and fall back to LLM classification when confidence is low.
Key insight: Semantic routing is an architectural pattern, not a library. The libraries (LangChain's
SemanticRouter, custom dispatchers) implement the pattern, but understanding the pattern lets you build it, debug it, and extend it regardless of which tools you use.
Before writing a single line of routing code, you need to design your intent taxonomy. This is the most important step and the one most people rush. A poorly designed schema leads to overlapping routes that confuse the classifier, gaps that cause unmatched queries, and handler mismatches that degrade response quality.
Let's use a concrete example throughout this lesson: an AI assistant for a data analytics SaaS platform. After analyzing 500 real user queries (which you should always do before designing routes), you identify these intent categories:
| Route Name | Description | Handler |
|---|---|---|
sql_generation |
User wants a SQL query written or explained | Code-gen chain |
data_lookup |
User wants specific data from their warehouse | Database agent |
documentation |
How-to questions about the product | Docs RAG index |
billing_support |
Questions about invoices, plans, payments | Billing RAG + CRM tool |
bug_report |
Something isn't working correctly | Support ticketing agent |
account_management |
Password resets, permissions, user management | Account management chain |
data_export |
Requests to export, download, or share data | Export workflow agent |
escalate_human |
Frustrated users, complex complaints, legal mentions | Human handoff |
Two design principles to internalize:
Routes should be mutually exclusive at the decision point. If a query could legitimately belong to two routes, either merge them or add a secondary routing step. "My SQL query is returning wrong data — is this a bug?" could be sql_generation or bug_report. You need to decide: does the system try to fix the SQL first, or file a bug report? Pick one as the default and be explicit about it.
Routes should be collectively exhaustive with a fallback. Every query that doesn't match a known route should hit a graceful fallback — usually a general-purpose chain with a disclaimer about its limitations, not a cryptic error.
The embedding router works by creating a set of example queries for each route, embedding them all, and then comparing any incoming query against those reference embeddings using cosine similarity.
Here's the full implementation:
import numpy as np
from openai import OpenAI
from dataclasses import dataclass, field
from typing import Optional
import json
client = OpenAI()
@dataclass
class Route:
name: str
description: str
examples: list[str]
embeddings: list[list[float]] = field(default_factory=list)
def embed_examples(self, client: OpenAI, model: str = "text-embedding-3-small"):
"""Pre-compute embeddings for all examples at startup."""
response = client.embeddings.create(
input=self.examples,
model=model
)
self.embeddings = [item.embedding for item in response.data]
def similarity_score(self, query_embedding: list[float]) -> float:
"""Return the max cosine similarity across all example embeddings."""
if not self.embeddings:
raise ValueError(f"Route '{self.name}' has no embeddings. Call embed_examples() first.")
query_vec = np.array(query_embedding)
similarities = []
for emb in self.embeddings:
ref_vec = np.array(emb)
cosine_sim = np.dot(query_vec, ref_vec) / (
np.linalg.norm(query_vec) * np.linalg.norm(ref_vec)
)
similarities.append(float(cosine_sim))
# Use max rather than mean — one strong match is enough
return max(similarities)
class EmbeddingRouter:
def __init__(self, routes: list[Route], confidence_threshold: float = 0.75):
self.routes = routes
self.confidence_threshold = confidence_threshold
self.embed_model = "text-embedding-3-small"
def initialize(self):
"""Pre-compute all route embeddings. Call once at startup."""
print("Initializing embedding router...")
for route in self.routes:
route.embed_examples(client, self.embed_model)
print(f"Router ready with {len(self.routes)} routes.")
def _embed_query(self, query: str) -> list[float]:
response = client.embeddings.create(
input=[query],
model=self.embed_model
)
return response.data[0].embedding
def route(self, query: str) -> tuple[Optional[str], float]:
"""
Returns (route_name, confidence_score).
Returns (None, best_score) if confidence is below threshold.
"""
query_embedding = self._embed_query(query)
scores = {
route.name: route.similarity_score(query_embedding)
for route in self.routes
}
best_route = max(scores, key=scores.get)
best_score = scores[best_route]
if best_score >= self.confidence_threshold:
return best_route, best_score
else:
return None, best_score
Now let's define the routes with realistic examples:
routes = [
Route(
name="sql_generation",
description="User wants a SQL query written, explained, or debugged",
examples=[
"Write me a SQL query to find all users who signed up last month",
"How do I join the events table with users on customer_id?",
"My GROUP BY isn't working, what am I doing wrong?",
"Can you write a query that calculates 30-day rolling retention?",
"What's the SQL syntax for a window function in Snowflake?",
"Show me how to write a CTE for this problem",
"Why is my query returning duplicates?",
]
),
Route(
name="data_lookup",
description="User wants to retrieve specific data from their warehouse",
examples=[
"How many active users do we have right now?",
"What was our revenue last Tuesday?",
"Show me the top 10 customers by spend this quarter",
"How many events did we track yesterday?",
"What's the conversion rate for the onboarding funnel?",
"Give me the MRR breakdown by plan type",
]
),
Route(
name="documentation",
description="User has how-to questions about using the product",
examples=[
"How do I create a new dashboard?",
"Where do I configure data source connections?",
"What's the difference between a metric and a dimension?",
"How do I share a report with my team?",
"Can I schedule reports to run automatically?",
"What file formats can I import?",
]
),
Route(
name="billing_support",
description="Questions about invoices, pricing, plans, or payments",
examples=[
"I was charged twice this month",
"Can I get a copy of my invoice for Q2?",
"What's included in the enterprise plan?",
"I need to upgrade my subscription",
"Is there a discount for annual billing?",
"My payment method is failing",
]
),
Route(
name="escalate_human",
description="Frustrated users, complex complaints, or legal/compliance concerns",
examples=[
"This is absolutely unacceptable and I want to speak to someone",
"I'm going to cancel my account if this isn't fixed immediately",
"I think you may have exposed my customer data — this is a GDPR issue",
"I've been waiting 3 days and nobody has responded",
"I want to make a formal complaint",
"My entire team is blocked and we're losing money every hour",
]
),
]
# Initialize the router at application startup
router = EmbeddingRouter(routes=routes, confidence_threshold=0.75)
router.initialize()
# Test it
test_queries = [
"Write a query to show me churn by cohort",
"How many trials converted this month?",
"How do I set up a Looker-style explore?",
"I want a refund, this product is garbage",
]
for q in test_queries:
route_name, score = router.route(q)
print(f"Query: '{q[:50]}...'")
print(f" → Route: {route_name or 'FALLBACK'} (confidence: {score:.3f})\n")
Tip: The number and quality of your example queries per route matters more than anything else. Aim for at least 7-10 examples per route, and make sure they cover the lexical diversity you'd expect from real users — not just rephrasing the same sentence. Pull examples from actual user logs whenever possible.
When embedding similarity doesn't return a confident match, you escalate to an LLM. Rather than asking the model to freeform-describe the intent, you give it your routing schema and ask for structured output — this is much more reliable than parsing natural language responses.
If you need a refresher on getting reliable structured outputs from LLMs, check out our lesson on Structured Output: Getting JSON, Tables, and Code from LLMs.
from pydantic import BaseModel, Field
from typing import Literal
# Define the structured output schema
class RoutingDecision(BaseModel):
route: str = Field(
description="The name of the most appropriate route from the provided list"
)
confidence: Literal["high", "medium", "low"] = Field(
description="How confident you are in this routing decision"
)
reasoning: str = Field(
description="One sentence explaining why you chose this route"
)
secondary_route: Optional[str] = Field(
default=None,
description="If the query could also belong to another route, name it here"
)
class LLMRouter:
def __init__(self, routes: list[Route]):
self.routes = routes
self._build_schema_description()
def _build_schema_description(self):
"""Build a compact description of all routes for the system prompt."""
route_descriptions = "\n".join(
f"- {r.name}: {r.description}" for r in self.routes
)
self.schema_description = route_descriptions
def _build_system_prompt(self) -> str:
return f"""You are an intent classifier for a data analytics platform assistant.
Your job is to analyze user queries and determine which routing category best matches their intent.
Available routes:
{self.schema_description}
Rules:
- Choose exactly one primary route — the one that best represents the user's core need
- If the query could reasonably belong to multiple routes, pick the most actionable one and note the secondary
- If you're genuinely uncertain, use low confidence — don't guess
- Base your decision on intent, not surface-level keywords
- "escalate_human" should be used when the user shows strong frustration or raises compliance/legal concerns"""
def classify(self, query: str, conversation_context: str = "") -> RoutingDecision:
"""
Classify a query using LLM-based intent detection.
conversation_context: recent message history for context-dependent routing
"""
user_content = f"User query: {query}"
if conversation_context:
user_content = f"Conversation context:\n{conversation_context}\n\nCurrent query: {query}"
response = client.beta.chat.completions.parse(
model="gpt-4o-mini", # Use a cheap, fast model for routing
messages=[
{"role": "system", "content": self._build_system_prompt()},
{"role": "user", "content": user_content}
],
response_format=RoutingDecision,
temperature=0, # Deterministic routing
)
return response.choices[0].message.parsed
# Usage
llm_router = LLMRouter(routes=routes)
result = llm_router.classify(
"The export I scheduled last week still hasn't shown up in my inbox and I've emailed support twice",
conversation_context="User has been waiting for a data export for 7 days."
)
print(f"Route: {result.route}")
print(f"Confidence: {result.confidence}")
print(f"Reasoning: {result.reasoning}")
print(f"Secondary: {result.secondary_route}")
# Output:
# Route: escalate_human
# Confidence: high
# Reasoning: User shows significant frustration and has already attempted support contact without resolution.
# Secondary: data_export
Notice a few things about this design:
We use gpt-4o-mini not gpt-4o. Routing decisions should be fast and cheap — you're not generating content, you're making a binary decision. For tips on model selection by task complexity, see Implementing LLM Router Architecture.
We use temperature=0. Routing should be deterministic. You don't want the same query to route differently on different calls.
We include conversation context. Without it, "Can you fix that?" is unroutable. With it, the classifier knows whether "that" refers to a SQL query or a billing issue.
Warning: Don't use a routing prompt that describes routes vaguely. "Handle billing questions" is much weaker than "Questions about invoices, pricing, plans, payment failures, or subscription changes." The LLM needs enough signal to distinguish routes that might be adjacent in meaning.
Now we combine both strategies. The embedding router runs first (it's fast and free). If it returns a high-confidence result, we dispatch immediately. If it's uncertain, we escalate to the LLM classifier.
import time
import logging
from dataclasses import dataclass
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("semantic_router")
@dataclass
class RoutingResult:
route: str
confidence_score: float
strategy_used: str # "embedding" | "llm" | "fallback"
reasoning: Optional[str] = None
secondary_route: Optional[str] = None
latency_ms: float = 0.0
class HybridSemanticRouter:
"""
Two-stage router: embedding similarity first, LLM classification as fallback.
"""
def __init__(
self,
routes: list[Route],
embedding_confidence_threshold: float = 0.78,
fallback_route: str = "documentation",
):
self.embedding_router = EmbeddingRouter(
routes=routes,
confidence_threshold=embedding_confidence_threshold
)
self.llm_router = LLMRouter(routes=routes)
self.fallback_route = fallback_route
self.routes = routes
def initialize(self):
self.embedding_router.initialize()
def route(
self,
query: str,
conversation_context: str = "",
force_llm: bool = False
) -> RoutingResult:
start_time = time.time()
# Stage 1: Try embedding router (unless forced to LLM)
if not force_llm:
embed_route, embed_score = self.embedding_router.route(query)
if embed_route is not None:
latency = (time.time() - start_time) * 1000
logger.info(
f"Embedding router matched '{embed_route}' "
f"(score={embed_score:.3f}, latency={latency:.1f}ms)"
)
return RoutingResult(
route=embed_route,
confidence_score=embed_score,
strategy_used="embedding",
latency_ms=latency
)
logger.info(
f"Embedding confidence too low ({embed_score:.3f}), "
f"escalating to LLM classifier"
)
# Stage 2: LLM classification
try:
llm_decision = self.llm_router.classify(query, conversation_context)
# Map LLM confidence string to numeric score
confidence_map = {"high": 0.90, "medium": 0.70, "low": 0.50}
numeric_confidence = confidence_map[llm_decision.confidence]
latency = (time.time() - start_time) * 1000
logger.info(
f"LLM classifier matched '{llm_decision.route}' "
f"(confidence={llm_decision.confidence}, latency={latency:.1f}ms)"
)
# If LLM is also low confidence, route to fallback
if llm_decision.confidence == "low":
return RoutingResult(
route=self.fallback_route,
confidence_score=numeric_confidence,
strategy_used="fallback",
reasoning=f"Low confidence routing: {llm_decision.reasoning}",
latency_ms=latency
)
return RoutingResult(
route=llm_decision.route,
confidence_score=numeric_confidence,
strategy_used="llm",
reasoning=llm_decision.reasoning,
secondary_route=llm_decision.secondary_route,
latency_ms=latency
)
except Exception as e:
latency = (time.time() - start_time) * 1000
logger.error(f"LLM routing failed: {e}. Using fallback route.")
return RoutingResult(
route=self.fallback_route,
confidence_score=0.0,
strategy_used="fallback",
reasoning=f"Router error: {str(e)}",
latency_ms=latency
)
Note: The
force_llmparameter is useful for debugging and A/B testing. You can direct a random sample of traffic through the LLM router even when the embedding router would have matched, then compare outcomes to validate that embedding routing isn't making silent mistakes.
The router is only useful if it dispatches to real handlers. Let's build a dispatcher that connects routing results to actual execution logic:
from typing import Callable, Any
from abc import ABC, abstractmethod
class QueryHandler(ABC):
"""Base class for all route handlers."""
@abstractmethod
async def handle(self, query: str, context: dict) -> dict:
"""
Process a routed query.
Returns a dict with 'response', 'sources', and any handler-specific metadata.
"""
pass
class SQLGenerationHandler(QueryHandler):
async def handle(self, query: str, context: dict) -> dict:
# In production, this would call a specialized SQL-generation chain
# with schema context, dialect awareness, and query validation
system_prompt = """You are an expert SQL assistant for a Snowflake data warehouse.
The user's schema includes: users, events, sessions, revenue, subscriptions tables.
Write clean, well-commented SQL. Validate that all table and column references are real."""
response = client.chat.completions.create(
model="gpt-4o", # Use the powerful model for actual generation
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": query}
],
temperature=0.1
)
return {
"response": response.choices[0].message.content,
"handler": "sql_generation",
"model_used": "gpt-4o"
}
class DocumentationHandler(QueryHandler):
def __init__(self, vector_store):
self.vector_store = vector_store
async def handle(self, query: str, context: dict) -> dict:
# Retrieve relevant docs, then generate a grounded response
# See: Building a Production Document Q&A System with Vector Embeddings
docs = self.vector_store.similarity_search(query, k=5)
context_text = "\n\n".join([doc.page_content for doc in docs])
sources = [doc.metadata.get("source", "unknown") for doc in docs]
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Answer based on the provided documentation. If the answer isn't in the docs, say so."},
{"role": "user", "content": f"Documentation:\n{context_text}\n\nQuestion: {query}"}
],
temperature=0
)
return {
"response": response.choices[0].message.content,
"handler": "documentation",
"sources": sources
}
class HumanEscalationHandler(QueryHandler):
async def handle(self, query: str, context: dict) -> dict:
# Create a support ticket and return a holding message
ticket_id = self._create_ticket(query, context)
return {
"response": f"I can see this is urgent and I want to make sure you get the right help. "
f"I've created support ticket #{ticket_id} and flagged it as high priority. "
f"A team member will reach out within 2 hours. Is there anything else I can "
f"document to help them resolve this faster?",
"handler": "human_escalation",
"ticket_id": ticket_id,
"requires_human": True
}
def _create_ticket(self, query: str, context: dict) -> str:
# Integrate with your ticketing system (Zendesk, Linear, etc.)
# Returns ticket ID
return "TKT-" + str(hash(query))[:6].upper()
class RoutingDispatcher:
"""Connects routing results to handler execution."""
def __init__(self, router: HybridSemanticRouter):
self.router = router
self.handlers: dict[str, QueryHandler] = {}
self.fallback_handler: Optional[QueryHandler] = None
def register_handler(self, route_name: str, handler: QueryHandler):
self.handlers[route_name] = handler
def register_fallback(self, handler: QueryHandler):
self.fallback_handler = handler
async def dispatch(self, query: str, context: dict = {}) -> dict:
"""Route and handle a query end-to-end."""
# Get conversation context for context-aware routing
conversation_history = context.get("conversation_history", "")
# Route the query
routing_result = self.router.route(query, conversation_context=conversation_history)
# Select handler
handler = self.handlers.get(routing_result.route)
if handler is None:
if self.fallback_handler:
handler = self.fallback_handler
routing_result.route = "fallback"
else:
raise ValueError(f"No handler registered for route '{routing_result.route}'")
# Execute handler
handler_result = await handler.handle(query, context)
# Merge routing metadata with handler result
return {
**handler_result,
"routing": {
"route": routing_result.route,
"confidence": routing_result.confidence_score,
"strategy": routing_result.strategy_used,
"reasoning": routing_result.reasoning,
"latency_ms": routing_result.latency_ms
}
}
The routing metadata in every response is not optional in production. You need it for monitoring, debugging, and building your human preference dataset to improve routing quality over time — a topic covered in Building a Feedback Collection and Human Preference Dataset Pipeline.
Real users often have multi-intent queries: "Can you write me a query to pull last month's revenue and also explain how the billing dashboard works?" This is simultaneously sql_generation and documentation. You have three options:
Option 1: Primary route only. Pick the dominant intent and serve that. Mention at the end that you can also help with the secondary intent. This is the simplest approach and usually good enough.
Option 2: Sequential routing. Decompose the query into sub-queries, route each one, execute in sequence, and synthesize the results. This is exactly what query decomposition pipelines do — see Implementing Query Decomposition and Sub-Question Synthesis for Complex RAG Pipelines for the full pattern.
Option 3: Parallel routing. Route to multiple handlers simultaneously and merge results. Effective when the sub-queries are independent and latency matters.
Here's a minimal parallel routing implementation:
import asyncio
async def dispatch_parallel(
dispatcher: RoutingDispatcher,
query: str,
routes: list[str],
context: dict = {}
) -> list[dict]:
"""Dispatch the same query to multiple handlers in parallel."""
async def run_handler(route_name: str) -> dict:
handler = dispatcher.handlers.get(route_name)
if handler:
result = await handler.handle(query, context)
result["_route"] = route_name
return result
return {"_route": route_name, "error": "No handler found"}
results = await asyncio.gather(*[run_handler(r) for r in routes])
return list(results)
# Example: query has secondary route that warrants parallel execution
async def smart_dispatch(dispatcher: RoutingDispatcher, query: str, context: dict = {}):
routing_result = dispatcher.router.route(query)
# If there's a secondary route and primary confidence is medium, run both
if (
routing_result.secondary_route
and routing_result.confidence_score < 0.85
and routing_result.confidence_score > 0.65
):
results = await dispatch_parallel(
dispatcher,
query,
[routing_result.route, routing_result.secondary_route],
context
)
return {"type": "multi_intent", "results": results}
else:
return await dispatcher.dispatch(query, context)
Build a semantic router for a financial data assistant. The assistant should handle four types of queries:
market_data — Requests for stock prices, market indices, or financial metricsportfolio_analysis — Questions about portfolio composition, performance, or riskregulatory_compliance — Questions about reporting requirements, audits, or regulationstrade_execution — Requests to place, modify, or cancel tradesYour tasks:
Part 1: Define a Route for each category with at least 8 realistic example queries. Think about edge cases — where do the category boundaries get fuzzy? Write down at least 3 queries per boundary that you'd expect to be ambiguous.
Part 2: Initialize an EmbeddingRouter with your routes and test it against the ambiguous queries you identified. What threshold value gives you the best balance of coverage versus accuracy? Plot confidence scores for 20 test queries to visualize the distribution.
Part 3: Implement an LLMRouter that includes a custom system prompt explaining the financial context and the business consequence of misrouting (e.g., routing a trade execution query to the wrong handler could have real money implications). Does adding that context improve classification?
Part 4: Wire up a HybridSemanticRouter and add logging that writes each routing decision to a local JSONL file, including: timestamp, query (hashed for privacy), route chosen, strategy used, and confidence score. After running 50 test queries, analyze the log to find which routes have the lowest average confidence — those are candidates for adding more examples.
If you set the embedding threshold at 0.90, you'll send almost everything through the LLM classifier and lose the performance benefit. If you set it at 0.60, you'll get false positive route matches for genuinely ambiguous queries.
Fix: Start with 0.75, run 100+ representative queries, plot the score distribution, and look for a natural gap between the "clearly matched" cluster and the "uncertain" cluster. Your threshold belongs in that gap.
If two routes share overlapping example queries, the embedding space gets confused. Route A examples like "How do I download my data?" and Route B examples like "Can I export my data?" will produce embeddings that compete.
Fix: Review your examples for semantic overlap and make each route's examples as distinctive as possible. If two routes genuinely look the same to an embedding model, they probably belong together as one route.
A query like "Can you fix that?" will fail to route correctly without knowing what "that" refers to. The embedding router will return low confidence (correctly), but even the LLM classifier can't route it without context.
Fix: Always pass recent conversation history to your router. A rolling window of the last 3-5 exchanges is usually sufficient. Longer windows cost more tokens without proportional benefit — see our discussion of token costs in Understanding Tokens.
In production, your routing will silently degrade as user behavior shifts. New product features introduce new query types. Marketing campaigns drive traffic from different personas. Without monitoring, you won't know your router is failing until users complain.
Fix: Log every routing decision with its confidence score, strategy used, and — when available — user feedback. Set up alerts when the average confidence score drops more than 5 points week-over-week, or when fallback route usage spikes. Check out Implementing LLM Observability for the full monitoring stack.
Warning: Never trust your routing layer blindly in security-sensitive contexts. A user crafting a query specifically to get routed to a privileged handler (like
trade_executionoradmin_tools) is a real threat. Combine routing with handler-level authorization checks and input validation — ideally using a dedicated guardrails layer.
You add a new route definition but forget to register a handler. Or you rename a route in the router but not in the dispatcher. These mismatches fail silently or produce confusing errors.
Fix: Add a startup validation check:
def validate_dispatcher(dispatcher: RoutingDispatcher, routes: list[Route]):
"""Ensure every defined route has a registered handler."""
route_names = {r.name for r in routes}
handler_names = set(dispatcher.handlers.keys())
unhandled = route_names - handler_names
unused_handlers = handler_names - route_names
if unhandled:
raise ValueError(f"Routes without handlers: {unhandled}")
if unused_handlers:
logger.warning(f"Handlers without routes (may be intentional): {unused_handlers}")
print("✓ Dispatcher validation passed")
The embedding-based routing path costs you one embedding API call per query. At text-embedding-3-small pricing, that's roughly $0.00002 per query — essentially free. The computation to compare the embedded query against your route reference embeddings is pure numpy math that runs in microseconds.
The LLM fallback path costs you a full inference call. With gpt-4o-mini and a typical routing prompt (under 500 tokens), that's approximately $0.0003 per query. Still cheap, but 15x more expensive than the embedding path, and with significantly higher latency (500ms-1500ms vs 100-200ms for embeddings).
This is exactly why the hybrid approach matters: if your embedding router handles 80% of queries confidently, you only pay LLM costs for the remaining 20%. At scale, that's a meaningful difference.
Key insight: Improving your embedding router's coverage (by adding better examples per route) has a direct cost and latency impact. Every percentage point you shift from the LLM fallback to the embedding match saves money and cuts response time. Treat example quality as infrastructure investment.
For very high-traffic systems, you can go further by caching routing decisions. If 40% of your traffic is semantically similar recurring queries, you can cache the routing result by semantic fingerprint using a Redis-backed semantic cache — covered thoroughly in Implementing LLM Response Caching with Redis.
A routing schema is not a one-time design decision. It evolves as your product evolves. Here's a process for keeping it sharp:
Weekly: Review routing decisions where the LLM was used (because embedding confidence was low). Are there recurring query patterns that consistently fall through? Those are candidates for new example phrases in an existing route, or a new route entirely.
Monthly: Sample routing decisions where human feedback indicates a bad response. Trace back to the routing decision — was the query sent to the wrong handler? If so, was it a routing failure or a handler failure?
Quarterly: Audit your route definitions against actual traffic distribution. If billing_support is seeing 50% of traffic but data_export sees 2%, either the routes need rebalancing or the product usage patterns have shifted dramatically.
This feedback loop is essentially building a preference dataset specifically for your routing layer. Over time, it lets you evaluate whether routing changes actually improve outcomes — not just whether they change routing decisions.
Semantic routing transforms your LLM application from a monolithic pipeline into a modular, expert system. Instead of asking one model to be a SQL expert, a billing support agent, a documentation bot, and a customer de-escalation specialist simultaneously, you dispatch each query to the handler built specifically for it.
The key principles to carry forward:
Where to go next from here: