Niezawodność LangChain: strumieniowanie, pakietowanie, ponawianie prób, rozwiązania awaryjne, middleware

Do tej pory pracownik obsługi klienta miał celowo ograniczoną swobodę działania. Może przeczytać informacje o odpowiedzialności za usługę oraz ostatnich incydentach, ale zanim napisze notatkę do zgłoszenia, musi się zatrzymać, przedstawić argumenty, które zamierza użyć, i czekać na ludzkiego pracownika. Taki układ funkcjonuje tak długo, dopóki każde wezwanie modelu lub narzędzia zwraca wynik szybko. W praktyce tak nie jest.

Jedno wezwanie modelu może zająć piętnaście sekund. Dostawca może odrzucić żądanie na całą minutę z powodu ograniczeń ilościowych. Jedno zgłoszenie w partii może zawierać błędne dane wejściowe. Model awaryjny może odpowiadać inaczej niż model główny, a użytkownik może zamknąć przeglądarkę w trakcie działania zatwierdzonego narzędzia. Każdy z tych przypadków wymaga określonego zachowania opartego na testach.

To kroku dodaje pięć ograniczeń dla pracownika. W przypadku pojedynczego żądania:

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

Dla serii niepowiązanych zapytań:

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

Zasada jest jedna: nie ukrywaj błędu za wesołą odpowiedzią.

Ticket, który ciągle testujemy

Sytuacja testowa się nie zmienia: żądania płatności w UE przestają działać po aktualizacji o godzinie 14:05, a w logach checkout-api pojawia się komunikat „database connection refused”. Użytkownik prosi agenta o sprawdzenie właściciela i ostatnich incydentów, zasugerowanie dalszych kroków oraz dodanie notatki do INC-2048 po jego zatwierdzeniu.

W prostym scenariuszu agent czyta informacje o właścicielu i ostatnich incydentach, pisze rekomendację, proponuje notatkę, czeka na zatwierdzenie, wpisuje informację i przekazuje wynik. Utrudnijmy sytuację: zgłoś trzy dodatkowe ticketi jednocześnie, spraw, by jedna próba połączenia z dostawcą zakończyła się niepowodzeniem dwa razy, uznaj jeden ticket za nieważny i przerwij połączenie z klientem w trakcie pierwszego żądania. Wszystko poniżej zostało zaprojektowane, aby poradzić sobie właśnie z takimi sytuacjami.

Najpierw nadawaj błędom nazwy

„Połączenie nie powiodło się” nie podaje wskazówek, co robić. Podziel błędy na kategorie, które prowadzą do różnych działań.

• Tymczasowy błąd dostawcy. Konieczne jest ponowne nawiązanie połączenia, czas oczekiwania na odpowiedź wygasł lub dostawca jasno prosi o ponowną próbę później. Spróbuj kilka razy z przerwami pomiędzy próbami.
• Limity częstotliwości. Dostawca chroni określoną kwotę lub swoją pojemność. Postępuj zgodnie z instrukcjami dotyczącymi ponownych prób, zwolnij tempo i zmniejsz jednoczesność zapytań. Natychmiastowe powtarzanie tylko pogorszy sytuację.
• Nieważne dane wejściowe. Bilet nie ma numeru ID, schemat nie jest zgodny lub tekst przekracza ustalone limity aplikacji. Odpowiedz jasnym błędem klienta i nie wysyłaj ponownie tego samego błędnego zapytania.
• Błąd uprawnień. Osoba wysyłająca zapytanie nie ma prawa do odczytu dokumentacji ani do edycji biletu. Zatrzymaj się w tym miejscu; nie zmieniaj danych dostępowych ani nie prosz innych modeli o obejście tej blokady.

Błąd w biznesowym mechanizmie narzędzia. Ticket został już zamknięty, przejście nie jest dozwolone lub istnieje ta sama notatka. Należy przekształcić odpowiedź narzędzia w sensowny błąd aplikacji, ponieważ ponowna próba może okazać się błędna.

