`federated_rag` — Enterprise API Reference

┌─────────────────────────────────────────────────────────┐
│                      FederatedRAG                        │
│                                                          │
│  ┌─────────┐   ┌──────────────┐   ┌───────────────────┐ │
│  │TTLCache │   │LifecycleHooks│   │AggregationEngine  │ │
│  └─────────┘   └──────────────┘   └───────────────────┘ │
│                                                          │
│  ┌────────────────────────────────────────────────────┐  │
│  │               FederatedSource  (×N)                │  │
│  │  name · rag_system · weight · timeout              │  │
│  │  CircuitBreaker · stats tracking                   │  │
│  └────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘
         │ parallel asyncio.gather / asyncio tasks
         ▼
  [ RAG Backend A ]  [ RAG Backend B ]  [ RAG Backend C ]

from fennec_community.rag.types.federated_rag import FederatedRAG, AggregationMethod

# 1. Instantiate the federation
fed = FederatedRAG(
    aggregation_method=AggregationMethod.WEIGHTED,
    cache_ttl=120.0,
)

# 2. Register your RAG backends
fed.add_source("docs",   docs_rag,   weight=1.5, timeout=8.0)
fed.add_source("wiki",   wiki_rag,   weight=1.0, timeout=5.0)
fed.add_source("arxiv",  arxiv_rag,  weight=0.8, timeout=12.0)

# 3a. Buffered (synchronous)
result = fed.query("What is Retrieval-Augmented Generation?")
print(result["answer"])

# 3b. Buffered (async)
result = await fed.query_async("What is RAG?")

# 3c. Streaming (sync)
for chunk in fed.stream("Explain transformers"):
    print(chunk.text, end="", flush=True)

# 3d. Streaming (async)
async for chunk in fed.astream("Explain transformers"):
    print(chunk.text, end="", flush=True)

class AggregationMethod(str, Enum):
    WEIGHTED = "weighted"
    VOTING   = "voting"
    RANKING  = "ranking"
    SIMPLE   = "simple"
    MERGE    = "merge"

@dataclass
class StreamChunk:
    text:     str
    source:   str
    done:     bool                     = False
    metadata: Optional[Dict[str, Any]] = None
    error:    Optional[str]            = None

async for chunk in fed.astream("What is RAG?"):
    if chunk.error:
        print(f"\n[ERROR from {chunk.source}]: {chunk.error}")
    elif chunk.done:
        print(f"\n[{chunk.source} finished — {chunk.metadata}]")
    else:
        print(chunk.text, end="", flush=True)

from fennec_community.rag.types.federated_rag import FederatedRAG

FederatedRAG(
    aggregation_method: AggregationMethod | str = AggregationMethod.WEIGHTED,
    min_sources:         int   = 1,
    cache_ttl:           float = 60.0,
    cache_max_size:      int   = 256,
    stream_race_timeout: float = 2.0,
    pre_query_hook:  Optional[Callable[[str, Optional[Dict]], None]] = None,
    post_query_hook: Optional[Callable[[Dict[str, Any]], None]]      = None,
)

import logging
from fennec_community.rag.types.federated_rag import FederatedRAG, AggregationMethod

logging.basicConfig(level=logging.INFO)

def before(query, ctx):
    print(f"[PRE] query={query!r}  context={ctx}")

def after(result):
    print(f"[POST] sources={result['sources_used']}  cached={result['cached']}")

fed = FederatedRAG(
    aggregation_method=AggregationMethod.RANKING,
    min_sources=2,
    cache_ttl=300.0,
    cache_max_size=512,
    stream_race_timeout=1.5,
    pre_query_hook=before,
    post_query_hook=after,
)

fed.add_source(
    name:       str,
    rag_system: Any,
    weight:     float = 1.0,
    timeout:    float = 10.0,
    metadata:   Optional[Dict] = None,
) -> FederatedRAG

