RAG-система на LangChain для 1С: ответы по своей базе знаний

RAG-система — это способ заставить языковую модель отвечать по вашим данным, а не по тому, что она запомнила при обучении. Для 1С это самый частый запрос: «хотим бота, который знает наш регламент, наш справочник номенклатуры, нашу базу вопросов поддержки». И почти всегда следом идёт вторая фраза — «давайте дообучим модель на наших данных». Не надо. В подавляющем большинстве задач нужен RAG, а не дообучение, и собирается связка RAG + LLM на LangChain за вечер. Ниже — рабочий код: индексация, ретривер, цепочка с историей диалога, свой ретривер под номенклатуру и то, о чём почти не пишут — как потом убедиться, что система отвечает правильно.

Почему «дообучить модель на своих данных» — не тот ответ

Дообучение (fine-tuning) меняет поведение модели: стиль, формат, следование инструкции. Оно почти не добавляет ей знаний, зато требует датасета, видеокарты и повторного прогона при каждом изменении данных. Поменялась цена в справочнике — переобучать?

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

Две половины системы: индексация и запрос

Любая RAG-система — это два независимых конвейера.

Индексация (запускается редко, по расписанию или по кнопке):

• взять документы — выгрузку номенклатуры, регламент, базу вопрос-ответ;
• нарезать на чанки;
• превратить каждый чанк в вектор (эмбеддинг);
• сложить векторы в хранилище.

Запрос (на каждое сообщение пользователя):

• превратить вопрос в вектор той же моделью;
• найти ближайшие чанки;
• собрать промпт: системная инструкция + найденный контекст + история диалога + вопрос;
• отдать модели и вернуть ответ.

Дальше разберём обе половины на коде. Про нарезку и про векторы есть отдельные разборы — как резать текст на чанки и семантический поиск на эмбеддингах и FAISS; здесь я беру из них минимум, нужный для сборки целого.

Индексация: документы, из которых что-то получится

Самая недооценённая часть. Качество RAG на 80% определяется тем, что вы положили в базу, а не тем, какую модель взяли.

Возьмём реальный пример — справочник образовательных программ в Markdown, где # это факультет, а ## — направление:

# Государство

## Эффективное государственное управление

Проходной балл бюджет: 300
Минимальный проходной бал по ЕГЭ:
- Русский язык 55/70
- Обществознание 55/70

Наивная нарезка по 500 символов разорвёт это в клочья, и в базу попадёт чанк «Проходной балл бюджет: 300» — сам по себе бессмысленный. Поиск его не найдёт, а если найдёт, модель не поймёт, к чему он относится.

Рабочий приём — резать по заголовкам и вклеивать метаданные в сам текст чанка:

from langchain_core.documents import Document
from langchain_text_splitters import MarkdownHeaderTextSplitter


def generate_documents() -> List[Document]:
    # Открываем файл с программами университета и считываем его содержимое
    with open("rag/programs.md", encoding="utf-8") as f:
        programs_text = f.read()

    # Определяем заголовки, по которым будем разделять текст
    headers_to_split_on = [("#", "department"), ("##", "direction")]
    splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)

    # Разделяем текст на части по заголовкам
    chunks = splitter.split_text(programs_text)

    # Каждый чанк несёт в себе факультет и направление — иначе он не ищется
    documents_list = []
    for chunk in chunks:
        doc = (
            f"Факультет: {chunk.metadata['department']}\n"
            f"Направление: {chunk.metadata['direction']}\n"
            f"Информация:{chunk.page_content}"
        )
        documents_list.append(
            Document(page_content=doc.lower(), metadata=chunk.metadata)
        )

    return documents_list

Два момента, которые тут легко пропустить.

Первое: заголовок попадает и в metadata (для фильтрации), и в page_content (для поиска). Векторный поиск смотрит только на page_content — то, чего в тексте нет, он не найдёт, как бы красиво оно ни лежало в метаданных.

