`core` in Rag Module — Public API Reference

Table of Contents

1. Module Overview
2. RAGConfig
3. RAGSystem
   • Constructor
   • Document Ingestion
   • Retrieval
   • Generation
   • Prompt Management
   • Persistence
   • Cache Management
   • Statistics & Diagnostics
   • Async API
   • Context Manager
4. BaseRAGSystem
   • Abstract Methods
   • Concrete Methods
   • Async Methods
5. PromptRouter
6. Reranker
7. RerankerConfig
8. QueryExpander
9. RetrievalCache
10. Logging Utilities
11. Exception Hierarchy
12. Data Classes
13. Module-level Helpers
14. Enumerations
15. Quick-Start Example

1. Module Overview

The core package is the heart of the Fennec Community library. It provides a complete, production-ready pipeline for Retrieval-Augmented Generation (RAG): indexing documents into a vector database, retrieving the most relevant passages for any given query, and generating grounded natural-language answers through an LLM.

Key capabilities:

• Plug-in architecture — bring your own vector DB, LLM, chunker, and context manager.
• Smart query expansion (LLM + curated synonym table, bilingual Arabic/English).
• Intelligent prompt routing — selects the optimal prompt template based on detected language and question type.
• Multi-mode reranking — heuristic, LLM-assisted, or hybrid.
• TTL-aware LRU retrieval cache for repeated queries.
• Full async support with streaming generation.
• Structured exception hierarchy for precise error handling.
• Centralized, structured logging with optional JSON output.

Publicly exported symbols:

from fennec_community.rag.core import (
    RAGConfig,
    BaseRAGSystem,
    RAGSystem,
    RerankerConfig,
    RerankMode,
    Reranker,
    PromptRouter,
    QuestionType,
)

2. RAGConfig

from fennec_community.rag.core import RAGConfig

RAGConfig is a dataclass that holds all tunable parameters for a RAGSystem instance. Every field has a sensible default, so you only need to override what you want to change.

Constructor

RAGConfig(
    chunk_size: int = 512,
    overlap: int = 128,
    top_k: int = 5,
    min_score: float = 0.0,
    enable_reranking: bool = False,
    rerank_mode: str = "heuristic",
    rerank_top_k: Optional[int] = None,
    rerank_original_weight: float = 0.40,
    rerank_llm_weight: float = 0.40,
    rerank_diversity_weight: float = 0.10,
    rerank_length_weight: float = 0.10,
    enable_prompt_routing: bool = True,
    prompt_language: Optional[str] = None,
    include_few_shot: bool = True,
    context_max_length: int = 4000,
    enable_retrieval_cache: bool = True,
    cache_maxsize: int = 256,
    cache_ttl: float = 300.0,
    ingest_batch_size: int = 64,
)

Constraint: The four rerank weights (rerank_original_weight + rerank_llm_weight + rerank_diversity_weight + rerank_length_weight) must sum to exactly 1.0.

validate()

config.validate() -> None

Purpose: Validates all configuration fields and raises ValueError for any illegal combination (e.g., non-positive chunk_size, overlap ≥ chunk_size, invalid rerank_mode, weights that do not sum to 1.0).

Called automatically by BaseRAGSystem.__init__, so manual calls are rarely needed.

Parameters: None.

Returns: None

Raises: ValueError with a descriptive message for any invalid field.

3. RAGSystem

from fennec_community.rag.core import RAGSystem

RAGSystem is the main public class you interact with. It wires together a vector database, an LLM, a chunker, and a context manager into a complete RAG pipeline.

3.1 Constructor

RAGSystem(
    vector_db: Any,
    llm: Any,
    chunker: Any,
    context_manager: Any,
    config: Optional[RAGConfig] = None,
    prompt: Optional[PromptInput] = None,
    enable_query_expansion: bool = True,
    query_expansion_variants: int = 3,
)

Purpose: Instantiate a fully wired RAG system. All four core components are required.

Returns: RAGSystem instance.

Raises: RAGInitializationError if any of the four required components is None.

Example:

from fennec_community.rag.core import RAGSystem, RAGConfig

