bifrost-rag 0.1.0

Production RAG pipeline with vector retrieval, configurable chunking, and evaluation harness

bifrost-rag

Production RAG pipeline with vector retrieval, configurable chunking, and evaluation harness.

Features

• Chunking strategies — Fixed-size, sentence-based, and recursive splitting
• Vector stores — ChromaDB (local) with pluggable store interface
• Retrieval — Top-K retrieval with score thresholds
• Evaluation harness — Precision@K, Recall@K, F1, MRR metrics
• Pipeline orchestrator — Ingest, retrieve, and query in one interface

Architecture

┌─────────────┐    ┌──────────────┐    ┌──────────────┐
│   Chunking   │───▶│  Embeddings  │───▶│ Vector Store │
│  strategies  │    │   (Voyage)   │    │  (ChromaDB)  │
└─────────────┘    └──────────────┘    └──────┬───────┘
                                              │
                   ┌──────────────┐    ┌──────▼───────┐
                   │  Evaluation  │◀───│  Retrieval   │
                   │   harness    │    │   pipeline   │
                   └──────────────┘    └──────────────┘

Installation

pip install bifrost-rag

Quick Start

from bifrost_rag import ChromaStore, Document, RAGPipeline, FixedSizeChunker

# Create a vector store
store = ChromaStore(collection_name="my-docs")

# Chunk and ingest documents
chunker = FixedSizeChunker(chunk_size=500, overlap=50)
text = "Your long document text here..."
chunks = chunker.chunk(text, metadata={"source": "doc1"})

# Add documents with embeddings (from your embedding provider)
docs = [
    Document(id=f"chunk-{c.index}", text=c.text, embedding=your_embeddings[i])
    for i, c in enumerate(chunks)
]
pipeline = RAGPipeline(store=store, top_k=5)
pipeline.ingest(docs)

# Query
result = pipeline.query(query_embedding=query_embed)
print(result.sources)

Chunking Strategies

Strategy	Description
FixedSizeChunker	Split by character count with overlap
SentenceChunker	Split by sentence boundaries
RecursiveChunker	Hierarchical split: paragraphs → sentences → characters

Evaluation

from bifrost_rag import RAGEvaluator, Document

evaluator = RAGEvaluator()

# Evaluate retrieval quality
dataset = [
    {
        "retrieved": [Document(id="d1", text="..."), Document(id="d2", text="...")],
        "relevant_ids": {"d1"},
    },
]
result = evaluator.evaluate(dataset, k=5)
print(result.summary())

License

MIT