BRT Platform
Hybrid RAG Consulting Infrastructure

# platform/retrieval/pipeline.py
class HybridRetriever:
    def __init__(self, tenant_id: str):
        self.qdrant = QdrantClient(host="localhost")
        self.collection = f"tenant_{tenant_id}"
        self.dense_model = SentenceTransformer("BAAI/bge-large-en-v1.5")
        self.sparse_model = SparseTextEmbedding("prithivida/Splade_PP_Efficient_v1")
        self.reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

    def retrieve(self, query: str, top_k: int = 5) -> list[ScoredPassage]:
        # Stage 1: Dense search
        dense_vec = self.dense_model.encode(query, normalize_embeddings=True)
        dense_hits = self.qdrant.search(self.collection, dense_vec, limit=20)

        # Stage 2: Sparse search
        sparse_vec = next(self.sparse_model.embed([query]))
        sparse_hits = self.qdrant.search(self.collection,
            query_sparse_vector=SparseVector(sparse_vec.indices, sparse_vec.values), limit=20)

        # Stage 3: RRF fusion
        candidates = rrf_fuse(dense_hits, sparse_hits, k=60)[:20]

        # Stage 4: Cross-encoder rerank
        pairs = [(query, c.payload["text"]) for c in candidates]
        scores = self.reranker.predict(pairs)
        reranked = sorted(zip(candidates, scores), key=lambda x: -x[1])
        return [ScoredPassage(c, s) for c, s in reranked[:top_k]]

def rrf_fuse(hits_a: list, hits_b: list, k: int = 60) -> list:
    scores: dict[str, float] = {}
    for rank, hit in enumerate(hits_a):
        scores[hit.id] = scores.get(hit.id, 0) + 1.0 / (k + rank + 1)
    for rank, hit in enumerate(hits_b):
        scores[hit.id] = scores.get(hit.id, 0) + 1.0 / (k + rank + 1)
    all_hits = {h.id: h for h in hits_a + hits_b}
    return [all_hits[id] for id in sorted(scores, key=lambda i: -scores[i])]

# platform/core/resilience.py
class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.state = "closed"
        self.failures = 0
        self.last_failure = 0
        self.threshold = failure_threshold
        self.timeout = recovery_timeout

    def call(self, fn, *args, fallback=None, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure > self.timeout:
                self.state = "half-open"
            else:
                return fallback(*args, **kwargs) if fallback else None
        try:
            result = fn(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failures = 0
            return result
        except Exception:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.threshold:
                self.state = "open"
            return fallback(*args, **kwargs) if fallback else None

CREATE VIRTUAL TABLE fts_index USING fts5(
    doc_id, text, tenant_id,
    content='documents', content_rowid='rowid'
);
-- WAL mode for concurrent reads during write
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;

#!/usr/bin/env bash
# instant-poc.sh — runs hybrid RAG on client sample docs, no infra needed
# Usage: ./instant-poc.sh /path/to/docs "your query here"
DOCS_DIR=$1; QUERY=$2
pip install -q sentence-transformers qdrant-client fastembed
python3 - <<EOF
from platform.retrieval.pipeline import HybridRetriever
from platform.ingestion import ingest_directory

r = HybridRetriever(tenant_id="poc")
ingest_directory(r, "$DOCS_DIR")
for hit in r.retrieve("$QUERY", top_k=3):
    print(f"[{hit.score:.3f}] {hit.payload['text'][:200]}")
EOF

# deploy-client.sh — provision full stack for a new client tenant
CLIENT_ID=$1; DOCS_DIR=$2
docker compose up -d qdrant redis
python3 -m platform.ingestion.bulk_ingest \
  --tenant "$CLIENT_ID" \
  --source "$DOCS_DIR" \
  --batch-size 64
echo "Tenant $CLIENT_ID ready. Collection: tenant_${CLIENT_ID}"
echo "Endpoint: http://localhost:8000/v1/retrieve?tenant=${CLIENT_ID}"