Multi-Hop RAG

Overview

MultiHopRAG answers complex questions by performing several retrieval hops, tracking a ReasoningState (what's known, what's missing) between them, and deciding what to search for next based on identified gaps — rather than retrieving once and generating.

HopStrategy (Enum)

Defined in multihop_strategy. Referenced values in multi_hop include HopStrategy.ENTITY_EXPANSION (the default for HopResult), and the class docstring is simply "Hop strategy" — inspect multihop_strategy directly for the complete member list if you need every strategy name.

ReasoningState

@dataclass
class ReasoningState:
    known_facts: List[str] = field(default_factory=list)
    missing_info: List[str] = field(default_factory=list)
    reasoning_chain: List[str] = field(default_factory=list)
    confidence: float = 0.0
    answer_complete: bool = False

Accumulated across hops by _update_reasoning_state.

HopResult

@dataclass
class HopResult:
    query: str
    chunks: List[Tuple]
    hop_number: int
    extracted_entities: List[str]
    strategy: HopStrategy = HopStrategy.ENTITY_EXPANSION
    reasoning_step: str = ""
    gap_filled: Optional[str] = None

One record per hop, aggregated into the final answer via _aggregate_chunks and _build_reasoned_context.

QueryDecomposer

class QueryDecomposer:
    def __init__(self, use_stanza: bool = True, language: str = "ar"): ...
    def decompose(self, query: str) -> List[str]: ...
    def extract_entities(self, text: str) -> List[str]: ...
    def get_entities_by_type(self, text: str) -> dict: ...

MultiHopRAG

Constructor

def __init__(
    self,
    rag_system,
    max_hops: int = 3,
    min_score_threshold: float = 0.3,
    enable_query_decomposition: bool = True,
    use_stanza_ner: bool = True,
    language: str = "ar",
    confidence_threshold: float = 0.75,
)

query(question, hops=None, return_intermediate=False)

Internal workflow:

1. _analyze_question_requirements(question) classifies the question type (comparison / causal / descriptive / temporal via _classify_question_type) and determines what information is needed for a complete answer.
2. _select_initial_strategy and _prepare_queries set up the first hop.
3. _perform_reasoning_hops(original_question, queries, num_hops, requirements) — the core loop. Each hop:
   • Retrieves via the backend, filters already-seen chunks (_filter_chunks against a seen_chunks set to avoid redundant re-processing),
   • Extracts facts and entities (_extract_facts_from_chunks, _extract_entities_from_chunks),
   • Updates ReasoningState (_update_reasoning_state),
   • Identifies what's still missing (_identify_missing_info),
   • Computes running confidence (_calculate_confidence),
   • Plans the next hop's query (_plan_next_hop), building either a gap-filling query (_build_gap_filling_query) targeting a specific missing piece of information, or a bridging query (_build_bridge_query) connecting what's been learned back to the original question.
4. Stops early if ReasoningState.confidence crosses confidence_threshold or answer_complete becomes True, otherwise continues until max_hops (or the hops override) is exhausted.
5. _generate_reasoned_answer builds the final context (_build_reasoned_context) and prompt (_build_reasoned_prompt — which enforces "no info outside provided context" and auto-replies in the detected question language) before calling the backend/LLM.

If return_intermediate=True, hop-by-hop detail (formatted via _format_hop_result) is included in the return value alongside the final answer.

_detect_language(text) -> str (staticmethod)

Detects Arabic vs. English based on the ratio of Arabic characters in the text — a simple heuristic, not a full language-ID model.

_safe_attr(obj, attr, default=None) (staticmethod)

Generic safe attribute accessor returning default instead of raising AttributeError — used throughout the hop pipeline to tolerate heterogeneous chunk/result object shapes.

get_stats() -> dict

Running statistics updated by _update_stats after every query (e.g. average hops used).

Async API

async def aquery(self, question: str, max_hops: int = None, return_intermediate: bool = False)

Per its docstring, this runs the blocking query() in a thread pool rather than reimplementing the hop loop with native await — so CPU-bound parts of MultiHopRAG genuinely run off the event loop, but true concurrent I/O across hops is not implemented here (unlike, say, FederatedRAG's async design). Plus __aenter__/__aexit__.

Complete example

from zerda.rag.types.multi_hop import MultiHopRAG

multi_hop = MultiHopRAG(backend, max_hops=4, language="en", confidence_threshold=0.8)

result = multi_hop.query(
    "How did the printing press influence the Reformation, and what came after?",
    return_intermediate=True,
)
print(result)

Best practices

• Increase max_hops cautiously — each hop is a full retrieve (+ optional LLM) round trip; 3–4 hops is a reasonable ceiling for interactive latency.
• Set language explicitly rather than relying on _detect_language's character-ratio heuristic for short or code-mixed queries, where it is more likely to misclassify.

Common mistakes

• Expecting aquery to parallelize hops — it offloads the whole blocking call to a thread, it does not run hops concurrently (hops are inherently sequential/dependent on each other's results anyway).
• Setting confidence_threshold very low and being surprised the system stops after one hop — early stopping is intentional and driven directly by this threshold.

Related APIs

• Self-Improving RAG's RecursiveRAG — a related but distinct question-decomposition approach (splits into independent sub-questions rather than sequential dependent hops).