RAG-система для документации 1С: архитектура MCP-серверов documents1c и metadata1c

Druzd3 часа назад

RAG-система для документации 1С: архитектура MCP-серверов documents1c и metadata1c

Уровень сложностиПростойВремя на прочтение11 минОхват и читатели2.5K1С*ОбзорВведение

Когда разработчик 1С задаёт вопрос AI-ассистенту, ему нужен точный, актуальный ответ со ссылкой на источник — документацию по 1С или описание объекта конфигурации. Именно для этого созданы два MCP-сервера: documents1c (поиск по документации 1С) и metadata1c (поиск по конфигурациям 1С). Оба сервера реализуют многоуровневую RAG-систему с несколькими слоями ранжирования.

В этой статье я разберу каждый слой: почему он появился, как работает и какой код за ним стоит. И на какие грабли наступил при построении системы.

Почему не векторная база с деревом

Первый вопрос, который возникает при проектировании RAG: почему не использовать специализированную векторную базу (Qdrant, Weaviate, Pinecone) с иерархическим деревом документов?

Причина 1: документ 1С не самодостаточен без контекста.

Например статья из документации о конфигурации Зарплата и управление персоналом может называться «Настройка» — а в контексте статьи нет отсылки о чем она. Объект конфигурации Справочник.Сотрудники имеет смысл только в связке с реквизитами, табличными частями и модулями. Дерево в векторной базе не решает эту проблему — оно лишь добавляет навигацию, но не контекст внутри чанка.

Также стоить отметить что дерево в векторной базе хорошо для статичной информации, когда нет постоянных изменений, обновлений информации, а уж тем более конфигураций 1С. Например “Библиотека стандартных подсистем” множество релизов, или описание синтаксиса языка 1С (новые методы, свойства, обработчики постоянно добавляются с выходом платформы).

Решение: хлебные крошки прямо в тексте чанка — heading_path вшивается в начало каждого чанка при индексировании, и модель эмбеддингов сразу видит полный контекст. Они могут быть как перечнем структуры разделов, текстовым ключом для связки, ключ описания объекта метаданных.

Причина 2: стоимость обновления.

Документация по 1С обновляется регулярно. При изменении статьи нужно пересчитать эмбеддинги затронутых чанков. В специализированной векторной базе это дополнительные затраты пересборки дерева. В нашем случае PostgreSQL с расширением pgvector уже используется как основная БД — добавить колонку embedding halfvec(1024) к таблице article_chunks значительно дешевле.

Слой 1: Построение чанков с хлебными крошками

Первый и самый важный слой — правильное разбиение статей на чанки. Плохой чанк испортит всё последующее ранжирование.

Алгоритм split_markdown_into_chunks

def split_markdown_into_chunks(
text: str,
max_len: int = 800,
overlap: int = 220,
include_heading_path: str = "full_path",
) -> list[str]:
"""
Markdown-aware разбиение текста на чанки для RAG.
1. Убирает YAML frontmatter
2. Разбивает по заголовкам (#, ##, ###)
3. Секции > max_len разбиваются по абзацам
4. Блоки кода не разрываются
5. Добавляет overlap между соседними чанками
"""
body = _strip_frontmatter(text)
sections = _split_by_headers(body)
chunks: list[str] = []

for section, heading_path, current_heading in sections:
# heading_path = ["Зарплата", "Начисление", "Настройка НДФЛ"]
context_heading = " > ".join(heading_path) # режим full_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)Ключевое решение — режим full_path: в начало каждого чанка добавляется полная цепочка заголовков. Например:

Зарплата и управление персоналом > Начисление зарплаты > Настройка НДФЛ

Для корректного расчёта НДФЛ необходимо указать ставку налога...Это позволяет модели эмбеддингов понять контекст даже для коротких технических фрагментов.

Индексирование с хлебными крошками

При сохранении чанка в БД heading_path сохраняется отдельно в поле meta — для последующей фильтрации и формирования ссылок:

embed_input = (
f"{chunk_record['heading_path']}\n\n{chunk_text}"
if chunk_record["heading_path"]
else chunk_text
)
chunk_embedding = await embed_text(embed_input)

chunk_obj = ArticleChunk(
article_id=article_id,
chunk_text=chunk_text,
embedding=chunk_embedding,
meta={
"chunk_index": idx,
"total_chunks": total_chunks,
"heading_path": chunk_record["heading_path"],
"heading_level": chunk_record["heading_level"],
},
)Обратите внимание: в эмбеддинг подаётся heading_path + chunk_text, а в БД хранится только chunk_text. Это позволяет при поиске использовать семантику заголовков, не раздувая хранимый текст.