Niepowodzenie walidacji wyniku modelu. Dostawca zwrócił dane, które naruszają umowę dotyczącą wyniku. W przypadku wyniku tylko do odczytu jedna próba naprawy może być akceptowalna, ale nadal musi istnieć ścieżka końcowego niepowodzenia.

Nieznane niepowodzenie. Aplikacja nie była w stanie sklasyfikować błędu. Należy zamknąć niepowodzenie, zarejestrować typ błędu i przeprowadzić dochodzenie, nie traktując każdego wyjątku jako tymczasowego.

Wspólnie te kategorie określają, które niepowodzenia podlegają ponownej próbie, które mogą być rozwiązane alternatywnymi metodami, a które trafiają bezpośrednio do użytkownika.

Struktura i konfiguracja projektu

Zdjęcie stanu z tego kroku zawiera:

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

Zainstaluj paczki:

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

Testy nigdy nie wywołują dostawcy.

Striming dostarcza dane; nie sprawdza ich

Pusta ekran podczas działania narzędzi oraz czas oczekiwania modelu to kiepskie doświadczenie użytkownika. Striming rozwiązuje ten problem poprzez wysyłanie zdarzeń w miarę postępów pracy. Obecny mechanizm strimingu agenta LangChain oferuje trzy widoki:

• updates: zakończone kroki agenta
• messages: tokeny modelu wraz z metadanymi
• custom: dane o postępach zdefiniowane przez twoją aplikację

Zawierają one różne rodzaje informacji, więc nie należy pokazywać ich wszystkich tak, jakby stanowiły jedną odpowiedź.

Bazowy cykl strumieniowania

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

Rozbierając ten mechanizm: agent.stream(...) uruchamia tę samą sekwencję działania, co invoke(...), ale w trakcie wykonywania zwraca poszczególne fragmenty. Pierwszym argumentem jest aktualizacja stanu, która w tym przypadku to jedna wiadomość od użytkownika. config przechowuje ustawienia specyficzne dla żądania, takie jak identyfikator wątku, a sekrety nie powinny się tam znajdować. Parametr stream_mode=["updates", "custom"] żąda danych o zakończonych krokach oraz własnych komunikatów o postępie, więc w tym przykładzie nie otrzymujemy tekstu token po tokenie. version="v2" wybiera obecny format fragmentów, w którym każdy z nich zawiera typ, przestrzeń nazw oraz treść danych. send_to_client(chunk) oznacza Twój własny kod, który może wysyłać Server-Sent Events, wiadomości typu WebSocket lub linie tekstowe do terminala.

Tłumaczenie surowych fragmentów dla interfejsu użytkownika

Snapshot konwertuje fragmenty przed tym, jak dotrą do interfejsu użytkownika:

@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,
                )

Ten adapter zapewnia interfejsowi użytkownika mały, stabilny kontrakt. Może wyświetlać takie linie:

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

To, czego nie powinno pokazywać przed faktycznym zakończeniem wyścigu, to następujące:

Done.

Częściowy tekst to nie tekst ostateczny

Tokeny strumieniowe sprawiają wrażenie szybszej odpowiedzi, ale nie wpływają na bezpieczeństwo ani dokładność. Model może zacząć pisać zdanie, a potem przenieść się gdzie indziej, narzędzie może zawieść po wysłaniu tokenów, a sprawdzenie zasad może odrzucić proponowaną akcję.

Używaj wyraźnych stanów zdarzeń: running, waiting_for_approval, completed, failed oraz cancelled. Tylko completed oznacza wynik ostateczny.

Określ, co oznacza przerwanie połączenia

Zamknięty przeglądarka nie dowodzi, że serwer przestał działać. Ustal zasady z góry:

• Anuluj prace tylko do odczytu, gdy istnieje możliwość anulowania i nikt nie czeka na odpowiedź.
• Pozwól zakończyć zatwierdzoną, trwałą pracę, jeśli użytkownik planuje do niej wrócić.
• Nigdy nie zakładaj, że operacja zapisu nie odbyła się tylko dlatego, że jej odpowiedź zaginęła.

