An agentic RAG system can do something a simple retrieval pipeline cannot: notice missing evidence, rewrite the query, fetch more context, verify contradictions, and try again.
That is also exactly how it quietly becomes expensive.
A demo looks great. The agent decomposes the question, retrieves from three sources, reranks, reflects, retrieves again, and produces a careful answer. Then production traffic arrives. Half the questions are simple. The agent still plans, retrieves, expands, reranks, and reflects. Latency climbs. Token usage climbs. Retrieval calls multiply. The system is now paying an agentic tax on questions that a single retrieval step could have answered.
The problem is not that agentic RAG is bad. The problem is that an agent loop without budget controls behaves like an unbounded search process. It can keep improving evidence until the marginal gain is tiny, while the cost curve keeps going up.
TL;DR
A simple RAG system usually looks like this:
query → retrieve → build prompt → generate
An agentic RAG system often looks more like this:
query
→ plan
→ retrieve
→ reflect
→ rewrite query
→ retrieve again
→ rerank
→ summarize evidence
→ detect gap
→ retrieve again
→ generate
Each arrow can cost something:
The dangerous part is that these costs compound. A second retrieval step is not just one extra retrieval call. It may require the model to process more context, rewrite the query, rerank a larger candidate set, and then reason over a larger evidence bundle.
Agentic RAG is powerful because it can recover from weak first-pass retrieval. But that same recovery mechanism can become a loop that keeps spending while accuracy improves only slightly.
The engineering challenge is not “make the agent smarter.” It is:
How do we make the agent stop at the right time?
Scenario:
Your agent is allowed to “keep searching until it is confident.” For hard questions, it retrieves ten times, expands each query, reranks every result, and appends everything to the context. For easy questions, it does nearly the same thing.
Why it matters:
Most teams cap model output tokens but forget to cap the loop itself. The model may stop generating, but the orchestration layer can keep calling tools.
A production agentic retrieval loop needs explicit budgets.
Solution:
Model the budget as part of the task.
from dataclasses import dataclass
@dataclass(frozen=True)
class RetrievalBudget:
max_steps: int
max_retrieval_calls: int
max_unique_queries: int
max_rerank_calls: int
max_context_tokens: int
max_latency_ms: int
class BudgetExhausted(Exception):
pass
class LoopGuard:
def __init__(self, budget: RetrievalBudget):
self.budget = budget
self.steps = 0
self.retrieval_calls = 0
self.unique_queries: set[str] = set()
self.rerank_calls = 0
self.context_tokens = 0
def charge(
self,
*,
steps: int = 0,
retrieval_calls: int = 0,
unique_queries: list[str] | None = None,
rerank_calls: int = 0,
context_tokens: int = 0,
) -> None:
self.steps += steps
self.retrieval_calls += retrieval_calls
self.rerank_calls += rerank_calls
self.context_tokens += context_tokens
if unique_queries:
self.unique_queries.update(unique_queries)
if self.steps > self.budget.max_steps:
raise BudgetExhausted("Too many agent steps.")
if self.retrieval_calls > self.budget.max_retrieval_calls:
raise BudgetExhausted("Too many retrieval calls.")
if len(self.unique_queries) > self.budget.max_unique_queries:
raise BudgetExhausted("Too many unique queries.")
if self.rerank_calls > self.budget.max_rerank_calls:
raise BudgetExhausted("Too many reranking calls.")
if self.context_tokens > self.budget.max_context_tokens:
raise BudgetExhausted("Context budget exceeded.")
The exact numbers depend on your product, but the categories matter. You need separate limits for:
Why this works:
It turns “do whatever it takes” into “do what is necessary within bounds.” The agent can still be adaptive, but it cannot spend without limit.
💡 Practical note:
Do not use one global budget for every task. A customer-support FAQ and a legal policy comparison need very different budgets.
Scenario:
The agent already retrieved a document that answers the question. Then it says, “Let me verify further,” retrieves three more chunks, and accidentally introduces contradictory text. The final answer becomes worse.
Why it matters:
Model confidence is not a reliable stop signal. A model can sound confident while wrong, and it can sound uncertain while having enough evidence.
If the loop stops based on vibes, cost and quality both become unpredictable.
Solution:
Define evidence requirements for the task.
For example, a billing question may require:
from dataclasses import dataclass, field
@dataclass(frozen=True)
class EvidenceRequirement:
key: str
min_sources: int = 1
requires_citation: bool = True
@dataclass
class EvidenceState:
facts: dict[str, list[str]] = field(default_factory=dict)
contradictions: bool = False
def evidence_is_sufficient(
requirements: list[EvidenceRequirement],
state: EvidenceState,
) -> bool:
if state.contradictions:
return False
for requirement in requirements:
sources = state.facts.get(requirement.key, [])
if len(sources) < requirement.min_sources:
return False
if requirement.requires_citation and not sources:
return False
return True
This is intentionally simple, but it changes the loop’s behavior. The agent now asks:
Why this works:
The loop stops when the task contract is satisfied, not when the model produces a convincing sentence.
For open-ended questions where requirements are fuzzy, use a softer rule: stop when additional retrieval produces diminishing evidence gain. If two consecutive retrieval steps add no new cited facts, stop or escalate.
Scenario:
The agent first searches:
“What is the refund window for annual plans?”
Two steps later, it searches:
“What is the refund period for annual subscriptions?”
The wording is different, but the intent is the same. Your retrieval system treats it as a new query, fetches similar chunks again, reranks them again, and appends them again.
Why it matters:
Agentic systems often rewrite queries. That is useful, but it creates duplicate work. If the cache key is the raw query string, small rephrasings defeat the cache.
Solution:
Normalize retrieval requests and cache by intent, filters, and source version.
import hashlib
import json
def retrieval_cache_key(
query: str,
filters: dict,
top_k: int,
source_version: str,
) -> str:
normalized_query = " ".join(query.lower().split())
payload = {
"query": normalized_query,
"filters": filters,
"top_k": top_k,
"source_version": source_version,
}
serialized = json.dumps(payload, sort_keys=True)
return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
For stronger deduplication, you can also track:
The key idea is that retrieval identity should be based on the request that matters, not the exact surface text.
Why this works:
It prevents the loop from paying repeatedly for the same evidence. This is especially important when reranking is expensive or when retrieval calls hit external APIs.
⚠️ Gotcha:
Cache invalidation must include source version. If the knowledge base changes, old evidence may no longer be valid.
Scenario:
A user asks a straightforward question. The agent generates five query variants, retrieves all five, reranks the combined results, and then answers. The answer is fine. The cost is five times higher than it needed to be.
Why it matters:
Query expansion improves recall, but it multiplies retrieval work. It is most valuable when the first query is ambiguous or when the corpus uses different vocabulary than the user.
It is least valuable when the first query is already precise.
Solution:
Use adaptive expansion.
Retrieve with the canonical query first. Expand only if the first pass looks weak or ambiguous.
@dataclass
class RetrievalHit:
chunk_id: str
score: float
text: str
def should_expand_query(hits: list[RetrievalHit]) -> bool:
if not hits:
return True
top_score = hits[0].score
if top_score < 0.55:
return True
if len(hits) >= 3:
margin = hits[0].score - hits[2].score
if margin < 0.05:
return True
return False
The thresholds are illustrative, not universal. The pattern is what matters: expansion should be triggered by evidence weakness, not applied by default.
Good expansion triggers include:
Why this works:
It makes the agent spend extra retrieval budget only when uncertainty justifies it.
🔍 Why this matters:
Query expansion can drift. If the agent generates speculative queries that are not grounded in the original question, it may retrieve plausible but irrelevant evidence.
Scenario:
Each loop iteration appends retrieved chunks to the conversation. By step four, the context contains repeated fragments, near-duplicate sections, and one crucial fact buried in the middle. The answer gets worse even though the agent “found more information.”
Why it matters:
More context is not always better. In retrieval-augmented systems, context quality matters more than context volume.
As the loop grows, the model has to deal with:
Solution:
Treat context assembly as a budgeting problem.
Keep an evidence ledger outside the prompt, then select only the strongest evidence for the model.
from typing import Callable
@dataclass
class EvidenceItem:
source_id: str
chunk_id: str
text: str
score: float
authority_tier: int
def assemble_context(
evidence: list[EvidenceItem],
estimate_tokens: Callable[[str], int],
max_tokens: int,
) -> list[EvidenceItem]:
selected: list[EvidenceItem] = []
used_tokens = 0
evidence.sort(
key=lambda item: (item.authority_tier, -item.score),
)
seen_chunks: set[str] = set()
for item in evidence:
if item.chunk_id in seen_chunks:
continue
tokens = estimate_tokens(item.text)
if used_tokens + tokens > max_tokens:
continue
selected.append(item)
used_tokens += tokens
seen_chunks.add(item.chunk_id)
return selected
This example sorts by authority and score, avoids duplicates, and respects a token budget. A production system may also consider:
Why this works:
The agent retains a full evidence history for auditing, but the model only sees the strongest subset.
🧠 The important part:
If every loop step appends raw retrieval results to the prompt, you are not building an evidence system. You are building a context landfill.
Scenario:
You cache final answers for common questions. Then a policy changes. Now the cached answer is wrong, and you do not know which retrieved evidence produced it.
Or the opposite happens: the final answer is too user-specific to cache, but the underlying evidence is stable.
Why it matters:
Final answers are often context-dependent. They may depend on:
Evidence is often more reusable than the answer.
Solution:
Cache retrieved evidence separately from generated responses.
from datetime import datetime
@dataclass
class CachedEvidence:
cache_key: str
chunks: list[RetrievalHit]
source_version: str
acl_fingerprint: str
created_at: datetime
When the agent issues a retrieval request, check whether the evidence bundle is still valid:
If yes, reuse the retrieved chunks. Then generate the answer using the current prompt, user context, and policy.
Why this works:
You reduce retrieval and reranking cost while preserving the ability to personalize or regenerate the final answer.
This is especially useful for multi-step agents. The same evidence bundle may be used for:
🚨 Production warning:
Never return cached evidence without rechecking permissions. A cache that ignores access control can become a data leak.
Scenario:
A user asks, “Where can I find the API key page?” The system launches an agentic loop: decompose, retrieve, reflect, rerank, verify. The answer is one sentence.
Why it matters:
Most production question distributions are skewed. Many questions are simple. A smaller set is genuinely multi-hop, ambiguous, or analytical.
If every question goes through the most powerful loop, you pay maximum cost for minimum necessary complexity.
Solution:
Route queries by complexity class.
from enum import Enum
class QueryClass(Enum):
SIMPLE_FACT = "simple_fact"
PROCEDURAL = "procedural"
MULTI_HOP = "multi_hop"
INVESTIGATIVE = "investigative"
def route_query(query_class: QueryClass) -> str:
if query_class == QueryClass.SIMPLE_FACT:
return "single_shot_rag"
if query_class == QueryClass.PROCEDURAL:
return "bounded_agentic_rag"
if query_class == QueryClass.MULTI_HOP:
return "bounded_agentic_rag"
return "supervised_agentic_rag"
In practice, the router may use:
A practical routing model often looks like this:
Why this works:
It preserves the power of agentic retrieval where it matters and avoids wasting it on trivial questions.
Scenario:
A new agentic loop improves answer quality slightly, but doubles retrieval calls and triples token usage. The team celebrates the accuracy gain until the monthly bill arrives.
Why it matters:
Accuracy alone is not enough. Production systems have constraints:
A system that is 2% more accurate but 5x more expensive may be worse for the product.
Solution:
Evaluate cost-adjusted performance.
Track metrics such as:
@dataclass
class TaskResult:
correct: bool
cost_usd: float
retrieval_calls: int
latency_ms: int
def cost_per_correct_answer(results: list[TaskResult]) -> float:
correct = [result for result in results if result.correct]
if not correct:
return float("inf")
total_cost = sum(result.cost_usd for result in correct)
return total_cost / len(correct)
You can also use a simple decision score:
def net_score(
accuracy: float,
average_cost_usd: float,
cost_penalty: float = 0.2,
) -> float:
return accuracy - cost_penalty * average_cost_usd
Do not treat that formula as universal truth. It is a way to force the tradeoff into the open.
Why this works:
It prevents teams from optimizing one dimension while ignoring the operational cost of the retrieval loop.
A useful evaluation table might look like this:
| Metric | What it reveals |
|---|---|
| Accuracy | Is the answer correct? |
| Groundedness | Is the answer supported by retrieved evidence? |
| Retrieval calls per task | How hard did the loop work? |
| Context tokens per task | How much evidence reached the model? |
| Budget exhaustion rate | How often tasks hit limits |
| Cost per correct answer | Is the accuracy worth the spend? |
| Latency p95 | Does the loop feel acceptable to users? |
Not every system needs the same amount of agency.
The right choice depends on the question distribution, the risk of being wrong, the cost of retrieval, and the tolerance for latency.
| Approach | Best for | Cost profile | Main risk | When to avoid |
|---|---|---|---|---|
| Single-shot RAG | Simple lookups, FAQs, stable docs | Low | Weak on ambiguous or multi-hop questions | Complex research tasks |
| Bounded agentic RAG | Most production assistants | Medium | Needs clear stop rules | Very simple high-volume traffic |
| Full agentic RAG | Research, analysis, multi-source investigation | High | Runaway loops and context bloat | Latency-sensitive or low-cost use cases |
| Human-in-the-loop agentic RAG | High-stakes decisions | Highest | Slow and operationally heavy | Casual or low-risk queries |
A reasonable default for many products is:
This is not a compromise. It is system design.
Agentic RAG should be used like a specialist tool, not like a default setting.
Before enabling an agentic retrieval loop in production, I would want these controls in place.
The core idea is simple:
Agentic RAG is not just a retrieval pattern. It is a spending pattern.
Used well, it buys better answers for hard questions. Used carelessly, it buys marginal accuracy gains at a price your system may not be able to sustain.
The goal is not to stop agents from retrieving. The goal is to make every retrieval step earn its place.