Why Your Agentic RAG System Costs 10x More Than It Should

# Wrong way: wrap every query in the agent loop regardless of complexity# This pattern ships in most agentic RAG tutorials.from langgraph.graph import StateGraph, ENDfrom typing import TypedDict, Annotatedimport operatorclass AgentState(TypedDict):    query: str    retrieved_docs: Annotated[list[str], operator.add]    answer: str    iterations: intdef retrieve(state: AgentState) -> AgentState:    """Retrieve documents for the current query."""    docs = retriever.invoke(state["query"])    return {"retrieved_docs": [d.page_content for d in docs]}def evaluate_sufficiency(state: AgentState) -> str:    """Ask the LLM if it has enough information."""    # This is where the governance vacuum lives.    # The agent evaluates its own confidence.    # There is no ceiling on how many times it can say "not enough".    # There is no token budget tracked across iterations.    # There is no circuit breaker.    prompt = f"""Given this context:{state['retrieved_docs']}Can you answer: {state['query']}? Reply SUFFICIENT or INSUFFICIENT."""        result = llm.invoke(prompt).content.strip()    return "generate" if result == "SUFFICIENT" else "retrieve"def generate(state: AgentState) -> AgentState:    context = "\n\n".join(state["retrieved_docs"])    answer = llm.invoke(        f"Context:\n{context}\n\nQuestion: {state['query']}\nAnswer:"    ).content    return {"answer": answer}# Build graph - no iteration limit, no token budget, no circuit breakergraph = StateGraph(AgentState)graph.add_node("retrieve", retrieve)graph.add_node("evaluate", evaluate_sufficiency)graph.add_node("generate", generate)graph.set_entry_point("retrieve")graph.add_edge("retrieve", "evaluate")graph.add_conditional_edges("evaluate", evaluate_sufficiency)graph.add_edge("generate", END)# Every query - simple FAQ, complex multi-hop, everything -# goes through this loop with no differentiation.# Simple queries: pay for 1-3 iterations they did not need.# Edge cases: may loop indefinitely until the context window fills.# Token budget per session: undefined.# Cost per query: unknown until the billing statement arrives.agent = graph.compile()

from enum import Enumfrom langchain_openai import ChatOpenAIclass QueryIntent(Enum):    DIRECT = "direct"        # No retrieval: LLM can answer from knowledge    SIMPLE = "simple"        # Single-pass: one retrieval round sufficient    COMPLEX = "complex"      # Agent loop: multi-hop reasoning requireddef classify_intent(query: str, fast_llm: ChatOpenAI) -> QueryIntent:    """    Classify query complexity using a fast, cheap model.    Route to the cheapest pipeline that can answer.        Use a small model here (gpt-4o-mini, claude-haiku) not your primary LLM.    The classification call cost is ~$0.000015 - trivial vs the savings    from skipping the agent loop on 60-70% of queries.        Calibrate against your production query distribution:    - Target: SIMPLE handles ~60-70% of queries    - COMPLEX handles ~20-30% (multi-hop, synthesis, comparison)    - DIRECT handles ~5-10% (greetings, basic facts)    """    prompt = """Classify this query for retrieval routing.Query: {query}DIRECT: answerable from general knowledge, no documents neededSIMPLE: requires retrieving from the knowledge base, single lookup sufficientCOMPLEX: requires multiple retrieval steps, sub-query decomposition, or synthesisReply with one word: DIRECT, SIMPLE, or COMPLEX""".format(query=query)        result = fast_llm.invoke(prompt).content.strip().upper()    try:        return QueryIntent(result.lower())    except ValueError:        return QueryIntent.SIMPLE  # Safe defaultdef route_query(query: str, fast_llm, primary_llm, retriever, agent) -> str:    """    Route based on intent classification.    Only COMPLEX queries enter the agent loop.    SIMPLE queries: single-pass retrieve and generate.    DIRECT queries: skip retrieval entirely.        fast_llm: cheap small model for classification (gpt-4o-mini, claude-haiku)    primary_llm: your main generation model (gpt-4o, claude-sonnet)    """    intent = classify_intent(query, fast_llm)        if intent == QueryIntent.DIRECT:        return primary_llm.invoke(query).content        elif intent == QueryIntent.SIMPLE:        # Single-pass: Part 1-4 pipeline, no agent loop        docs = retriever.invoke(query)        context = "\n\n".join([d.page_content for d in docs[:5]])        return primary_llm.invoke(            f"Context:\n{context}\n\nQuestion: {query}\nAnswer:"        ).content        else:  # QueryIntent.COMPLEX        # Agent loop - with budget governance (see Component 2)        return agent.invoke({"query": query, "budget_tokens": 8000})

