RAG i Python: Bygg ett GDPR-säkert dokumentsök-API med EU-hostad inferens

Bygg ett produktionsklart retrieval-augmented generation (RAG)-system i Python som håller all data inom EU. Den här guiden täcker dokumentinmatning med PyMuPDF, vektorlagring med Qdrant och LLM-inferens via Juice Factorys privata EU-API — allt sammankopplat i en FastAPI-tjänst.

När du är klar har du ett fungerande dokumentsök-API som:

• Extraherar text från PDF-filer med PyMuPDF
• Genererar embeddings och lagrar dem i Qdrant
• Besvarar frågor med hjälp av hämtad kontext + EU-hostad LLM-inferens
• Aldrig skickar användardata utanför EU

Förutsättningar

• Python 3.10+
• Docker (för Qdrant)
• En Juice Factory API-nyckel (skaffa en här)

Arkitekturöversikt

┌──────────────┐     ┌───────────────┐     ┌──────────────────┐
│  PDF Upload  │────▶│  PyMuPDF      │────▶│  Qdrant          │
│  (FastAPI)   │     │  Text Extract │     │  Vector Store    │
└──────────────┘     └───────────────┘     └──────────────────┘
                                                    │
┌──────────────┐     ┌───────────────┐              │
│  User Query  │────▶│  Embedding    │──── search ──┘
│  (FastAPI)   │     │  (EU API)     │
└──────────────┘     └───────┬───────┘
                             │
                     ┌───────▼───────┐     ┌──────────────────┐
                     │  Context +    │────▶│  LLM Inference   │
                     │  Query        │     │  (EU-hosted)     │
                     └───────────────┘     └──────────────────┘

Systemet följer en standard RAG-pipeline, men varje komponent som hanterar användardata körs inom EU-infrastruktur. Qdrant är självhostad, och både embeddings och LLM-inferens routas genom Juice Factorys EU-endpoints.

Steg 1: Projektuppställning

Skapa projektkatalogen och installera beroenden:

mkdir rag-document-search && cd rag-document-search
python -m venv .venv
source .venv/bin/activate

Installera de nödvändiga paketen:

pip install fastapi uvicorn pymupdf qdrant-client openai python-multipart

Skapa projektstrukturen:

rag-document-search/
├── main.py              # FastAPI application
├── ingest.py            # Document ingestion pipeline
├── search.py            # Query and retrieval logic
├── config.py            # Configuration
└── requirements.txt

requirements.txt:

fastapi==0.115.0
uvicorn==0.30.0
pymupdf==1.24.0
qdrant-client==1.11.0
openai==1.50.0
python-multipart==0.0.9

Steg 2: Konfiguration

Konfigurera med dina Juice Factory API-uppgifter:

# config.py
import os

# Juice Factory EU API (OpenAI-compatible)
API_BASE_URL = "https://api.juicefactory.ai/v1"
API_KEY = os.environ.get("JUICEFACTORY_API_KEY", "your-api-key")

# Embedding model
EMBEDDING_MODEL = "text-embedding-3-small"
EMBEDDING_DIMENSIONS = 1536

# Chat model for RAG responses
CHAT_MODEL = "gpt-4"

# Qdrant configuration (self-hosted in EU)
QDRANT_HOST = os.environ.get("QDRANT_HOST", "localhost")
QDRANT_PORT = int(os.environ.get("QDRANT_PORT", "6333"))
COLLECTION_NAME = "documents"

# Chunk settings
CHUNK_SIZE = 500       # tokens per chunk (approximate)
CHUNK_OVERLAP = 50     # overlap between chunks
TOP_K = 5              # number of chunks to retrieve

Steg 3: Starta Qdrant med Docker

Kör Qdrant lokalt (eller på din EU-server):

docker run -d \
  --name qdrant \
  -p 6333:6333 \
  -p 6334:6334 \
  -v qdrant_storage:/qdrant/storage \
  qdrant/qdrant:latest

Qdrant lagrar all data lokalt — inga externa anrop, ingen telemetri, full kontroll över var datan hamnar.

