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

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

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