Asystent dokumentów RAG z testowalnymi usługami i czatem w Next.js

Asystent dokumentów typu RAG z integracją HTTP od usług czystego Pythona, udokumentowane pliki z wskazanymi fragmentami, FastAPI w tle oraz czat z podziałem na panele w Next.js.

Wydawca: REACTAPP.TOPDodane na tej stronie 26 wrz 202618 min czytania4200 słów

Źródła: Building a RAG-Powered Document Assistant from Scratch

Pomocnik dokumentacyjny, który podaje źródła, to produkt typu RAG, a nie tylko skóra dla chatbotu. Ta wersja oddziela ścieżki HTTP od czystych usług w Pythonie, przechowuje embeddingi do pobierania danych oraz prezentuje interfejs czatu w Next.js z podziałem na panel transkrypcji i cytatów.

Cele

Zapisywanie materiałów, ich indeksowanie, odpowiadanie na pytania przy użyciu odpowiednich fragmentów oraz zapewnienie możliwości testowania logiki podstawowej bez uruchamiania serwera.

Budowa backendu

Ścieżki API obsługują wyłącznie HTTP; usługi zajmują się przyjmowaniem danych, tworzeniem embeddingów, pobieraniem informacji oraz generowaniem wyników. Taki podział umożliwia używanie pytest do testowania mechanizmów sortowania bez konieczności korzystania z ASGI.

Budowa frontendu

Next.js w połączeniu z Tailwind: czat w podzielonym panelu, tematy ciemne i jasne korzystające z tego samego klienta API, elementy cytowania umożliwiające szybkie przejście do odpowiednich fragmentów.

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...

Konfiguracja musi być jawna – nazwy narzędzi do tworzenia embeddingów, ścieżki indeksowania oraz identyfikatory modeli powinny znajdować się w pliku env, a nie być ustawione jako ciągi znaków bezpośrednio w kodzie.

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

Zintegruj ten krok przed rozbudową interfejsu użytkownika; błędy przy pobieraniu danych są łatwiejsze do znalezienia w środowisku 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)

Zachowaj jasność konfiguracji — nazwy narzędzi wgrzewania, ścieżki indeksów oraz identyfikatory modeli powinny znajdować się w pliku env, a nie być ustawione jako ciągi znaków bez możliwości zmiany.

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

Zintegruj ten krok przed rozbudową interfejsu użytkownika; błędy przy pobieraniu danych są łatwiejsze do znalezienia w środowisku 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()

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
)

Zintegruj ten krok przed rozbudową interfejsu użytkownika; błędy przy pobieraniu danych są łatwiejsze do znalezienia w środowisku 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))])
)

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)

Zintegruj ten krok przed rozbudową interfejsu użytkownika; błędy przy pobieraniu danych są łatwiejsze do znalezienia w środowisku thin client.

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

Niech konfiguracja będzie jawna – nazwy embederów, ścieżki indeksu oraz identyfikatory modeli powinny znajdować się w środowisku, a nie w ciągach tekstowych zaimplementowanych bezpośrednio.

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

Zintegruj ten krok przed rozbudową interfejsu użytkownika; błędy przy pobieraniu danych są łatwiejsze do wykrycia w środowisku 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}

Niech konfiguracja będzie jawna – nazwy embederów, ścieżki indeksu oraz identyfikatory modeli powinny znajdować się w środowisku, a nie w ciągach tekstowych zaimplementowanych bezpośrednio.

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

Zintegruj ten krok przed rozbudową interfejsu użytkownika; błędy przy pobieraniu danych są łatwiejsze do wykrycia w środowisku 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"]}'

Niech konfiguracja będzie jawna – nazwy embederów, ścieżki indeksu oraz identyfikatory modeli powinny znajdować się w środowisku, a nie w ciągach tekstowych zaimplementowanych bezpośrednio.

# 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"

Zintegruj ten krok przed rozbudową interfejsu użytkownika; błędy przy pobieraniu danych są łatwiejsze do wykrycia w środowisku 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"

