MultiHop-RAG Modular `multi_hop` — Enterprise API Reference

┌──────────────────────────────────────────────────────────────────────┐
│                          MultiHopRAG                                  │
│                                                                        │
│  ┌───────────────────┐        ┌──────────────────────────────────┐    │
│  │  QueryDecomposer  │        │        ReasoningState            │    │
│  │  (Stanza / Regex) │        │  known_facts · missing_info      │    │
│  └────────┬──────────┘        │  reasoning_chain · confidence    │    │
│           │ decompose()       └──────────────┬───────────────────┘    │
│           │                                  │ updated every hop      │
│           ▼                                  │                        │
│  ┌────────────────────────────────────────────────────────────────┐   │
│  │                   Reasoning Hop Loop                            │   │
│  │                                                                  │   │
│  │  ① _analyze_question_requirements  → question type + slots     │   │
│  │  ② _select_initial_strategy        → HopStrategy               │   │
│  │  ③ rag.retrieve(query)             → raw chunks                 │   │
│  │  ④ _filter_chunks                  → deduplicated + scored      │   │
│  │  ⑤ _extract_entities_from_chunks  → NER entities               │   │
│  │  ⑥ _update_reasoning_state        → facts + gaps + confidence  │   │
│  │  ⑦ Early stop check               → break if answer_complete   │   │
│  │  ⑧ _plan_next_hop                 → next query + strategy      │   │
│  └────────────────────────────────────────────────────────────────┘   │
│           │                                                            │
│           ▼                                                            │
│  _aggregate_chunks → _build_reasoned_context → _build_reasoned_prompt │
│           │                                                            │
│           ▼                                                            │
│        rag.llm.generate(prompt)  →  Final Answer                      │
└──────────────────────────────────────────────────────────────────────┘

from fennec_community.rag.types.multi_hop import MultiHopRAG, HopStrategy

# Wrap any RAG system that exposes .retrieve() and .llm.generate()
mh = MultiHopRAG(
    rag_system=my_rag,
    max_hops=3,
    language="ar",
    confidence_threshold=0.75,
)

# Simple answer
answer = mh.query("What caused the French Revolution and what were its results?")
print(answer)

# Full reasoning trace
result = mh.query(
    "Compare the economic policies of Germany and France after 2008",
    return_intermediate=True,
)
print(result["answer"])
print(result["reasoning_chain"])

# Async usage
answer = await mh.aquery("Why did the Ottoman Empire collapse?")

from fennec_community.rag.types.multi_hop import HopStrategy

class HopStrategy(Enum):
    ENTITY_EXPANSION  = "entity_expansion"
    RELATION_BRIDGING = "relation_bridging"
    CLARIFICATION     = "clarification"
    VERIFICATION      = "verification"

from fennec_community.rag.types.multi_hop import ReasoningState

@dataclass
class ReasoningState:
    known_facts:     List[str] = []     # Discovered relevant sentences
    missing_info:    List[str] = []     # Named knowledge gaps still open
    reasoning_chain: List[str] = []     # Per-hop reasoning step summaries
    confidence:      float     = 0.0    # Current answer completeness estimate [0, 1]
    answer_complete: bool      = False  # True when confidence ≥ threshold and facts > 0

result = mh.query("Why did Rome fall?", return_intermediate=True)

state = result["knowledge_state"]
print("Known facts:", state["known_facts"])
print("Missing info:", state["missing_info"])
print("Confidence:", state["confidence"])

from fennec_community.rag.types.multi_hop import HopResult

@dataclass
class HopResult:
    query:               str               # The search query used for this hop
    chunks:              List[Tuple]        # Retrieved (chunk, score) pairs
    hop_number:          int                # 1-based hop index
    extracted_entities:  List[str]          # Entities extracted from this hop's chunks
    strategy:            HopStrategy        # Strategy used for this hop
    reasoning_step:      str               # Human-readable summary of this hop
    gap_filled:          Optional[str]      # The knowledge gap this hop targeted, if any