Zanim ponownie spróbujesz zapisać dane, sprawdź tę operację pod kątem klucza idempotencji.

Strumieniowanie i anulowanie są powiązane, ale nie stanowią jednej funkcji.

Zbieranie niezależnych danych we grupy

Komponenty LangChain udostępniają funkcje batch i abatch. Domyślna funkcja batch wykonywa niezależne żądania w puli wątków, co nadaje się do wielu operacji związanych z siecią; natomiast komponent dostarczony przez konkretnego dostawcę może zastąpić ją wbudowaną API do grupowania danych. Pracownik obsługi klienta przyjmuje kilka niepowiązanych zgłoszeń:

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

tickets to uporządkowana lista danych wejściowych. max_concurrency określa, ilu żądań może być wykonywanych jednocześnie – wartość cztery jest tu tylko przykładem; należy ją obliczyć na podstawie limitów dostawcy, możliwości narzędzia, kosztów oraz zmierzonej opóźnienia. Przy użyciu return_exceptions=True awaria jednego elementu nie niszczy wyników pozostałych. Następnie aplikacja dopasowuje każdy wynik do odpowiedniego danych wejściowych:

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)

To zapewnia jeden wynik na ticket.

Pomijaj procesowanie równoległe w następujących przypadkach: drugi element zależy od pierwszego; każdy element aktualizuje ten sam rekord bez kontroli współbieżności; kolejność wykonywania jest samą w sobie zasadą biznesową; dostawca oferuje transakcje masowe z innymi gwarancjami; lub jeden użytkownik może wysłać wystarczającą liczbę elementów, aby wyczerpać wspólną przepustowość. Procesowanie grupowe jedynie przenosi obciążenie, ale nie sprawia, że znika ono.

Ponowić próbę przy najmniejszej bezpiecznej granicy

Ponowna próba oznacza wykonanie pracy ponownie, więc uważaj, co powtarzasz. Ponowne uruchomienie całego procesu może skutkować ponownym pobieraniem danych, ponownymi wywołaniami modelu oraz ponowną zapisem informacji na ticketu. Bezpieczniejszym podejściem jest ponowna próba tylko kroku odczytu, który wiadomo, że może zawieść tymczasowo. Dla tego celu istnieje dedykowany błąd w formie zrzutu stanu:

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

i otacza jeden wykonywalny plik:

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

RunnableLambda(primary) przekształca zwykłą funkcję Pythona w obiekt wykonywalny. retry_if_exception_type działa jak lista dozwolonych wartości, dzięki czemu nieważny ticket nie zostanie dopuszczony i jest od razu zwracany do wywołującego. Ustawienie wait_exponential_jitter=False sprawia, że test offline przebiega szybko i w sposób deterministyczny. stop_after_attempt=3 liczy każdą próbę, co oznacza łącznie trzy próby, a nie trzy powtórki dodatkowo do pierwszej. W przypadku rzeczywistego usługi sieciowej eksponencjalne opóźnienie z jitterem jest zazwyczaj lepszym wyborem, ponieważ zapobiega to tymczasowemu obciążeniu usługi przez wszystkie procesory jednocześnie.

Zapisy nie nadają się do ślepego powtarzania prób

Załóżmy, że usługa biletowa dodaje notatkę, a następnie połączenie zostanie przerwane przed nadejściem odpowiedzi o sukcesie. Osoba dzwoniąca widzi błąd czasu wygaśnięcia, mimo że zapis może już istnieć, a naiwna próba ponownego wysłania wiadomości doprowadziłaby do dodania notatki dwukrotnie. Należy użyć klucza idempotencji:

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

Służba obsługi biletów przechowuje ten klucz razem z pierwszym udanym zapisem. Jeśli ten sam klucz pojawi się ponownie, zwraca przechowywany wynik i nie aplikuje zmiany drugi raz. Gdy system zewnętrzny nie zapewnia właściwości idempotencji, należy sprawdzić jego aktualny stan przed ponowną próbą i zaakceptować fakt, że niektóre okna awarii nie mogą zostać w pełni zamknięte.