Zachowaj wyraźność konfiguracji — nazwy narzędzi do wstawiania treści, ścieżki indeksów oraz identyfikatory modeli powinny znajdować się w środowisku, a nie być ustawione jako ciągi znaków bezpośrednio w kodzie.

Ocena

Zamroź dwadzieścia pytań wraz z oczekiwanymi identyfikatorami segmentów. Śledź wskaźnik trafień oraz obecność cytatów. Poprawki w interfejsie użytkownika nigdy nie naprawią błędnego wyboru najważniejszych wyników.

Uwagi dotyczące wdrożenia

Uruchamiaj API oraz procesory oddzielnie, jeśli praca z danymi jest obciążająca. Sprawdzania stanu powinny zawieść się, gdy magazyn wektorów jest niedostępny, a nie tylko wtedy, gdy połączenie HTTP działa poprawnie.

Zakończenie

Czyste rozgraniczenia — trasy vs usługi, pobieranie danych vs ich generowanie, interfejs użytkownika vs API — przekształcają demonstrację RAG na weekend w coś, co zespoły mogą testować i obsługiwać na bieżąco.

Zaktualizuj umowy API wraz z identyfikatorami narzędzi do wstawiania treści.

Dla testów jednostkowych oddziel przetwarzacze HTTP od czystych usług.

Tematy interfejsu w trybie ciemnym i jasnym nie mogą wpływać na logikę wyszukiwania danych.

Zapisuj identyfikatory pobrań wraz z kolejnościami rozmów, aby ułatwić późniejszą analizę.

Odbuduj indeksy w tempie odpowiadającym zmianom w źródle danych.

Odrzuć prośby, gdy najważniejsze segmenty wydają się słabe.

Zaktualizuj umowy API obok identyfikatorów embederów.

Rozdziel przetwarzacze HTTP od czystych usług w celu testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą oddzielać logiki pobierania danych.

Zapisz identyfikatory pobrań wraz z kolejnościami rozmów do dalszej analizy.

Odbuduj indeksy w tempie odpowiadającym zmianom w źródle danych.

Odrzuć prośby, gdy najważniejsze segmenty wydają się słabe.

Zaktualizuj umowy API obok identyfikatorów embederów.

Rozdziel przetwarzacze HTTP od czystych usług w celu testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą oddzielać logiki pobierania danych.

Zapisz identyfikatory pobrań wraz z kolejnościami rozmów do dalszej analizy.

Odbuduj indeksy w tempie odpowiadającym zmianom w źródle danych.

Odrzuć prośby, gdy najważniejsze segmenty wydają się słabe.

Zaktualizuj umowy API obok identyfikatorów embederów.

Rozdziel przetwarzacze HTTP od czystych usług w celu testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą rozgałęziać logiki pobierania danych.

Zapisuj identyfikatory pobrań wraz z kolejnościami rozmów w celu późniejszej analizy.

Odbudowuj indeksy zgodnie z tempem zmian w źródłach danych.

Odrzucaj prośby, gdy najważniejsze segmenty wydają się słabe.

wersjonuj umowy API obok identyfikatorów embederów.

Rozdzielaj obsługę HTTP od czystych usług w celu testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą rozgałęziać logiki pobierania danych.

Zapisuj identyfikatory pobrań wraz z kolejnościami rozmów w celu późniejszej analizy.

Odbudowuj indeksy zgodnie z tempem zmian w źródłach danych.

Odrzucaj prośby, gdy najważniejsze segmenty wydają się słabe.

wersjonuj umowy API obok identyfikatorów embederów.

Rozdzielaj obsługę HTTP od czystych usług w celu testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą rozgałęziać logiki pobierania danych.

Zapisuj identyfikatory pobrań wraz z kolejnościami rozmów w celu późniejszej analizy.

Odbudowuj indeksy zgodnie z tempem zmian w źródłach danych.

Odrzucaj prośby, gdy najważniejsze segmenty wydają się słabe.

Wersje umów API obok identyfikatorów embederów.

Rozdzielaj handlery HTTP od czystych usług w celu przeprowadzania testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą oddzielać logiki pobierania danych.