Второе: .lower(). Регистр даёт разные векторы. Если вы приводите документы к нижнему регистру при индексации, то и вопрос обязаны приводить при поиске — иначе получите тихую деградацию качества, которую потом неделю ищете.

Для 1С источником обычно служит выгрузка: номенклатура с характеристиками, регламенты в Markdown, накопленная база вопрос-ответ из поддержки. Последний вариант, кстати, вообще не требует нарезки — об этом ниже.

Векторы и хранилище: минимум для сборки

Нарезанные документы превращаются в векторы и складываются в хранилище. В учебном проекте это FAISS — библиотека, которая живёт прямо в процессе Python и хранит индекс в файле:

faiss_manager = FAISSManager(
    vector_store_name="rag",
    embedding_type="huggingface",
    embedding_kwargs={
        "model_name": "ai-sage/Giga-Embeddings-instruct",
        "model_kwargs": {"trust_remote_code": True, "device": "cpu"},
    },
)

# Один раз собрать базу из документов
# faiss_manager.generate_vector_store(documents=generate_documents())

# В обычном запуске — просто загрузить готовую
faiss_manager.load_vector_store()

ai-sage/Giga-Embeddings-instruct — русскоязычная модель эмбеддингов, работает локально, данные никуда не уходят. Для 1С-контура это часто решающий аргумент. Как устроен сам поиск, чем similarity отличается от mmr, как читать score и выбирать порог — в отдельном разборе про эмбеддинги и FAISS.

Из хранилища делается ретривер — объект, который по строке возвращает документы:

retriever = faiss_manager.vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3},
)

k=3 — сколько чанков подложить в промпт. Меньше — рискуете не дать модели нужный кусок. Больше — размываете внимание модели и платите за токены. Три-пять для начала, дальше подбирается замером.

Ретривер, который помнит разговор

Вот здесь ломается большинство самодельных RAG-ботов. Пользователь спрашивает: «Интересует эффективное государственное управление». Потом: «Сколько баллов по русскому надо набрать?».

Второй вопрос уходит в векторный поиск как есть — и находит балл по русскому во всех программах разом, потому что в самом вопросе нет ни слова про госуправление. Контекст остался в предыдущей реплике, а поиск про неё не знает.

Лечится это отдельным вызовом модели перед поиском: переписать вопрос так, чтобы он был самодостаточным.

# Определяем системный промпт для суммаризации истории чата
summary_system_prompt = (
    "Учитывая историю чата и последний вопрос пользователя, "
    "который может ссылаться на контекст в истории чата, сформулируйте отдельный вопрос, "
    "который можно понять без истории чата. "
    "НЕ отвечайте на вопрос, просто переформулируйте его, если необходимо, "
    "и в противном случае верните его как есть."
)

summary_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", summary_system_prompt),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)

# Ретривер, который сначала переписывает вопрос, а потом ищет
history_aware_retriever = create_history_aware_retriever(
    model, retriever, summary_prompt
)

create_history_aware_retriever делает ровно это: гоняет вопрос через модель с историей, получает «Сколько баллов по русскому нужно для направления „Эффективное государственное управление“?» и уже эту строку отдаёт в векторный поиск.

Цена — лишний вызов LLM на каждое сообщение. Обратите внимание на формулировку «НЕ отвечайте на вопрос»: без неё модель начинает отвечать вместо переписывания, и в поиск улетает готовый ответ вместо запроса.

Сборка цепочки

Дальше собирается вторая половина — промпт ответа и сама цепочка:

# Промпт, который запрещает выдумывать
qa_system_prompt = (
    "Ты AI консультант университета. Отвечай на вопросы Только Строго по источникам информации. "
    "Если в источниках нет ответа на вопрос, скажи: информация отсутствует.\n"
    "Тебе даны следующие источники информации:"
    "\n\n"
    "{context}"
)

qa_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", qa_system_prompt),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)