config = RAGConfig(top_k=8, enable_reranking=True, rerank_mode="hybrid")
rag = RAGSystem(
    vector_db=my_vector_db,
    llm=my_llm,
    chunker=my_chunker,
    context_manager=my_ctx_mgr,
    config=config,
)

3.2 Document Ingestion

add_text()

rag.add_text(
    text: str,
    doc_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> int

Purpose: Index a single plain-text string directly — the simplest ingestion path for short texts (employee records, FAQ entries, notes, etc.).

Returns: int — number of chunks created from the text.

Example:

n = rag.add_text(
    "Name: Ahmed Al-Farsi, Phone: +966 50 123 4567",
    doc_id="emp_001",
    metadata={"department": "Engineering"},
)
print(f"Indexed in {n} chunk(s)")

add_texts()

rag.add_texts(
    texts: Dict[str, str],
    metadata: Optional[Dict[str, Dict[str, Any]]] = None,
) -> Dict[str, int]

Purpose: Batch-index multiple documents from a {doc_id: text} dictionary. Preferred over calling add_text() in a loop because it performs a single vector DB write.

Returns: Dict[str, int] — mapping of document ID → number of chunks created (0 for failed or empty documents).

Example:

chunks = rag.add_texts(
    {
        "doc_ar_1": "اسم: محمد، الهاتف: 0501234567",
        "doc_ar_2": "اسم: سارة، الهاتف: 0557654321",
    },
    metadata={
        "doc_ar_1": {"team": "Sales"},
        "doc_ar_2": {"team": "HR"},
    },
)
# → {"doc_ar_1": 1, "doc_ar_2": 1}

add_documents()

rag.add_documents(docs: List[LoadedDocument]) -> Dict[str, int]

Purpose: Low-level ingestion entry point used internally by add_text and add_texts. Accepts a list of LoadedDocument objects, chunks each document, and writes all chunks to the vector DB in a single batch.

Returns: Dict[str, int] — {doc_id: num_chunks}. Documents that fail chunking or are empty receive a count of 0.

Notes:

• The retrieval cache is automatically invalidated after successful ingestion.
• Documents with empty or whitespace-only page_content are skipped with a warning.

3.3 Retrieval

retrieve()

rag.retrieve(
    query: str,
    top_k: Optional[int] = None,
) -> List[Tuple[Any, float]]

Purpose: Retrieve the most relevant document chunks for a query. Internally applies smart query expansion, merges results across all variants, and optionally reranks.

Pipeline:

1. Check the retrieval cache — return immediately on a hit.
2. Expand the query into semantically equivalent variants.
3. Search the vector DB for each variant.
4. Merge results, keeping the highest score per unique chunk.
5. Optionally rerank the merged results.
6. Store the result in the cache.

Returns: List[Tuple[Any, float]] — list of (chunk, score) pairs sorted by score descending. Empty list if the query is blank or no results are found.

Raises: RAGRetrievalError on unexpected search failures.

Example:

results = rag.retrieve("ما رقم هاتف أحمد؟", top_k=3)
for chunk, score in results:
    print(f"[{score:.2f}] {chunk.text}")

3.4 Generation

generate()

rag.generate(
    query: str,
    include_sources: bool = False,
    language: Optional[str] = None,
    **llm_kwargs: Any,
) -> str

Purpose: End-to-end RAG inference: retrieves relevant chunks, builds a context block, selects an appropriate prompt, and calls the LLM to produce a grounded answer.

Returns: str — the LLM's answer. Never raises — errors are returned as human-readable strings (e.g., "❌ Search error occurred.").

Behaviour on edge cases:

• Empty query → "⚠️ Please enter a valid question."
• No relevant chunks found → "No relevant information found."
• Retrieval error → "❌ Search error occurred."

Example:

answer = rag.generate(
    "What is the phone number of Ahmed?",
    include_sources=True,
    temperature=0.2,
)
print(answer)

ask()

rag.ask(
    question: str,
    include_sources: bool = False,
    language: Optional[str] = None,
    **llm_kwargs: Any,
) -> str

Purpose: Friendly alias for generate() — identical behaviour with a more conversational method name.

Returns: str — the generated answer.

Example:

answer = rag.ask("ما اسم المدير العام؟")

3.5 Prompt Management

set_prompt()

rag.set_prompt(prompt: Optional[PromptInput]) -> None

Purpose: Replace or remove the custom prompt template at runtime, without recreating the entire RAGSystem. Passing None restores the default PromptRouter behaviour (if prompt routing is enabled in config).

Returns: None

Example:

# Use a custom format string
rag.set_prompt("Context:\n{context}\n\nQuestion: {question}\nAnswer:")

# Restore PromptRouter
rag.set_prompt(None)

3.6 Persistence

save()

rag.save(path: str) -> None

Purpose: Persist the vector database index and system statistics to disk so the system can be restored later without re-indexing all documents.

Returns: None

Raises: VectorDBPersistenceError if the save operation fails.

Directory layout created:

<path>/
├── vector_db/    ← serialized vector database
└── stats.json    ← query and document statistics

load() (class method)

RAGSystem.load(
    path: str,
    vector_db: Any,
    llm: Any,
    chunker: Any,
    context_manager: Any,
    config: Optional[RAGConfig] = None,
    enable_query_expansion: bool = True,
) -> RAGSystem

Purpose: Reconstruct a previously saved RAGSystem from disk, restoring the vector index and statistics. No re-indexing is required.

Returns: RAGSystem — fully initialized and ready to serve queries.

Raises: FileNotFoundError if the path does not exist. VectorDBPersistenceError if the index cannot be loaded.

Example:

rag = RAGSystem.load(
    path="./saved_rag",
    vector_db=fresh_vector_db,
    llm=my_llm,
    chunker=my_chunker,
    context_manager=my_ctx_mgr,
)
answer = rag.ask("What documents are in this system?")

remove_document()

rag.remove_document(doc_id: str) -> int

Purpose: Remove all chunks belonging to a specific document from the vector database, and update internal statistics accordingly.

Returns: int — number of chunks removed. Returns 0 if the document was not found or if removal failed.

3.7 Cache Management

invalidate_cache()

rag.invalidate_cache() -> None

Purpose: Manually flush the entire retrieval cache. Useful when documents are updated or removed externally (e.g., via a direct vector DB operation that bypasses RAGSystem).

Parameters: None.

Returns: None

get_cache_stats()

rag.get_cache_stats() -> Dict[str, Any]

Purpose: Return runtime statistics for the retrieval cache to help diagnose performance and tune cache parameters.

Parameters: None.

Returns: Dict[str, Any] with the following keys:

Returns an empty dict if the cache is disabled (enable_retrieval_cache=False).

3.8 Statistics & Diagnostics

get_stats()

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

Purpose: Return a snapshot of all system-wide statistics including document counts, query counts, component types, and cache metrics. Useful for monitoring and dashboards.

Parameters: None.

Returns: Dict[str, Any] containing:

validate_connection()

rag.validate_connection(test_query: str) -> Dict

Purpose: Health-check the entire pipeline by running a test retrieval and generation. Returns a structured result indicating which components are working.

Returns: Dict with the following structure:

# Success
{
    "success": True,
    "reason": "✅ System working successfully",
    "components": {"retrieval": True, "generation": True}
}

# Partial (retrieval works, generation unavailable)
{
    "success": True,
    "reason": "✅ Retrieval works, generation unavailable",
    "components": {"retrieval": True, "generation": False, "generation_error": "..."}
}

# Failure
{
    "success": False,
    "reason": "❌ Error: ...",
    "components": None
}

# No documents
{
    "success": False,
    "reason": "⚠️ No documents in the system",
    "components": None
}

reset_stats()

rag.reset_stats() -> None

Purpose: Reset query-level counters (total_queries, successful_queries, failed_queries) back to zero. Document and chunk counts are preserved.

Parameters: None.

Returns: None

3.9 Async API

All async methods mirror their synchronous counterparts but are safe for use in asyncio event loops.

aadd_documents()

await rag.aadd_documents(docs: Dict[str, str]) -> Dict[str, int]

Purpose: Async document ingestion. Chunks all documents concurrently using asyncio.gather, then writes to the vector DB (using the native async method aadd if available, otherwise asyncio.to_thread).

Returns: Dict[str, int] — {doc_id: num_chunks}.

aretrieve()

await rag.aretrieve(
    query: str,
    top_k: Optional[int] = None,
) -> List[Tuple[Any, float]]

Purpose: Async retrieval with concurrent query expansion. All variant searches are fired simultaneously via asyncio.gather.

Returns: List[Tuple[Any, float]] — scored chunks, sorted descending.

agenerate()

await rag.agenerate(
    query: str,
    include_sources: bool = False,
    language: Optional[str] = None,
    **llm_kwargs: Any,
) -> str

Purpose: Async end-to-end generation. Uses the async generate method of the LLM (generate_async) if available.

Returns: str — the generated answer.

astream()

async for token in rag.astream(
    query: str,
    language: Optional[str] = None,
    **llm_kwargs: Any,
):
    print(token, end="", flush=True)

Purpose: Async streaming generation — yields tokens as they are produced by the LLM. Enables real-time display of long answers.

Yields: str — individual tokens or word chunks.

3.10 Context Manager

RAGSystem supports both synchronous and asynchronous context managers:

# Synchronous
with RAGSystem(vector_db, llm, chunker, ctx_mgr) as rag:
    rag.add_texts({"doc1": "some content"})
    print(rag.ask("What is in doc1?"))
# cleanup() called automatically on exit

# Asynchronous
async with RAGSystem(vector_db, llm, chunker, ctx_mgr) as rag:
    await rag.aadd_documents({"doc1": "some content"})
    print(await rag.agenerate("What is in doc1?"))

cleanup()

rag.cleanup() -> None

Purpose: Release resources held by all four components (calls cleanup() on each if the method exists). Called automatically when exiting a context manager.

Parameters: None.

Returns: None

4. BaseRAGSystem

from fennec_community.rag.core import BaseRAGSystem

BaseRAGSystem is the abstract base class that all RAG system implementations must extend. It defines the required interface and provides default implementations of utility methods.

4.1 Abstract Methods

Subclasses must implement these:

4.2 Concrete Methods

remove_document()

system.remove_document(doc_id: str) -> int

Purpose: Remove a document by ID. The base implementation raises NotImplementedError; override in subclasses.

Returns: int — number of deleted chunks.

save()

system.save(path: str) -> None

Purpose: Persist the system to disk. Raises NotImplementedError in the base class.

load() (class method)

BaseRAGSystem.load(path: str, **kwargs) -> BaseRAGSystem

Purpose: Load a previously saved system. Raises NotImplementedError in the base class.

get_stats()

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

Purpose: Return a snapshot of system statistics.

Returns: Dict with keys: total_queries, successful_queries, failed_queries, total_documents, total_chunks.

reset_stats()

system.reset_stats() -> None

Purpose: Zero out query counters while preserving document and chunk counts.

validate_connection()

system.validate_connection(test_query: str) -> Dict

Purpose: Run a pipeline health-check. See §3.8 for the full return schema.

cleanup()

system.cleanup() -> None

Purpose: Release resources. Override in subclasses as needed.

4.3 Async Methods

5. PromptRouter

from fennec_community.rag.core import PromptRouter, QuestionType

PromptRouter analyses each user query, detects its language and semantic type, and selects the best-fit prompt template. It supports bilingual operation (Arabic and English) with optional few-shot examples.

Constructor

PromptRouter(include_few_shot: bool = True)

detect_language()

router.detect_language(text: str) -> str

Purpose: Determine whether a text is predominantly Arabic or English using Unicode character ratio analysis.

Returns: "ar" if the ratio of Arabic Unicode characters exceeds 25%, otherwise "en". Returns "ar" for empty strings.

Example:

router = PromptRouter()
print(router.detect_language("ما هي عاصمة مصر؟"))  # → "ar"
print(router.detect_language("What is the capital of Egypt?"))  # → "en"

detect_question_type()

router.detect_question_type(
    query: str,
    language: Optional[str] = None,
) -> QuestionType

Purpose: Classify a query into one of eight semantic question types using regex pattern matching on language-specific keyword sets.

Returns: QuestionType enum value. Falls back to QuestionType.GENERAL if no pattern matches.

Supported types:

build()

router.build(
    query: str,
    context: str,
    language: Optional[str] = None,
    question_type: Optional[QuestionType] = None,
) -> str

Purpose: Assemble a complete, ready-to-send LLM prompt by combining the appropriate instruction, optional few-shot example, retrieved context, and the user's query.

Returns: str — a fully formatted prompt string ready to be passed to llm.generate().

Example:

router = PromptRouter(include_few_shot=True)
prompt = router.build(
    query="How does RAG work?",
    context="RAG combines retrieval with generation...",
)
answer = llm.generate(prompt)

6. Reranker

from fennec_community.rag.core import Reranker

Reranker re-scores and re-orders an initial list of retrieved (chunk, score) pairs to improve answer quality. It supports three modes and includes near-duplicate filtering.

Constructor

Reranker(
    config: Optional[RerankerConfig] = None,
    llm: Any = None,
)

rerank()

reranker.rerank(
    query: str,
    chunks: List[Tuple[Any, float]],
) -> List[Tuple[Any, float]]

Purpose: The main entry point. Applies near-duplicate filtering (if enabled), scores each chunk according to the configured mode, and returns the top results sorted by composite score.

Scoring modes:

• HEURISTIC: Combines original vector score with a length optimality bonus and a diversity/term-overlap bonus. Fast, no LLM calls.
• LLM: Calls the LLM in parallel to rate each chunk's relevance on a 0–10 scale, then combines with the original score and length bonus.
• HYBRID: Combines all factors — original score, LLM relevance, length bonus, and diversity bonus.

Returns: List[Tuple[Any, float]] — re-ranked (chunk, composite_score) pairs, descending. Length is min(len(input), config.top_k).

Example:

from fennec_commuinty.rag.core import Reranker, RerankerConfig, RerankMode

rc = RerankerConfig(mode=RerankMode.HYBRID, top_k=5)
reranker = Reranker(config=rc, llm=my_llm)

raw_results = vector_db.search("phone number of Ahmed", top_k=20)
reranked = reranker.rerank("phone number of Ahmed", raw_results)

clear_cache()

reranker.clear_cache() -> None

Purpose: Clear the internal LRU cache that stores LLM relevance scores for (query, text) pairs. Useful to free memory or force re-evaluation.

Parameters: None.

Returns: None

7. RerankerConfig

from fennec_community.rag.core import RerankerConfig, RerankMode

Dataclass holding all parameters for the Reranker.

Constructor

RerankerConfig(
    mode: RerankMode = RerankMode.HEURISTIC,
    top_k: Optional[int] = None,
    original_score_weight: float = 0.40,
    llm_score_weight: float = 0.40,
    diversity_weight: float = 0.10,
    length_weight: float = 0.10,
    llm_eval_language: str = "ar",
    ideal_min_words: int = 30,
    ideal_max_words: int = 300,
    dedup_enabled: bool = True,
    dedup_threshold: float = 0.85,
)

Constraint: original_score_weight + llm_score_weight + diversity_weight + length_weight must equal 1.0.

effective_weights()

config.effective_weights(mode: RerankMode) -> Dict[str, float]

Purpose: Return mode-adjusted weight dictionary. In HEURISTIC mode, llm_score_weight is folded into original_score_weight so no LLM weight is wastefully applied.

Returns: Dict[str, float] — keys: orig, llm, len, div.

8. QueryExpander

from fennec_community.rag.core import QueryExpander

QueryExpander generates semantically equivalent rephrasings of a query to improve recall. It uses an LLM as the primary strategy and falls back to a curated bilingual synonym table when the LLM is unavailable.

Constructor

QueryExpander(
    llm: Optional[Any] = None,
    max_variants: int = 3,
    use_llm: bool = True,
    fallback_synonyms: bool = True,
)

expand()

expander.expand(query: str) -> List[str]

Purpose: Return the original query plus up to max_variants semantically equivalent alternatives. The first element is always the original query.

Strategy:

1. If use_llm=True, call the LLM with a language-appropriate expansion prompt.
2. On LLM failure, fall back to the synonym table.
3. If both are disabled, return [original_query].

Returns: List[str] — [original, variant_1, variant_2, ...]. Always contains at least one element.

Example:

expander = QueryExpander(llm=my_llm, max_variants=3)
queries = expander.expand("رقم التليفون")
# → ["رقم التليفون", "رقم الهاتف", "رقم الجوال", "رقم الموبايل"]

merge_retrieval_results() (module-level function)

from fennec_community.rag.core import merge_retrieval_results

merge_retrieval_results(
    results_per_query: List[List[Tuple[Any, float]]],
    top_k: int,
) -> List[Tuple[Any, float]]

Purpose: Combine retrieval results from multiple query variants into a single deduplicated list. When the same chunk appears across multiple variant results, the highest score is kept.

Returns: List[Tuple[Any, float]] — merged, deduplicated, score-sorted list of up to top_k results.

Example:

results_a = vector_db.search("phone number", top_k=5)
results_b = vector_db.search("contact number", top_k=5)
merged = merge_retrieval_results([results_a, results_b], top_k=5)

9. RetrievalCache

from fennec_community.rag.core import RetrievalCache

RetrievalCache is a thread-safe, TTL-aware LRU cache for storing retrieval results. It is managed internally by RAGSystem but can also be used standalone.

Constructor

RetrievalCache(
    maxsize: int = 256,
    ttl: Optional[float] = 300.0,
)

get()

cache.get(query: str, top_k: int) -> Optional[List[Tuple[Any, float]]]

Purpose: Retrieve a cached result for a query and top_k combination. Returns None on a cache miss or if the entry has expired. Updates the LRU order on a hit.

Returns: Cached List[Tuple[Any, float]] on a hit, None on a miss or expiry.

set()

cache.set(
    query: str,
    top_k: int,
    results: List[Tuple[Any, float]],
) -> None

Purpose: Store a retrieval result in the cache. If the cache is at capacity, the oldest entry is evicted.

Returns: None

invalidate()

cache.invalidate(query: str, top_k: int) -> bool

Purpose: Remove a specific entry from the cache by query and top_k.

Returns: True if the entry existed and was removed; False otherwise.

clear()

cache.clear() -> None

Purpose: Evict all entries and reset hit/miss counters.

Returns: None

stats()

cache.stats() -> Dict[str, Any]

Purpose: Return current cache metrics.

Returns: Dict with keys size, maxsize, ttl, hits, misses, hit_rate.

10. Logging Utilities

from fennec_community.rag.core.logger import get_logger, setup_logging, FennecLogger

setup_logging()

setup_logging(
    level: str = "INFO",
    log_file: Optional[str] = None,
    json_format: bool = False,
    use_colour: bool = True,
    max_bytes: int = 10_485_760,   # 10 MB
    backup_count: int = 3,
    propagate: bool = False,
) -> None

Purpose: Configure the library-wide fennec logger once at application startup. Sets up a colour-coded console handler and optionally a rotating file handler. Calling this more than once is safe — it clears and rebuilds handlers each time.

Returns: None

Example:

from fennec_community.rag.core.logger import setup_logging

setup_logging(
    level="DEBUG",
    log_file="logs/fennec.log",
    json_format=True,
)

get_logger()

get_logger(name: str) -> logging.Logger

Purpose: Return a child logger registered under the fennec hierarchy. Performs lazy bootstrap if setup_logging() has not been called yet. All library modules call this internally.

Returns: logging.Logger bound under fennec.<name> (or fennec if name already starts with fennec).

Example:

from fennec_community.rag.core.logger import get_logger

logger = get_logger(__name__)
logger.info("RAG system initialised with %d documents", n)
logger.debug("Query variants: %s", variants)

FennecLogger (class)

A thin static wrapper for users who prefer a class-based logging API:

FennecLogger.setup(level="DEBUG", log_file="fennec.log")
log = FennecLogger.get("mymodule")
log.info("processing started")

11. Exception Hierarchy

from fennec_community.rag.core.exceptions import *

All exceptions inherit from FennecError, enabling callers to catch either the specific subclass or the general base.

FennecError
├── ConfigurationError
│   ├── MissingAPIKeyError(provider)
│   └── InvalidConfigValueError(field, value, reason)
├── DocumentError
│   ├── EmptyDocumentError(doc_id)
│   ├── DocumentNotFoundError(doc_id)
│   └── ChunkingError(doc_id, reason)
├── VectorDatabaseError
│   ├── VectorDBConnectionError(db_type, reason)
│   ├── VectorDBWriteError
│   ├── VectorDBSearchError
│   └── VectorDBPersistenceError(operation, path, reason)
├── EmbeddingError
│   ├── EmbeddingModelNotFoundError(model_name)
│   ├── EmbeddingDimensionMismatchError(expected, got)
│   └── EmbeddingAPIError(provider, reason)
├── LLMError
│   ├── LLMConnectionError(provider, reason)
│   ├── LLMGenerationError(provider, reason)
│   ├── LLMTimeoutError(provider, timeout_seconds)
│   └── LLMRateLimitError(provider)
├── RAGError
│   ├── RAGInitializationError(reason)
│   ├── RAGRetrievalError(query, reason)
│   ├── RAGGenerationError(reason)
│   ├── NoRelevantDocumentsError(query)
│   └── QueryExpansionError(reason)
├── PromptError
│   └── PromptFormattingError(template_name, missing_vars)
├── CacheError
│   ├── CacheReadError
│   └── CacheWriteError
└── LoaderError
    ├── UnsupportedFileTypeError(extension)
    └── FileReadError(filepath, reason)

Base class attributes:

Example error handling:

from fennec_community.rag.core.exceptions import FennecError, RAGRetrievalError, LLMRateLimitError

try:
    answer = rag.generate("What is X?")
except LLMRateLimitError as e:
    print(f"Rate limited by {e.provider} — retry later")
except RAGRetrievalError as e:
    print(f"Retrieval failed: {e.reason}")
except FennecError as e:
    print(f"Unexpected Fennec error [{e.code}]: {e.message}")

12. Data Classes

LoadedDocument

from fennec_community.rag.core import LoadedDocument

Represents a single document ready for ingestion.

Methods:

• to_dict() -> Dict[str, Any] — serialize to a plain dictionary.

13. Module-level Helpers

merge_retrieval_results()

Documented in §8. Exported from core.query_expansion.

14. Enumerations

QuestionType

from fennec_community.rag.core import QuestionType

RerankMode

from fennec_community.rag.core import RerankMode

15. Quick-Start Example

The following example demonstrates the complete workflow from installation to querying:

from fennec_community.rag.core import RAGSystem, RAGConfig
from fennec_community.rag.core.logger import setup_logging

# 1. Configure logging
setup_logging(level="INFO", log_file="logs/app.log", json_format=False)

# 2. Build configuration
config = RAGConfig(
    top_k=8,
    enable_reranking=True,
    rerank_mode="hybrid",
    enable_retrieval_cache=True,
    cache_ttl=600.0,
)

# 3. Initialise the system (supply your own components)
rag = RAGSystem(
    vector_db=my_vector_db,
    llm=my_llm,
    chunker=my_chunker,
    context_manager=my_ctx_mgr,
    config=config,
    enable_query_expansion=True,
    query_expansion_variants=3,
)

# 4. Index documents
rag.add_texts({
    "hr_001": "Name: Ahmed Al-Farsi, Role: Software Engineer, Phone: +966 50 111 2222",
    "hr_002": "Name: Sara Al-Qahtani, Role: HR Manager, Phone: +966 55 333 4444",
})

# 5. Ask questions
answer = rag.ask("What is Ahmed's phone number?", include_sources=True)
print(answer)

# 6. Check system health
health = rag.validate_connection("test query")
print(health)

# 7. Inspect statistics
print(rag.get_stats())

# 8. Persist and reload
rag.save("./saved_rag")
rag2 = RAGSystem.load(
    "./saved_rag",
    vector_db=fresh_vector_db,
    llm=my_llm,
    chunker=my_chunker,
    context_manager=my_ctx_mgr,
)

# 9. Async streaming
import asyncio

async def stream_answer():
    async for token in rag.astream("Describe Ahmed's role"):
        print(token, end="", flush=True)

asyncio.run(stream_answer())

Simple Real Example

from fennec_community.llm import GeminiInterface
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

loader = TextLoader("./data_kn/faq.txt").load()
chunker = ArabicTextChunker(chunk_size=100, overlap=20)
embedder = OllamaEmbedder()
vector_db = FAISSVectorDatabase(embedder=embedder)
llm = GeminiInterface(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)
query = "ما هي طرق الدفع المتاحة؟"
response = rag_system.generate(query,include_sources=True)
print("Response:", response)