Rozwiązanie awaryjne to inna ścieżka

Gdy wykorzystane zostaną wszystkie dozwolone próby, może przejąć kontrolę program rozwiązywania awarii:

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

Działa wyłącznie dla określonego tymczasowego błędu, a to, co zwraca, jest następujące:

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",
)

Odpowiedź przyznaje, co się stało. Nie twierdzi ona, że przeszukano dokumentację operacyjną ani że aktualizowano zgłoszenie.

Przełączenie się na inny model nie jest tym samym co ponowne uruchomienie pierwszego. Drugi model może zachowywać się inaczej podczas wywoływania narzędzi, mieć mniejszy limit kontekstu, w inny sposób obsługiwać strukturyzowany wynik, stosować inne zasady bezpieczeństwa, różnić się opóźnieniami i ceną lub podlegać innej polityce danych dostawcy. Traktuj trasę awaryjną jako odrębną konfigurację i testuj ją osobno. Najlepszą opcją awaryjną jest czasami w ogóle brak modelu – tylko stała, konserwatywna wiadomość lub kolejka do przeglądu przez człowieka.

Czasochłonności istnieją na kilku poziomach

Jedna wartość timeoutu nie wystarcza. Należy przydzielić oddzielne limity na ustanawianie połączenia, pojedynczą prośbę o model, pojedynczą prośbę o narzędzie, cały proces działania agenta, czekanie na zatwierdzenie oraz połączenie z klientem. Limit dla całego procesu musi przewyższać limity poszczególnych wywołań, ale nadal wymaga określenia granicy.

Należy umieścić zasady obsługi timeoutów obok operacji, które ograniczają: timeouty dostawcy w klientach dostawcy, timeouty narzędzia w klientach tego narzędzia oraz termin realizacji żądania webowego na granicy usługi. Gdy któryś z nich zostanie aktywowany, należy odnotować, która warstwa zakończyła pracę. Proste stwierdzenie „timeout” bez określenia granicy utrudnia naprawę problemu.

Middleware otacza wywołania agenta

W poprzednich krokach używano głównie stałych łańcuchów, natomiast krok zatwierdzenia wprowadził middleware agenta. Middleware stanowi również doskonałe miejsce na reguły, które powinny obejmować wiele wywołań modeli lub narzędzi: ponawianie prób w przypadku błędów modeli, ograniczanie liczby wywołań modeli i narzędzi, filtrowanie narzędzi dla aktualnego użytkownika, dodawanie metadanych żądania do celów logowania i śledzenia, maskowanie wybranych danych osobowych oraz zatrzymywanie działania po wyczerpaniu budżetu aplikacji. Snapshot wykorzystuje wbudowany middleware:

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 otacza wywołania modeli, a początkowe wywołanie wraz z dwoma próbami ponownymi daje maksymalnie trzy próby. ModelCallLimitMiddleware zapobiega nieskończonemu działaniu pętli modelu. ToolCallLimitMiddleware ogranicza nadmierne użycie narzędzi podczas jednego wykonywania. Te mechanizmy pełnią odrębne funkcje: limit wywołań nie zastępuje czasu oczekiwania, czas oczekiwania nie zastępuje zatwierdzenia, a zatwierdzenie nie zastępuje uprawnień.

Kolejność ma znaczenie

Haki, które działają przed wywołaniem, są umieszczane w kolejności listy, te, które działają po nim – w odwrotnej kolejności, a haki otaczające siebie są układane w sposób podobny do wywołań funkcji. Gdy middleware A otacza B, ponowna próba w A może powtórzyć działanie B, dlatego należy zawsze określić kolejność, gdy dwa komponenty mogą próbować ponownie, zmieniać modele lub modyfikować stan. Zatwierdzenie operacji zapisu powinno znajdować się poza wszelkimi próbami ponownymi, które mogłyby powtórzyć zatwierdzone efekty uboczne.

Spersonalizowane haki to zwykły kod

