RAG Pipeline Development

from qdrant_client import QdrantClient
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter

async def ingest_document(doc_path: str, metadata: dict) -> int:
    """Chunk, embed, and index a document into Qdrant."""

    # Load and chunk with overlap for cross-boundary context
    raw_text = extract_text(doc_path)        # PDF, DOCX, HTML, plain text
    splitter  = RecursiveCharacterTextSplitter(
        chunk_size=600, chunk_overlap=120,
        separators=["

", "
", ". ", " "],
    )
    chunks = splitter.split_text(raw_text)

    # Embed all chunks in a single batched API call
    embedder   = OpenAIEmbeddings(model="text-embedding-3-large")
    embeddings = await embedder.aembed_documents(chunks)

    # Upsert into Qdrant with source metadata for citation
    points = [
        PointStruct(
            id=str(uuid4()),
            vector=embedding,
            payload={
                "text": chunk,
                "source": doc_path,
                "page": metadata.get("page"),
                "section": metadata.get("section"),
                "doc_id": metadata["doc_id"],
            },
        )
        for chunk, embedding in zip(chunks, embeddings)
    ]
    client.upsert(collection_name=COLLECTION, points=points)
    return len(chunks)


async def query(question: str, top_k: int = 8) -> Answer:
    """Hybrid search: dense + BM25, then rerank, then generate."""

    # Dense retrieval
    q_embed  = await embedder.aembed_query(question)
    dense    = client.search(COLLECTION, q_embed, limit=top_k * 2)

    # BM25 keyword retrieval (catches exact-match queries)
    keyword  = bm25_index.search(question, top_n=top_k * 2)

    # Merge and rerank with cross-encoder
    combined = dedupe_and_merge(dense, keyword)
    reranked = cross_encoder.rerank(question, combined)[:top_k]

    # Generate with source citations
    context  = "

".join(c.text for c in reranked)
    answer   = await llm.acomplete(PROMPT.format(context=context, question=question))
    return Answer(text=answer, sources=[c.source for c in reranked])