BaseRAGSystem

core.base_rag_system.BaseRAGSystem

Overview

BaseRAGSystem is an abstract base class (abc.ABC) that defines the unified contract every RAG implementation in the library must follow. It owns configuration validation, a stats dictionary, and a set of concrete convenience methods (context-manager support, async wrappers, connection validation) that subclasses get "for free". RAGSystem is the only concrete subclass in the analyzed source.

Why it exists: to let callers write code against a stable interface (add_documents / retrieve / generate) regardless of which concrete backend is used, and to centralize cross-cutting behavior (stats tracking, sync→async bridging, context-manager protocol) so subclasses don't have to reimplement it.

When to use it directly: subclass it when you want to build a RAG system with a different internal architecture than RAGSystem (e.g. a different persistence model) but still want it to be interchangeable with RAGSystem from the caller's point of view.

When not to use it: if RAGSystem's duck-typed dependency model (vector_db/llm/chunker/context_manager) already fits your needs, use RAGSystem directly instead of subclassing.

Exact signature

class BaseRAGSystem(ABC):
    def __init__(self, config: Optional[RAGConfig] = None): ...

Constructor

Parameters

Initialization flow

1. self.config = config or RAGConfig() — falls back to defaults if None.
2. self.config.validate() — raises ValueError immediately if the config is invalid (see RAGConfig.validate()).
3. self.stats is initialized to:

   {'total_queries': 0, 'successful_queries': 0, 'failed_queries': 0,
    'total_documents': 0, 'total_chunks': 0}
   

4. Logs 🚀 Initializing {ClassName} at INFO level via the module-level logging.getLogger(__name__) logger (note: this module uses the standard library logger directly, not core.logger.get_logger).

Raises

Attributes

Abstract methods (must be implemented by subclasses)

add_documents(docs: Dict[str, str]) -> Dict[str, int]

Add documents to the system. Args: docs — {doc_id: text}. Returns {doc_id: num_chunks}.

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

Retrieve the nearest chunks to the query. Returns a list of (DocumentChunk, score) tuples.

generate(query: str, **kwargs) -> str

Generate an answer for the query. Returns the answer text.

Concrete methods (inherited as-is unless overridden)

remove_document(doc_id: str) -> int

Default behavior: raises NotImplementedError("❌ This feature is not supported in this system"). RAGSystem overrides this with a real implementation — see RAGSystem.remove_document.

save(path: str) -> None

Default behavior: raises NotImplementedError. Overridden by RAGSystem.

load(path: str, **kwargs) (classmethod)

Default behavior: raises NotImplementedError. Overridden by RAGSystem.load.

get_stats() -> Dict[str, Any]

Returns a shallow copy (self.stats.copy()) of the stats dict, so callers cannot mutate internal state through the returned object.

reset_stats()

Resets total_queries, successful_queries, and failed_queries to 0, but preserves total_documents and total_chunks (read via .get(..., 0) from the current stats before rebuilding the dict). Logs 📊 Statistics reset at INFO.

validate_connection(test_query: str) -> Dict

Runs a live smoke test against the concrete subclass's own retrieve() and generate() methods.

Algorithm:

1. Call self.retrieve(test_query, top_k=1).
2. If no chunks are returned → {"success": False, "reason": "⚠️ No documents in the system", "components": None}.
3. Otherwise, attempt self.generate(test_query):
   • If it returns a truthy answer → {"success": True, "reason": "✅ System working successfully", "components": {"retrieval": True, "generation": True}}.
   • If it raises → {"success": True, "reason": "✅ Retrieval works, generation unavailable", "components": {"retrieval": True, "generation": False, "generation_error": str(exc)}} (retrieval is still reported successful).
4. If retrieve() itself raises → {"success": False, "reason": f"❌Error: {exc}", "components": None}.

Returns: dict with keys success (bool), reason (str), components (dict or None).

cleanup()

Logs 🧹Cleaning up system resources at INFO. Does not release any specific resource itself — subclasses (like RAGSystem) override this to actually clean up components.

__repr__() -> str

{ClassName}(docs={total_documents}, chunks={total_chunks}, queries={total_queries})

__enter__ / __exit__

Implements the synchronous context-manager protocol. __exit__ calls self.cleanup() and does not suppress exceptions (implicitly returns None, which is falsy).

Async API (default implementations — thread-wrapped)

These default implementations wrap the synchronous methods using asyncio.to_thread, so subclasses get async support "for free" without writing native async code. RAGSystem overrides all three with native async implementations for better concurrency (see RAGSystem Async API).

async aadd_documents(docs: Dict[str, str]) -> Dict[str, int]

return await asyncio.to_thread(self.add_documents, docs)

async aretrieve(query: str, top_k=None) -> List[Tuple]

return await asyncio.to_thread(self.retrieve, query, top_k)

async agenerate(query: str, **kwargs) -> str

return await asyncio.to_thread(self.generate, query, **kwargs)

async __aenter__ / async __aexit__

Async context-manager protocol; __aexit__ calls the synchronous self.cleanup() (not an async cleanup) and returns False.

Thread safety

Not documented or enforced by the class itself. self.stats is a plain dict mutated without any locking — concurrent calls to methods that increment stats (as done in RAGSystem) are not guaranteed to be free of race conditions under true multi-threaded (not asyncio) concurrency. asyncio.to_thread-based async methods run the synchronous method in a thread pool, so the same caveat applies there too.

Complexity

All methods on BaseRAGSystem itself are O(1) except where they delegate to subclass-provided retrieve/generate implementations, whose complexity depends entirely on the subclass.

Common mistakes

• Assuming remove_document, save, or load work out of the box — they raise NotImplementedError unless the subclass overrides them.
• Mutating the dict returned by get_stats() and expecting it to affect the live system — it's a copy.
• Relying on stats thread-safety under concurrent synchronous access.

Related APIs

• RAGSystem — the concrete implementation.
• RAGConfig — the configuration object every BaseRAGSystem requires.