RAG: inteligentne wyszukiwanie w dokumentach dla programistów

Retrieval-Augmented Generation: inteligentna wyszukiwarka dokumentów w firmie

RAG łączy przeszukiwanie wektorowe z generatywnym AI, aby dostarczać precyzyjne odpowiedzi na pytania dotyczące zamkniętych baz dokumentów. System wyodrębnia istotne fragmenty z plików PDF, Excel, kodu i generuje wyjaśnienia wyłącznie na podstawie źródeł, minimalizując halucynacje.

Proces rozpoczyna się od przetwarzania wstępnego: dokumenty są parsowane, dzielone na kawałki uwzględniając typ treści i konwertowane na embeddingi do przechowywania w bazie wektorowej.

Wyodrębnianie tekstu z formatów

System obsługuje parsowanie wielu formatów bez utraty danych. Dla każdego formatu stosowany jest specjalistyczny extractor.

SUPPORTED = {
    '.pdf':  lambda f: ''.join(p.extract_text() or '' for p in PyPDF2.PdfReader(f).pages),
    '.docx': lambda f: '\n'.join(p.text for p in Document(f).paragraphs),
    '.pptx': lambda f: '\n'.join(
        shape.text
        for slide in Presentation(f).slides
        for shape in slide.shapes
        if hasattr(shape, "text") and shape.text.strip()
    ),
    '.xlsx': extract_xlsx,
    '.csv':  extract_csv,
    # .txt, .json, .py, .js itp.
}

Na wyjściu — czysty tekst gotowy do dalszej obróbki.

Inteligentne dzielenie na fragmenty

Tekst nie jest dzielony według stałej długości, ale według semantyki. Najpierw wykrywany jest typ:

def detect_content_type(text):
    if re.search(r'<table|</td>|<tr>', text, re.IGNORECASE):
        return 'table'
    tab_lines = sum(1 for line in text.splitlines() if '\t' in line)
    if tab_lines > max(3, len(text.splitlines()) * 0.4):
        return 'table'
    if re.search(r'function\\s+\\w+\\s*\(|\\bdef\\s+\\w+\\s*\(|\\bclass\\s+\\w+\\b', text):
        return 'code'
    return 'prose'

• Tabele: linie z powtarzającym się nagłówkiem w każdym fragmencie
• Kod: dzielenie według funkcji/klas bez naruszania logiki
• Proza: granice semantyczne oparte na embeddingach zdań

Dla tabel:

def split_table_by_rows(text, chunk_size):
    lines = [l for l in text.splitlines() if l.strip()]
    header = lines[0]
    chunks = []
    cur_lines = [header]
    for line in lines[1:]:
        if sum(len(l) for l in cur_lines) + len(line) >= chunk_size:
            chunks.append('\n'.join(cur_lines))
            cur_lines = [header, line]
        else:
            cur_lines.append(line)
    if cur_lines:
        chunks.append('\n'.join(cur_lines))
    return chunks

Semantyczne dzielenie wykorzystuje podobieństwo kosinusowe sąsiednich zdań z adaptacyjnym progiem (25. percentyl).

Generowanie embeddingów

Fragmenty są wektoryzowane modelem typu text-embedding-3-small (rozmiar 1536). Wektory + metadane są zapisywane w bazie.

def get_embeddings(texts, batch_size=100):
    embeddings = []
    for i in range(0, len(texts), batch_size):
        resp = requests.post(
            "https://openrouter.ai/api/v1/embeddings",
            headers=HEADERS,
            json={
                "model": "openai/text-embedding-3-small",
                "input": texts[i:i + batch_size],
                "encoding_format": "float"
            },
        ).json()
        embeddings.extend(d["embedding"] for d in resp.get("data", []))
    return np.array(embeddings)

Wieloetapowe wyszukiwanie

Zapytanie jest konwertowane na wektor, generowane są jego przeformułowania (3 warianty przez LLM), a następnie wykonane zostaje hybrydowe wyszukiwanie.

Hybrydowe wyszukiwanie: wektory + BM25

• Wektory: podobieństwo kosinusowe w przestrzeni embeddingów
• BM25: tf-idf z parametrami k1=1.5, b=0.75 dla dokładnych dopasowań

Połączenie poprzez RRF:

# Fuzja RRF
for rank, idx in enumerate(sorted_by_vector):
    rrf_scores[idx] += 1.0 / (k_rrf + rank)
for rank, idx in enumerate(sorted_by_bm25):
    rrf_scores[idx] += bm25_normalized[idx] * (1.0 / (k_rrf + rank))

Reranking i filtry

Top-50 kandydatów ocenia LLM, wybierając 5–10 najlepszych. Przymusowe uwzględnienie:

• Cytaty w „kawałkach”
• Liczby/nr (regex: \b(№|N|numer)\\s*\\d+\\b)

quote_matches = re.findall(r'«([^»]+)»|"([^"]+)"', query)
for match in quote_matches:
    quote = (match[0] or match[1]).strip()
    if len(quote) > 5:
        for i, chunk in enumerate(chunks_data):
            if quote in chunk['content']:
                forced_indices.append(i)
                break

Optymalizacje wydajności

Hierarchiczne wyszukiwanie

Najpierw top-dokumenty po anotacjach, potem fragmenty wewnątrz nich.

doc_summaries = [chunks_data_by_doc[doc_id][0]['content'][:500] for doc_id in all_doc_ids]
doc_embs = get_embeddings(doc_summaries)
sims = np.dot(doc_embs_normalized, query_embedding_normalized)
top_doc_ids = [all_doc_ids[i] for i in np.argsort(sims)[::-1][:5]]
chunks_data = [ch for ch in chunks_data if ch['doc_id'] in top_doc_ids]

Iteracyjne wyszukiwanie

Dla skomplikowanych zapytań — kilka rund z precyzją poprzez LLM.

Co jest ważne

• RAG minimalizuje halucynacje, generując odpowiedzi wyłącznie na podstawie istotnych fragmentów
• Hybrydowe wyszukiwanie (wektory + BM25 + RRF) zapewnia dokładność o 20–30% wyższą niż czyste przeszukiwanie wektorowe
• Semantyczne dzielenie zachowuje kontekst tabel i kodu, zmniejszając szum
• Hierarchiczne podejście przyspiesza wyszukiwanie w dużych zbiorach (>10k dokumentów)
• Przymusowe filtry gwarantują zwracanie faktów z liczbami i cytowaniami

— Editorial Team