Слой 2: Redis — кеш эмбеддингов запросов

Генерация эмбеддинга через локальную LM Studio модель (text-embedding-qwen3-embedding-4b) занимает 50–200 мс. При повторяющихся запросах (а в RAG-системе агенты часто задают похожие вопросы) это накапливается.

Решение: Redis-кеш эмбеддингов с TTL.

def _cache_key(text: str, model: str, dimensions: int | None) -> str:
"""SHA256 от model:dimensions:text — инвалидируется при смене модели."""
payload = f"{model}:{dimensions}:{text}"
return "embed:" + hashlib.sha256(payload.encode()).hexdigest()

async def cache_get(text: str, model: str, dimensions: int | None) -> list[float] | None:
r = get_redis()
if r is None:
return None
try:
data = await r.get(_cache_key(text, model, dimensions))
return json.loads(data) if data else None
except Exception:
return NoneИнтеграция в embed_text:

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

# Проверяем Redis-кеш перед вызовом API
if _cache_enabled():
cached = await cache_get(input_text, model, dims)
if cached is not None:
return cached # cache hit — 0 мс вместо 100+ мс

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

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

return vectorКлюч кеша включает имя модели и размерность — при смене модели эмбеддингов старые записи автоматически становятся невалидными.

Слой 3: Векторный поиск (pgvector)

Основной поиск — косинусное расстояние через оператор <=> расширения pgvector. Используется тип halfvec для экономии памяти (16-битные float вместо 32-битных):

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 — чем ближе к 1, тем релевантнее чанк.

Слой 4: RRF (Reciprocal Rank Fusion)

Векторный поиск хорошо находит семантически близкие тексты, но плохо работает с точными терминами — например, &НаКлиентеНаСервере или УстановитьПривилегированныйРежим. Для таких запросов нужен полнотекстовый поиск (FTS).

RRF объединяет два ранжированных списка — векторный и FTS — без необходимости нормализовывать их оценки к единой шкале:

@staticmethod
def _rrf_merge(
vector_results: list[dict],
fts_results: list[dict],
top_k: int,
k: int = 60,
) -> list[dict]:
"""
score_rrf = 1/(k + rank_vector) + 1/(k + rank_fts)
k=60
"""
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)

# Нормализация min-max для совместимости со шкалой cosine score
s_min, s_max = min(scores.values()), max(scores.values())
score_range = s_max - s_min if s_max > s_min else 1.0

result = []
for cid in sorted_ids[:top_k]:
entry = dict(items[cid])
entry["score"] = round((scores[cid] - s_min) / score_range, 6)
result.append(entry)
return resultДокумент, найденный обоими методами, получает два слагаемых и поднимается выше. Документ только из одного списка — одно слагаемое.

Подмешивание чанков встроенного языка 1С (HBK-буст)

Если агент вызвал инструмент без фильтра по разделам статей к результатам RRF добавляются топ-чанки из описания встроенного языка 1С (load_mode='hbk'). Это критично: если агент спрашивает «как работает ТаблицаЗначений», он хочет увидеть описание из синтакс-помощника, а не только сводные статьи 1С.

# При поиске без destination подмешиваем HBK-чанки
if destination is None:
hbk_items = self._hbk_boost_search(query, query_emb, hbk_limit, release)
existing_ids = {item["id"] for item in merged}
boost_items = [item for item in hbk_items if item["id"] not in existing_ids]
merged = boost_items[:boost_count] + merged # HBK-чанки в начало

Слой 5: BM25 Reranking (опционально)

После RRF у нас есть 50–100 кандидатов. BM25 reranker уточняет их порядок, учитывая точное совпадение терминов запроса с текстом чанка.

Когда BM25 помогает

BM25 особенно эффективен когда:

• Запрос содержит специфические термины 1С (ОбщийМодуль, РегистрСведений, &НаСервере)
• Запрос содержит точные имена объектов (Справочник.Номенклатура)
• Векторная модель «размывает» смысл коротких технических запросов

Токенизация для 1С