from typing import TypedDict, Annotatedfrom langgraph.graph import StateGraph, ENDimport operatorimport tiktokenclass BudgetedAgentState(TypedDict):    query: str    retrieved_docs: Annotated[list[str], operator.add]    answer: str    iterations: int    tokens_used: int    budget_tokens: int   # Hard ceiling set at session start    terminated_early: bool    next_step: str       # Set by check_budget_and_evaluate, read by routerdef count_tokens(text: str, model: str = "gpt-4o") -> int:    """Count tokens for budget tracking."""    enc = tiktoken.encoding_for_model(model)    return len(enc.encode(text))def check_budget_and_evaluate(state: BudgetedAgentState) -> BudgetedAgentState:    """    Node function: evaluates agent confidence and updates state with routing signal.    Returns state update dict - does NOT return routing string directly.        Sets state["next_step"] which the router reads to pick the next edge.    """    MAX_ITERATIONS = 4        # Hard ceilings - terminate without LLM call    if state["tokens_used"] >= state["budget_tokens"]:        return {"next_step": "generate_with_warning"}        if state["iterations"] >= MAX_ITERATIONS:        return {"next_step": "generate_with_warning"}        # Confidence evaluation    prompt = f"""Context gathered so far:{chr(10).join(state['retrieved_docs'][-10:])}Can you answer: {state['query']}?Reply SUFFICIENT or INSUFFICIENT."""        tokens_this_call = count_tokens(prompt)        # Pre-flight: will this call push us over budget?    if state["tokens_used"] + tokens_this_call > state["budget_tokens"]:        return {            "next_step": "generate_with_warning",            "tokens_used": state["tokens_used"] + tokens_this_call,        }        result = llm.invoke(prompt).content.strip()    next_step = "generate" if result == "SUFFICIENT" else "retrieve"    return {        "next_step": next_step,        "tokens_used": state["tokens_used"] + tokens_this_call,    }def route_from_budget_check(state: BudgetedAgentState) -> str:    """    Pure routing function: reads state["next_step"] set by the node above.    Returns the edge key for conditional routing.    This is separate from the node function - LangGraph requires this separation.    """    return state.get("next_step", "retrieve")def budgeted_retrieve(state: BudgetedAgentState) -> BudgetedAgentState:    """Retrieve and track token spend."""    query = state["query"]    docs = retriever.invoke(query)    doc_texts = [d.page_content for d in docs[:5]]        # Track tokens consumed by this retrieval round    tokens_this_round = sum(count_tokens(t) for t in doc_texts)        return {        "retrieved_docs": doc_texts,        "iterations": state["iterations"] + 1,        "tokens_used": state["tokens_used"] + tokens_this_round,        "terminated_early": False,    }def budgeted_generate(state: BudgetedAgentState) -> BudgetedAgentState:    """Generate from whatever context was assembled within budget."""    context = "\n\n".join(state["retrieved_docs"])    answer = llm.invoke(        f"Context:\n{context}\n\nQuestion: {state['query']}\nAnswer:"    ).content    return {"answer": answer, "terminated_early": False}def generate_with_budget_warning(    state: BudgetedAgentState,) -> BudgetedAgentState:    """    Generate from partial context when budget ceiling is hit.    Flags the response for downstream monitoring.        Returning a flagged answer is better than looping indefinitely.    The flag surfaces in Langfuse/LangSmith traces as a metric    to monitor: what fraction of agent sessions hit the ceiling?    If that fraction is high, raise the ceiling or improve retrieval    so fewer iterations are needed.    """    context = "\n\n".join(state["retrieved_docs"])    answer = llm.invoke(        f"Context (budget limit reached after {state['iterations']} "        f"iterations):\n{context}\n\nQuestion: {state['query']}\n"        f"Answer based on available context:"    ).content    return {        "answer": answer,        "terminated_early": True,  # Flagged for monitoring    }def build_budgeted_agent():    """    Assemble the budget-governed agent graph.        Node separation (LangGraph requirement):    - check_budget_and_evaluate: state-updating node, sets state["next_step"]    - route_from_budget_check: pure routing function, reads state["next_step"]        The routing function is passed to add_conditional_edges, NOT added as a node.    """    graph = StateGraph(BudgetedAgentState)        graph.add_node("retrieve", budgeted_retrieve)    graph.add_node("check_budget_and_evaluate", check_budget_and_evaluate)    graph.add_node("generate", budgeted_generate)    graph.add_node("generate_with_warning", generate_with_budget_warning)        graph.set_entry_point("retrieve")    graph.add_edge("retrieve", "check_budget_and_evaluate")        # Routing function reads state["next_step"] set by the node above    graph.add_conditional_edges(        "check_budget_and_evaluate",        route_from_budget_check,  # Pure router - not a node        {            "retrieve": "retrieve",            "generate": "generate",            "generate_with_warning": "generate_with_warning",        }    )    graph.add_edge("generate", END)    graph.add_edge("generate_with_warning", END)        return graph.compile()

