How to build a router query engine in LlamaIndex

Applications that search different knowledge sources need one entry point that chooses a retrieval path for each question. A LlamaIndex router query engine provides that entry point by selecting a query engine tool before forwarding the question.

Each QueryEngineTool pairs a query engine with metadata that describes its subject. The selector returns the chosen tool index and a reason, while RouterQueryEngine exposes that decision through the response's selector_result metadata.

The KeywordSelector class treats billing and refund separately from incident and outage. It rejects questions that match both term sets or neither set, so an unsupported question cannot silently fall through to a default route. An application can replace this deterministic selector with PydanticSingleSelector after its language model is configured.

1. Install the LlamaIndex core package in the active Python environment.

   $ python3 -m pip install --upgrade llama-index-core

   Related: How to install LlamaIndex with pip

2. Create router_query_engine.py with the imports, mock model, and deterministic query engine class.

   router_query_engine.py

   from llama_index.core import Settings
   from llama_index.core.base.base_selector import MultiSelection, SingleSelection
   from llama_index.core.llms import MockLLM
   from llama_index.core.query_engine import CustomQueryEngine, RouterQueryEngine
   from llama_index.core.selectors import BaseSelector
   from llama_index.core.tools import QueryEngineTool
    
    
   Settings.llm = MockLLM(max_tokens=32)
    
    
   class StaticQueryEngine(CustomQueryEngine):
       label: str
       answer: str
    
       def custom_query(self, query_str):
           return f"{self.label}: {self.answer}"

   MockLLM prevents RouterQueryEngine from resolving a hosted default model during this credential-free routing check. The static engines isolate the routing decision from retrieval and generation behavior.

3. Append the keyword selector with four explicit routing boundaries.

   class KeywordSelector(BaseSelector):
       def _get_prompts(self):
           return {}
    
       def _update_prompts(self, prompts):
           return None
    
       def _select(self, choices, query):
           query_text = query.query_str.lower()
           billing_match = any(
               term in query_text for term in ("billing", "refund")
           )
           incident_match = any(
               term in query_text for term in ("incident", "outage")
           )
    
           if billing_match and incident_match:
               raise ValueError("Question matched both routes.")
           if not billing_match and not incident_match:
               raise ValueError("Question matched no route.")
    
           if billing_match:
               selection = SingleSelection(
                   index=0,
                   reason="Billing terms matched the billing tool metadata.",
               )
           else:
               selection = SingleSelection(
                   index=1,
                   reason="Incident terms matched the operations tool metadata.",
               )
           return MultiSelection(selections=[selection])
    
       async def _aselect(self, choices, query):
           return self._select(choices, query)

   Billing-only questions select index 0, incident-only questions select index 1, mixed questions raise an ambiguity error, and unrelated questions raise an unmatched error. The asynchronous method delegates to the same decision.

4. Append the billing and operations query engine instances.

   billing_engine = StaticQueryEngine(
       label="billing_policy",
       answer="Refund requests are reviewed before a credit is issued.",
   )
   operations_engine = StaticQueryEngine(
       label="operations_playbook",
       answer="Priority incidents page the on-call engineer within 15 minutes.",
   )

5. Wrap both query engines in QueryEngineTool objects with distinct descriptions.

   tools = [
       QueryEngineTool.from_defaults(
           query_engine=billing_engine,
           name="billing_policy",
           description="Answers billing policy and refund process questions.",
       ),
       QueryEngineTool.from_defaults(
           query_engine=operations_engine,
           name="operations_playbook",
           description="Answers incident response and outage questions.",
       ),
   ]

   An LLM-backed selector evaluates these descriptions, so describe the questions each engine can answer rather than repeating only its name.

6. Instantiate RouterQueryEngine with the selector and both query engine tools.

   router = RouterQueryEngine(selector=KeywordSelector(), query_engine_tools=tools)

7. Append fail-capable assertions for the billing-only and incident-only routes.

   route_checks = [
       ("Who reviews refund requests?", "billing_policy"),
       ("What happens during a priority incident?", "operations_playbook"),
   ]
    
   for question, expected_tool in route_checks:
       response = router.query(question)
       selector_result = response.metadata["selector_result"]
       selected_tool = tools[selector_result.ind].metadata.name
       assert selected_tool == expected_tool
    
       print(f"question={question}")
       print(f"selected_tool={selected_tool}")
       print(f"answer={response}")

   Each assertion stops the program if a recognized question reaches the wrong query engine.

8. Append fail-capable assertions for ambiguous and unmatched questions.

   rejection_checks = [
       (
           "How does billing respond to an outage?",
           "Question matched both routes.",
       ),
       ("Where is the staff handbook?", "Question matched no route."),
   ]
    
   for question, expected_message in rejection_checks:
       try:
           router.query(question)
       except ValueError as error:
           assert str(error) == expected_message
           print(f"question={question}")
           print(f"route_error={error}")
       else:
           raise AssertionError(f"Expected route error: {expected_message}")

   The else branch stops the program if either rejected question is accepted, while the message assertion distinguishes the ambiguous and unmatched boundaries.

9. Run router_query_engine.py to verify all four routing boundaries.

   $ python3 router_query_engine.py
   question=Who reviews refund requests?
   selected_tool=billing_policy
   answer=billing_policy: Refund requests are reviewed before a credit is issued.
   question=What happens during a priority incident?
   selected_tool=operations_playbook
   answer=operations_playbook: Priority incidents page the on-call engineer within 15 minutes.
   question=How does billing respond to an outage?
   route_error=Question matched both routes.
   question=Where is the staff handbook?
   route_error=Question matched no route.