Надзяйнасць 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 через помилку „database connection refused“ у журналах checkout-api. Користувач просить агента знайти власника та останні інциденти, запропонувати наступний крок та додати примітку до INC-2048 після його схвалення.

У оптимістичному сценарії агент читає інформацію про власника та останні інциденти, складає рекомендацію, пропонує примітку, чекає на схвалення, записує результат та повертає його. Щоб ускладнити ситуацію: подайте ще три тикети одночасно, зробіть так, щоб один дзвінок до постачальника зазнав двох невдач, зробіть один тикет недійсним та перервіть з’єднання з клієнтом під час першого запиту. Усе наведене нижче розроблено саме для вирішення таких ситуацій.

Спочатку називайте проблеми

„Звонак не выйшоў“ не паведамляе, што робіць. Раздзеліце неудачы на катэгоріі, якія ведуць да разных дзеянняў.

• Тымчасовая неудача прадавца. Патрэбна перзапуск з’яўлення звязку, тайма-аут запиту або адпаведзь прадавца, якай чытка прасіць вернуцца пазней. Праказваце спробы колькасць разоў з перапаузамі межы ўсімі.
• Ліміт частоты. Прадавець сябе захоўвае ад перэкрасці квоты або сваёй ўместнасці. Дазвольце інфармацыі пра праказванне спроб, уповольніце швальнасць з’яўлення звязку і зменшыце адночаснасць. Негайная паўтарэнне толькі пагоршыце ситуацыю.
• Некоректны вхідны дадзеныя. У заявцы няма ID, схема не падходзіць або текст перасягае ліміт прыемлівання. Адпаведзце чыткай памятаючы пра адміністрацыйныя проблемы кліента і не праказваце зноў той жа некоректны запит.
• Неудача автарызаціі. У з’яўлення звязку немае прав на чытанне інструкцый або запіс у заявку. Зупініцеся тут; не меняйце атрыбуты автарызаціі і не прасіце іншую модель абы як парадзіць з гэтым адхіленнем.

Адміністратыўная памылка апрантка. Квітанцыя вже закрыта, пераход не дазволены або існуе тая ж зместка. Ператворыце адпаведную адпаведнасць апрантка на значымую памылку пры выкорыстоўванні, таму што пракатка можа быць некоректной.

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

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

Разам гэтыя катагорыі вядомаюць, якія аббярэнні павінны быць пракаткаваны знову, якія можна пераходзіць на альтэрнатыўны спосаб і якія павінны ідзці безпасова да вызывальніка.

Разметка і налашоўкі проекта

Снэйпшот для гэтага крока містіць:

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 мае настройкі, прызначаныя конкрэтна для запиту, такія як ID потока, а секрэты не павінны там быць. 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 выконвае незалежныя запыты ў пулі потакоў, што падходзіць для багатых операцый, залежных ад сеті, а runnable, спецыязаваны для певнага прадаўцы, можа заменіць його на власную API для групавання. Спеціяліст службы падтрымкі прыймае кальку незв’язаных запытоў:

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

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