Надежность LangChain: потоковая обработка, групповая обработка, повторные попытки, альтернативные решения, промежуточное программное обеспечение

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

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

Этот шаг добавляет пять ограничений для агента. Для одного запроса:

one request
  -> stream honest progress
  -> retry a small transient failure
  -> use a labeled fallback if attempts are exhausted
  -> stop at time and call limits

Для набора не связанных между собой запросов:

many independent requests
  -> run a bounded batch
  -> keep one result or error beside each input

Основное правило: не скрывайте сбои за веселым ответом.

Тикет, который мы продолжаем тестировать

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

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

Сначала дайте сбоям названия

Фраза «Звонок не удался» не указывает, что делать дальше. Разделите сбои на категории, соответствующие разным действиям.

• Временный сбой у поставщика. Необходимо перезапустить соединение, произошел тайм-аут запроса или поставщик ясно попросил вернуться позже. Попробуйте еще несколько раз с перерывами между попытками.
• Ограничение скорости. Поставщик ограничивает количество запросов или свою пропускную способность. Следуйте инструкциям по повторной попытке, снизьте частоту запросов и уменьшите одновременные операции. Немедленное повторение только усугубит ситуацию.
• Некорректные данные. В заявке отсутствует идентификатор, структура данных не совпадает или объем текста превышает установленный лимит. Ответьте с указанием четкой ошибки клиента и не отправляйте тот же некорректный запрос снова.
• Сбой авторизации. У пользователя нет прав на чтение информации или запись в заявку. Прекратите действия; не меняйте учетные данные и не просите другую систему обойти отказ.

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

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

Неизвестная ошибка. Приложение не смогло классифицировать ошибку. Закройте операцию с ошибкой, зарегистрируйте тип ошибки и проведите расследование; никогда не считайте каждую исключение временной.

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

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

Снимок состояния на этом этапе содержит:

langchain-helpdesk/
├── app.py
├── reliability.py
├── batch_runner.py
├── stream_view.py
├── middleware_config.py
└── tests/
    ├── test_reliability.py
    └── test_stream_view.py

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

python -m pip install -U \
  langchain \
  langchain-core \
  langgraph \
  pytest

Тесты никогда не вызывают поставщика услуг.

Стриминг передаёт данные; он не проверяет их

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

• updates: завершённые шаги агента
• messages: токены модели вместе с метаданными
• custom: данные о ходе выполнения, определённые вашим приложением

Они содержат разные виды информации, поэтому не следует отображать их все как единый ответ.

Основной цикл стриминга

for chunk in agent.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": request_text,
            }
        ]
    },
    config,
    stream_mode=["updates", "custom"],
    version="v2",
):
    send_to_client(chunk)

Разбор функции: agent.stream(...) запускает тот же процесс, что и invoke(...), но при этом генерирует части данных в процессе выполнения. Первым аргументом является обновление состояния, которое здесь представляет собой одно сообщение пользователя. Параметр config хранит настройки, связанные с конкретным запросом, такие как идентификатор потока; секретные данные в него не включаются. Параметр stream_mode=["updates", "custom"] запрашивает данные о завершённых шагах вместе с собственными сообщениями о прогрессе, поэтому в данном примере не возвращается текст по отдельным токенам. Параметр version="v2" определяет текущий формат блоков данных, при котором каждый блок содержит тип, пространство имён и нагрузку данных. Функция send_to_client(chunk) представляет собой ваш собственный код, который может отправлять Server-Sent Events, сообщения через WebSocket или строки в терминале.

Перевод сырых блоков данных для интерфейса

Функция кэширования преобразует блоки данных до того, как они достигнут пользовательского интерфейса:

@dataclass(frozen=True)
class ProgressEvent:
    kind: str
    label: str
    payload: Any

def readable_events(chunks):
    for chunk in chunks:
        chunk_type = chunk.get("type")
        data = chunk.get("data")        if chunk_type == "updates" and isinstance(data, dict):
            for step_name, step_data in data.items():
                yield ProgressEvent(
                    kind="step",
                    label=step_name,
                    payload=step_data,
                )

Этот адаптер обеспечивает интерфейсу небольшой, стабильный контракт. Он может отображать такие строки:

Checking service ownership...
Reading recent incidents...
Waiting for ticket-note approval...
Ticket note approved and written.

То, что нельзя показывать до завершения запуска, — это:

Done.

Частичный текст — это не окончательный текст

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

Используйте чёткие состояния событий: running, waiting_for_approval, completed, failed и cancelled. Только состояние completed означает наличие окончательного результата.

