Главная / Статьи / Ищите руководства, а не вставляйте их: небольшой путь RAG с цитированием в LangChain

langchainragretrievalvector-searchpython

Ищите руководства, а не вставляйте их: небольшой путь RAG с цитированием в LangChain

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

2886 слов

Ищите в руководствах по эксплуатации вместо того, чтобы заполнять запрос

На этом этапе приложение службы поддержки разветвляется в зависимости от типа заявки. Для базовых заявок загружаются каталог услуг, недавние инциденты и последние развертывания одновременно, а для заявок на доступ используется другой маршрут. Однако по-прежнему отсутствуют подробные инструкции, лежащие в основе рекомендаций.

Вы можете вставить все руководства по эксплуатации в запрос. Для трех небольших файлов это сработает, но при увеличении объема коллекции такой подход перестает быть эффективным. Большие запросы стоят дороже, обрабатываются медленнее и затрудняют определение того, какой фрагмент текста послужил основанием для ответа. Кроме того, они могут передавать модели документы, к которым текущему пользователю никогда не было разрешено иметь доступ. Решение — поиск.

Создаваемый здесь путь состоит из двух этапов. Во время индексации руководства в формате Markdown превращаются в объекты типа Document с исходными метаданными, после чего эти объекты делятся на более мелкие фрагменты:

indexing time
Markdown runbooks
  -> Document objects with source metadata
  -> smaller chunks
  -> embeddings
  -> vector store

При запросе система классификации билетов формирует целенаправленный запрос к механизму поиска:

request time
ticket classification
  -> focused search query
  -> retriever
  -> a few runbook passages with source IDs
  -> recommendation chain
  -> source-ID check

Люди часто называют этот процесс двухэтапным генерированием с усилением на основе поиска, или двухэтапным RAG. Название имеет меньшее значение, чем порядок действий: сначала происходит поиск, а затем генератор получает только те фрагменты текста, которые были найдены. Модель никогда не решает, следует ли проводить поиск, и она не может выбирать произвольный источник данных. Агент не требуется.

Пример тикета, используемый на протяжении всего текста

Приведённый пример — это тикет с описанием проблемы: запросы на оплату в ЕС перестали работать после обновления в 14:05, а логи checkout-api содержат сообщение «database connection refused». Его структурированная классификация выглядит следующим образом:

{
  "ticket_id": "INC-2048",
  "summary": "EU checkout fails after a deployment.",
  "category": "database",
  "severity": "SEV2",
  "service": "checkout-api",
  "confidence": 0.91,
  "evidence": ["database connection refused"],
  "missing_information": ["Affected request percentage"]
}

Хороший поиск возвращает руководство по работе с базой данных для процесса оплаты, а не руководство по задержкам платежей или руководство по доступу к идентификаторам. Он также должен содержать достаточно метаданных, чтобы ответить на четыре вопроса: из какого файла взят текст, кому принадлежит этот сервис, где в файле начинается соответствующий фрагмент и какие идентификаторы источника были использованы в рекомендации. С текстом без метаданных источника трудно работать.

Шесть этапов извлечения информации

Каждый этап требует определения перед его использованием.

Документ. Под Документом следует понимать блок текста с привязанными к нему метками. Текст может находиться в файле Markdown, PDF, веб-странице, вики или даже в строке таблицы, а метки могут содержать идентификатор источника, название, сервис, путь, владельца, версию или информацию о том, кто может его просматривать.

Чанкирование. Документы делятся на отдельные фрагменты, которые можно находить независимо друг от друга. Один руководство может описывать одновременно процедуры диагностики, восстановления и отката, однако при запросе на учетные данные базы данных нет необходимости каждый раз возвращать материалы для отката.

Встраивание. Модель встраивания преобразует текст в список чисел, при этом тексты с похожим значением должны оказываться рядом друг с другом согласно выбранному методу сравнения. Это не проверка фактов — похожий текст может быть неверным, устаревшим или недоступным.

Хранилище векторов. Хранилище векторов хранит векторы документов и находит близкие к ним. Некоторые из них хранятся только в памяти, другие — в базе данных или на облачном сервисе. Место хранения векторов не заменяет механизмы контроля доступа.