result = mh.query("Who invented the telephone and when?", return_intermediate=True)

for hop in result["hops"]:
    print(f"Hop {hop['hop_number']} [{hop['strategy']}]: {hop['query']}")
    print(f"  Chunks: {hop['chunks_found']}  Entities: {hop['entities']}")
    print(f"  Scores: {hop['top_scores']}")
    print(f"  Gap filled: {hop['gap_filled']}")

from fennec_community.rag.types.multi_hop import QueryDecomposer

QueryDecomposer(
    use_stanza: bool = True,
    language:   str  = "ar",
)

use_stanza=True  AND  stanza installed  AND  model loads  →  Stanza NER (best accuracy)
use_stanza=True  AND  stanza installed  AND  model fails  →  Regex fallback (logged warning)
use_stanza=True  AND  stanza NOT installed               →  Regex fallback (logged warning)
use_stanza=False                                          →  Regex fallback (explicit)

pip install stanza
python -c "import stanza; stanza.download('ar')"

# Best accuracy (requires stanza + model)
decomposer = QueryDecomposer(use_stanza=True, language="ar")

# English, Stanza NER
decomposer_en = QueryDecomposer(use_stanza=True, language="en")

# Fast startup, no model required
decomposer_fast = QueryDecomposer(use_stanza=False)

decomposer.decompose(query: str) -> List[str]

# Composite question → two sub-queries
parts = decomposer.decompose("من هو أينشتاين ثم ما أهم اكتشافاته؟")
# ["من هو أينشتاين", "ما أهم اكتشافاته؟"]

# Atomic question → single element
parts = decomposer.decompose("What is the capital of France?")
# ["What is the capital of France?"]

decomposer.extract_entities(text: str) -> List[str]

score = entity_type_weight            # base weight by type (see table below)
      + (len(entity_text) / 10)      # longer = more specific
      + (count_in_text × 0.5)        # frequency bonus
      + 1.0 (if contains English chars)
      + 0.5 (if contains digits)

# With Stanza (Arabic)
entities = decomposer.extract_entities(
    "ولد نابليون بونابرت عام 1769 في جزيرة كورسيكا وأصبح إمبراطوراً لفرنسا."
)
# ["نابليون بونابرت", "كورسيكا", "فرنسا", "1769"]

# Without Stanza (regex fallback)
entities = decomposer.extract_entities(
    'The "Eiffel Tower" was built in 1889 by Gustave Eiffel in Paris.'
)
# ["Eiffel Tower", "Gustave Eiffel", "Paris", "1889"]

decomposer.get_entities_by_type(text: str) -> Dict[str, List[str]]

typed = decomposer.get_entities_by_type(
    "Albert Einstein was born in Ulm, Germany in 1879 and worked at Princeton University."
)
# {
#   "PERS": ["Albert Einstein"],
#   "LOC":  ["Ulm", "Germany"],
#   "ORG":  ["Princeton University"],
#   "DATE": ["1879"],
# }

# Downstream usage example:
persons = typed.get("PERS", [])
dates   = typed.get("DATE", [])

from fennec_community.rag.types.multi_hop import MultiHopRAG

MultiHopRAG(
    rag_system:                  Any,
    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,
)

from fennec_community.rag.types.multi_hop import MultiHopRAG

mh = MultiHopRAG(
    rag_system=my_rag,         # Required: any RAG backend
    max_hops=4,
    min_score_threshold=0.25,
    enable_query_decomposition=True,
    use_stanza_ner=True,
    language="ar",
    confidence_threshold=0.75,
)

mh.query(
    question:            str,
    hops:                Optional[int] = None,
    return_intermediate: bool          = False,
) -> str | Dict[str, Any]

