Agentic RAG

Overview

AgenticRAG wraps a backend rag_system in an autonomous decision loop. Instead of a fixed retrieve-then-generate pipeline, it repeatedly chooses one of five actions (RETRIEVE, REFINE_QUERY, CHECK_SUFFICIENCY, GENERATE, STOP) until it has enough context or exhausts max_iterations. Use it when a single retrieval pass is often insufficient — ambiguous queries, queries needing rewriting, or cases where you want automatic self-correction on the generated answer.

Exceptions

class AgenticRAGError(Exception): ...
class RetrievalError(AgenticRAGError): ...
class GenerationError(AgenticRAGError): ...
class MaxIterationsExceeded(AgenticRAGError): ...

AgentAction (Enum)

class AgentAction(Enum):
    RETRIEVE = "retrieve"
    GENERATE = "generate"
    REFINE_QUERY = "refine_query"
    CHECK_SUFFICIENCY = "check_sufficiency"
    STOP = "stop"

ReasoningDepth (str, Enum)

class ReasoningDepth(str, Enum):
    SHALLOW = "shallow"
    MEDIUM = "medium"
    DEEP = "deep"

Controls prompt complexity for LLM calls (used wherever the agent constructs refinement/sufficiency prompts).

AgenticConfig

@dataclass
class AgenticConfig:
    max_iterations: int = 5
    min_confidence: float = 0.7
    min_docs_required: int = 2
    enable_query_refinement: bool = True
    enable_self_correction: bool = True
    enable_sufficiency_check: bool = True
    reasoning_depth: ReasoningDepth = ReasoningDepth.MEDIUM
    cache_ttl_seconds: int = 300
    retry_attempts: int = 2
    retry_backoff_base: float = 0.5

Document

@dataclass
class Document:
    text: str
    score: float = 0.0
    metadata: Dict[str, Any] = field(default_factory=dict)
    source: str = ""

    def snippet(self, max_chars: int = 300) -> str: ...

Unified document representation used internally once raw backend output is normalised. snippet() returns the text truncated to max_chars, appending "…" if truncated.

ActionStep

@dataclass
class ActionStep:
    iteration: int
    action: AgentAction
    query: str
    details: str = ""
    duration_ms: float = 0.0
    confidence: float = 0.0
    refined_query: Optional[str] = None

One record per decision the agent makes; the full ordered list is returned as AgenticResult.action_history and is what on_action callbacks receive.

AgenticResult

@dataclass
class AgenticResult:
    answer: str
    retrieved_docs: List[Document]
    action_history: List[ActionStep]
    iterations: int
    reasoning: str
    confidence: float = 0.0
    cached: bool = False

    def to_dict(self) -> Dict[str, Any]: ...

to_dict() returns {"answer", "docs_used", "iterations", "confidence" (rounded to 4 decimals), "cached", "reasoning"} — note it summarizes retrieved_docs as a count (docs_used), not the full document list.

TTLCache

class TTLCache:
    def __init__(self, ttl_seconds: int = 300) -> None: ...
    def get(self, query: str) -> Optional[List[Document]]: ...
    def set(self, query: str, docs: List[Document]) -> None: ...
    def clear(self) -> None: ...

In-process, not thread-safe (explicitly documented as such in the docstring), SHA-256-keyed cache of retrieval results, used internally by AgenticRAG to avoid re-retrieving identical queries within ttl_seconds.

AgenticRAG

Constructor

def __init__(
    self,
    rag_system: Any,
    config: Optional[AgenticConfig] = None,
    llm: Optional[Any] = None,
    on_action: Optional[Callable[[ActionStep], None]] = None,
) -> None

query(query, context=None) -> AgenticResult

Runs the full decision loop synchronously.

Internal workflow (from _decide_action's documented priority order):

1. First iteration → always RETRIEVE.
2. No documents yet → RETRIEVE.
3. At max_iterations - 1 → force GENERATE (guarantees an answer is produced before the cap).
4. Documents present and sufficiency checking enabled → CHECK_SUFFICIENCY.
5. Low average document score, refinement enabled, and still early → REFINE_QUERY.
6. Otherwise → GENERATE.

Retrieval goes through _retrieve_with_retry (exponential backoff, raises RetrievalError after retry_attempts failures) and then _normalise_docs, which accepts Document objects, (chunk, score) tuples (e.g. FAISS output), dicts with text/content + score/relevance keys, or a {"results": [...]} wrapper — so your rag_system.retrieve() can return any of these shapes.

If enable_self_correction is on, the final answer passes through _self_correct, which asks the LLM to validate/improve it and silently falls back to the original answer on any error (this method never raises).

Raises: MaxIterationsExceeded if the loop overruns the cap (see note above on why this is designed to be rare); RetrievalError if every retrieval retry fails and no cached documents exist; GenerationError if every generation retry fails.

add_document(text, metadata=None) -> Any / add_documents(documents, metadatas=None) -> Any

Thin pass-through helpers that add content to the underlying rag_system's index (single vs. bulk).

clear_cache() -> None

Flushes both the retrieval TTLCache and the internal result cache.

aquery(query, **kwargs) -> AgenticResult

Async version of query().

astream_query(query)

Runs the same agent logic but streams the final answer token by token after the reasoning loop completes (the reasoning steps themselves are not streamed — only the last generation step is).

async for token in agent.astream_query("What is AI?"):
    print(token, end="", flush=True)

Async context manager

AgenticRAG implements __aenter__/__aexit__, so async with AgenticRAG(rag) as agent: is valid.

Complete example

from zerda.rag.types.agentic_rag import AgenticRAG, AgenticConfig, AgentAction

def trace(step):
    print(f"[{step.iteration}] {step.action.value} -> {step.details}")

config = AgenticConfig(max_iterations=4, min_confidence=0.75, reasoning_depth="deep")
agent = AgenticRAG(rag_system, config=config, on_action=trace)

result = agent.query("How did the industrial revolution affect urbanization?")
print(result.answer)
print(f"confidence={result.confidence:.2f} iterations={result.iterations}")
for step in result.action_history:
    print(step.action, step.confidence)

Best practices

• Keep max_iterations low (3–5) in latency-sensitive paths; each REFINE_QUERY or CHECK_SUFFICIENCY action costs an LLM round-trip.
• If your backend doesn't separate retrieval from generation, you may see reduced benefit from REFINE_QUERY, since the refined query only affects the next retrieve() call — not generate().
• Provide on_action in production to build audit trails of why a particular answer was produced.

Common mistakes

• Assuming MaxIterationsExceeded is common — the loop is designed to force a GENERATE before hitting the cap, so treat this exception as a signal of a config or backend problem, not routine control flow.
• Relying on _self_correct to raise on failure — it never does; check result.confidence instead if you need to detect low-quality answers.

Related APIs

• Self-Improving RAG — a different self-critique loop with an explicit confidence threshold and refinement history.
• Federated RAG — can wrap an AgenticRAG instance as one of its sources.