Système RAG pour 1C : architecture du serveur MCP

Architecture RAG Multi-Niveaux pour la Recherche dans la Documentation 1C sur les Serveurs MCP

Les serveurs MCP documents1c et metadata1c implémentent un système RAG multi-niveaux pour une recherche précise dans la documentation et les configurations 1C. Chaque couche de classement aborde des défis spécifiques : du contexte des segments à la combinaison de la recherche vectorielle et en texte intégral. Le système utilise PostgreSQL avec pgvector, Redis pour la mise en cache et une tokenisation personnalisée pour la terminologie 1C.

Pourquoi Nous Avons Abandonné les Bases de Données Vectorielles avec Hiérarchies

Les bases de données vectorielles standard comme Qdrant ou Weaviate peinent avec la nature dynamique de la documentation 1C. Les documents ne sont pas autonomes sans contexte : un article intitulé "Configuration" nécessite de préciser la section, et un objet comme Reference.Employees a besoin des descriptions de ses attributs et modules.

La documentation est mise à jour fréquemment — nouvelles versions BSP, méthodes de plateforme. Reconstruire l'arborescence dans une base de données vectorielle est coûteux. La solution : PostgreSQL avec pgvector + heading_path intégré dans chaque segment. Les embeddings sont générés pour heading_path + text, seul le texte du segment est stocké.

Économies : Une colonne embedding halfvec(1024) dans la table article_chunks au lieu d'une base de données vectorielle séparée.

Niveau 1 : Segments avec Contexte d'En-tête

La division du markdown en segments tient compte de la structure : le YAML est supprimé, division par en-têtes, les blocs de code ne sont pas cassés, chevauchement de 220 caractères.

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)

Exemple de segment :

Paie et Gestion RH > Calcul de la Paie > Configuration de l'Impôt sur le Revenu des Personnes Physiques

Pour calculer correctement l'impôt sur le revenu des personnes physiques, vous devez spécifier le taux d'imposition...

Dans la base de données, meta stocke heading_path séparément pour les filtres et les liens. Embedding : heading_path + chunk_text.

Niveau 2 : Cache d'Embeddings dans Redis

La génération d'embeddings (text-embedding-qwen3-embedding-4b) prend 50–200 ms. Pour les requêtes répétées — Redis avec TTL, clé SHA256 de 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

Le cache est invalidé automatiquement lorsque le modèle change.

Niveau 3 : Recherche Vectorielle avec halfvec

Distance cosinus via <=> dans pgvector, type halfvec(1024) économise la mémoire.

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.

Niveau 4 : Fusion des Classements avec RRF

La recherche vectorielle est faible pour les termes précis comme &OnClientOnServer. RRF fusionne les listes vectorielles et 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)
    # Normalisation et retour top_k

Boost HBK : Pour les requêtes sans filtre, les segments du langage intégré 1C (load_mode='hbk') sont ajoutés au début de la liste.

• Top segments HBK par query_emb
• Les doublons sont exclus
• Placés avant les principaux résultats

Niveau 5 : Reclassement BM25 avec Tokenisation 1C

Pour 50–100 candidats de RRF, BM25 est appliqué avec des pondérations (vectoriel 0.6, bm25 0.4). Efficace pour :

• Termes spécifiques : CommonModule, InformationRegister, &OnServer
• Noms d'objets : Reference.Nomenclature
• Requêtes techniques courtes

Tokenisation personnalisée :

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

Sépare le CamelCase, le snake_case, extrait les directives et GUID.

Points Clés à Retenir

• heading_path dans les segments fournit un contexte sans arborescence de documents
• Cache Redis réduit la latence d'embedding de 100+ ms à 0 ms
• RRF combine sémantique et recherche exacte sans normalisation
• BM25 avec tokenisation 1C booste les termes techniques pertinents
• halfvec(1024) économise la mémoire dans pgvector tout en maintenant la précision cosinus

— Editorial Team