Sistema RAG para 1C: arquitectura de servidor MCP

Arquitectura RAG Multinivel para Búsqueda en Documentación 1C en Servidores MCP

Los servidores MCP documents1c y metadata1c implementan un sistema RAG multinivel para búsqueda precisa en documentación y configuraciones de 1C. Cada capa de clasificación aborda desafíos específicos: desde el contexto de fragmentos hasta la combinación de búsqueda vectorial y de texto completo. El sistema utiliza PostgreSQL con pgvector, Redis para caché y tokenización personalizada para terminología 1C.

Por qué Abandonamos las Bases de Datos Vectoriales con Jerarquías

Los almacenes vectoriales estándar como Qdrant o Weaviate tienen dificultades con la naturaleza dinámica de la documentación de 1C. Los documentos no son autónomos sin contexto: un artículo titulado "Configuración" requiere especificar la sección, y un objeto como Reference.Employees necesita descripciones de sus atributos y módulos.

La documentación se actualiza con frecuencia: nuevos lanzamientos de BSP, métodos de plataforma. Reconstruir el árbol en una base de datos vectorial es costoso. La solución: PostgreSQL con pgvector + heading_path incrustado en cada fragmento. Los embeddings se generan para heading_path + text, solo se almacena el texto del fragmento.

Ahorro: Una columna embedding halfvec(1024) en la tabla article_chunks en lugar de una base de datos vectorial separada.

Capa 1: Fragmentos con Contexto de Encabezado

La división de markdown en fragmentos considera la estructura: se elimina YAML, división por encabezados, los bloques de código no se rompen, superposición de 220 caracteres.

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)

Ejemplo de fragmento:

Nómina y Gestión de RRHH > Cálculo de Nómina > Configuración del Impuesto sobre la Renta Personal

Para calcular correctamente el impuesto sobre la renta personal, debe especificar la tasa impositiva...

En la base de datos, meta almacena heading_path por separado para filtros y enlaces. Embedding: heading_path + chunk_text.

Capa 2: Caché de Embeddings en Redis

Generar embeddings (text-embedding-qwen3-embedding-4b) toma 50–200 ms. Para consultas repetidas: Redis con TTL, clave 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

La caché se invalida automáticamente cuando cambia el modelo.

Capa 3: Búsqueda Vectorial con halfvec

Distancia coseno mediante <=> en pgvector, tipo halfvec(1024) ahorra memoria.

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;

Puntuación = 1 - distancia.

Capa 4: Fusión de Clasificaciones con RRF

La búsqueda vectorial es débil para términos precisos como &OnClientOnServer. RRF fusiona listas vectoriales y 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)
    # Normalización y retorno de top_k

Impulso HBK: Para consultas sin filtro, los fragmentos del lenguaje incorporado de 1C (load_mode='hbk') se agregan al principio de la lista.

• Fragmentos HBK principales por query_emb
• Se excluyen duplicados
• Colocados antes de los resultados principales

Capa 5: Reclasificación BM25 con Tokenización 1C

Para 50–100 candidatos de RRF, se aplica BM25 con pesos (vector 0.6, bm25 0.4). Efectivo para:

• Términos específicos: CommonModule, InformationRegister, &OnServer
• Nombres de objetos: Reference.Nomenclature
• Consultas técnicas cortas

Tokenización personalizada:

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

Divide CamelCase, snake_case, extrae directivas y GUIDs.

Conclusiones Clave

• heading_path en fragmentos proporciona contexto sin un árbol de documentos
• Caché Redis reduce la latencia de embedding de 100+ ms a 0 ms
• RRF combina semántica y búsqueda exacta sin normalización
• BM25 con tokenización 1C impulsa términos técnicos relevantes
• halfvec(1024) ahorra memoria en pgvector manteniendo precisión coseno

— Editorial Team