Поиск. Если передать поисковой системе текст запроса в свободной форме, она вернёт объекты Document. Это более широкое понятие, чем векторный поиск: за одинаковым интерфейсом может скрываться сопоставление ключевых слов, хранилище векторов, гибридная поисковая служба или собственный API поиска.

Генерация. На последнем этапе берутся выбранные фрагменты, и генерируется текстовый ответ. Что бы она ни написала, она не может притворяться, что поиск вернул то, чего на самом деле не было.

Поиск — это не память

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

Структура проекта и настройка

Снимок на этом этапе имеет следующую структуру:

langchain-helpdesk/
├── app.py
├── chains.py
├── embeddings.py
├── retrieval.py
├── schemas.py
├── workflow.py
├── runbooks/
│   ├── checkout-database.md
│   ├── identity-access.md
│   └── payment-latency.md
└── tests/
    └── test_retrieval.py

Установите пакеты:

python -m pip install -U \
  langchain \
  langchain-openai \
  langchain-text-splitters \
  pydantic \
  python-dotenv \
  pytest

Приложение вызывает встроенные компоненты поставщика. В тестах используется небольшой локальный заменитель, поэтому вызовы поставщика не происходят.

Шаг 1: написание учетных книг операций с возможностью их идентификации

Учетная книга операций оформления заказа начинается с строки с заголовком:

# Checkout database connection failures

За этим следуют сервис, владелец и кузов:

Service: checkout-api
Owner: commerce-platformUse this runbook when checkout-api reports database connection refused,
pool exhaustion or authentication failures.## First checks1. Compare the active database host, port and database name with the
   previous successful deployment.
2. Confirm that the application secret points to the expected
   credential version.
3. Test network reachability from the application environment without
   printing the credential.
4. Compare connection-pool limits with the database connection limit.

Заголовок, услуга и владелец представляют собой обычный текст, причем загружающий модуль также копирует выбранные значения в метаданные. В корпоративной системе никогда не следует позволять заголовку определять права доступа. Метки доступа должны поступать из надежной системы и проверяться при каждом запросе к данным.

Шаг 2: загрузка каждого файла в документ

Добавьте функцию load_runbooks в файл retrieval.py. Для её работы сначала потребуется вспомогательная функция для получения путей:

from pathlib import Path

А теперь сам загрузчик:

from langchain_core.documents import Document
def load_runbooks(runbook_dir: Path) -> list[Document]:
    documents: list[Document] = []    for path in sorted(runbook_dir.glob("*.md")):
        text = path.read_text(encoding="utf-8")
        title = text.splitlines()[0].removeprefix("# ").strip()
        service_line = next(
            line for line in text.splitlines()
            if line.startswith("Service:")
        )
        service = service_line.split(":", 1)[1].strip()        documents.append(
            Document(
                page_content=text,
                metadata={
                    "source_id": path.stem,
                    "title": title,
                    "service": service,
                    "path": str(path),
                },
            )
        )    if not documents:
        raise ValueError(f"No Markdown runbooks found in {runbook_dir}")    return documents

Пошаговое описание:

  • glob("*.md") выбирает файлы формата Markdown в одном известном каталоге.
  • sorted(...) обеспечивает воспроизводимость порядка локального индекса. Качество поиска не должно зависеть от порядка файлов в файловой системе, но воспроизводимый входной материал упрощает отладку.
  • Файл декодируется в формате UTF-8.
  • Извлечение заголовка и информации о сервисе делается намеренно примитивным, поскольку примерные файлы следуют одной фиксированной структуре. Реальный загрузчик должен проверять метаданные и сообщать о повреждённых файлах вместо того, чтобы делать предположения.
  • page_content хранит текст, доступный для поиска.
  • metadata содержит поля, которые должны присутствовать во всех последующих фрагментах.
  • Финальная проверка предотвращает ситуацию, когда пустой индекс выглядит как успешный результат.

Шаг 3: разделение документов

Используйте RecursiveCharacterTextSplitter:

from langchain_text_splitters import RecursiveCharacterTextSplitter

def split_runbooks(documents: list[Document]) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=420,
        chunk_overlap=60,
        add_start_index=True,
    )
    return splitter.split_documents(documents)