# Chained registration
fed = (
    FederatedRAG()
    .add_source("primary_docs", docs_rag,   weight=2.0, timeout=5.0)
    .add_source("wiki",         wiki_rag,   weight=1.0, timeout=8.0)
    .add_source("arxiv",        arxiv_rag,  weight=0.5, timeout=15.0,
                metadata={"domain": "academic"})
)

fed.remove_source(name: str) -> None

fed.remove_source("arxiv")
# "arxiv" will no longer be queried

fed.enable_source(name: str, enabled: bool = True) -> None

# Disable a source during maintenance
fed.enable_source("arxiv", enabled=False)

result = fed.query("What is RAG?")  # Only docs and wiki are queried

# Re-enable after maintenance
fed.enable_source("arxiv", enabled=True)

fed.get_stats() -> Dict[str, Dict]

stats = fed.get_stats()
for source_name, s in stats.items():
    print(
        f"{source_name}: "
        f"{s['total_queries']} calls | "
        f"avg {s['avg_latency_ms']} ms | "
        f"error rate {s['error_rate']:.1%} | "
        f"circuit {s['circuit_state']}"
    )

primary_docs: 120 calls | avg 243.7 ms | error rate 1.7% | circuit closed
wiki:          89 calls | avg 510.2 ms | error rate 5.6% | circuit closed
arxiv:         45 calls | avg 1420.0 ms | error rate 20.0% | circuit open

fed.query(
    query:   str,
    context: Optional[Dict] = None,
    top_k:   int = 5,
) -> Dict[str, Any]

result = fed.query(
    "What are the main types of neural networks?",
    context={"language": "en", "user_id": "u-42"},
    top_k=10,
)

print(result["answer"])
print("Sources used:", result["sources_used"])
print("Cached?", result["cached"])

await fed.query_async(
    query:   str,
    context: Optional[Dict] = None,
    top_k:   int = 5,
) -> Dict[str, Any]

# Alias:
await fed.aquery(query, context, top_k)

import asyncio
from fennec_community.rag.types.federated_rag import FederatedRAG, AggregationMethod

async def main():
    fed = FederatedRAG(aggregation_method=AggregationMethod.MERGE)
    fed.add_source("src_a", rag_a)
    fed.add_source("src_b", rag_b)

    result = await fed.query_async("Explain attention mechanisms", top_k=8)
    print(result["answer"])

asyncio.run(main())

fed.stream(
    query:   str,
    context: Optional[Dict] = None,
    top_k:   int = 5,
) -> Generator[StreamChunk, None, None]

print("Answer: ", end="")
for chunk in fed.stream("Describe the transformer architecture", top_k=5):
    if chunk.error:
        print(f"\n[ERROR: {chunk.error}]")
    elif not chunk.done:
        print(chunk.text, end="", flush=True)
print()  # newline at end

async def astream(
    query:   str,
    context: Optional[Dict] = None,
    top_k:   int = 5,
) -> AsyncGenerator[StreamChunk, None]

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/stream")
async def stream_answer(q: str):
    async def generate():
        async for chunk in fed.astream(q):
            if not chunk.done and not chunk.error:
                yield chunk.text
    return StreamingResponse(generate(), media_type="text/plain")

async for chunk in fed.astream("What is BERT?"):
    if chunk.done:
        meta = chunk.metadata or {}
        print(f"\n✓ {chunk.source} — {meta.get('latency_ms')} ms")
    elif not chunk.error:
        print(chunk.text, end="", flush=True)

class MyRAGBackend:
    def generate(self, query: str) -> str | dict:
        ...

class FullRAGBackend:
    def retrieve(self, query: str, top_k: int = 5) -> List[Tuple[Document, float]]:
        """Returns list of (document, score) tuples."""
        ...

    async def aretrieve(self, query: str, top_k: int = 5) -> List[Tuple[Document, float]]:
        """Async variant of retrieve."""
        ...

    def generate(self, query: str, language: str = "en") -> str | dict:
        """Generate an answer. Return str or dict with at least {"answer": str}."""
        ...

    async def agenerate(self, query: str, language: str = "en") -> str | dict:
        """Async variant of generate."""
        ...

    def stream(self, query: str, top_k: int = 5) -> Iterable[str | dict]:
        """Yield text fragments or {"text": str, "score": float} dicts."""
        ...

    async def astream(self, query: str, top_k: int = 5) -> AsyncIterable[str | dict]:
        """Async variant of stream."""
        ...