Steg 4: Dokumentinmatning med PyMuPDF

Inmatningspipelinen extraherar text från PDF-filer, delar upp den i chunks, genererar embeddings via EU-API:et och lagrar allt i Qdrant.

# ingest.py
import fitz  # PyMuPDF
from openai import OpenAI
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid
import config


def get_openai_client() -> OpenAI:
    """Create OpenAI client pointing to Juice Factory EU API."""
    return OpenAI(
        api_key=config.API_KEY,
        base_url=config.API_BASE_URL,
    )


def get_qdrant_client() -> QdrantClient:
    """Create Qdrant client."""
    return QdrantClient(host=config.QDRANT_HOST, port=config.QDRANT_PORT)


def extract_text_from_pdf(pdf_bytes: bytes) -> list[dict]:
    """Extract text from PDF, page by page."""
    doc = fitz.open(stream=pdf_bytes, filetype="pdf")
    pages = []
    for page_num, page in enumerate(doc):
        text = page.get_text("text").strip()
        if text:
            pages.append({
                "page": page_num + 1,
                "text": text,
            })
    doc.close()
    return pages


def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Split text into overlapping chunks by word count."""
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunk = " ".join(words[start:end])
        chunks.append(chunk)
        start = end - overlap
    return chunks


def generate_embeddings(texts: list[str], client: OpenAI) -> list[list[float]]:
    """Generate embeddings using Juice Factory EU API."""
    response = client.embeddings.create(
        model=config.EMBEDDING_MODEL,
        input=texts,
    )
    return [item.embedding for item in response.data]


def ensure_collection(qdrant: QdrantClient):
    """Create Qdrant collection if it doesn't exist."""
    collections = [c.name for c in qdrant.get_collections().collections]
    if config.COLLECTION_NAME not in collections:
        qdrant.create_collection(
            collection_name=config.COLLECTION_NAME,
            vectors_config=VectorParams(
                size=config.EMBEDDING_DIMENSIONS,
                distance=Distance.COSINE,
            ),
        )


def ingest_pdf(pdf_bytes: bytes, filename: str) -> int:
    """Full ingestion pipeline: PDF → chunks → embeddings → Qdrant."""
    openai_client = get_openai_client()
    qdrant = get_qdrant_client()
    ensure_collection(qdrant)

    # Extract text from PDF
    pages = extract_text_from_pdf(pdf_bytes)

    # Chunk all pages
    all_chunks = []
    for page_data in pages:
        chunks = chunk_text(
            page_data["text"],
            chunk_size=config.CHUNK_SIZE,
            overlap=config.CHUNK_OVERLAP,
        )
        for chunk in chunks:
            all_chunks.append({
                "text": chunk,
                "page": page_data["page"],
                "filename": filename,
            })

    if not all_chunks:
        return 0

    # Generate embeddings (batch)
    texts = [c["text"] for c in all_chunks]
    embeddings = generate_embeddings(texts, openai_client)

    # Store in Qdrant
    points = [
        PointStruct(
            id=str(uuid.uuid4()),
            vector=embedding,
            payload={
                "text": chunk["text"],
                "page": chunk["page"],
                "filename": chunk["filename"],
            },
        )
        for chunk, embedding in zip(all_chunks, embeddings)
    ]

    qdrant.upsert(
        collection_name=config.COLLECTION_NAME,
        points=points,
    )

    return len(points)

Viktiga punkter:

• PyMuPDF (fitz) extraherar text utan externa beroenden eller molnanrop
• Embeddings genereras via Juice Factorys EU-API — samma OpenAI SDK, EU-endpoint
• Qdrant lagrar vektorer lokalt utan telemetri

Steg 5: Sök och RAG-frågor

Sökmodulen embeddar användarens fråga, hämtar relevanta chunks och skickar dem tillsammans med frågan till LLM:en.

# search.py
from openai import OpenAI
from qdrant_client import QdrantClient
import config
from ingest import get_openai_client, get_qdrant_client, generate_embeddings


