babylon.rag.rag_pipeline

Main RAG pipeline service that orchestrates document ingestion and query processing.

Uses VectorStoreProtocol from babylon.persistence for backend-agnostic vector storage. ChromaDB has been removed; use PgVectorStore (Feature 037).

Classes

IngestionResult(**data)	Result of document ingestion process.
RagConfig(**data)	Configuration for RAG pipeline.
RagPipeline(vector_store[, config, ...])	Main RAG pipeline that orchestrates ingestion and retrieval.

class babylon.rag.rag_pipeline.IngestionResult(**data)

Bases: BaseModel

Result of document ingestion process.

Parameters:

• success (bool)

• chunks_processed (int)

• chunks_stored (int)

• processing_time_ms (float)

• embedding_time_ms (float)

• storage_time_ms (float)

• errors (list[str])

• source_files (list[str])

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

success: bool

chunks_processed: int

chunks_stored: int

processing_time_ms: float

embedding_time_ms: float

storage_time_ms: float

errors: list[str]

source_files: list[str]

class babylon.rag.rag_pipeline.RagConfig(**data)

Bases: BaseModel

Configuration for RAG pipeline.

Parameters:

• chunk_size (int)

• chunk_overlap (int)

• min_chunk_length (int)

• max_chunk_length (int)

• embedding_batch_size (int)

• max_concurrent_embeds (int)

• default_top_k (int)

• default_similarity_threshold (float)

• collection_name (str)

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

chunk_size: int

chunk_overlap: int

min_chunk_length: int

max_chunk_length: int

embedding_batch_size: int

max_concurrent_embeds: int

default_top_k: int

default_similarity_threshold: float

collection_name: str

class babylon.rag.rag_pipeline.RagPipeline(vector_store, config=None, embedding_manager=None)

Bases: object

Main RAG pipeline that orchestrates ingestion and retrieval.

Requires a VectorStoreProtocol implementation (e.g. PgVectorStore).

Parameters:

• vector_store (VectorStoreProtocol)

• config (RagConfig | None)

• embedding_manager (EmbeddingManager | None)

__init__(vector_store, config=None, embedding_manager=None)

Initialize the RAG pipeline.

Parameters:

• vector_store (VectorStoreProtocol) – VectorStoreProtocol implementation for storage.

• config (RagConfig | None) – RAG configuration (uses default if None).

• embedding_manager (EmbeddingManager | None) – Embedding manager (creates new if None).

async aingest_text(content, source_id, metadata=None)

Asynchronously ingest text content into the RAG system.

Parameters:

• content (str) – Text content to ingest

• source_id (str) – Unique identifier for the content source

• metadata (dict[str, Any] | None) – Optional metadata to attach to chunks

Return type:

IngestionResult

Returns:

IngestionResult with processing statistics

Raises:

RagError – If ingestion fails

ingest_text(content, source_id, metadata=None)

Synchronously ingest text content into the RAG system.

Parameters:

• content (str) – Text content to ingest

• source_id (str) – Unique identifier for the content source

• metadata (dict[str, Any] | None) – Optional metadata to attach to chunks

Return type:

IngestionResult

Returns:

IngestionResult with processing statistics

async aingest_file(file_path, encoding='utf-8')

Asynchronously ingest a text file into the RAG system.

Parameters:

• file_path (str) – Path to the text file

• encoding (str) – File encoding (default: utf-8)

Return type:

IngestionResult

Returns:

IngestionResult with processing statistics

Raises:

• FileNotFoundError – If file doesn’t exist

• RagError – If ingestion fails

ingest_file(file_path, encoding='utf-8')

Synchronously ingest a text file into the RAG system.

Parameters:

• file_path (str) – Path to the text file

• encoding (str) – File encoding (default: utf-8)

Return type:

IngestionResult

Returns:

IngestionResult with processing statistics

async aingest_files(file_paths, encoding='utf-8', max_concurrent=5)

Asynchronously ingest multiple files concurrently.

Parameters:

• file_paths (list[str]) – List of file paths to ingest

• encoding (str) – File encoding (default: utf-8)

• max_concurrent (int) – Maximum number of concurrent ingestions

Return type:

list[IngestionResult]

Returns:

List of IngestionResult objects

ingest_files(file_paths, encoding='utf-8', max_concurrent=5)

Synchronously ingest multiple files.

Parameters:

• file_paths (list[str]) – List of file paths to ingest

• encoding (str) – File encoding (default: utf-8)

• max_concurrent (int) – Maximum number of concurrent ingestions

Return type:

list[IngestionResult]

Returns:

List of IngestionResult objects

async aquery(query, top_k=None, similarity_threshold=None, metadata_filter=None)

Asynchronously query the RAG system for relevant content.

Parameters:

• query (str) – Query text to search for

• top_k (int | None) – Number of results to return (uses config default if None)

• similarity_threshold (float | None) – Minimum similarity score (uses config default if None)

• metadata_filter (dict[str, Any] | None) – Optional filters for chunk metadata

Return type:

QueryResponse

Returns:

QueryResponse with search results

Raises:

RagError – If query processing fails

query(query, top_k=None, similarity_threshold=None, metadata_filter=None)

Synchronously query the RAG system for relevant content.

Parameters:

• query (str) – Query text to search for

• top_k (int | None) – Number of results to return (uses config default if None)

• similarity_threshold (float | None) – Minimum similarity score (uses config default if None)

• metadata_filter (dict[str, Any] | None) – Optional filters for chunk metadata

Return type:

QueryResponse

Returns:

QueryResponse with search results

get_stats()

Get statistics about the RAG system.

Return type:

dict[str, Any]

Returns:

Dictionary with system statistics

clear_collection()

Clear all documents from the collection.

WARNING: This will delete all ingested documents!

Return type:

None

async aclose()

Asynchronously close the RAG pipeline and clean up resources.

Return type:

None

close()

Synchronously close the RAG pipeline and clean up resources.

Return type:

None

__enter__()

Context manager entry.

Return type:

Self

__exit__(exc_type, exc_val, exc_tb)

Context manager exit with cleanup.

Return type:

None

Parameters:

• exc_type (type[BaseException] | None)

• exc_val (BaseException | None)

• exc_tb (TracebackType | None)