CLOSED ──(≥ threshold failures)──▶ OPEN ──(recovery_timeout elapsed)──▶ HALF_OPEN
   ▲                                                                          │
   └────────────────(success in HALF_OPEN)──────────────────────────────────┘
                              │
                    (failure in HALF_OPEN)
                              ▼
                            OPEN

for name, s in fed.get_stats().items():
    if s["circuit_state"] == "open":
        print(f"⚠️  {name} is circuit-open — check that backend!")

fed._cache.invalidate()   # clears all entries

{
    "answer":       "<human-readable message>",
    "error":        "<error_code>",
    "sources_used": [],
    "num_sources":  0,
}

import logging
from fennec_community.rag.types.federated_rag import FederatedRAG, AggregationMethod

logging.basicConfig(level=logging.INFO)

fed = FederatedRAG(aggregation_method=AggregationMethod.RANKING)
fed.add_source("docs",  docs_rag,  weight=2.0, timeout=5.0)
fed.add_source("wiki",  wiki_rag,  weight=1.0, timeout=8.0)

result = fed.query("What is retrieval-augmented generation?", top_k=5)

if "error" in result:
    print("Federation error:", result["error"])
else:
    print("Answer:", result["answer"])
    print("Primary source:", result.get("primary_source"))
    print("Ranked sources:", [r["source"] for r in result.get("ranked_results", [])])

print("\nSource Stats:")
for src, s in fed.get_stats().items():
    print(f"  {src}: {s['avg_latency_ms']} ms avg, {s['error_rate']:.0%} error rate")

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fennec_community.rag.types.federated_rag import FederatedRAG, AggregationMethod

app = FastAPI()

fed = FederatedRAG(aggregation_method=AggregationMethod.WEIGHTED)
fed.add_source("primary", primary_rag, weight=2.0)
fed.add_source("fallback", fallback_rag, weight=0.5)

@app.get("/ask")
async def ask(q: str, lang: str = "en"):
    async def token_stream():
        ctx = {"language": lang}
        async for chunk in fed.astream(q, context=ctx, top_k=5):
            if chunk.done:
                yield f"\n[done: {chunk.source}]"
            elif chunk.error:
                yield f"\n[error from {chunk.source}: {chunk.error}]"
            else:
                yield chunk.text

    return StreamingResponse(token_stream(), media_type="text/plain")

import asyncio
from fennec_community.rag.types.federated_rag import FederatedRAG, AggregationMethod

async def research(question: str):
    fed = FederatedRAG(aggregation_method=AggregationMethod.MERGE)
    fed.add_source("papers",    papers_rag,    weight=3.0)
    fed.add_source("textbooks", textbooks_rag, weight=2.0)
    fed.add_source("web",       web_rag,       weight=1.0)

    result = await fed.query_async(question, top_k=10)
    # result["answer"] will contain [papers] ...\n\n[textbooks] ...\n\n[web] ...
    return result["answer"]

answer = asyncio.run(research("Explain the attention mechanism in transformers"))
print(answer)

from fennec_community.rag.types.federated_rag import FederatedRAG

fed = FederatedRAG()
fed.add_source("src_a", rag_a, weight=1.0)
fed.add_source("src_b", rag_b, weight=1.5)
fed.add_source("src_c", rag_c, weight=0.8)

# Disable a source for maintenance
fed.enable_source("src_c", enabled=False)
result = fed.query("Who invented the internet?")   # Only src_a and src_b queried

# Re-enable after maintenance
fed.enable_source("src_c", enabled=True)

# Add a new source at runtime
fed.add_source("src_d", rag_d, weight=2.0)

# Remove a decommissioned source
fed.remove_source("src_a")