Определите, что означает потеря связи

Закрытие браузера не доказывает, что сервер остановился. Заранее утвердите правила:

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

Перед повторной попыткой записи найдите операцию по ключу идемпотентности.

Стриминг и отмена связаны между собой, но это не одна и та же функция.

Группировка независимых входных данных

Инструменты LangChain предоставляют функции batch и abatch. По умолчанию функция batch выполняет независимые запросы в пуле потоков, что подходит для многих операций, зависящих от сети; однако реализация, специфичная для конкретного поставщика, может заменить её на встроенный API для группировки. Сотрудник службы поддержки принимает несколько независимых заявок:

outputs = runnable.batch(
    tickets,
    config={"max_concurrency": 4},
    return_exceptions=True,
)

tickets — это упорядоченный список входных данных. max_concurrency определяет максимальное количество одновременно выполняемых запросов, причем значение 4 носит иллюстративный характер; его следует рассчитать на основе квот поставщика, мощности инструмента, затрат и измеренной задержки. При установке return_exceptions=True сбой одного элемента больше не приводит к потере результатов остальных. Затем приложение сопоставляет каждый результат с соответствующим входным данным:

for ticket, output in zip(tickets, outputs, strict=True):
    if isinstance(output, BaseException):
        save_failure(ticket.ticket_id, type(output).__name__)
    else:
        save_recommendation(ticket.ticket_id, output)

Это обеспечивает один результат на каждый тикет.

Игнорируйте параллельную обработку в следующих случаях: второй элемент зависит от первого; каждый элемент обновляет одну и ту же запись без контроля конкуренции; порядок выполнения сам по себе является бизнес-правилом; у поставщика есть транзакция с большим объемом данных с иными гарантиями; или один пользователь может отправить достаточное количество элементов, чтобы исчерпать общую пропускную способность. Группировка задач лишь перераспределяет нагрузку, но не устраняет её.

Попробуйте снова в самых безопасных пределах

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

class TemporaryProviderError(RuntimeError):
    """A short-lived error that may succeed when tried again."""

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

primary_call = RunnableLambda(primary).with_retry(
    retry_if_exception_type=(TemporaryProviderError,),
    wait_exponential_jitter=False,
    stop_after_attempt=3,
)

RunnableLambda(primary) преобразует обычную функцию на Python в объект, способный к выполнению. retry_if_exception_type функционирует как список разрешенных типов исключений, поэтому недопустимый тикет не совпадает и сразу возвращается вызывающему коду. Установка wait_exponential_jitter=False обеспечивает быстрые и предсказуемые тесты в автономном режиме. Параметр stop_after_attempt=3 учитывает каждую попытку, то есть в сумме происходит три попытки, а не три повторных запуска после первой. При работе с реальным сетевым сервисом экспоненциальное откладывание попыток с джиттером обычно является лучшим выбором, поскольку это предотвращает одновременный запрос к сервису со стороны всех рабочих процессов.

Повторные попытки при записи не всегда безопасны

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

operation_id = "ticket-note:INC-2048:approval-7331"

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

Альтернативный вариант — это другой путь

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

reliable = primary_call.with_fallbacks(
    [RunnableLambda(conservative_fallback)],
    exceptions_to_handle=(TemporaryProviderError,),
)

Он работает только для указанной временной ошибки, и то, что он возвращает, — это:

Recommendation(
    ticket_id=ticket.ticket_id,
    text=(
        "The recommendation service is temporarily unavailable. "
        "Keep the ticket open and compare the failed deployment "
        "with the last known-good release."
    ),
    source="fallback",
)

В ответе признаётся факт произошедшего. В нём не утверждается, что производился поиск в руководствах по эксплуатации или что тикет обновлялся.

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

Таймауты существуют на нескольких уровнях

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

Укажите поведение при истечении времени ожидания рядом с операцией, которую оно контролирует: лимиты времени ожидания у поставщика — на клиенте поставщика, лимиты времени ожидания у инструмента — на клиенте инструмента, а крайний срок выполнения веб-запроса — на границе сервиса. Когда срабатывает один из этих механизмов, необходимо записать, какой слой прервал работу. Простое сообщение «время истекло» без указания конкретной границы затрудняет устранение проблемы.

Промежуточный слой оборачивает вызовы агента

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

def reliability_middleware():
    return [
        ModelRetryMiddleware(
            max_retries=2,
            retry_on=(
                TemporaryProviderError,
                TimeoutError,
                ConnectionError,
            ),
            on_failure="error",
            initial_delay=0.5,
            backoff_factor=2.0,
            max_delay=4.0,
            jitter=True,
        ),
        ModelCallLimitMiddleware(
            run_limit=8,
            exit_behavior="error",
        ),
        ToolCallLimitMiddleware(
            run_limit=6,
            exit_behavior="end",
        ),
    ]