① _analyze_question_requirements()  → question type, need flags, seed entities
② _prepare_queries()                → [sub_query_1, sub_query_2, ...]
③ _perform_reasoning_hops()         → List[HopResult], ReasoningState
④ _aggregate_chunks()               → ranked, deduplicated (chunk, score) list
⑤ _generate_reasoned_answer()       → context + prompt + LLM call
⑥ _update_stats()                   → rolling average hops

answer = mh.query("What were the causes of World War I?")
print(answer)

# Use only 1 hop for a simple factual question
answer = mh.query("What is the capital of France?", hops=1)

result = mh.query(
    "Compare the industrial revolutions of Britain and Germany",
    return_intermediate=True,
)

print("Answer:", result["answer"])
print()
print("Reasoning chain:")
for step in result["reasoning_chain"]:
    print(" •", step)
print()
print("Hops executed:", len(result["hops"]))
for hop in result["hops"]:
    print(f"  Hop {hop['hop_number']} [{hop['strategy']}]: {hop['query']}")
    print(f"    Chunks: {hop['chunks_found']}  Top scores: {hop['top_scores']}")
    print(f"    Entities: {hop['entities']}")
    print(f"    Gap filled: {hop['gap_filled']}")
print()
print("Confidence:", result["knowledge_state"]["confidence"])
print("Known facts:", len(result["knowledge_state"]["known_facts"]))
print("Total chunks:", result["total_chunks"])

async def aquery(
    question:            str,
    max_hops:            Optional[int] = None,
    return_intermediate: bool          = False,
) -> str | Dict[str, Any]

async with MultiHopRAG(rag_system=my_rag) as mh:
    answer = await mh.aquery("Why did the Roman Empire fall?")
    print(answer)

from fastapi import FastAPI
from fennec_community.rag.types.multi_hop import MultiHopRAG

app = FastAPI()
mh = MultiHopRAG(rag_system=my_rag, max_hops=3, language="en")

@app.get("/ask")
async def ask(q: str, hops: int = None, trace: bool = False):
    result = await mh.aquery(q, max_hops=hops, return_intermediate=trace)
    if trace:
        return result          # dict with full trace
    return {"answer": result}  # plain string

@app.get("/ask/arabic")
async def ask_arabic(q: str):
    return {"answer": await mh.aquery(q)}

mh.get_stats() -> Dict[str, Any]

# After running several queries
stats = mh.get_stats()

print(f"Total queries:       {stats['total_queries']}")
print(f"Average hops:        {stats['average_hops']:.2f}")
print(f"Early stops:         {stats['early_stops']} "
      f"({stats['early_stops']/max(stats['total_queries'],1):.0%} of queries)")
print(f"Decomposed queries:  {stats['decomposed_queries']}")
print(f"Entities extracted:  {stats['entities_extracted']}")
print(f"NER engine:          {stats['ner_method']}")
print(f"Confidence threshold:{stats['confidence_threshold']}")

Total queries:        47
Average hops:         1.87
Early stops:          31 (66% of queries)
Decomposed queries:   12
Entities extracted:   284
NER engine:           Stanza
Confidence threshold: 0.75

{
    "answer": str,                      # LLM-generated final answer

    "reasoning_chain": List[str],       # Per-hop summaries, e.g.:
    # ["hop 1: looking for '…'. found 4 facts. confidence: 60%. missing: causal relationship.",
    #  "hop 2: looking for '…'. found 6 facts. confidence: 80%. missing: none."]

    "hops": List[Dict],                 # Per-hop details (see below)

    "knowledge_state": {
        "known_facts": List[str],       # All relevant sentences discovered
        "missing_info": List[str],      # Remaining knowledge gaps (empty = complete)
        "confidence": float,            # Final confidence score [0.0, 1.0]
    },

    "total_chunks": int,                # Total unique chunks collected across all hops

    "stats": Dict,                      # Output of get_stats() at this moment
}

