This llamaindex routerqueryengine tutorial walks you through building a production-ready multi-source RAG system that routes queries to the right index automatically. You’ll learn to configure selector prompts, handle hybrid retrieval, and add observability — all with runnable code you can adapt immediately.

Prerequisites

Python 3.10+
An OpenAI API key (or any LLM provider supported by LlamaIndex)
A vector database — we’ll use Chroma for the example, but Pinecone, Weaviate, or Qdrant work the same way
Familiarity with basic LlamaIndex concepts: documents, nodes, indexes, query engines

Install the dependencies:

pip install llama-index llama-index-vector-stores-chroma llama-index-llms-openai chromadb

Set your API key:

export OPENAI_API_KEY="sk-..."

Project structure

multi_source_rag/
├── data/
│   ├── technical_docs/     # API references, specs
│   └── product_docs/       # Marketing, FAQs, pricing
├── src/
│   ├── build_indexes.py
│   ├── router_engine.py
│   └── eval.py
└── requirements.txt

Create the directories and add sample documents. For this tutorial, we’ll synthesize content programmatically so you can run it immediately.

Step 1: Build separate indexes per source

Each source gets its own index and query engine. This keeps retrieval scoped and lets the router make clean decisions.

# src/build_indexes.py
import os
from pathlib import Path
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    StorageContext,
    Settings,
)
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

# Configure global settings
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

PERSIST_DIR = Path("./storage")
PERSIST_DIR.mkdir(exist_ok=True)

chroma_client = chromadb.PersistentClient(path=str(PERSIST_DIR / "chroma"))

def build_index(collection_name: str, data_dir: Path) -> VectorStoreIndex:
    """Build or load a vector index for a document collection."""
    vector_store = ChromaVectorStore(
        chroma_collection=chroma_client.get_or_create_collection(collection_name)
    )
    storage_context = StorageContext.from_defaults(vector_store=vector_store)

    # Check if collection already has data
    collection = chroma_client.get_collection(collection_name)
    if collection.count() > 0:
        print(f"Loading existing index: {collection_name} ({collection.count()} vectors)")
        return VectorStoreIndex.from_vector_store(vector_store, storage_context=storage_context)

    print(f"Building new index: {collection_name}")
    documents = SimpleDirectoryReader(str(data_dir), recursive=True).load_data()
    print(f"  Loaded {len(documents)} documents")

    index = VectorStoreIndex.from_documents(
        documents, storage_context=storage_context, show_progress=True
    )
    print(f"  Indexed {len(documents)} documents")
    return index