ModelRetryMiddleware оборачивает вызовы моделей, причем первоначальный вызов плюс две попытки повтора дают в общей сложности не более трех попыток. ModelCallLimitMiddleware предотвращает бесконечную работу цикла модели. ToolCallLimitMiddleware ограничивает чрезмерное использование инструментов в ходе одной операции. Эти механизмы контроля выполняют отдельные функции: ограничение количества вызовов не заменяет таймаут, таймаут не заменяет утверждение, а утверждение не заменяет авторизацию.

Порядок имеет значение

Хуки, которые выполняются до вызова, идут в порядке списка, а те, что выполняются после, — в обратном порядке; кроме того, хуки, оборачивающие другие, упаковываются так же, как функции. Когда middleware A оборачивает B, попытка повтора в A может выполнить работу B заново, поэтому всегда определяйте порядок действий, если два компонента могут одновременно пытаться повторить действия, изменить модели или состояние. Утверждение записи должно происходить до любых попыток повтора, которые могут вновь вызвать уже утвержденные последствия.

Пользовательские хуки — это обычный код

LangChain предоставляет хуки в стиле узлов (before_model, after_model) и хуки в стиле обёрток (wrap_model_call, wrap_tool_call). Хуки в стиле узлов подходят для проверок и обновления состояния в определённых моментах. Хуки в стиле обёрток используются для кода, который должен окружать один вызов или определять, будет ли обработчик выполняться нуль раз, один раз или несколько раз. Однако не каждое правило приложения должно превращаться в промежуточный компонент. Переходы состояния заявок, решения по авторизации и правила владения данными проще управлять как названные сервисы приложения.

Тестирование путей сбоя в офлайн-режиме

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

Временная ошибка успешно решается при второй попытке:

result = runnable.invoke(ticket())

assert calls == 2
assert result.source == "primary"

При неудачных попытках используется указанный вариант запаса:

assert calls == 3
assert result.source == "fallback"
assert "temporarily unavailable" in result.text

Неверный ввод не повторяется и не скрывается:

with pytest.raises(InvalidTicketError):
    runnable.invoke(ticket())

assert calls == 1

Одна неудача в партии не отменяет другого успеха:

assert results[0].status == "completed"
assert results[1].status == "failed"
assert results[1].error_type == "RuntimeError"

Прогресс отдельно от токенов модели:

assert [(event.kind, event.label) for event in events] == [
    ("step", "model"),
    ("progress", "custom"),
    ("token", "model"),
]

Запустите их:

pytest -q

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

5 passed

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

Запишите матрицу сбоев

Перед выпуском добавьте в код или документацию таблицу вроде этой:

failure                         retry?   fallback?   client result
connection reset               yes      maybe       503 if exhausted
provider rate limit            yes      maybe       503 or queued
invalid ticket                 no       no          422
authorization denied           no       no          403
structured output invalid      limited  maybe       502 if exhausted
ticket already closed          no       no          409
unknown exception              no       no          500 + incident ID

Конкретные коды состояния зависят от вашего API. Важно то, что для каждой категории сбоев существует четко определенная политика обработки.

Ещё не хватает

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

Состояние процесса обслуживания в службе поддержки

В настоящее время процесс имеет видимые состояния и ограниченные возможности восстановления:

request
  -> progress events
  -> small retry boundary
  -> labeled fallback when allowed
  -> model and tool call limits
  -> approval before write
  -> final result or named failure

batch
  -> bounded concurrency
  -> one outcome per ticket

Никакой механизм контроля не может полностью устранить неопределенность. Цель состоит в том, чтобы она оставалась видимой, имела определенные границы и поддавалась тестированию. Далее рассматривается весь процесс работы службы поддержки, связанный с API: настройки, конфиденциальные данные, идентификаторы запросов, таймауты, логи и механизмы отслеживания, затем детерминистичные тесты, случаи оценки и проверки при развертывании, а также список того, что реальная компания должна предоставить перед тем, как сервис начнет обрабатывать реальные запросы. Всё вышеизложенное было проверено в соответствии с документацией LangChain на Python по темам потоковой обработки, возможностей повторных попыток, резервных решений, обработки пакетов данных, промежуточного программного обеспечения и тестирования 31 августа 2026 года; при этом API пакетов и типы ошибок поставщиков могут изменяться.