{
    "hop_number":   int,          # 1-based hop index
    "query":        str,          # Query sent to the RAG backend
    "strategy":     str,          # HopStrategy value string
    "reasoning_step": str,        # Human-readable summary of this hop
    "gap_filled":   Optional[str],# Gap this hop targeted, or None
    "chunks_found": int,          # Number of accepted chunks
    "entities":     List[str],    # Entities extracted from top chunks
    "top_scores":   List[float],  # Relevance scores of top 3 chunks
}

query(question)
│
├─ 1. _analyze_question_requirements(question)
│      ├─ _classify_question_type()       → "comparison" | "causal" | "temporal" | "procedural" | "factual"
│      ├─ Flag: needs_comparison          → any of [مقارنة, compare, difference, ...]
│      ├─ Flag: needs_causality           → any of [لماذا, why, cause, ...]
│      ├─ Flag: needs_timeline            → any of [متى, when, date, ...]
│      ├─ Flag: needs_multi_entity        → any of [و, مع, and, both, ...]
│      └─ entities                        → QueryDecomposer.extract_entities(question)
│
├─ 2. _prepare_queries(question)
│      └─ decompose(question)             → ["sub_q1", "sub_q2"] or ["question"]
│
├─ 3. _perform_reasoning_hops(question, sub_queries, num_hops, requirements)
│      └─ For each hop:
│           ├─ rag.retrieve(current_query)
│           ├─ _filter_chunks()           → deduplicate + score filter
│           ├─ _extract_entities_from_chunks()   → top-5 entities from top-3 chunks
│           ├─ _extract_facts_from_chunks()      → relevant sentences
│           ├─ _update_reasoning_state()
│           │    ├─ known_facts += new facts
│           │    ├─ missing_info = _identify_missing_info()
│           │    ├─ confidence  = _calculate_confidence()
│           │    └─ answer_complete = confidence ≥ threshold AND facts > 0
│           ├─ [EARLY STOP if answer_complete]
│           └─ _plan_next_hop()           → next (query, strategy)
│
├─ 4. _aggregate_chunks(hop_results)
│      └─ weight = (1 / hop_number) + (0.1 if CLARIFICATION)
│         Sort by weighted_score → deduplicate by chunk_id
│
├─ 5. _generate_reasoned_answer(question, chunks, hop_results, reasoning_state)
│      ├─ _build_reasoned_context()       → reasoning chain + facts + gaps + sources
│      ├─ _build_reasoned_prompt()        → language-aware strict-grounding prompt
│      └─ rag.llm.generate(prompt, max_tokens=512)
│
└─ 6. _update_stats(num_hops)

Initial strategy:
  needs_comparison  → RELATION_BRIDGING
  needs_causality   → CLARIFICATION
  otherwise         → ENTITY_EXPANSION

Next hop strategy (_plan_next_hop):
  Priority 1: Use next pre-decomposed sub-query → ENTITY_EXPANSION
  Priority 2: missing_info is non-empty         → CLARIFICATION  (gap-filling query)
  Priority 3: comparison + missing entity       → RELATION_BRIDGING (bridge query)
  Priority 4: entities available                → RELATION_BRIDGING (bridge query)
  Priority 5: no path found                     → return None (stop hopping)

base_confidence = min(len(known_facts) / 5.0 , 0.85)
gap_penalty     = len(missing_info) × 0.15
confidence      = max(0.0, base_confidence - gap_penalty)

# Bonus: if no gaps and ≥ 3 facts
if not missing_info and len(known_facts) >= 3:
    confidence = max(confidence, 0.8)

confidence = min(confidence, 1.0)

answer_complete = (confidence >= confidence_threshold AND len(known_facts) > 0)
              OR  (len(missing_info) == 0 AND len(known_facts) > 0)

def retrieve(query: str) -> List[Tuple[chunk, float]] | List[chunk]:
    """
    Retrieve relevant chunks for a query.

    Returns either:
    - List of (chunk, score) tuples   ← preferred
    - List of chunk objects            ← score defaults to 1.0

    Each chunk object must have:
    - chunk.chunk_id : str    (unique identifier — chunks without this are silently skipped)
    - chunk.text     : str    (text content for context building)
    - chunk.doc_id   : str    (document identifier for context headers)
    """