# Цепочка «сложить найденные документы в {context} и спросить модель»
question_answer_chain = create_stuff_documents_chain(model, qa_prompt)

# Цепочка «найти документы → ответить по ним»
rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

{context} — служебная подстановка: create_stuff_documents_chain сам склеит туда найденные чанки. Стратегия stuff означает «запихнуть всё найденное в один промпт»; она простая и на k=3..5 работает нормально.

Запуск — обычный цикл с накоплением истории:

chat_history = []

while True:
    user_input = input("Введите сообщение: ")
    if user_input.lower() == "q":
        break

    result = rag_chain.invoke({"input": user_input, "chat_history": chat_history})
    print(f"AI: {result['answer']}")
    chat_history.append(HumanMessage(content=user_input))
    chat_history.append(AIMessage(content=result["answer"]))

Всё. Это полная RAG-система примерно на 60 строк.

Инструкция «Отвечай Только Строго по источникам. Если в источниках нет ответа — скажи: информация отсутствует» делает больше для качества, чем смена модели на более дорогую. Без неё модель охотно достроит недостающее из общих знаний, и вы получите уверенный ответ про несуществующий регламент.

Модель за фабрикой, а не в коде

Провайдера LLM меняют чаще, чем кажется: начали на GigaChat, упёрлись в лимиты, ушли на локальную Ollama, для сравнения качества дёрнули OpenAI. Если модель создаётся прямо в цепочке — переезд превращается в правку десятка мест.

class LlmEnum(str, Enum):
    OpenAI = "openai"
    GigaChat = "giga_chat"
    Ollama = "ollama"


class LLMFactory:
    """Фабрика для создания экземпляров языковых моделей (LLM)."""

    @staticmethod
    def create(model_type: LlmEnum, model_name: str, api_key: str = None, ...):
        if model_type == LlmEnum.OpenAI:
            return ChatOpenAI(api_key=api_key, model=model_name)
        elif model_type == LlmEnum.GigaChat:
            return GigaChat(
                credentials=api_key,
                verify_ssl_certs=verify_ssl_certs,
                model=model_name,
                temperature=temperature,
            )
        elif model_type == LlmEnum.Ollama:
            return OllamaLLM(model=model_name)
        raise ValueError(f"Модель LLM '{model_type}' не поддерживается")

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

Когда стандартного ретривера мало: поиск по номенклатуре

Учебная схема ломается на реальном справочнике 1С. Диалог с менеджером выглядит так:

— Нужны вентиляторы
— настольные
— девятьсот
— а телевизоры есть?

Ни одна из этих реплик по отдельности не является поисковым запросом. Нужен свой ретривер — LangChain позволяет унаследоваться от BaseRetriever и подставить его в ту же цепочку:

class ProductRetriever(BaseRetriever):
    llm: Any
    embedding_manager: Any
    chat_history: Any
    reranker: Any

    def _get_relevant_documents(self, query: str, *, run_manager) -> List[Document]:
        # Модель собирает из истории список законченных запросов:
        # ["вентиляторы настольные 900", "телевизоры"]
        products_list = self._summary_query(query=query)
        if len(products_list) == 0:
            products_list.append(query)

        # Ищем по последнему товару широко...
        similarity_search_result = self.embedding_manager.vector_store.similarity_search(
            query=products_list[-1], k=10
        )
        # ...и отжимаем реранкером до действительно релевантных
        return self._filter_relevant_passages(
            query=products_list[-1], documents=similarity_search_result
        )

Логика внутри _summary_query — это промпт с примерами, который учит модель отличать уточнение («девятьсот» → дописать к текущему товару) от нового запроса («а телевизоры есть?» → закрыть предыдущий, начать новый).