chunk_size=420 означает, что куски будут содержать примерно 420 символов, что подходит для этого небольшого примера. chunk_overlap=60 заставляет небольшой участок текста повторяться между соседними кусками, что позволяет сохранять предложение рядом с его объяснением, если граница попадает посередине. Большее перекрытие не всегда лучше: это приводит к большему объёму хранимого текста, увеличению нагрузки на обработку векторных представлений и большему количеству почти одинаковых результатов.

При использовании параметра add_start_index=True каждый кусок записывает смещение символов, в котором он начинается, что позволяет возвращаться к исходному файлу. Функция split_documents возвращает новые объекты типа Document, причём каждый кусок наследует метаданные своего родителя.

Выбор размера куска

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

Шаг 4: создание механизма поиска в векторном хранилище

Чтобы пример оставался простым, в приложении используется InMemoryVectorStore:

from langchain_core.vectorstores import InMemoryVectorStore

def build_vector_retriever(runbook_dir: Path, embeddings, *, k: int = 3):
    chunks = split_runbooks(load_runbooks(runbook_dir))
    store = InMemoryVectorStore.from_documents(
        chunks,
        embedding=embeddings,
    )
    return store.as_retriever(search_kwargs={"k": k})

from_documents встраивает и хранит фрагменты, а as_retriever предоставляет стандартный интерфейс поиска. Параметр k=3 запрашивает три документа. Это не означает, что все три являются релевантными, и не гарантирует наличия среди них нужного отрывка.

Приложение создаёт встраиваемые представления поставщиков:

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model=os.environ["EMBEDDING_MODEL"]
)
retriever = build_vector_retriever(
    Path(__file__).parent / "runbooks",
    embeddings,
)

Доступные модели и их названия зависят от учетной записи вашего поставщика, поэтому сохраняйте название модели в конфигурации. Кроме того, InMemoryVectorStore теряет свой индекс при завершении процесса. Это подходит для учебных примеров и небольших тестов, но это не индекс для совместного использования в производственных условиях.

Шаг 5: сформировать целенаправленный запрос на основе надежных полей

Не используйте для поиска огромный набор данных всего состояния приложения. Выберите необходимые поля для классификации:

def build_query(classification: TicketClassification) -> str:
    values = [
        classification.service or "unknown service",
        classification.category,
        classification.summary,
        *classification.evidence,
    ]
    return " | ".join(values)

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

checkout-api | database | EU checkout fails after a deployment. |
database connection refused

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

Шаг 6: передача метаданных в процессе конвертации

В результате работы инструмента возвращаются объекты Document, каждый из которых должен превратиться в текстовый фрагмент с указанием типа:

class RetrievedPassage(BaseModel):
    source_id: str
    title: str
    service: str
    start_index: int
    text: str

Затем сопоставьте с ним Document:

def to_passage(document: Document) -> RetrievedPassage:
    return RetrievedPassage(
        source_id=str(document.metadata["source_id"]),
        title=str(document.metadata["title"]),
        service=str(document.metadata["service"]),
        start_index=int(document.metadata.get("start_index", 0)),
        text=document.page_content.strip(),
    )

Обязательные поля обозначаются квадратными скобками, поэтому фрагмент без source_id, title или service сразу вызывает ошибку. Для start_index задаётся значение по умолчанию, так как фрагменты, сохранённые ранее, могут его не содержать, и при производственной миграции следует решить, является ли такой запасной вариант приемлемым.

Шаг 7: создание узкоспециализированного входных данных для рекомендаций

Функция, предназначенная для работы с моделью, получает следующий контракт:

class RecommendationInput(BaseModel):
    ticket_id: str
    ticket_summary: str
    category: Category
    severity: Severity
    service: str | None
    evidence: list[str]
    passages: list[RetrievedPassage] = Field(
        min_length=1,
        max_length=4,
    )

Когда никакие результаты не возвращаются, рекомендаций не существует:

def prepare_recommendation_input(
    classification: TicketClassification,
    passages: list[RetrievedPassage],
) -> RecommendationInput:
    if not passages:
        raise ValueError("At least one runbook passage is required")

    return RecommendationInput(
        ticket_id=classification.ticket_id,
        ticket_summary=classification.summary,
        category=classification.category,
        severity=classification.severity,
        service=classification.service,
        evidence=classification.evidence,
        passages=passages[:4],
    )