Стандартная токенизация не подходит для технической документации 1С. Реализована специальная:

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]:
"""
Токенизация для русского языка и технической документации 1С.
- GUID и версии платформы как «якорные» токены
- Директивы препроцессора (&НаКлиенте, #Если)
- Разбивка CamelCase, snake_case, dotted.names
- Расширение по словарю сокращений 1С (только для запроса)
"""
text = _norm_text(text) # ё→е, 1c→1с, унификация тире
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 or (len(t) == 1 and not t.isdigit()):
continue
# Разбиваем идентификаторы: ОбщийМодуль.Процедура → ["общиймодуль", "процедура"]
for p in _split_identifier(t):
out.append(p.lower())

return out

Комбинирование векторного и BM25 score

def rerank(self, query: str, candidates: list[dict], top_k: int,
vector_weight: float = 0.6, bm25_weight: float = 0.4) -> list[dict]:
query_tokens = self._tokenize(query, expand_query=True)
bm25_scores = self.bm25.get_scores(query_tokens)

# Нормализация BM25 к [0, 1]
max_bm25 = max(bm25_scores) if len(bm25_scores) > 0 else 1.0
normalized_bm25 = [
min(s / max(max_bm25, 10.0), 1.0) if s > 0 else 0.0
for s in bm25_scores
]

reranked = []
for i, candidate in enumerate(candidates):
vector_score = candidate.get("score", 0.0)
bm25_score = normalized_bm25[i]
# Взвешенное среднее: 60% вектор + 40% BM25
combined = vector_weight * vector_score + bm25_weight * bm25_score
reranked.append({**candidate, "score": combined,
"vector_score": vector_score, "bm25_score": bm25_score})

reranked.sort(key=lambda x: x["score"], reverse=True)
return reranked[:top_k]Веса 1.2 / 1.0 подобраны эмпирически на тестовых запросах. При необходимости настраиваются через переменные окружения VECTOR_SCORE_WEIGHT и BM25_SCORE_WEIGHT.

Слой 6: Cross-encoder реранжинг по эмбеддингам

После BM25 reranking (опционально) применяется более дорогой, но точный cross-encoder — модель (bge-reranker-v2-m3), которая оценивает пару (запрос, чанк) совместно, а не по отдельности.

Когда используется: На слабом железе этот шаг опционален — он добавляет 200–500 мс на запрос.

Когда помогает: cross-encoder видит взаимодействие между запросом и текстом и лучше оценивает релевантность для длинных, многосмысловых запросов.

# Применяется после BM25, к топ-N кандидатам
if use_embed_reranker and self.embed_reranker:
embed_opts = get_embed_rerank_settings(self.db)
if embed_opts["enable_embed_reranker"] and embed_opts["model"]:
results = await self.embed_reranker.rerank(
query=query,
candidates=results[:embed_opts["candidates"]],
top_k=top_k,
model=embed_opts["model"],
base_url=embed_opts["base_url"],
endpoint=embed_opts.get("endpoint", "rerank"),
)

Слой 7: Постобработка через рассуждающую модель

Финальный слой — RagPostprocessor. Он принимает топ-K чанков и формирует структурированный ответ через LLM с рассуждением.

Зачем нужен: чанки — это фрагменты документации. Пользователь хочет готовый ответ, а не список отрывков. Постпроцессор:

• Синтезирует ответ из нескольких чанков
• Выделяет ключевые тезисы (key_points)
• Формирует список источников (sources) со ссылками
• Добавляет предупреждения (warnings) при противоречияхКонтракт ответа:

def normalize_rag_to_contract(semantic_result: dict, postprocess_used: bool = False) -> dict:
"""
Контракт: {
"answer": str,
"key_points": list,
"sources": list,
"warnings": list,
"postprocess_used": bool
}
"""Для длинных ответов используется стратегия Map-Reduce:

• Map: каждый батч чанков обрабатывается отдельно
• Reduce: промежуточные ответы объединяются в финальныйЗачем кешировать результаты постобработки:

Постобработка — самый дорогой шаг, поэтому лучше кешировать результаты запрос-ответ