def search_documents(query: str, top_k: int = None) -> list[dict]:
    """Search for relevant document chunks."""
    if top_k is None:
        top_k = config.TOP_K

    openai_client = get_openai_client()
    qdrant = get_qdrant_client()

    # Embed the query
    query_embedding = generate_embeddings([query], openai_client)[0]

    # Search Qdrant
    results = qdrant.search(
        collection_name=config.COLLECTION_NAME,
        query_vector=query_embedding,
        limit=top_k,
    )

    return [
        {
            "text": hit.payload["text"],
            "page": hit.payload["page"],
            "filename": hit.payload["filename"],
            "score": hit.score,
        }
        for hit in results
    ]


def rag_query(question: str) -> dict:
    """Full RAG pipeline: embed query → retrieve context → generate answer."""
    # Retrieve relevant chunks
    chunks = search_documents(question)

    if not chunks:
        return {
            "answer": "No relevant documents found. Please upload documents first.",
            "sources": [],
        }

    # Build context from retrieved chunks
    context_parts = []
    for i, chunk in enumerate(chunks, 1):
        context_parts.append(
            f"[Source {i}: {chunk['filename']}, page {chunk['page']}]\n{chunk['text']}"
        )
    context = "\n\n".join(context_parts)

    # Generate answer using EU-hosted LLM
    openai_client = get_openai_client()
    response = openai_client.chat.completions.create(
        model=config.CHAT_MODEL,
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a document assistant. Answer questions based on the "
                    "provided context. Always cite which source and page number your "
                    "answer comes from. If the context doesn't contain enough "
                    "information to answer, say so clearly."
                ),
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {question}",
            },
        ],
        temperature=0.1,
        max_tokens=1000,
    )

    return {
        "answer": response.choices[0].message.content,
        "sources": [
            {
                "filename": c["filename"],
                "page": c["page"],
                "score": round(c["score"], 4),
                "excerpt": c["text"][:200] + "..." if len(c["text"]) > 200 else c["text"],
            }
            for c in chunks
        ],
        "model": response.model,
        "usage": {
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
        },
    }

Funktionen rag_query är kärnan i systemet:

1. Embeddar användarens fråga via EU-API:et
2. Hämtar de top-K mest relevanta chunks ur Qdrant
3. Skickar kontext + fråga till den EU-hostade LLM:en
4. Returnerar svaret med källhänvisningar

Steg 6: FastAPI-applikation

Koppla ihop allt med en FastAPI-tjänst:

# main.py
from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
from ingest import ingest_pdf
from search import rag_query, search_documents

app = FastAPI(
    title="GDPR-Safe Document Search API",
    description="RAG-powered document search with EU-hosted inference",
    version="1.0.0",
)


class QueryRequest(BaseModel):
    question: str
    top_k: int = 5


class QueryResponse(BaseModel):
    answer: str
    sources: list[dict]
    model: str | None = None
    usage: dict | None = None


@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """Upload a PDF document for indexing."""
    if not file.filename.lower().endswith(".pdf"):
        raise HTTPException(status_code=400, detail="Only PDF files are supported")

    pdf_bytes = await file.read()
    if len(pdf_bytes) > 50 * 1024 * 1024:  # 50MB limit
        raise HTTPException(status_code=400, detail="File too large (max 50MB)")

    num_chunks = ingest_pdf(pdf_bytes, file.filename)

    return {
        "filename": file.filename,
        "chunks_indexed": num_chunks,
        "status": "indexed",
    }


@app.post("/query", response_model=QueryResponse)
async def query_documents(request: QueryRequest):
    """Ask a question about uploaded documents."""
    if not request.question.strip():
        raise HTTPException(status_code=400, detail="Question cannot be empty")

    result = rag_query(request.question)
    return QueryResponse(**result)


@app.post("/search")
async def search_only(request: QueryRequest):
    """Search for relevant chunks without generating an answer."""
    results = search_documents(request.question, top_k=request.top_k)
    return {"results": results}


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {"status": "ok", "data_residency": "EU"}

Steg 7: Kör och testa

Starta API-servern:

export JUICEFACTORY_API_KEY="your-api-key"
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