import timefrom dataclasses import dataclass, fieldfrom langfuse import Langfuselangfuse = Langfuse()@dataclassclass SessionMetrics:    session_id: str    query: str    intent: str    iterations: int = 0    tokens_used: int = 0    latency_ms: float = 0.0    terminated_early: bool = False    cost_usd: float = 0.0def track_agent_session(    session_id: str,    query: str,    intent: str,    agent_result: dict,    start_time: float,    token_price_per_million: float = 3.00,  # gpt-4o as of 2026) -> SessionMetrics:    """    Capture per-session metrics for cost governance monitoring.        Key signals to watch in production:        1. Average iterations per session by intent bucket       - If COMPLEX queries average 4+ iterations, your retrieval         quality is poor - the agent is searching repeatedly because         single-pass results are insufficient. Fix upstream (Parts 1-4).        2. terminated_early rate       - If >10% of sessions hit the budget ceiling, your ceiling         is too tight OR your corpus coverage is too sparse.         Distinguish the two by checking context_recall (Part 5).        3. Cost per session by intent       - SIMPLE sessions should cost 3-10x less than COMPLEX.         If they do not, your intent router is mis-classifying.        4. Token accumulation rate per iteration       - If context grows faster than 1,000 tokens per iteration,         your retrieval is returning too many documents.         Reduce top-k or tighten the reranker threshold.    """    metrics = SessionMetrics(        session_id=session_id,        query=query,        intent=intent,        iterations=agent_result.get("iterations", 0),        tokens_used=agent_result.get("tokens_used", 0),        latency_ms=(time.time() - start_time) * 1000,        terminated_early=agent_result.get("terminated_early", False),        cost_usd=agent_result.get("tokens_used", 0)        / 1_000_000        * token_price_per_million,    )        # Log to Langfuse for observability    langfuse.trace(        name="agentic_rag_session",        id=session_id,        metadata={            "intent": intent,            "iterations": metrics.iterations,            "tokens_used": metrics.tokens_used,            "terminated_early": metrics.terminated_early,            "cost_usd": metrics.cost_usd,            "latency_ms": metrics.latency_ms,        },    )        return metrics