def build_cache_key(
query: str,
rag_result: dict,
rag_postprocess_model: str,
rag_postprocess_temperature: float,
rag_postprocess_context_window: int,
) -> str:
"""
Детерминированный ключ: SHA256 от (запрос + чанки + параметры модели).
При изменении любого параметра — промах кеша.
"""
payload = {
"query": query.strip(),
"rag_result": _normalize_rag_result(rag_result),
"rag_postprocess_model": rag_postprocess_model.strip(),
"rag_postprocess_temperature": float(rag_postprocess_temperature),
"rag_postprocess_context_window": int(rag_postprocess_context_window),
}
raw = json.dumps(payload, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()Ключ кеша включает параметры модели — при смене модели или температуры старые записи автоматически инвалидируются.

Полная схема RAG-пайплайна

Запрос пользователя/агента
│
▼
┌─────────────────┐
│ Redis-кеш │ ← cache hit → сразу возвращаем эмбеддинг
│ эмбеддингов │
└────────┬────────┘
│ cache miss
▼
┌─────────────────┐
│ LM Studio │ ← локальная модель эмбеддингов
│ embed_text() │
└────────┬────────┘
│ вектор запроса
▼
┌─────────────────────────────────────────────────────┐
│ Параллельный поиск │
│ ┌──────────────────┐ ┌──────────────────────────┐ │
│ │ Векторный поиск │ │ Полнотекстовый поиск │ │
│ │ pgvector <=> │ │ PostgreSQL FTS (tsvector) │ │
│ │ top-120 чанков │ │ top-120 чанков │ │
│ └────────┬─────────┘ └────────────┬─────────────┘ │
│ └──────────┬──────────────┘ │
│ ▼ │
│ ┌──────────────┐ │
│ │ RRF Merge │ + HBK-буст (синтакс) │
│ │ top-120 │ │
│ └──────┬───────┘ │
└─────────────────────┼───────────────────────────────┘
│
▼
┌────────────────┐
│ BM25 Reranker │ ← 50-100 кандидатов → top-K
│ (rank-bm25) │
└───────┬────────┘
│
▼
┌────────────────┐
│ Cross-encoder │ ← опционально, дорого
│ Embed Reranker │
└───────┬────────┘
│
▼
┌────────────────────────┐
│ Кеш постобработки │ ← cache hit → готовый ответ
│ (PostgreSQL) │
└───────────┬────────────┘
│ cache miss
▼
┌────────────────────────┐
│ RagPostprocessor │ ← рассуждающая LLM
│ Map-Reduce │ (локальная модель)
└───────────┬────────────┘
│
▼
{ answer, key_points, sources, warnings }

Итоги

Каждый слой решает конкретную проблему:

СлойПроблемаРешениеХлебные крошкиЧанк без контекста непонятенheading_path в начале чанкаRedis-кеш эмбеддинговПовторные запросы медленныеTTL-кеш по SHA256 от текста+моделиpgvectorСемантический поискhalfvec + cosine distanceRRFТочные термины теряютсяОбъединение векторного + FTSHBK-бустСинтаксис 1С не находитсяПодмешивание чанков из синтакс-помощникаBM25Неточное ранжирование терминовВзвешенное среднее vector+BM25Cross-encoderМногосмысловые запросыСовместная оценка пары (запрос, чанк)ПостобработкаЧанки ≠ готовый ответMap-Reduce через рассуждающую LLMКеш постобработкиLLM медленнаяSHA256 от (запрос + чанки + параметры)Всё это работает на ноутбуке ноутбук MSI Katana HX: Intel Core i5-14450HX (2.40 GHz), ОЗУ 32 ГБ, NVidia GeForce RTX 4060 (8 Гб) Ti Laptop GPU. — локальная модель для эмбеддингов (text-embedding-qwen3-embedding-4b) через LM Studio.

• PostgreSQL v17 в Docker.
• модель для Cross-encoder (bge-reranker-v2-m3) в Docker с отдельным эндпойнтом /v1/reranker.
• Redis для кеша в Docker.
• модель для Постобработки (qwen3.6-plus-preview) через openrouter.ai.В базе:

• разделов ~ 30: загруженные книги, встроенный синтаксис помощник, полное руководство платформы 1С, различные библиотеки (Библиотека стандартных подсистем, Библиотека интернет-поддержки пользователей и.т.п.), разнородная информация с профильных сайтов, различные конфигурации 1С.
• статей ~ 80, 000
• чанков ~ 1 500 000
• информация о кодовой базе различных конфигураций ~ 5 000 000.На вопрос “какой latency?” - мне хватает)

• для documents1c ~ 25 секунд по всему пайплайну, основная боль это постобработка.
• для metadata1c ~ 6 секунд, т.к. там урезанный пайплайн.Ну и не забываем это все крутится на ноутбуке.

Всем удачи при проектировании RAG для 1С!Теги:• rag
• 1с
• 1c
• mcpХабы:• 1С