Inicio / Artículos / RAG Document Assistant with Testable Services and Next.js Chat

Este artículo está publicado en inglés.

RAG Document Assistant with Testable Services and Next.js Chat

Split HTTP from pure Python services, index documents, cite segments, and ship a split-panel UI without forking retrieval.

4200 palabras

A document assistant that cites its sources is a RAG product, not a chatbot skin. This build separates HTTP routes from pure Python services, stores embeddings for fetch, and fronts a Next.js chat UI with a split transcript/citation pane.

Goals

Upload materials, index them, answer with grounded segments, and keep the core logic unit-testable without booting a server.

Backend shape

API routes handle HTTP only; services own ingest, embed, fetch, and generate. That split lets pytest exercise ranking without ASGI.

Frontend shape

Next.js plus Tailwind: split panel chat, dark/light themes that share the same API client, citation chips that deep-link into fetched segments.

from pypdf import PdfReader, errors
from fastapi import HTTPException, UploadFile

@router.post("/ingest")
async def ingest_pdf(file: UploadFile, settings: Settings = Depends(get_settings)):
    if file.content_type != "application/pdf":
        raise HTTPException(400, "Only PDF files accepted")
    contents = await file.read()
    try:
        reader = PdfReader(BytesIO(contents))
    except errors.DocumentLoadError:
        # If PDF is an image-scanned file, pypdf may fail here
        raise HTTPException(400, "Failed to load PDF (maybe it's scanned/image-only)")
    text = ""
    for page in reader.pages:
        page_text = page.extract_text()
        if page_text:
            text += page_text + "\n"
    if not text.strip():
        raise HTTPException(400, "PDF contained no extractable text")
    # Now text holds the raw content to index...

Keep configuration explicit—embedder names, index paths, and model ids belong in env, not hard-coded strings.

import tiktoken

ENCODING = tiktoken.get_encoding("cl100k_base")
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50

def chunk_text(text: str) -> list[str]:
    tokens = ENCODING.encode(text)
    chunks = []
    for i in range(0, len(tokens), CHUNK_SIZE - CHUNK_OVERLAP):
        chunk_tokens = tokens[i : i + CHUNK_SIZE]
        # Decode back to string; words may split, but overlap handles continuity
        chunks.append(ENCODING.decode(chunk_tokens))
        if i + CHUNK_SIZE >= len(tokens):
            break
    return chunks

Integrate this step before expanding the UI; retrieval bugs are cheaper to find behind a thin client.

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
# (In actual code, we'd use a singleton or lru_cache to avoid re-loading)

Keep configuration explicit—embedder names, index paths, and model ids belong in env, not hard-coded strings.

chunk_vectors = model.encode(chunks, show_progress_bar=False)

Integrate this step before expanding the UI; retrieval bugs are cheaper to find behind a thin client.

import httpx

HF_API_URL = "https://api-inference.huggingface.co/models/sentence-transformers/all-MiniLM-L6-v2"

async def embed_texts(texts: list[str], api_key: str) -> list[list[float]]:
    headers = {"Authorization": f"Bearer {api_key}"}
    async with httpx.AsyncClient() as client:
        response = await client.post(
            HF_API_URL,
            headers=headers,
            json={"inputs": texts, "options": {"wait_for_model": True}}
        )
        response.raise_for_status()
        return response.json()

Keep configuration explicit—embedder names, index paths, and model ids belong in env, not hard-coded strings.

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, PointStruct

qdrant = QdrantClient(url=settings.QDRANT_URL, api_key=settings.QDRANT_API_KEY)
# Recreate the collection for a clean slate (or check if exists)
qdrant.recreate_collection(
    collection_name="documents",
    vector_size=384,
    distance=Distance.COSINE
)

Integrate this step before expanding the UI; retrieval bugs are cheaper to find behind a thin client.

from qdrant_client.models import Filter, FieldCondition, MatchValue

qdrant.delete_points(
    collection_name="documents",
    filter=Filter(must=[FieldCondition(key="source", match=MatchValue(value=source_id))])
)

Keep configuration explicit—embedder names, index paths, and model ids belong in env, not hard-coded strings.

from uuid import uuid4

points = []
for chunk_text, vector in zip(chunks, chunk_vectors):
    points.append(
        PointStruct(
            id=str(uuid4()),
            vector=vector,
            payload={"text": chunk_text, "source": source_id}
        )
    )
qdrant.upsert(collection_name="documents", points=points)

Integrate this step before expanding the UI; retrieval bugs are cheaper to find behind a thin client.

question_vec = model.encode([question])[0]
search_result = qdrant.search(
    collection_name="documents",
    query_vector=question_vec,
    limit=5
)

Keep configuration explicit—embedder names, index paths, and model ids belong in env, not hard-coded strings.

from groq import Groq

client = Groq()
system_msg = (
    "You are a knowledgeable assistant. Answer the user's question using **only** the provided context. "
    "Include citations like [source:chunk_id] for any statements."
)
messages = [
    {"role": "system", "content": system_msg},
    {"role": "user", "content": user_question}
]
response = client.chat.completions.create(
    model="llama-3.1-8b-instant",
    messages=messages,
    temperature=0.1
)
answer = response.choices[0].message.content

Integrate this step before expanding the UI; retrieval bugs are cheaper to find behind a thin client.

from fastapi import FastAPI, Depends

app = FastAPI()

@app.post("/api/v1/ingest")
async def ingest(files: list[UploadFile], settings: Settings = Depends(get_settings)):
    # For each PDF: parse -> chunk -> embed -> upsert (as shown above).
    return {"success": True, "indexed": len(files)}

@app.post("/api/v1/query")
async def query(q: QueryRequest, settings: Settings = Depends(get_settings)):
    # q.question is the string from the user
    chunks = retrieve_top_chunks(q.question)   # search Qdrant
    answer = call_llm(q.question, chunks)      # prompt Llama
    return {"answer": answer, "sources": [c["source"] for c in chunks], "chunks": chunks}

Keep configuration explicit—embedder names, index paths, and model ids belong in env, not hard-coded strings.

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",
        "https://rag-banking-health-assistant-c8f7.vercel.app",
    ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Integrate this step before expanding the UI; retrieval bugs are cheaper to find behind a thin client.

curl -X POST "https://api-inference.huggingface.co/models/sentence-transformers/all-MiniLM-L6-v2" \
  -H "Authorization: Bearer $HF_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"inputs": ["test sentence"]}'

Keep configuration explicit—embedder names, index paths, and model ids belong in env, not hard-coded strings.

# Before (dead)
HF_API_URL = f"https://api-inference.huggingface.co/pipeline/feature-extraction/{MODEL_NAME}"

# After (working)
HF_API_URL = f"https://router.huggingface.co/hf-inference/models/{MODEL_NAME}/pipeline/feature-extraction"

Integrate this step before expanding the UI; retrieval bugs are cheaper to find behind a thin client.

class Settings(BaseSettings):
        QDRANT_URL: str
        QDRANT_API_KEY: str
        GROQ_API_KEY: str
        HF_API_KEY: str
        class Config:
            env_file = ".env"

Keep configuration explicit—embedder names, index paths, and model ids belong in env, not hard-coded strings.

Evaluation

Freeze twenty questions with expected segment ids. Track hit rate and citation presence. UI polish never fixes a wrong top-k.

Deployment notes

Run API and worker separately if ingest is heavy. Health checks should fail when the vector store is unreachable, not only when HTTP is up.

Closing

Clean boundaries—routes vs services, fetch vs generate, UI vs API—turn a weekend RAG demo into something teams can test and operate.