Главная / Статьи / Помощник для документов RAG с тестируемыми сервисами и чатом на Next.js

Помощник для документов RAG с тестируемыми сервисами и чатом на Next.js

Отделите HTTP-часть от чисто Python-сервисов, индексируйте документы, указывайте местоположение фрагментов и реализуйте интерфейс с разделенными панелями без необходимости изменения механизма поиска.

4200 слов

Помощник с документами, указывающий источники, является продуктом типа RAG, а не оболочкой чат-бота. В этой версии HTTP-маршруты отделены от чистых Python-сервисов, хранятся эмбеддинги для загрузки, а интерфейс чата в Next.js оснащен разделенной панелью для транскрипта и цитат.

Цели

Загружать материалы, индексировать их, отвечать с использованием соответствующих фрагментов и сохранять возможность тестирования основной логики с помощью unit-тестов без запуска сервера.

Структура бэкенда

API-маршруты обрабатывают только HTTP-запросы; сами сервисы отвечают за ввод данных, генерацию эмбеддингов, их загрузку и создание результатов. Такое разделение позволяет использовать pytest для тестирования алгоритмов ранжирования без ASGI.

Структура фронтенда

Next.js в сочетании с Tailwind: чат в разделенных панелях, темы «темно»/«светло», использующие один и тот же API-клиент, а также элементы цитат, позволяющие переходить по глубоким ссылкам на загруженные фрагменты.

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

Конфигурация должна быть явной — имена инструментов генерации эмбеддингов, пути к индексам и идентификаторы моделей должны храниться в переменных окружения, а не быть закодированы как строки.

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

Интегрируйте этот шаг до расширения интерфейса; ошибки получения данных легче найти в среде тонкого клиента.

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)

Сохраняйте конфигурацию явной — имена встраивателей, пути к индексам и идентификаторы моделей должны находиться в переменных окружения, а не быть закодированными как строки.

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

Интегрируйте этот шаг до расширения интерфейса; ошибки получения данных легче найти в среде тонкого клиента.

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
)

Интегрируйте этот шаг до расширения интерфейса; ошибки получения данных легче найти в среде тонкого клиента.

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)

Интегрируйте этот шаг до расширения интерфейса; ошибки получения данных легче найти в среде тонкого клиента.

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

Сохраняйте конфигурацию явной — имена встраивателей, пути к индексам и идентификаторы моделей должны находиться в среде окружения, а не быть жестко закодированными строками.

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

Интегрируйте этот шаг до расширения интерфейса пользователя; ошибки получения данных легче найти при использовании тонкого клиента.

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}

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

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"]}'

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

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

Сохраняйте конфигурацию явной — имена инструментов встраивания, пути к индексам и идентификаторы моделей должны находиться в переменных окружения, а не быть закодированными как строки.

Оценка

Заморозьте двадцать вопросов с предполагаемыми идентификаторами сегментов. Отслеживайте процент точных ответов и наличие цитат. Улучшения интерфейса никогда не могут исправить ошибки в алгоритме выбора топ-к.

Замечания по развертыванию

Если объем данных большой, запускайте API и рабочие процессы отдельно. Проверки работоспособности должны завершаться с ошибкой, когда хранилище векторов недоступно, а не только тогда, когда HTTP-соединение активно.

Заключение

Четкое разделение функций — маршрутов и сервисов, операций получения данных и генерации результатов, интерфейса и API — позволяет превратить демо-версию RAG, созданную за выходные, в что-то, что команды могут тестировать и использовать в работе.

Обновляйте контракты API одновременно с идентификаторами инструментов встраивания.

Для тестирования на единицы разделяйте обработчики HTTP-запросов от чистых сервисов.

Темы интерфейса «темная»/«светлая» не должны влиять на логику поиска данных.

Записывайте идентификаторы операций получения данных вместе с информацией о сообщениях в чате для последующего анализа.

Пересоздавайте индексы с такой частотой, которая соответствует изменчивости исходных данных.

Отклоняйте запросы, если лидирующие сегменты выглядят слабыми.

Версионируйте контракты API рядом с идентификаторами встраивателей.

Для тестирования на единицы отделяйте обработчики HTTP от чистых сервисов.

Темы интерфейса «темная/светлая» не должны изменять логику получения данных.

Записывайте идентификаторы запросов вместе с этапами чата для последующей обработки.