Если поиск ничего не находит, модель не должна импровизировать, опираясь только на общие знания. Перенаправьте заявку к конкретному человеку или используйте альтернативный вариант, который был заранее одобрен.

Шаг 8: запрашивайте идентификаторы источников в структурированном выводе

Схема результата включает ссылки на источники:

class GroundedRecommendation(BaseModel):
    explanation: str = Field(min_length=20, max_length=500)
    next_checks: list[str] = Field(min_length=1, max_length=4)
    source_ids: list[str] = Field(min_length=1, max_length=4)
    needs_human_review: bool

В запросе написано:

Use only the supplied ticket fields and runbook passages.
Every recommended check must be supported by at least one supplied passage.
Return the source_id values you used.
Do not claim that a check or production change was performed.

Инструкции к промпту помогают, но проверки на уровне приложения по-прежнему важны.

Шаг 9: отклонять цитаты, которые не были возвращены при поиске

После того как автор выполнил задание, сравните его идентификаторы источников с полученными отрывками:

available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available

if unknown:
    raise ValueError(
        "Recommendation cited sources that retrieval did not return: "
        + ", ".join(sorted(unknown))
    )

Это позволяет обнаружить поддельный идентификатор источника, но не более того. Модель может сослаться на документ, который действительно был возвращен, но всё равно ошибаться в его интерпретации, поэтому оценка и проверка человеком по-прежнему необходимы.

Шаг 10: подключение исправленного рабочего процесса

Код, используемый во время запроса, прост:

def run_retrieval_workflow(
    classification: TicketClassification,
    retriever,
    recommendation_chain,
) -> RetrievalResult:
    query = build_query(classification)
    passages = retrieve_passages(retriever, query)
    values = prepare_recommendation_input(classification, passages)
    recommendation = recommendation_chain.invoke(values)

    available = {passage.source_id for passage in passages}
    unknown = set(recommendation.source_ids) - available
    if unknown:
        raise ValueError("Recommendation cited an unavailable source")    return RetrievalResult(
        query=query,
        recommendation=recommendation,
        passages=passages,
    )

Здесь нет скрытого цикла. Поиск выполняется один раз, обработка текста — один раз, а проверка исходного кода — тоже один раз.

Результат

Результат выглядит так:

{
  "query": "checkout-api | database | EU checkout fails after a deployment. | database connection refused",
  "recommendation": {
    "explanation": "The error matches the checkout database connection runbook.",
    "next_checks": [
      "Compare the active database host, port and database name with the previous successful deployment.",
      "Confirm that the application secret points to the expected credential version."
    ],
    "source_ids": ["checkout-database"],
    "needs_human_review": false
  },
  "passages": [
    {
      "source_id": "checkout-database",
      "title": "Checkout database connection failures",
      "service": "checkout-api",
      "start_index": 0
    }
  ]
}

API может преобразовать каждый идентификатор источника в ссылку, но только после проверки того, что текущий вызывающий пользователь имеет право её открыть.

Тестирование получения данных без оплаты за эмбеддинги

В комплекте идёт пример KeywordEmbeddings — небольшой тестовый заменитель, использующий фиксированный словарь:

VOCABULARY = (
    "database",
    "connection",
    "checkout",
    "deployment",
    "credential",
    "pool",
    "payment",
    "latency",
    "timeout",
    "identity",
    "access",
    "role",
    "claim",
)

Не путайте его с полноценным встраиваемым инструментом. Его единственная цель — обеспечить быстрые, офлайн-тесты с предсказуемым результатом. Первый тест проверяет действующий тикет:

passages = retrieve_passages(
    retriever,
    build_query(classification()),
)

assert passages[0].source_id == "checkout-database"
assert passages[0].service == "checkout-api"
assert passages[0].start_index >= 0

Вторые показатели были зафиксированы в двух из трех тестовых запросов:

score = hit_at_k(
    retriever,
    [
        ("checkout database connection refused", "checkout-database"),
        ("payment provider latency timeout", "payment-latency"),
        ("identity role claim access denied", "identity-access"),
    ],
)

assert score == 1.0