if __name__ == "__main__":
    # Create sample data directories
    tech_dir = Path("./data/technical_docs")
    product_dir = Path("./data/product_docs")
    tech_dir.mkdir(parents=True, exist_ok=True)
    product_dir.mkdir(parents=True, exist_ok=True)

    # Write sample technical docs
    (tech_dir / "api_reference.md").write_text("""
# Authentication API

## POST /auth/token
Obtain an access token using client credentials.

**Request:**
```json
&#123;"client_id": "string", "client_secret": "string", "grant_type": "client_credentials"&#125;

Response:

&#123;"access_token": "eyJ...", "token_type": "Bearer", "expires_in": 3600&#125;

Error codes:

400: Invalid client credentials
429: Rate limit exceeded (100 req/min) “”“)

(tech_dir / “rate_limits.md”).write_text(“”“

Rate Limiting

All API endpoints enforce rate limits per organization.

Tier	Requests/min	Burst
Free	60	10
Pro	300	50
Enterprise	1000	200

Headers returned:

X-RateLimit-Limit
X-RateLimit-Remaining
X-RateLimit-Reset “”“)

Write sample product docs

(product_dir / “pricing.md”).write_text(“”“

Pricing Plans

Free Tier

1,000 API calls/month
Community support
1 project

Pro ($49/month)

100,000 API calls/month
Email support (24h SLA)
10 projects
Custom domains

Enterprise (custom)

Unlimited API calls
Dedicated support (1h SLA)
Unlimited projects
SSO, audit logs, SLA “”“)

(product_dir / “faq.md”).write_text(“”“

Frequently Asked Questions

Q: Can I upgrade from Free to Pro anytime? A: Yes, upgrades take effect immediately. Prorated billing applies.

Q: What happens if I exceed my rate limit? A: Requests return 429 with Retry-After header. Implement exponential backoff.

Q: Do you offer discounts for nonprofits? A: Yes, 50% off Pro for verified 501(c)(3) organizations. “”“)

Build both indexes

tech_index = build_index(“technical_docs”, tech_dir) product_index = build_index(“product_docs”, product_dir)

print(“\nIndexes ready. Run router_engine.py next.”)


Run it:

```bash
python src/build_indexes.py

Expected output:

Building new index: technical_docs
  Loaded 2 documents
  Indexed 2 documents
Building new index: product_docs
  Loaded 2 documents
  Indexed 2 documents

Indexes ready. Run router_engine.py next.

Step 2: Create the router query engine

The RouterQueryEngine uses an LLM-based selector to pick the right query engine. We’ll use LLMSingleSelector for single-best routing and LLMMultiSelector for queries that need multiple sources.

# src/router_engine.py
from llama_index.core import (
    VectorStoreIndex,
    StorageContext,
    Settings,
    QueryBundle,
)
from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector, LLMMultiSelector
from llama_index.core.tools import QueryEngineTool
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
from pathlib import Path

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

PERSIST_DIR = Path("./storage")
chroma_client = chromadb.PersistentClient(path=str(PERSIST_DIR / "chroma"))

def load_index(collection_name: str) -> VectorStoreIndex:
    vector_store = ChromaVectorStore(
        chroma_collection=chroma_client.get_collection(collection_name)
    )
    storage_context = StorageContext.from_defaults(vector_store=vector_store)
    return VectorStoreIndex.from_vector_store(vector_store, storage_context=storage_context)

def build_router_engine():
    # Load indexes
    tech_index = load_index("technical_docs")
    product_index = load_index("product_docs")

    # Create query engines with different retrieval configs
    tech_engine = tech_index.as_query_engine(
        similarity_top_k=4,
        response_mode="compact",
    )
    product_engine = product_index.as_query_engine(
        similarity_top_k=4,
        response_mode="compact",
    )

    # Define tools with descriptions the selector uses
    tech_tool = QueryEngineTool.from_defaults(
        query_engine=tech_engine,
        name="technical_docs",
        description=(
            "Use for technical questions: API endpoints, authentication, rate limits, "
            "error codes, SDK usage, integration guides, webhooks, and system architecture."
        ),
    )
    product_tool = QueryEngineTool.from_defaults(
        query_engine=product_engine,
        name="product_docs",
        description=(
            "Use for product questions: pricing, plans, features, billing, FAQs, "
            "discounts, trial periods, and commercial terms."
        ),
    )

    # Single selector for unambiguous queries
    single_selector = LLMSingleSelector.from_defaults()

    # Multi selector for queries spanning sources
    multi_selector = LLMMultiSelector.from_defaults()

    # Router with both selectors — tries single first, falls back to multi
    router = RouterQueryEngine(
        selector=single_selector,
        query_engine_tools=[tech_tool, product_tool],
        # Enable multi-select for complex queries
        selector=multi_selector,
        verbose=True,
    )
    return router

def test_queries(router):
    test_cases = [
        # Technical queries
        "How do I authenticate with the API?",
        "What are the rate limit headers?",
        # Product queries
        "How much does the Pro plan cost?",
        "Do you offer nonprofit discounts?",
        # Hybrid queries — should hit both
        "What's the rate limit for the Pro plan?",
        "Can I use custom domains on the Free tier?",
    ]

    for query in test_cases:
        print(f"\n{'='*60}")
        print(f"QUERY: {query}")
        print(f"{'='*60}")
        response = router.query(query)
        print(f"RESPONSE:\n{response}")
        print(f"SOURCE NODES: {[n.node.metadata.get('file_name', 'unknown') for n in response.source_nodes]}")

if __name__ == "__main__":
    router = build_router_engine()
    test_queries(router)

Run it:

python src/router_engine.py

Expected output (truncated for readability):

============================================================
QUERY: How do I authenticate with the API?
============================================================
RESPONSE:
To authenticate with the API, use the POST /auth/token endpoint with client credentials...
SOURCE NODES: ['api_reference.md']

============================================================
QUERY: What's the rate limit for the Pro plan?
============================================================
RESPONSE:
The Pro plan allows 300 requests per minute with a burst of 50...
SOURCE NODES: ['rate_limits.md', 'pricing.md']

Notice the hybrid query "What's the rate limit for the Pro plan?" retrieves from both rate_limits.md (technical) and pricing.md (product). The LLMMultiSelector correctly identifies this needs multiple sources.

Step 3: Customize selector prompts for your domain

The default selector prompt works well, but domain-specific prompts improve routing accuracy. Here’s how to inject a custom prompt:

# src/custom_selector.py
from llama_index.core.selectors import LLMSingleSelector
from llama_index.core.prompts import PromptTemplate

CUSTOM_SELECTOR_PROMPT = PromptTemplate(
    """You are a routing agent for a developer platform with two knowledge bases:

1. TECHNICAL_DOCS: API references, authentication flows, rate limits, error codes, SDKs, webhooks, system architecture
2. PRODUCT_DOCS: Pricing tiers, feature comparisons, billing, FAQs, discounts, trials, commercial terms

A query may require ONE or BOTH sources. Respond with a JSON object:
{
  "choice": "TECHNICAL_DOCS" | "PRODUCT_DOCS" | "BOTH",
  "reason": "brief explanation"
}

Query: {query_str}
"""
)

def get_custom_selector():
    return LLMSingleSelector.from_defaults(
        prompt_template=CUSTOM_SELECTOR_PROMPT
    )

Plug it into the router:

# In router_engine.py, replace the selector line:
from custom_selector import get_custom_selector

router = RouterQueryEngine(
    selector=get_custom_selector(),
    query_engine_tools=[tech_tool, product_tool],
    verbose=True,
)

Test the custom selector with ambiguous queries:

# Add to test_queries in router_engine.py
ambiguous_cases = [
    "What limits apply to my account?",
    "How do I get more API calls?",
    "What's included in the Enterprise plan?",
]

for query in ambiguous_cases:
    print(f"\n{'='*60}")
    print(f"AMBIGUOUS: {query}")
    print(f"{'='*60}")
    response = router.query(query)
    print(f"RESPONSE:\n{response}")

Output shows the selector reasoning:

============================================================
AMBIGUOUS: What limits apply to my account?
============================================================
RESPONSE:
Your account limits depend on your plan tier. The Free tier allows 60 requests/minute...
SOURCE NODES: ['rate_limits.md', 'pricing.md']

The custom prompt’s explicit BOTH option helps the model admit when it needs multiple sources rather than guessing.

Step 4: Add hybrid retrieval per source

Different sources benefit from different retrieval strategies. Technical docs often need keyword matching for error codes; product docs work better with semantic search.

# src/hybrid_retrieval.py
from llama_index.core.retrievers import VectorIndexRetriever, KeywordTableRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core import get_response_synthesizer
from llama_index.core.postprocessor import SimilarityPostprocessor

def build_hybrid_engine(index, name: str):
    """Build a query engine that combines vector + keyword retrieval."""
    # Vector retriever (semantic)
    vector_retriever = VectorIndexRetriever(
        index=index,
        similarity_top_k=4,
    )

    # Keyword retriever (exact matches for codes, IDs, terms)
    keyword_retriever = KeywordTableRetriever(
        index=index,
        num_chunks_per_query=3,
    )

    # Fuse results — simple concatenation with deduplication
    class HybridRetriever:
        def __init__(self, *retrievers):
            self.retrievers = retrievers

        def retrieve(self, query_bundle):
            all_nodes = []
            seen = set()
            for r in self.retrievers:
                nodes = r.retrieve(query_bundle)
                for n in nodes:
                    key = n.node.node_id
                    if key not in seen:
                        seen.add(key)
                        all_nodes.append(n)
            return all_nodes

    hybrid_retriever = HybridRetriever(vector_retriever, keyword_retriever)

    # Post-process: filter by similarity threshold
    postprocessor = SimilarityPostprocessor(similarity_cutoff=0.7)

    response_synthesizer = get_response_synthesizer(response_mode="compact")

    return RetrieverQueryEngine(
        retriever=hybrid_retriever,
        response_synthesizer=response_synthesizer,
        node_postprocessors=[postprocessor],
    )

Replace the engine creation in router_engine.py:

# Replace:
# tech_engine = tech_index.as_query_engine(...)
# product_engine = product_index.as_query_engine(...)

# With:
tech_engine = build_hybrid_engine(tech_index, "technical")
product_engine = build_hybrid_engine(product_index, "product")

Now queries like `“42 like “error code 429” hit the keyword retriever for exact matches, while “how does authentication work” uses semantic search.

Step 5: Observability and logging

Production systems need visibility into routing decisions. LlamaIndex’s callback system captures this.

# src/observability.py
from llama_index.core.callbacks import CallbackManager, LlamaDebugHandler
from llama_index.core import Settings
import json
import time

debug_handler = LlamaDebugHandler(print_trace_on_end=True)
Settings.callback_manager = CallbackManager([debug_handler])

class RoutingLogger:
    """Log routing decisions for analysis."""
    def __init__(self):
        self.events = []

    def log_routing(self, query: str, selected_tools: list, latency_ms: float):
        self.events.append({
            "timestamp": time.time(),
            "query": query,
            "selected_tools": selected_tools,
            "latency_ms": latency_ms,
        })

    def export(self, path: str):
        with open(path, "w") as f:
            json.dump(self.events, f, indent=2)

# Usage in router_engine.py:
routing_logger = RoutingLogger()

# Wrap router.query to capture routing
original_query = router.query

def logged_query(query_str):
    start = time.time()
    response = original_query(query_str)
    latency = (time.time() - start) * 1000

    # Extract selected tools from response metadata
    selected = []
    if hasattr(response, 'metadata') and 'selector_result' in response.metadata:
        selected = [r.tool_name for r in response.metadata['selector_result'].selections]

    routing_logger.log_routing(query_str, selected, latency)
    return response

router.query = logged_query

After running test queries, export the log:

routing_logger.export("./routing_logs.json")

Sample log entry:

{
  "timestamp": 1704067200.123,
  "query": "What's the rate limit for the Pro plan?",
  "selected_tools": ["technical_docs", "product_docs"],
  "latency_ms": 1247.3
}

This lets you measure routing accuracy, latency per route, and identify queries that consistently hit the wrong source.

Step 6: Evaluation harness

Route correctness is measurable. Build a small eval set and score routing decisions.

# src/eval.py
from router_engine import build_router_engine
import json

EVAL_SET = [
    {"query": "How do I get an access token?", "expected": ["technical_docs"]},
    {"query": "What is the Free tier limit?", "expected": ["product_docs"]},
    {"query": "Rate limit for Enterprise plan", "expected": ["technical_docs", "product_docs"]},
    {"query": "Nonprofit discount eligibility", "expected": ["product_docs"]},
    {"query": "Webhook signature verification", "expected": ["technical_docs"]},
    {"query": "Can I upgrade mid-cycle?", "expected": ["product_docs"]},
    {"query": "X-RateLimit-Reset header format", "expected": ["technical_docs"]},
    {"query": "Custom domain setup guide", "expected": ["technical_docs", "product_docs"]},
]

def evaluate_routing():
    router = build_router_engine()
    results = []

    for case in EVAL_SET:
        response = router.query(case["query"])

        # Extract selected tools
        selected = []
        if hasattr(response, 'metadata') and 'selector_result' in response.metadata:
            selected = [r.tool_name for r in response.metadata['selector_result'].selections]

        # Compute metrics
        expected_set = set(case["expected"])
        selected_set = set(selected)

        precision = len(expected_set & selected_set) / len(selected_set) if selected_set else 0
        recall = len(expected_set & selected_set) / len(expected_set) if expected_set else 0
        f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0

        result = {
            "query": case["query"],
            "expected": case["expected"],
            "selected": selected,
            "precision": precision,
            "recall": recall,
            "f1": f1,
        }
        results.append(result)
        print(f"Q: {case['query']}")
        print(f"  Expected: {case['expected']}")
        print(f"  Selected: {selected}")
        print(f"  F1: {f1:.2f}\n")

    avg_f1 = sum(r["f1"] for r in results) / len(results)
    print(f"Average F1: {avg_f1:.2f}")

    with open("./eval_results.json", "w") as f:
        json.dump(results, f, indent=2)

if __name__ == "__main__":
    evaluate_routing()