Zapisuj identyfikatory pobrań wraz z kolejnościami rozmów w celu późniejszej analizy.

Odbudowuj indeksy zgodnie z tempem zmian w źródłach danych.

Odrzucaj prośby, gdy najważniejsze segmenty wydają się słabe.

Wersje umów API obok identyfikatorów embederów.

Rozdzielaj handlery HTTP od czystych usług w celu przeprowadzania testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą oddzielać logiki pobierania danych.

Zapisuj identyfikatory pobrań wraz z kolejnościami rozmów w celu późniejszej analizy.

Odbudowuj indeksy zgodnie z tempem zmian w źródłach danych.

Odrzucaj prośby, gdy najważniejsze segmenty wydają się słabe.

Wersje umów API obok identyfikatorów embederów.

Rozdzielaj handlery HTTP od czystych usług w celu przeprowadzania testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą oddzielać logiki pobierania danych.

Zapisuj identyfikatory pobrań wraz z kolejnościami rozmów w celu późniejszej analizy.

Odbuduj indeksy w tempie odpowiadającym zmianom w źródle danych.

Odrzuć prośby, gdy najważniejsze segmenty wydają się słabe.

Zaktualizuj umowy API obok identyfikatorów embederów.

Rozdziel przetwarzacze HTTP od czystych usług w celu testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą oddzielać logiki pobierania danych.

Zapisz identyfikatory pobrań wraz z kolejnościami rozmów do dalszej analizy.

Odbuduj indeksy w tempie odpowiadającym zmianom w źródle danych.

Odrzuć prośby, gdy najważniejsze segmenty wydają się słabe.

Zaktualizuj umowy API obok identyfikatorów embederów.

Rozdziel przetwarzacze HTTP od czystych usług w celu testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą oddzielać logiki pobierania danych.

Zapisz identyfikatory pobrań wraz z kolejnościami rozmów do dalszej analizy.

Odbuduj indeksy w tempie odpowiadającym zmianom w źródle danych.

Odrzuć prośby, gdy najważniejsze segmenty wydają się słabe.

Zaktualizuj umowy API obok identyfikatorów embederów.

Rozdziel przetwarzacze HTTP od czystych usług w celu testów jednostkowych.

Tematy interfejsu ciemnego/jasnego nie mogą modyfikować logiki pobierania danych.

Zapisuj identyfikatory pobrań zgodnie z kolejnością rozmów w czacie w celu późniejszej analizy.

Odbudowuj indeksy w tempie odpowiadającym zmianom w źródle danych.

Odrzuć operację, jeśli najważniejsze segmenty wydają się słabe.

Zachowuj pliki konfiguracyjne, osoby odpowiedzialne oraz notatki o odwróceniu zmian obok procesów pobierania i wdrażania zmian.

Zachowaj dane referencyjne, osoby odpowiedzialne oraz notatki dotyczące odwrócenia zmian obok procesów pobierania i wdrażania zmian.

Literatura pokrewna

Stwórz assistenta do rozmów w Wikipedii z użyciem LlamaIndex i ReAct — Indeksuj strony Wikipedii na żądanie, wstawiaj fragmenty tekstu za pomocą małego kodera i umożliwij agencie ReAct wywoływanie narzędzia do przetwarzania zapytań, aby odpowiedzi Chainlit były oparte na uzyskanych dowodach.
Projektowanie SaaS do rozmów z AI przy użyciu Next.js, FastAPI, Credits i SSE — Przejrzyj architekturę rozmów z AI typu full-stack, w której wykorzystano bramkarza FastAPI, rejestr kredytów, ograniczenia szybkości działania Redis oraz strumieniowanie SSE, a także wskazane luki wymagające naprawy.

PDF z pytaniami i odpowiedziami o charakterze semantycznym z Document Intelligence, embeddingami oraz Redis — LangGraph wyodrębnia pliki PDF z harmonogramem opłat, dzieli je na fragmenty i integruje z Azure OpenAI, a także odpowiada na pytania dotyczące opłat w języku naturalnym za pomocą lokalnego indeksu wektorowego Redis.

← Wszystkie artykuły Wyślij brief