ractogateway.rag.stores.weaviate_store

Weaviate vector store (lazy import).

Install with: pip install ractogateway[rag-weaviate]

class ractogateway.rag.stores.weaviate_store.WeaviateStore(class_name='RactoChunk', *, url=None, api_key=None, additional_headers=None, distance_metric='cosine', batch_size=100)

Bases: BaseVectorStore

Vector store backed by Weaviate.

Supports embedded (local, no server needed), local server, and Weaviate Cloud (WCS) connections.

Parameters:

• class_name (str) – Weaviate class (collection) name.

• url (str | None) – Weaviate server URL. None = use embedded Weaviate.

• api_key (str | None) – Weaviate Cloud API key.

• additional_headers (dict[str, str] | None) – Extra HTTP headers (e.g. for OpenAI API key pass-through to Weaviate).

• distance_metric (str) – "cosine" or "l2-squared".

• batch_size (int) – Objects per batch import.

add(chunks)

Add chunks (with embeddings) to the store.

Parameters:

chunks (list[Chunk]) – Chunks to index. Each chunk must have a non-None embedding.

Raises:

ValueError – If any chunk has embedding=None.

Return type:

None

search(embedding, top_k=5, filters=None)

Search for the top_k most similar chunks.

Parameters:

• embedding (list[float]) – Query embedding vector.

• 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 list of results (rank 1 = most similar).

delete(chunk_ids)

Remove chunks with the given IDs from the store.

Return type:

None

clear()

Remove all chunks from the store.

Return type:

None

count()

Return the total number of indexed chunks.

Return type:

int