print(f"Active federation: {fed}")  # FederatedRAG(sources=['src_b', 'src_c', 'src_d'], ...)
print(f"Total sources: {len(fed)}")

import time
from fennec_community.rag.types.federated_rag import FederatedRAG

request_log = []

def pre_hook(query: str, context):
    request_log.append({"query": query, "ts": time.time(), "status": "started"})
    print(f"[METRIC] query started: {query[:50]!r}")

def post_hook(result: dict):
    request_log.append({
        "sources": result["sources_used"],
        "cached": result["cached"],
        "ts": time.time(),
        "status": "done",
    })
    print(f"[METRIC] query done | sources={result['sources_used']} cached={result['cached']}")

fed = FederatedRAG(
    pre_query_hook=pre_hook,
    post_query_hook=post_hook,
)
fed.add_source("main", my_rag)
fed.query("What is machine learning?")

import asyncio
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.federated_rag import FederatedRAG , AggregationMethod   

chunker = ArabicTextChunker(chunk_size=100, overlap=20)
embedder = OllamaEmbedder()
vector_db = FAISSVectorDatabase(embedder=embedder)
llm = MistralInterface(api_key=llm_api)
context_manager = ContextManager()

vdb1     = FAISSVectorDatabase(embedder=embedder)
rag_tech = RAGSystem(vector_db=vdb1, llm=llm, chunker=chunker, context_manager=context_manager)
rag_tech.add_text(text= "بايثون لغة برمجة عالية المستوى متعددة الاستخدامات، صُمِّمت لتكون سهلة القراءة والكتابة. "
            "تدعم البرمجة الإجرائية والكائنية والوظيفية. تُستخدم على نطاق واسع في علم البيانات، "
            "والذكاء الاصطناعي، وتطوير الويب، وأتمتة المهام. تتميز بمكتبة قياسية ضخمة ومجتمع "
            "نشط يوفر آلاف الحزم الجاهزة عبر PyPI.",doc_id="tech_python", metadata={"source": "tech_docs", "category": "programming"})
        
 
vdb2     = FAISSVectorDatabase(embedder=embedder)
rag_wiki = RAGSystem(vector_db=vdb2, llm=llm, chunker=chunker, context_manager=context_manager)
rag_wiki.add_text(text="بدأت بحوث الذكاء الاصطناعي رسمياً في مؤتمر دارتموث عام 1956. "
            "مرّ المجال بفترتين من شتاء الذكاء الاصطناعي بسبب توقف التمويل. "
            "انتعش مجدداً مطلع الألفية الثالثة بفضل البيانات الضخمة والقدرة الحوسبية， "
            "ثم شهد طفرة مع ظهور نماذج اللغة الكبيرة.", doc_id="wiki_ai_history", metadata={"source": "wikipedia", "category": "history"})
 
vdb3         = FAISSVectorDatabase(embedder=embedder)
rag_internal = RAGSystem(vector_db=vdb3, llm=llm, chunker=chunker, context_manager=context_manager)
rag_internal.add_text(text="شركتنا TechAI Solutions تأسست عام 2020 وتتخصص في حلول الذكاء الاصطناعي. "
            "نخدم أكثر من 150 عميلاً في قطاعات المال والصحة والتجزئة. "
            "يضم فريقنا 80 موظفاً منهم 45 مهندساً وباحثاً.", doc_id="internal_company_overview", metadata={"source": "internal", "category": "company_info"})
 

fed = FederatedRAG(
        aggregation_method  = AggregationMethod.WEIGHTED,
        min_sources         = 1,
        cache_ttl           = 120.0,
        stream_race_timeout = 3.0,
    )
fed.add_source("tech_docs", rag_tech,     weight=2.0, timeout=15.0) \
       .add_source("wikipedia", rag_wiki,     weight=1.0, timeout=10.0) \
       .add_source("internal",  rag_internal, weight=1.5, timeout=8.0)
 
print(f"عدد المصادر: {len(fed)}")
print(repr(fed))

result = fed.query("ما هو  لغه بايثون", top_k=5)
print(result)