ZerdaZerda

Self-Improving RAG

On this page

Overview

This module bundles three distinct strategies that share a dataclass vocabulary and a couple of internal helpers, but implement different ideas:

  1. SelfIRAG — iteratively retrieve → generate → evaluate → refine until a confidence threshold is met.
  2. HyDERAG — Hypothetical Document Embeddings: generate synthetic answer documents first, then retrieve using them as search queries.
  3. RecursiveRAG — decompose complex questions into sub-questions, answer each (recursively), then synthesize.

Shared exceptions

Python · 3 lines
class RAGSystemError(Exception): ...          # base for all errors in this module
class LLMCallError(RAGSystemError): ...        # every retry of an LLM call failed
class RetrievalError(RAGSystemError): ...      # document retrieval failed

[!WARNING] Name collision with agentic_rag.RetrievalError This RetrievalError is a separate class from agentic_rag.agentic_rag.RetrievalError. If you catch exceptions across both subsystems, import and reference each by its fully-qualified path.

Shared internal helpers (exported despite the underscore)

Python · 4 lines
def _doc_key(doc: Dict[str, Any]) -> str
def _deduplicate(docs: List[Dict[str, Any]]) -> List[Dict[str, Any]]
def _call_llm(llm: Any, prompt: str, retries: int = 2, backoff: float = 0.5) -> str
def _retrieve_safe(rag_system: Any, query: str, top_k: int = 5) -> List[Dict[str, Any]]
FunctionBehavior
_doc_key(doc)Stable hash key from the first 120 characters of a document's text — used for deduplication.
_deduplicate(docs)Removes duplicates by _doc_key, keeping the copy with the highest score.
_call_llm(llm, prompt, retries=2, backoff=0.5)Calls llm.generate(prompt) with exponential-backoff retries; raises LLMCallError if every attempt fails.
_retrieve_safe(rag_system, query, top_k=5)Normalizes any backend output shape to List[Dict] — handles List[Tuple[DocumentChunk, float]] (FAISS-style), already-normalized List[Dict], and {"results": [...]} wrapped dicts.

Dataclasses (self_rag_dataclass.py)

EvaluationResult

Python · 11 lines
@dataclass
class EvaluationResult:
    relevance: float = 0.5
    accuracy: float = 0.5
    completeness: float = 0.5
    clarity: float = 0.5
    issues: List[str] = field(default_factory=list)
    suggestions: List[str] = field(default_factory=list)

    @property
    def confidence(self) -> float: ...

confidence is documented as "geometric-mean-inspired" — deliberately not a plain arithmetic mean, so that a single very-low dimension (e.g. accuracy=0.1) drags the overall score down much more than an average would, which the docstring calls "more honest."

RefinementStep

Python · 8 lines
@dataclass
class RefinementStep:
    iteration: int
    answer: str
    evaluation: EvaluationResult

    @property
    def confidence(self) -> float: ...  # delegates to evaluation.confidence

SelfRAGResult

Python · 18 lines
@dataclass
class SelfRAGResult:
    answer: str
    confidence: float
    iterations: int
    history: List[RefinementStep]

    @property
    def final_answer(self) -> str: ...          # alias for answer
    @property
    def refinement_steps(self) -> int: ...       # alias for iterations
    @property
    def steps(self) -> List[RefinementStep]: ... # flattened, notebook-friendly view
    @property
    def initial_quality(self) -> float: ...      # confidence of first iteration
    @property
    def final_quality(self) -> float: ...        # confidence of last iteration
    def to_dict(self) -> Dict[str, Any]: ...

The multiple aliases (final_answer/answer, refinement_steps/ iterations) exist, per the source, for compatibility with example notebooks — either name is safe to use.

HyDEResult

Python · 6 lines
@dataclass
class HyDEResult:
    answer: str
    hypothetical_docs: List[str]
    num_retrieved: int
    num_ranked: int

SubAnswer / RecursiveRAGResult

Python · 13 lines
@dataclass
class SubAnswer:
    question: str
    answer: str
    depth: int

@dataclass
class RecursiveRAGResult:
    answer: str
    depth: int
    decomposed: bool
    sub_questions: List[str] = field(default_factory=list)
    sub_answers: List[SubAnswer] = field(default_factory=list)

Config classes (self_config.py)

SelfRAGConfig

Python · 9 lines
@dataclass
class SelfRAGConfig:
    max_refinement_iterations: int = 3
    confidence_threshold: float = 0.8
    enable_answer_refinement: bool = True
    enable_retrieval_refinement: bool = True
    llm_retries: int = 2
    fast_mode: bool = True
    skip_refinement_on_high_confidence: bool = True
FieldPurpose
fast_modeWhen True, generation and evaluation happen in one LLM call, saving roughly 50% of LLM calls per iteration.
skip_refinement_on_high_confidenceWhen True and the first answer already meets confidence_threshold, all refinement is skipped entirely.

RecursiveRAGConfig