def generate(prompt: str, max_tokens: int = 512, **kwargs) -> str:
    """
    Generate an answer from a prompt.
    Required for answer generation. Without it, query() returns
    "Language model not available".
    """

from dataclasses import dataclass

@dataclass
class Chunk:
    chunk_id: str
    text:     str
    doc_id:   str

class MinimalRAG:
    def __init__(self):
        self.llm = self   # expose .llm

    def retrieve(self, query: str):
        # Return (chunk, score) tuples
        return [(Chunk("c1", "Paris is the capital of France.", "doc_1"), 0.92)]

    def generate(self, prompt: str, max_tokens: int = 512, **kwargs) -> str:
        return "Answer based on context..."

rag = MinimalRAG()
mh = MultiHopRAG(rag_system=rag)

arabic_ratio = arabic_chars / len(question)
language     = "ar" if arabic_ratio > 0.2 else "en"

from fennec_community.rag.types.multi_hop import MultiHopRAG
from fennec_community.rag.core import RAGSystem

my_rag=RAGSystem()
mh = MultiHopRAG(
    rag_system=my_rag,
    max_hops=3,
    language="ar",
    confidence_threshold=0.75,
)

answer = mh.query("ما هي أسباب الثورة الفرنسية وما نتائجها؟")
print(answer)

result = mh.query(
    "Compare the economic impacts of World War I and World War II on Europe",
    return_intermediate=True,
)

print("=" * 60)
print("ANSWER")
print("=" * 60)
print(result["answer"])

print("\n" + "=" * 60)
print("REASONING CHAIN")
print("=" * 60)
for step in result["reasoning_chain"]:
    print("•", step)

print("\n" + "=" * 60)
print("HOP DETAILS")
print("=" * 60)
for hop in result["hops"]:
    print(f"\n[Hop {hop['hop_number']}] Strategy: {hop['strategy']}")
    print(f"  Query:      {hop['query']}")
    print(f"  Chunks:     {hop['chunks_found']}")
    print(f"  Entities:   {hop['entities']}")
    print(f"  Top scores: {[f'{s:.3f}' for s in hop['top_scores']]}")
    print(f"  Gap filled: {hop['gap_filled']}")
    print(f"  Reasoning:  {hop['reasoning_step']}")

print("\n" + "=" * 60)
print("KNOWLEDGE STATE")
print("=" * 60)
ks = result["knowledge_state"]
print(f"Confidence:    {ks['confidence']:.0%}")
print(f"Known facts:   {len(ks['known_facts'])}")
print(f"Missing info:  {ks['missing_info'] or 'None'}")
print(f"Total chunks:  {result['total_chunks']}")

from fastapi import FastAPI, Query
from fennec_community.rag.types.multi_hop import MultiHopRAG

app = FastAPI()

mh = MultiHopRAG(
    rag_system=my_rag,
    max_hops=3,
    language="ar",
    confidence_threshold=0.75,
)

@app.get("/query")
async def answer_question(
    q:     str,
    hops:  int  = Query(default=None, ge=1, le=5),
    trace: bool = False,
):
    result = await mh.aquery(q, max_hops=hops, return_intermediate=trace)
    if trace:
        return result
    return {"answer": result}


@app.get("/stats")
def system_stats():
    return mh.get_stats()

from fennec_community.rag.types.multi_hop import QueryDecomposer

# Arabic, NER-powered
decomposer = QueryDecomposer(use_stanza=True, language="ar")

# Decompose
parts = decomposer.decompose("من هو ابن سينا ثم ما أهم مؤلفاته؟")
print(parts)
# ["من هو ابن سينا", "ما أهم مؤلفاته؟"]

# Extract entities (flat, ranked)
entities = decomposer.extract_entities(
    "أسس جيف بيزوس شركة أمازون عام 1994 في مدينة سياتل الأمريكية."
)
print(entities)
# ["جيف بيزوس", "أمازون", "سياتل", "1994"]