LangChain oferuje haki w stylu węzła (before_model, after_model) oraz haki w stylu otoczenia (wrap_model_call, wrap_tool_call). Haki w stylu węzła nadają się do wykonywania sprawdzeń i aktualizacji stanu w określonych punktach. Haki w stylu otoczenia służą do kodu, który musi otaczać jedną wywołanie lub decydować o tym, czy obsłużyciel ma zostać uruchomiony zero, raz lub kilka razy. Nie każda zasada aplikacji powinna jednak stać się middleware’em. Przejścia stanu zleceń, decyzje autoryzacyjne oraz zasady własności danych są łatwiejsze do zarządzania jako nazwane usługi aplikacyjne.

Testowanie ścieżek awarii offline

Snapshot zawiera pięć testów, z których żaden nie wymaga dostawcy.

Tymczasowy błąd udaje się przy drugiej próbie:

result = runnable.invoke(ticket())

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

W przypadku nieudanych prób stosuje się oznaczony plan awaryjny:

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

Nieważne dane wejściowe nie są ani ponawiane, ani ukrywane:

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

assert calls == 1

Jedna nieudana partia nie znosi drugiego sukcesu:

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"),
]

Zrób to:

pytest -q

Oczekiwany wynik:

5 passed

Te testy potwierdzają poprawność Twoich lokalnych reguł kontroli. Nie dowodzą one jednak, że błędy dostawcy są prawidłowo mapowane – do tego potrzebny jest krótki test integracyjny z rzeczywistym SDK dostawcy oraz konto testowe pod kontrolą.

Zapisz macierz błędów

Umieść taki tabelę w kodzie lub dokumentacji przed wydaniem:

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

Kodы stanu zależą od używanej API. Ważne jest to, że każda kategoria błędów ma określoną politykę postępowania.

Nadal brakuje

Behawior w czasie wykonywania jest teraz jaśniejszy, ale nadal chodzi tu o zarys funkcjonowania w interfejsie linii poleceń. Brakuje tu autoryzowanej API HTTP, sprawdzeń uruchomienia i gotowości, ładowania tajemnic, ustrukturyzowanych dzienników zapytań, trwałego przechowywania zadań, konfiguracji wdrażania, rzeczywistych testów integracyjnych z dostawcami, zestawu do oceny przed wydaniem oraz przeglądu śladów z produkcji. Kształt usługi, który te elementy dodaje, zostanie przedstawiony później.

Sytuacja z procesem obsługi klienta

Proces ma teraz widoczne stany oraz ograniczone możliwości przywracania do normalnego funkcjonowania:

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

Żadna kontrola nie eliminuje niepewności. Celem jest utrzymanie jej w widocznym stanie, w określonych granicach i umożliwienie jej sprawdzenia. Następnie pojawia się cały proces obsługi klienta za pośrednictwem API, obejmujący konfigurację, tajne dane, identyfikatory zapytań, limity czasowe, logi oraz mechanizmy śledzenia, a także testy deterministyczne, przypadki oceny i sprawdzenia wdrożenia, plus lista tego, co rzeczywista firma musi dostarczyć, zanim usługa będzie mogła obsługiwać rzeczywiste zgłoszenia. Wszystko powyższe zostało sprawdzone w odniesieniu do dokumentacji Pythona LangChain dotyczącej przepływu danych, możliwości ponawiania prób, rozwiązań awaryjnych, przetwarzania zbiorczego, middleware’u oraz testowania z dnia 31 sierpnia 2026 roku, a API pakietów oraz typy błędów dostawców mogą ulec zmianie.

Literatura pokrewna

• LangChain @dynamic_prompt do instrukcji systemowych świadomych zadań — Stwórz jednego agenta HR, który wymienia prompty systemowe dotyczące urlopów, świadczeń i administracji w zależności od kontekstu działania — przy jednoczesnym zachowaniu uprawnień w kodzie aplikacji.
• Gate Agent Tools Per User z LangChain Wrap Middleware — Jeden agent kalkulatora, trzy poziomy dostępu: middleware typu wrap zastępuje listę narzędzi na podstawie user_type w czasie działania, zamiast klonować trzy oddzielne agenty.