Ladda upp ett dokument

curl -X POST http://localhost:8000/upload \
  -F "file=@contract.pdf"

Svar:

{
  "filename": "contract.pdf",
  "chunks_indexed": 47,
  "status": "indexed"
}

Ställ en fråga

curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What are the payment terms in the contract?"}'

Svar:

{
  "answer": "According to the contract (Source 1, page 4), payment terms are Net 30 from the date of invoice. Late payments accrue interest at 1.5% per month as specified in Section 5.2.",
  "sources": [
    {
      "filename": "contract.pdf",
      "page": 4,
      "score": 0.9234,
      "excerpt": "Payment Terms. The Client shall pay all invoices within thirty (30) days..."
    }
  ],
  "model": "gpt-4-0125-preview",
  "usage": {
    "prompt_tokens": 847,
    "completion_tokens": 89
  }
}

Checklista för GDPR-efterlevnad

Den här arkitekturen uppfyller GDPR-kraven i varje lager:

Viktiga garantier:

• Användarfrågor lämnar aldrig EU
• Ingen data används för modellträning
• Qdrant lagrar bara embeddings (inte råa frågor)
• LLM-inferens är tillståndslös — frågor sparas inte
• Du kontrollerar all loggning och datalagring

Produktionsöverväganden

Skala Qdrant

För produktionsdriftsättning med stora dokumentsamlingar:

# Run Qdrant with persistent storage and resource limits
docker run -d \
  --name qdrant \
  -p 6333:6333 \
  --memory=4g \
  -v /data/qdrant:/qdrant/storage \
  qdrant/qdrant:latest

För samlingar som överskrider 10 miljoner vektorer bör du överväga Qdrants distribuerade läge med sharding över flera EU-hostade noder.

Strategi för chunking

Den enkla ordräkningsbaserade chunkningen i den här guiden fungerar för de flesta dokument. För bättre resultat med strukturerade dokument:

• Semantisk chunkning: Dela vid styckes- eller avsnittsgränser
• Glidande fönster: Använd överlappande chunks för att undvika att klippa mitt i ett sammanhang
• Metadataberikning: Inkludera avsnittsrubriker, dokumenttitlar och datum i chunkens metadata

Felhantering

Lägg till retry-logik för API-anrop och hantera anslutningsfel till Qdrant:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))
def generate_embeddings_with_retry(texts, client):
    return generate_embeddings(texts, client)

Autentisering

Lägg till API-nyckelautentisering på dina FastAPI-endpoints i produktion:

from fastapi import Depends, Security
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_api_key(api_key: str = Security(api_key_header)):
    if api_key != os.environ.get("APP_API_KEY"):
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

@app.post("/query", dependencies=[Depends(verify_api_key)])
async def query_documents(request: QueryRequest):
    ...

Sammanfattning

Den här guiden visar en komplett RAG-pipeline som upprätthåller GDPR-efterlevnad genom hela flödet:

1. Dokumentinmatning: PyMuPDF extraherar text lokalt, inga molnberoenden
2. Embeddings: Genereras via Juice Factorys EU-API utan datalagring
3. Vektorlagring: Självhostad Qdrant håller all indexerad data under din kontroll
4. LLM-inferens: EU-hostad, tillståndslös bearbetning utan lagring av frågor
5. API-lager: FastAPI ger dig full kontroll över åtkomst, loggning och datahantering

Hela systemet kan driftsättas på EU-infrastruktur utan att data lämnar regionen. Att byta från en icke-kompatibel uppsättning är enkelt — ersätt API:ets bas-URL, peka embeddings mot EU-endpointen och självhosta din vektorlagring.

Relaterade guider

• GDPR-Safe AI Inference — Arkitekturguide för kompatibla AI-applikationer med RAG
• Replacing OpenAI with EU Infrastructure — Migreringsguide för att byta API-leverantör
• n8n + Private AI Automation — Arbetsflödesautomation med EU-hostad inferens
• Cursor AI BYOK Setup — Använd Juice Factory som din BYOK-leverantör i Cursor