Python · 7 lines
@dataclass
class RecursiveRAGConfig:
    max_depth: int = 3
    max_sub_questions: int = 4
    min_query_length: int = 40
    complexity_keywords: List[str] = field(default_factory=lambda: [...])  # Arabic + English markers
    llm_retries: int = 2

complexity_keywords defaults to a bilingual list (Arabic: و, أو, ثم, أسباب, نتائج, مقارنة, العلاقة, كيف, لماذا... plus English: and, or, causes, effects, compare, relationship, how, why, when).

SelfIRAG

Constructor

Python
def __init__(self, rag_system: Any, llm: Any, config: Optional[SelfRAGConfig] = None) -> None

query(query, context=None) -> SelfRAGResult

Loop: retrieve (_retrieve) → generate (_generate, or _generate_and_evaluate_fast when fast_mode=True) → evaluate (_evaluate, parsed via _parse_evaluation which "parses JSON evaluation output; falls back gracefully" on malformed LLM output) → if confidence is below confidence_threshold and iterations remain, _refine_answer and/or _adapt_query (which "appends top suggestion keywords to broaden the retrieval query") before the next iteration. The iteration with the highest confidence across the whole run is returned, not necessarily the last one.

Raises: propagates LLMCallError from _call_llm if every retry of an LLM call fails; propagates RetrievalError if _retrieve_safe cannot get any usable documents.

Other methods

add_document, get_stats() -> dict, aquery(query, context=None) -> SelfRAGResult (async wrapper), __aenter__/__aexit__.

HyDERAG

Constructor

Python · 11 lines
def __init__(
    self,
    rag_system: Any,
    llm: Any,
    num_hypothetical_docs: int = 3,
    use_original_query: bool = True,
    top_k: int = 5,
    max_final_docs: int = 10,
    llm_retries: int = 2,
    max_workers: int = 4,
) -> None

query(query, context=None) -> HyDEResult

  1. _generate_hypothetical_docs_parallel(query) — generates num_hypothetical_docs synthetic documents concurrently using a ThreadPoolExecutor (max_workers threads) to cut latency.
  2. Each hypothetical doc is used as a retrieval query; results are deduplicated (_deduplicate) and capped at max_final_docs.
  3. _generate(query, docs) builds the final answer directly from the real retrieved docs (not the hypothetical ones).
Python · 3 lines
hyde = HyDERAG(backend, llm, num_hypothetical_docs=3)
result = hyde.query("What are the benefits of exercise?")
print(result.answer, result.num_retrieved, result.num_ranked)

RecursiveRAG

Constructor

Python
def __init__(self, rag_system: Any, llm: Any, config: Optional[RecursiveRAGConfig] = None) -> None

query(query, context=None, _depth=0) -> RecursiveRAGResult

_depth is an internal recursion counter — do not set it manually.

  1. _is_complex(query) — heuristic: the query must be long enough and contain at least two of complexity_keywords.
  2. If not complex (or _depth >= max_depth) → _answer_directly.
  3. If complex → _decompose(query) asks the LLM to split it into sub-questions (parsed as a JSON array), each answered via a recursive query() call at _depth + 1, then _merge(query, sub_answers) asks the LLM to synthesize a final answer from all SubAnswers.
Python · 4 lines
rag = RecursiveRAG(backend, llm, config=RecursiveRAGConfig(max_depth=3))
result = rag.query("What caused the Industrial Revolution and what were its effects?")
print(result.answer)
print(result.sub_questions)

Complete example (combining all three)

Python · 14 lines
from zerda.rag.types.self_improving_rag import SelfIRAG, HyDERAG, RecursiveRAG, SelfRAGConfig

# Iterative refinement for a single, well-scoped question
self_rag = SelfIRAG(backend, llm, config=SelfRAGConfig(confidence_threshold=0.85))
r1 = self_rag.query("What is quantum entanglement?")
print(r1.final_answer, r1.final_quality)

# HyDE for vague/underspecified queries
hyde = HyDERAG(backend, llm, num_hypothetical_docs=4)
r2 = hyde.query("stuff about black holes")

# Recursive decomposition for compound questions
recursive = RecursiveRAG(backend, llm)
r3 = recursive.query("Compare the causes and effects of World War I and World War II.")

Best practices

  • Enable fast_mode (default) in SelfRAGConfig unless you specifically need generation and evaluation decoupled — it roughly halves LLM calls.
  • Use HyDERAG for queries where the phrasing of the question is unlikely to match document phrasing (a common recall problem in dense retrieval); use SelfIRAG when retrieval already works but answer quality is inconsistent; use RecursiveRAG for genuinely compound, multi-part questions.

Common mistakes

  • Passing a value for RecursiveRAG.query()'s _depth parameter manually — it exists purely for internal recursion bookkeeping.
  • Assuming SelfRAGResult.final_answer differs from .answer — they are the exact same value, just aliased for readability.
  • Multi-Hop RAG — a related but distinct decomposition strategy built around sequential, dependent hops rather than independent sub-questions merged at the end.
  • Agentic RAG — a different self-correction/refinement loop with its own confidence model.

Related pages