# Extract entities by type (requires Stanza)
typed = decomposer.get_entities_by_type(
    "أسس جيف بيزوس شركة أمازون عام 1994 في مدينة سياتل الأمريكية."
)
print(typed)
# {"PERS": ["جيف بيزوس"], "ORG": ["أمازون"], "LOC": ["سياتل"], "DATE": ["1994"]}

import time

mh = MultiHopRAG(rag_system=my_rag, max_hops=3, confidence_threshold=0.75)

questions = [
    "What is machine learning?",
    "Why did the Ottoman Empire fall?",
    "Compare GPU and CPU for deep learning",
    "When was the Eiffel Tower built and by whom?",
]

start = time.time()
for q in questions:
    answer = mh.query(q)
    print(f"Q: {q[:50]}")
    print(f"A: {answer[:100]}...\n")

elapsed = time.time() - start
stats = mh.get_stats()

print("\n--- System Stats ---")
print(f"Total queries:       {stats['total_queries']}")
print(f"Average hops/query:  {stats['average_hops']:.2f}")
print(f"Early stopping rate: {stats['early_stops']}/{stats['total_queries']} "
      f"({stats['early_stops']/stats['total_queries']:.0%})")
print(f"Total time:          {elapsed:.1f}s")
print(f"Avg time/query:      {elapsed/stats['total_queries']:.1f}s")
print(f"NER method:          {stats['ner_method']}")

from fennec_community.rag.types.multi_hop import MultiHopRAG

mh = MultiHopRAG(rag_system=my_rag, max_hops=5, confidence_threshold=0.75)

# Simple factual — 1 hop is enough
simple = mh.query("What is the capital of Japan?", hops=1)

# Causal — needs 2-3 hops
causal = mh.query("Why did the 2008 financial crisis happen?", hops=3)

# Complex comparison — use maximum hops
comparison = mh.query(
    "Compare the AI research output of the US and China between 2020 and 2024",
    hops=5,
    return_intermediate=True,
)
print("Comparison answer confidence:", comparison["knowledge_state"]["confidence"])

import asyncio
from fennec_community.rag.types.multi_hop import MultiHopRAG

async def batch_query(questions: list[str]) -> list[str]:
    async with MultiHopRAG(rag_system=my_rag, max_hops=3) as mh:
        results = []
        for q in questions:
            answer = await mh.aquery(q, return_intermediate=False)
            results.append(answer)
        return results

answers = asyncio.run(batch_query([
    "What is quantum computing?",
    "Why is the sky blue?",
    "When was the first iPhone released?",
]))

for q, a in zip(questions, answers):
    print(f"Q: {q}\nA: {a}\n")

from fennec_community.llm import MistralInterface
from fennec_community.document_loaders import TextLoader 
from fennec_community.vector_database import FAISSVectorDatabase
from fennec_community.chunks import ArabicTextChunker
from fennec_community.context import ContextManager
from fennec_community.embeddings import OllamaEmbedder
from fennec_community.rag.core import RAGSystem 
from fennec_community.rag.types.multi_hop import MultiHopRAG       

loader_1 = TextLoader("./data_kn/faq.txt").load()
chunker = ArabicTextChunker(chunk_size=100, overlap=20)
embedder = OllamaEmbedder()
vector_db = FAISSVectorDatabase(embedder=embedder)
llm = MistralInterface(api_key=llm_api)
context_manager = ContextManager()
rag_system = RAGSystem(llm=llm, vector_db=vector_db,chunker=chunker, context_manager=context_manager)


rag_system.add_documents(loader_1)

multi_hop = MultiHopRAG(
    rag_system=rag_system,
    max_hops=3,
    confidence_threshold=0.75,
    language="ar",
)

answer = multi_hop.query("ماهي طرق الدفع المتاحة؟")
print(answer)