LlamaIndex Tutorial (Python): building conditional routing for intermediate developers
This tutorial shows you how to build a conditional router in LlamaIndex that sends each query to the right tool based on simple rules. You need this when one index is not enough and you want deterministic control over where a request goes, instead of hoping an LLM picks the right path every time.
What You'll Need
- •Python 3.10+
- •
llama-indexinstalled - •An OpenAI API key set as
OPENAI_API_KEY - •Optional but useful:
- •
python-dotenvfor local environment variables - •A basic understanding of
QueryEngine,Retriever, andRouterQueryEngine
- •
Install the packages:
pip install llama-index python-dotenv
Set your API key:
export OPENAI_API_KEY="your-key-here"
Step-by-Step
- •Start by creating two different knowledge sources and turning them into query engines. In this example, one engine answers billing questions and the other answers technical support questions.
from llama_index.core import VectorStoreIndex, Document
billing_docs = [
Document(text="Billing policy: invoices are generated on the 1st of each month."),
Document(text="Refunds are processed within 5 business days."),
]
tech_docs = [
Document(text="Technical support: restart the service if the API returns 502."),
Document(text="If auth fails, check that the token has not expired."),
]
billing_index = VectorStoreIndex.from_documents(billing_docs)
tech_index = VectorStoreIndex.from_documents(tech_docs)
billing_engine = billing_index.as_query_engine()
tech_engine = tech_index.as_query_engine()
- •Define routing logic with a custom selector function. This is the conditional part: if the query contains billing-related terms, route to billing; otherwise route to technical support.
def select_route(query: str) -> str:
q = query.lower()
billing_keywords = ["invoice", "billing", "refund", "payment", "charge"]
if any(keyword in q for keyword in billing_keywords):
return "billing"
return "technical"
test_queries = [
"When will I get my refund?",
"Why am I getting a 502 error?",
]
for q in test_queries:
print(q, "->", select_route(q))
- •Wrap both engines in a router using LlamaIndex’s
RouterQueryEngine. The router takes a selector object, so we’ll implement one that uses our custom condition function and returns a matching tool.
from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata
billing_tool = QueryEngineTool.from_defaults(
query_engine=billing_engine,
metadata=ToolMetadata(
name="billing",
description="Answers questions about invoices, refunds, payments, and charges.",
),
)
tech_tool = QueryEngineTool.from_defaults(
query_engine=tech_engine,
metadata=ToolMetadata(
name="technical",
description="Answers questions about API errors, auth issues, and service outages.",
),
)
- •Build a simple conditional selector that matches LlamaIndex’s routing interface. The selector returns an index into the tool list, which lets
RouterQueryEnginedispatch to the correct engine.
from llama_index.core.base.base_selector import BaseSelector
from llama_index.core.base.llms.types import SelectorResult
class KeywordSelector(BaseSelector):
def _select(self, choices, query_str):
route = select_route(query_str)
if route == "billing":
return SelectorResult(ind=0, reason="Matched billing keyword.")
return SelectorResult(ind=1, reason="Defaulted to technical support.")
selector = KeywordSelector()
router = RouterQueryEngine(selector=selector, query_engine_tools=[billing_tool, tech_tool])
- •Run a few queries through the router and inspect the responses. If your routing is working correctly, billing questions should hit the billing engine and technical questions should hit the technical engine.
queries = [
"How do I get a refund?",
"What should I do if auth fails?",
]
for query in queries:
response = router.query(query)
print(f"\nQ: {query}")
print(f"A: {response}")
- •Add guardrails for ambiguous inputs by making your routing logic explicit. In production, you usually want a default route plus logging so you can see where misroutes happen.
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("router")
def safe_select_route(query: str) -> str:
route = select_route(query)
logger.info("routing_query=%r route=%s", query, route)
return route
for q in ["Can you help me?", "Why was my card charged twice?"]:
print(safe_select_route(q))
Testing It
Test with queries that clearly belong to one category and verify they land on the expected engine. Use at least one ambiguous question like “Can you help me?” to confirm your default path behaves predictably.
If you want stronger verification, add logs around each engine call or temporarily make each tool return a distinct prefix like [BILLING] or [TECH]. That makes routing mistakes obvious during development.
Also test edge cases:
- •empty strings
- •mixed-intent questions like “I was charged twice and now auth fails”
- •synonyms not covered by your keyword list
Next Steps
- •Replace keyword routing with an LLM-based selector using LlamaIndex selectors when your taxonomy gets larger.
- •Add more tools for compliance, account lookup, or claims triage.
- •Persist routing decisions and build evaluation sets so you can measure misroutes before shipping to production.
Keep learning
- •The complete AI Agents Roadmap — my full 8-step breakdown
- •Free: The AI Agent Starter Kit — PDF checklist + starter code
- •Work with me — I build AI for banks and insurance companies
By Cyprian Aarons, AI Consultant at Topiax.
Want the complete 8-step roadmap?
Grab the free AI Agent Starter Kit — architecture templates, compliance checklists, and a 7-email deep-dive course.
Get the Starter Kit