Два примера вызывают небольшой вопрос: находится ли ожидаемый источник среди двух лучших результатов? Настоящий набор тестов требует множества вопросов, сложных отрицательных примеров, похожих документов и текущих правил доступа. Идеальный результат на трех примерах доказывает лишь то, что эти три пройдены.

В третьем тесте используется фальшивая рекомендация с указанием not-returned, и рабочий процесс должен её отклонить. Запустите всё с:

pytest -q

Ожидаемый результат:

3 passed

Названия типов сбоев при поиске

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

  • Сбой индексации: файл не удалось прочитать, отсутствовали метаданные или с запросом на генерацию эмбеддингов возникла ошибка. Никогда не публикуйте неполностью сформированный индекс как готовый.
  • Сбой актуальности: изменились правила, но поисковая система продолжает использовать старые инструкции. Фиксируйте версии и временные метки, а также определяйте, как изменения и удаления влияют на индекс.
  • Сбой разрешений: индекс показал текст, на чтение которого у текущего пользователя нет прав. Применяйте фильтры доступа до того, как текст попадет к модели, ведь скрытие ссылки позже слишком поздно.
  • Сбой ранжирования: правильный документ индексирован, но отсутствует среди результатов поиска. Измените запрос, фильтры, способ разбиения на части, генерацию эмбеддингов или алгоритм ранжирования, затем снова проведите измерения.
  • Сбой генерации: получение текста прошло успешно, но модель пропустила или неправильно интерпретировала отрывок. Необходимо оценить оба этапа отдельно, чтобы выяснить, в каком из них возникла проблема.
  • Сбой цитирования: ответ содержит ссылку на источник, который не существует или не подтверждает утверждение. Проверка идентификаторов по возвращенному набору позволяет обнаружить только первую проблему.

    • Полученный текст недостоверен

    Руководства по работе содержат ошибки, а иногда в них присутствуют строки, похожие на команды для модели. Считайте полученные отрывки цитируемым источником материала, а не правилами приложения с более высоким приоритетом. Обеспечьте стабильное поведение модели с помощью системного промпта, ограничьте полученный текст и ограничьте использование инструментов в коде приложения. Ни одно руководство не должно иметь возможности авторизовать себя для запуска инструмента, предназначенного для записи текста.

    Когда векторный поиск — неподходящий инструмент

    Для некоторых поисков не требуются векторы. Идентификатор билета может быть найден с помощью обычного запроса к базе данных, а известное название сервиса напрямую соответствует записи в каталоге. Для точных кодов ошибок и редких идентификаторов более эффективен поиск по ключевым словам. Если в компании уже существует сервис поиска, который индексирует нужные документы и контролирует права доступа, следует использовать его. Векторный поиск — это лишь один из нескольких инструментов.

    Что ещё не хватает для промышленного использования

    Прежде чем компания сможет полагаться на этот подход, не хватает нескольких элементов:

    • постоянное хранилище вместо оперативной памяти
    • чёткое определение ответственного за каждый документ
    • фильтры доступа, применяемые во время запроса
    • механизмы обработки версий документов и их удаления
    • план действий на случай изменений модели встраивания
    • ограничение объёма возвращаемого текста
    • тесты с метками для проверки качества поиска
  • тесты генерации, основанные на извлечённых фрагментах текста
  • записи о затратах и задержках
  • журналы аудита, предотвращающие утечку конфиденциальной информации

    • Фреймворк связывает все компоненты, а приложение принимает решения по их использованию.

    Место приложения службы поддержки

    Теперь оно может искать в заданном наборе руководств по работе и сохранять их источники:

    classification
  -> focused query
  -> retriever
  -> typed passages with metadata
  -> recommendation
  -> source-ID check

    Далее следует разделение истории сообщений, сохранённых данных пользователей, текущего состояния рабочего процесса и точек контроля на отдельные компоненты. Определение таких границ также позволяет понять, достаточно ли простой цепи LangChain или лучше подходит LangGraph.

    Дополнительная литература:

    • https://docs.langchain.com/oss/python/langchain/retrieval
    • https://docs.langchain.com/oss/python/langchain/knowledge-base
    • https://docs.langchain.com/oss/python/integrations/splitters/index
    • https://docs.langchain.com/oss/python/integrations/retrievers
    Все статьиПрислать бриф