Второй приём здесь важнее первого: широкий поиск плюс реранкер. Векторный поиск на номенклатуре вида «Вентилятор настольный, Модель 901» / «Вентилятор оконный, модель 902» различает позиции плохо — они почти одинаковы по смыслу. Поэтому берём с запасом (k=10) и прогоняем через кросс-энкодер, который оценивает пару «запрос — документ» и отсеивает по порогу. Как это выглядит в готовом сервисе — в разборе ИИ-сервиса анализа цен, там гибридный поиск разобран в бою.

Что ломается в проде

Учебный скрипт и сервис — разные вещи. Что приходится дописывать.

Векторная база может не загрузиться. Файла нет, путь переехал, версия FAISS сменилась. Скрипт в этом месте падает; сервис должен подняться с пустым индексом и работать дальше:

def init_db(self):
    try:
        self.vector_store = FAISS.load_local(
            folder_path=self.emb_db_directory,
            index_name=self.emb_db_name,
            embeddings=self.embeddings,
            allow_dangerous_deserialization=True,
        )
        self.retriever = self.vector_store.as_retriever(search_kwargs={"k": 100})
        return True
    except Exception as e:
        error_message(f"Ошибка загрузки базы FAISS: {e}")

    # База не загрузилась — поднимаем пустой индекс нужной размерности
    index = faiss.IndexFlatL2(len(self.embeddings.embed_query("hello world")))
    self.vector_store = FAISS(
        embedding_function=self.embeddings,
        index=index,
        docstore=InMemoryDocstore(),
        index_to_docstore_id={},
    )

Размерность вектора здесь вычисляется честно — прогоном пробной строки через саму модель эмбеддингов. Хардкодить её нельзя: у разных моделей она разная.

История диалога живёт не в переменной. В консольном примере chat_history — это список в памяти. В сервисе история у каждого пользователя своя, переживает перезапуск и хранится в БД:

async def manager(self, user_id: int, chat_type: str, question: str, callback_handler=None) -> str:
    # 1. Ищем контекст в векторной базе
    answer, links = await bot_manager["embedding_qa"].find_examples(
        message=question, callback_handler=callback_handler
    )
    # 2. Спрашиваем модель с этим контекстом и историей
    ai_response = await self.process_request(
        user_id=user_id, chat_type=chat_type,
        question=question, context=answer, links=links,
        callback_handler=callback_handler,
    )
    # 3. Сохраняем обе реплики
    await self.history.save_human_message(user_id=user_id, chat_type=chat_type, content=question)
    await self.history.save_ai_message(user_id=user_id, chat_type=chat_type, content=ai_response)
    return ai_response

Ошибка не должна валить чат. Любое исключение внутри цепочки — это внешний сервис, который отвалился. Пользователю уходит «Извините, произошла ошибка, попробуйте позже», в лог — трассировка.

Промпты — в файлах, а не в коде. read_prompt_from_file("qa_service_system.txt"). Промпт правят в десять раз чаще, чем логику, и правит его обычно не тот, кто писал сервис.

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

for qa in all_qa:
    documents.append(
        Document(
            page_content=qa.question.lower(),
            metadata={"id": qa.id, "answer": qa.answer, "links": qa.links},
        )
    )

Поиск ищет по формулировке вопроса — то есть сравнивает вопрос с вопросом, а не вопрос с ответом. Точность заметно выше, и заодно появляется ссылка на документ-источник, которую можно показать пользователю.

Как понять, что RAG работает

Самый пропускаемый этап. Систему собрали, десять вопросов задали руками, вроде отвечает — выкатили. Через месяц выясняется, что на трети вопросов она отвечает мимо, и никто не знает, стало лучше или хуже после правки промпта.

Мерить это можно и нужно. Схема из трёх частей.

Первое — тестовый набор. Составлять его руками долго, поэтому вопросы генерируются из самой базы знаний. Giskard умеет делать это по типам:

from giskard.rag import generate_testset, KnowledgeBase
from giskard.rag.question_generators import (
    simple_questions,       # простой вопрос по отрывку
    complex_questions,      # перефразированный, усложнённый
    double_questions,       # два вопроса в одном
    distracting_questions,  # с отвлекающим элементом из базы
    situational_questions,  # с контекстом пользователя
)

