DEV Community

Mithilesh Kumar
Mithilesh Kumar

Posted on

Why Basic RAG Fails in Production and How Adaptive Query Routing Fixes It

#python #langchain #fastapi #ai

Most developers build Retrieval-Augmented Generation (RAG) pipelines assuming every user query needs a vector search. In production, this naive approach fails in three distinct scenarios:

Simple Queries: "Hi", "Who created this bot?", or general knowledge queries don't need expensive vector database lookups.

Ambiguous Queries: Vague user questions lead to noisy retrieval, diluting the LLM's context window with irrelevant chunks.

Out-of-Domain Queries: When the vector DB contains no relevant documents, naive RAG forces the LLM to hallucinate an answer based on poor context.

In this guide, I’ll break down how to implement Adaptive RAG with Dynamic Query Routing using LangChain, Vector Stores (Pinecone/Chroma), and FastAPI.

What is Adaptive RAG?
Instead of routing every request directly to vector retrieval, Adaptive RAG acts as an intent-aware orchestrator:

graph TD
A[User Query Received] --> B[Intent Classifier Node]
B -->|General Query| C[Direct LLM Response]
B -->|Internal Docs| D[Vector DB Retrieval]
B -->|External/News| E[Web Search Fallback]
D --> F[Hallucination Grader Node]

Classify Intent: Determine whether the query needs internal vector docs, web search, or a direct response.

Retrieve & Grade: Fetch documents, then evaluate their relevance score before generating the answer.

Fallback Circuit: If document relevance is low, trigger fallback web search (e.g., Tavily API) or ask the user for clarification.

Step 1: Building a Structured Router with Pydantic
We enforce a strict JSON output schema using Pydantic to ensure our routing decision is 100% deterministic.

from pydantic import BaseModel, Field
from typing import Literal

class RouteQuery(BaseModel):
"""Route a user query to the most appropriate data source."""

datasource: Literal["vectorstore", "web_search", "direct_llm"] = Field(
    ...,
    description="Given a user question, choose whether to route it to vectorstore, web search, or direct LLM."
)
reasoning: str = Field(
    ..., description="Brief explanation for the routing decision."
)
Enter fullscreen mode Exit fullscreen mode

Step 2: Query Classification Node
Using function calling / structured output capabilities of LLMs (like Google Gemini or OpenAI):

from langchain_core.prompts import ChatPromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash", temperature=0)
structured_router = llm.with_structured_output(RouteQuery)

system_prompt = """You are an expert at routing user queries.

  • Use 'vectorstore' for questions related to internal technical documents, architecture, or codebase.
  • Use 'web_search' for recent events, live news, or external context.
  • Use 'direct_llm' for greetings, general conversational queries, or basic coding syntax. """

route_prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{question}")
])

question_router = route_prompt | structured_router

Test the router

result = question_router.invoke({"question": "What is the API endpoint for CGC-NEXUS event registration?"})
print(f"Destination: {result.datasource} | Reason: {result.reasoning}")

Output: Destination: vectorstore | Reason: Query asks about specific internal project endpoints.

Step 3: Integrating into Async FastAPI Endpoint
Here is how to expose the adaptive pipeline via a FastAPI service:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Adaptive RAG Engine")

class QueryRequest(BaseModel):
question: str

@app.post("/api/v1/query")
async def process_query(request: QueryRequest):
try:
# Step 1: Route Query
decision = await question_router.ainvoke({"question": request.question})

    # Step 2: Execute based on intent
    if decision.datasource == "direct_llm":
        response = await llm.ainvoke(request.question)
        return {"source": "direct_llm", "answer": response.content}

    elif decision.datasource == "vectorstore":
        # Perform vector store search & hallucination check
        return {"source": "vectorstore", "answer": "Retrieved from vector database."}

    else:
        # Fallback to Web Search
        return {"source": "web_search", "answer": "Retrieved from web search fallback."}

except Exception as e:
    raise HTTPException(status_code=500, detail=str(e))
Enter fullscreen mode Exit fullscreen mode

Key Production Insights
Cost Optimization: Filtering out trivial questions before vector search reduces API calls and vector database read costs by up to 40%.

Zero Hallucination Loop: By running a fast grader node on retrieved chunks, you ensure irrelevant text never enters the final LLM prompt context.

Latency Reduction: Direct LLM calls bypass embedding generation and vector lookup entirely, responding in under 300ms.

Connect & Explore Code
Live Portfolio: https://mithilesh-kumar-ai-engineer.netlify.app/

GitHub Repository: https://github.com/mithxcode

LinkedIn: https://www.linkedin.com/in/mithileshkumar001

X (Twitter): https://x.com/MITHILESH_7781

How are you handling ambiguous queries in your RAG pipelines? Drop a comment below!

Top comments (0)