System RAG dla 1C: architektura serwerów MCP

Wielowarstwowa architektura RAG do wyszukiwania w dokumentacji 1C na serwerach MCP

Serwery MCP documents1c i metadata1c implementują wielowarstwowy system RAG do precyzyjnego wyszukiwania w dokumentacji 1C i konfiguracjach. Każda warstwa rankingowa rozwiązuje konkretne problemy: od kontekstu fragmentów po kombinację wyszukiwania wektorowego i pełnotekstowego. System wykorzystuje PostgreSQL z pgvector, Redis do buforowania oraz niestandardową tokenizację dla terminów 1C.

Dlaczego rezygnacja z baz wektorowych z hierarchią

Standardowe magazyny wektorowe, takie jak Qdrant czy Weaviate, nie radzą sobie z dynamiką dokumentacji 1C. Dokumenty nie są samowystarczalne bez kontekstu: artykuł "Konfiguracja" wymaga wskazania sekcji, obiekt Katalog.Pracownicy — opisu atrybutów i modułów.

Dokumentacja jest często aktualizowana — nowe wydania BSP, metody platformy. Przebudowa drzewa w bazie wektorowej jest kosztowna. Rozwiązanie: PostgreSQL z pgvector + heading_path wbudowany w każdy fragment. Embeddingi są generowane dla heading_path + tekst, przechowywany jest tylko tekst fragmentu.

Oszczędność: kolumna embedding halfvec(1024) w tabeli article_chunks zamiast osobnej bazy wektorowej.

Warstwa 1: Fragmenty z kontekstem nagłówków

Podział markdown na fragmenty uwzględnia strukturę: YAML jest usuwany, podział według nagłówków, bloki kodu nie są przerywane, nakładanie 220 znaków.

def split_markdown_into_chunks(
    text: str,
    max_len: int = 800,
    overlap: int = 220,
    include_heading_path: str = "full_path",
) -> list[str]:
    body = _strip_frontmatter(text)
    sections = _split_by_headers(body)
    chunks: list[str] = []

    for section, heading_path, current_heading in sections:
        context_heading = " > ".join(heading_path)
        section_with_context = f"{context_heading}\n\n{section}"

        if len(section_with_context) <= max_len:
            chunks.append(section_with_context)
        else:
            sub_chunks = _split_section_by_paragraphs(section_with_context, max_len)
            chunks.extend(sub_chunks)

    return _add_overlap(chunks, overlap)

Przykład fragmentu:

Płace i zarządzanie personelem > Naliczanie wynagrodzeń > Konfiguracja PIT

Do poprawnego obliczenia PIT należy określić stawkę podatku...

W bazie danych meta przechowuje heading_path osobno dla filtrów i linków. Embedding: heading_path + chunk_text.

Warstwa 2: Bufor embeddingów w Redis

Generowanie embeddingów (text-embedding-qwen3-embedding-4b) zajmuje 50–200 ms. Dla powtarzających się zapytań — Redis z TTL, klucz SHA256 od model:dimensions:text.

async def embed_text(text: str) -> list[float]:
    model = _get_embed_model_name()
    dims = _get_embed_dimensions()

    if _cache_enabled():
        cached = await cache_get(text, model, dims)
        if cached is not None:
            return cached

    response = await to_thread(client.embeddings.create, model=model, input=text)
    vector = response.data[0].embedding

    if _cache_enabled():
        await cache_set(text, model, dims, vector, _cache_ttl())

    return vector

Bufor jest unieważniany automatycznie przy zmianie modelu.

Warstwa 3: Wyszukiwanie wektorowe z halfvec

Odległość cosinusowa przez <=> w pgvector, typ halfvec(1024) oszczędza pamięć.

SELECT
    ac.id, ac.article_id, ac.chunk_text, ac.meta,
    ac.embedding::halfvec(1024) <=> CAST(:query_embedding AS halfvec(1024)) AS distance
FROM article_chunks ac
INNER JOIN articles a ON ac.article_id = a.id
WHERE a.job_id IN (SELECT id FROM jobs WHERE load_mode IN ('hbk', 'docs'))
AND a.destination = :destination
ORDER BY distance
LIMIT :limit;

Score = 1 - distance.

Warstwa 4: Łączenie rankingów RRF

Wyszukiwanie wektorowe jest słabe dla precyzyjnych terminów, takich jak &NaKliencieNaSerwerze. RRF łączy listy wektorowe i FTS:

@staticmethod
def _rrf_merge(vector_results, fts_results, top_k: int, k: int = 60) -> list[dict]:
    scores: dict[int, float] = {}
    items: dict[int, dict] = {}

    for rank, item in enumerate(vector_results, start=1):
        chunk_id = item["id"]
        scores[chunk_id] = scores.get(chunk_id, 0.0) + 1.0 / (k + rank)
        items[chunk_id] = item

    for rank, item in enumerate(fts_results, start=1):
        chunk_id = item["id"]
        scores[chunk_id] = scores.get(chunk_id, 0.0) + 1.0 / (k + rank)
        if chunk_id not in items:
            items[chunk_id] = item

    sorted_ids = sorted(scores, key=lambda cid: scores[cid], reverse=True)
    # Normalizacja i zwrot top_k

HBK-boost: dla zapytań bez filtra dodawane są fragmenty wbudowanego języka 1C (load_mode='hbk') na początek listy.

• Top fragmenty HBK według query_emb
• Duplikaty są wykluczane
• Umieszczane przed głównymi wynikami

Warstwa 5: BM25 reranking z tokenizacją 1C

Dla 50–100 kandydatów z RRF stosowany jest BM25 z wagami (vector 0.6, bm25 0.4). Skuteczny dla:

• Specyficznych terminów: ModułOgólny, RejestrInformacji, &NaSerwerze
• Nazw obiektów: Katalog.Asortyment
• Krótkich zapytań technicznych

Niestandardowa tokenizacja:

GUID_RE = re.compile(r"\\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\\b", re.I)
VERSION_RE = re.compile(r"\\bv?8\\.\\d+(?:\\.\\d+){1,3}\\b", re.I)
DIRECTIVE_RE = re.compile(r"[&#][A-Z][A-Za-z0-9_]*", re.I)

def _tokenize(self, text: str, expand_query: bool = False) -> list[str]:
    text = _norm_text(text)
    out = []

    for g in GUID_RE.findall(text): out.append(g.lower())
    for v in VERSION_RE.findall(text): out.append(v.lower())
    for d in DIRECTIVE_RE.findall(text): out.append(d.lower())

    for t in TOKEN_RE.findall(text):
        if t in RU_STOP: continue
        for p in _split_identifier(t):
            out.append(p.lower())
    return out

Rozbija CamelCase, snake_case, wyodrębnia dyrektywy i GUID.

Co jest ważne

• heading_path w fragmentach zapewnia kontekst bez drzewa dokumentów
• Redis-bufor zmniejsza opóźnienie embeddingów z 100+ ms do 0 ms
• RRF łączy semantykę i precyzyjne wyszukiwanie bez normalizacji
• BM25 z tokenizacją 1C podnosi trafność terminów technicznych
• halfvec(1024) oszczędza pamięć w pgvector przy dokładności cosinus

— Editorial Team