ractogateway.rag.pipeline

RactoRAG — the unified RAG pipeline.

Combines file reading, text processing, chunking, embedding, vector storage, retrieval, and LLM-based generation into a single, ergonomic interface.

Example:

from ractogateway import openai_developer_kit as opd
from ractogateway.rag.pipeline import RactoRAG
from ractogateway.rag.embedders.openai_embedder import OpenAIEmbedder
from ractogateway.rag.stores.chroma_store import ChromaStore

kit = opd.OpenAIDeveloperKit(model="gpt-4o")
rag = RactoRAG(
    vector_store=ChromaStore(collection="docs"),
    embedder=OpenAIEmbedder(),
    llm_kit=kit,
)
rag.ingest("report.pdf")
response = rag.query("What were the key findings?")
print(response.answer.content)

class ractogateway.rag.pipeline.RactoRAG(vector_store=None, embedder=None, *, store=None, chunker=None, processors=None, llm_kit=None, context_template=None, reader_registry=None, default_prompt=None)

Bases: object

Production-grade RAG pipeline for RactoGateway.

Parameters:

• vector_store (BaseVectorStore | None) – Any BaseVectorStore instance.

• embedder (BaseEmbedder | None) – Any BaseEmbedder instance.

• chunker (BaseChunker | None) – How to split documents. Defaults to RecursiveChunker with chunk_size=512, overlap=50.

• processors (list[BaseProcessor] | None) – List of text processors applied to each chunk before embedding. Defaults to [TextCleaner()].

• llm_kit (Any | None) – Any developer kit (OpenAIDeveloperKit, GoogleDeveloperKit, or AnthropicDeveloperKit). Required for query() / aquery().

• context_template (str | None) – Template string for injecting retrieved context into the LLM prompt. Must contain {context} and {question} placeholders.

• reader_registry (FileReaderRegistry | None) – Custom FileReaderRegistry. Defaults to a registry with all built-in readers.

• default_prompt (RactoPrompt | None) – RACTO prompt used for generation. Falls back to a built-in RAG prompt.

ingest(path, **metadata)

Read, chunk, embed, and store a single file.

Parameters:

• path (str | Path) – Path to the file to ingest.

• **metadata (Any) – Extra metadata merged into each chunk’s ChunkMetadata.extra.

Return type:

list[Chunk]

Returns:

list[Chunk] – The chunks that were added to the vector store.

ingest_dir(directory, pattern='**/*', **metadata)

Recursively ingest all supported files in a directory.

Parameters:

• directory (str | Path) – Root directory to scan.

• pattern (str) – Glob pattern relative to directory.

• **metadata (Any) – Extra metadata merged into every chunk.

Return type:

list[Chunk]

Returns:

list[Chunk] – All chunks added across all ingested files.

ingest_text(text, source='manual', **metadata)

Ingest a raw text string directly (no file needed).

Parameters:

• text (str) – The text content to ingest.

• source (str) – A label identifying the source (stored in metadata).

• **metadata (Any) – Extra metadata merged into each chunk.

Return type:

list[Chunk]

async aingest(path, **metadata)

Async variant of ingest().

Return type:

list[Chunk]

async aingest_dir(directory, pattern='**/*', **metadata)

Async variant of ingest_dir().

Return type:

list[Chunk]

async aingest_text(text, source='manual', **metadata)

Async variant of ingest_text().

Return type:

list[Chunk]

retrieve(query, top_k=5, filters=None)

Embed query and retrieve the top-k most relevant chunks.

Parameters:

• query (str) – Natural-language question or search phrase.

• top_k (int) – Number of results to return.

• filters (dict[str, Any] | None) – Optional metadata filters (store-specific format).

Return type:

list[RetrievalResult]

Returns:

list[RetrievalResult] – Ranked results (rank 1 = most relevant).

async aretrieve(query, top_k=5, filters=None)

Async variant of retrieve().

Return type:

list[RetrievalResult]

query(question, top_k=5, filters=None, prompt=None, temperature=0.0, max_tokens=2048)

Retrieve relevant chunks and generate an answer.

Parameters:

• question (str) – The user’s question.

• top_k (int) – Number of context chunks to retrieve.

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

• prompt (RactoPrompt | None) – Override the default RACTO prompt for generation.

• temperature (float) – LLM temperature (default 0.0 for factual answers).

• max_tokens (int) – Maximum tokens in the generated answer.

Return type:

RAGResponse

Returns:

RAGResponse – Contains the generated answer plus the retrieved source chunks.

Raises:

RuntimeError – If no llm_kit was provided.

async aquery(question, top_k=5, filters=None, prompt=None, temperature=0.0, max_tokens=2048)

Async variant of query().

Return type:

RAGResponse

property store: BaseVectorStore

The underlying vector store.

property embedder: BaseEmbedder

The underlying embedder.

count()

Return the total number of indexed chunks.

Return type:

int

clear()

Remove all indexed chunks from the vector store.

Return type:

None