Шукайте посібники з виконання завдань, а не копіюйте їх: невеликий шлях RAG з посиланнями в LangChain

Шукайте в посібниках з експлуатації замість того, щоб заповнювати запит

На цьому етапі додаток служби підтримки розгалужується за типом заявки. Для базових заявок завантажуються каталог послуг, нещодавні інциденти та останні розгортання одночасно, а для заявок на доступ використовується інший шлях. Проте все ще відсутні детальні кроки виконання рекомендацій.

Ви могли б вставити всі посібники з експлуатації у запит. Для трьох невеликих файлів це працює, але коли кількість матеріалів зростає, цей підхід перестає ефективним. Великі запити коштують більше, виконуються повільніше та ускладнюють визначення того, який фрагмент тексту став основою відповіді. Крім того, вони можуть передавати моделі документи, які поточному користувачеві ніколи не дозволялося читати. Рішенням є пошук.

Складена тут схема має дві фази. Під час індексування посібники у форматі Markdown перетворюються на об’єкти документів із вихідними метаданими, а потім ці об’єкти діляться на менші частини:

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, веб-сторінки, wiki чи навіть з рядка в таблиці, а етикети можуть містити ідентифікатор джерела, назву, сервіс, шлях, власника, версію чи інформацію про те, хто може його бачити.

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

Вбудовування. Модель вбудовування перетворює текст на список чисел, і тексти з схожим значенням мають опинятися поруч один з одним за допомогою обраного методу порівняння. Це не перевірка фактів – схожий текст все одно може бути неправильним, застарілим або недоступним.

Сховище векторів. Сховище векторів зберігає вектори документів та знаходить схожі на них. Деякі з них зберігаються лише в пам’яті, інші – у базі даних чи на хостованій послузі. Місце зберігання векторів не замінює контроль доступу.

Отримання даних. Якщо передати ретріверу текст запиту у вільному форматі, він поверне об’єкти 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