knowledge_base = KnowledgeBase(data=df)

testset = generate_testset(
    knowledge_base,
    question_generators=[
        simple_questions, complex_questions, double_questions,
        distracting_questions, situational_questions,
    ],
    num_questions=50,
    language="ru",
    agent_description="Служба поддержки университета, отвечает на вопросы студентов строго на русском языке.",
)

testset.save("test-dataset.jsonl")

Отвлекающие и составные вопросы — самое ценное. Простые проходят почти всегда; система разваливается ровно на «Италия прекрасна, но какая столица Франции?».

Генератор, кстати, можно направить на локальную модель — не обязательно платить за OpenAI:

api_base = "http://10.10.1.10:11434"
giskard.llm.set_llm_model("ollama/gemma3:12b-it-qat", disable_structured_output=True, api_base=api_base)
giskard.llm.set_embedding_model("ollama/bge-m3:latest", api_base=api_base)

Второе — прогон и трассировка. Набор заливается в Langfuse как датасет, по нему запускается эксперимент. Каждый прогон видно целиком: какой вопрос пришёл, что нашёл векторный поиск, какой промпт ушёл в модель, что она ответила и сколько это стоило.

Третье — оценка ответов. Сравнивать строки бессмысленно: правильный ответ можно сформулировать сотней способов. Поэтому судьёй ставят другую модель — ей дают вопрос, полученный ответ и эталонный, и просят оценку от 1 до 10:

class EvaluationQuestion(BaseModel):
    """Оценка ответа на вопрос"""
    evaluate: int = Field(description="Оценка ответа числом от 1 до 10")


response = await bot_manager["judge_llm"].send_message(
    messages=messages, schema=EvaluationQuestion, callback_handler=handler
)
score = response.evaluate / 10

handler.trace.score(name="qa_score", value=score)

Оценка пишется в тот же трейс. Дальше правка промпта или смена модели перестают быть вопросом веры: прогнали 50 вопросов до, прогнали после, сравнили среднее.

Судью берите другой моделью, не той, что отвечает — иначе она щедро хвалит собственные ответы.

Как это подключается к 1С

Схема стандартная: RAG живёт отдельным сервисом на FastAPI, 1С обращается к нему обычным HTTP-запросом.

Сервис поднимает векторную базу и модель один раз при старте (в lifespan), а не на каждый запрос — загрузка модели эмбеддингов занимает секунды. Наружу торчит один эндпоинт: принял вопрос и идентификатор пользователя, вернул ответ. Со стороны 1С это HTTPСоединение, JSON в теле и разбор ответа — та же механика, что в любой интеграции с внешним сервисом.

Подробнее про обвязку — три архитектуры внедрения ИИ в 1С и разбор ИИ-агента для 1С на LangGraph, где сервис уже дёргается кнопкой из УТ.

Если нужен не разрозненный набор рецептов, а последовательная сборка такого сервиса — от Poetry и структуры проекта до подключения к типовой конфигурации, — это курс «ChatGPT и нейросети в 1С»: там весь путь разбирается на живых проектах.

Что дальше

RAG собирается быстро, а вот доводится до приемлемого качества долго — и почти всегда правками в индексации, а не в модели. Порядок, в котором стоит крутить ручки:

• Нарезка. Разберитесь, что реально попало в базу — про сплиттеры и метаданные в чанках.
• Поиск. Модель эмбеддингов, k, пороги отсечки — про эмбеддинги, FAISS и score.
• Реранкер. Когда документы похожи друг на друга, без него не обойтись.
• Промпт. Жёсткий запрет выдумывать — самая дешёвая правка с самым большим эффектом.
• Замер. Всё предыдущее бессмысленно, пока вы не измеряете результат.

И только потом — более дорогая модель.