Zurück zur Startseite

RAG-System für 1C: MCP-Serverarchitektur

Der Artikel beschreibt die Architektur der MCP-Server documents1c und metadata1c für RAG-Suche in 1C-Dokumentation. Analysierte Schichten: Chunks mit heading_path, Redis-Cache, pgvector-Vectorsuche, RRF, BM25 mit benutzerdefinierter Tokenisierung. Quellcode der Schlüsseldkomponenten bereitgestellt.

RAG-Architektur für die Suche in 1C-Dokumentation
Advertisement 728x90

Mehrstufige RAG-Architektur für die Suche in 1C-Dokumentation auf MCP-Servern

MCP-Server documents1c und metadata1c implementieren ein mehrstufiges RAG-System für präzise Suche in 1C-Dokumentation und -Konfigurationen. Jede Ranking-Ebene adressiert spezifische Herausforderungen: vom Chunk-Kontext bis zur Kombination von Vektor- und Volltextsuche. Das System nutzt PostgreSQL mit pgvector, Redis für Caching und benutzerdefinierte Tokenisierung für 1C-Terminologie.

Warum wir von Vektordatenbanken mit Hierarchien abgerückt sind

Standard-Vektorspeicher wie Qdrant oder Weaviate kämpfen mit der dynamischen Natur der 1C-Dokumentation. Dokumente sind ohne Kontext nicht eigenständig: Ein Artikel mit dem Titel "Einrichtung" erfordert die Angabe des Abschnitts, und ein Objekt wie Referenz.Mitarbeiter benötigt Beschreibungen seiner Attribute und Module.

Die Dokumentation wird häufig aktualisiert – neue BSP-Versionen, Plattformmethoden. Das Neuerstellen der Baumstruktur in einer Vektordatenbank ist kostspielig. Die Lösung: PostgreSQL mit pgvector + heading_path, eingebettet in jeden Chunk. Embeddings werden für heading_path + text generiert, nur der Chunk-Text wird gespeichert.

Google AdInline article slot

Einsparungen: Eine embedding halfvec(1024)-Spalte in der Tabelle article_chunks anstelle einer separaten Vektordatenbank.

Ebene 1: Chunks mit Überschriftenkontext

Das Aufteilen von Markdown in Chunks berücksichtigt die Struktur: YAML wird entfernt, Aufteilung nach Überschriften, Codeblöcke werden nicht unterbrochen, Überlappung von 220 Zeichen.

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)

Beispiel-Chunk:

Google AdInline article slot
Lohn- und Personalmanagement > Lohnabrechnung > Einstellung der Einkommensteuer

Um die Einkommensteuer korrekt zu berechnen, müssen Sie den Steuersatz angeben...

In der Datenbank speichert meta heading_path separat für Filter und Links. Embedding: heading_path + chunk_text.

Ebene 2: Embedding-Cache in Redis

Das Generieren von Embeddings (text-embedding-qwen3-embedding-4b) dauert 50–200 ms. Für wiederholte Abfragen – Redis mit TTL, Schlüssel SHA256 von 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

Der Cache wird automatisch ungültig, wenn sich das Modell ändert.

Google AdInline article slot

Ebene 3: Vektorsuche mit halfvec

Kosinus-Distanz via <=> in pgvector, halfvec(1024)-Typ spart Speicher.

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.

Ebene 4: Zusammenführung der Rankings mit RRF

Vektorsuche ist schwach für präzise Begriffe wie &OnClientOnServer. RRF fusioniert Vektor- und FTS-Listen:

@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)
    # Normalisierung und Rückgabe von top_k

HBK-Boost: Für Abfragen ohne Filter werden 1C-eigene Sprach-Chunks (load_mode='hbk') an den Anfang der Liste gesetzt.

  • Top-HBK-Chunks nach query_emb
  • Duplikate werden ausgeschlossen
  • Vor den Hauptergebnissen platziert

Ebene 5: BM25-Reranking mit 1C-Tokenisierung

Für 50–100 Kandidaten aus RRF wird BM25 mit Gewichtungen (Vektor 0,6, bm25 0,4) angewendet. Effektiv für:

  • Spezifische Begriffe: AllgemeinesModul, Informationsregister, &OnServer
  • Objektnamen: Referenz.Nomenklatur
  • Kurze technische Abfragen

Benutzerdefinierte Tokenisierung:

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"[&#][And-IA-Z][And-YaYoA-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

Teilt CamelCase, snake_case, extrahiert Direktiven und GUIDs.

Wichtige Erkenntnisse

  • heading_path in Chunks bietet Kontext ohne Dokumentenbaum
  • Redis-Cache reduziert Embedding-Latenz von 100+ ms auf 0 ms
  • RRF kombiniert Semantik und exakte Suche ohne Normalisierung
  • BM25 mit 1C-Tokenisierung verstärkt relevante technische Begriffe
  • halfvec(1024) spart Speicher in pgvector bei gleichbleibender Kosinus-Genauigkeit

— Editorial Team

Advertisement 728x90

Weiterlesen