Strona główna / Artykuły / Strumieniowanie LLM umożliwiające przerwanie z użyciem Redis Streams jako punktu spotkania na każdą turę

redis streaming llm agents gateway realtime

Strumieniowanie LLM umożliwiające przerwanie z użyciem Redis Streams jako punktu spotkania na każdą turę

Przetrwaj przerwy w połączeniu oraz kilkuminutowe pauzy narzędzia poprzez publikowanie zdarzeń agenta do strumienia Redis skonfigurowanego na poziomie rundy, dzięki czemu klienci mogą kontynuować pracę.

2049 słów

To, czego chcieliśmy

Odpowiedzi agenta mogą trwać dziesiątki sekund: rozumowanie, ładowanie umiejętności, wywoływanie narzędzi, czekanie, przesyłanie tokenów. W demonstracjach utrzymywana jest jedna otwarta połączenie. W środowisku produkcyjnym komunikacja zostaje przerwana w trakcie odpowiedzi, ponownie konfigurowane są bramy komunikacyjne, a oczekiwanie na wyniki z narzędzi trwa minuty. Celem jest przerwany transmisja w czasie rzeczywistym, aby klient mógł ponownie się połączyć i kontynuować tę samą rozmowę.

Przebieg rozwoju

Iteracja 1: klient komunikuje się bezpośrednio z agentem

Po prostu, ale kruche rozwiązanie. Każda awaria sieci kończy transmisję. Skalowanie poziome oznacza konieczność używania sesji trwałych lub utratę informacji.

Iteracja 2: transmisja gRPC pomiędzy usługami

Lepsze standardy wewnętrzne, ale nadal niewygodne dla klientów przeglądarkowych i wciąż słabe przy przerwach trwających kilka minut podczas restartów usług.

Dlaczego nie Kafka?

Doskonałe do trwałych zapisów; cięższe, niż to konieczne przy spotkaniach co turę z krótkim przechowywaniem i grupami konsumentów, które słabo pasują do modelu „jednej karty przeglądarki”.

Iteracja 3 (to, co zadziałało): nazwane spotkanie na każdą turę

// Agent output
{
  "type": "tool_result",
  "tool": "product_search",
  "data": {
    "items": [...]
  }
}

// Gateway -> TV
{
  "type": "product_carousel",
  "items": [...]
}

// Gateway -> Mobile
{
  "type": "product_list",
  "items": [...]
}

cursor = last_event_id or "0-0"

while True:
    entries = xread({key: cursor}, block=30_000)

    if not entries:          # the only timeout check point
        check_timeouts()
        continue

    for entry_id, event in entries:
        # writes to the socket; not an ack that the client received it
        sse.send(id=entry_id, data=event.payload)
        cursor = entry_id
        if event.type in TERMINAL:
            return

id: 1755600000123-0
data: {"type":"tool_selected","tool":"search"}

id: 1755600000871-0
data: {"type":"response_block","block":{...}}

GET /sessions/{sid}/turns/{tid}/stream
Last-Event-ID: 1755600000871-0

# turn starts: one atomic step (MULTI/EXEC, or a Lua script)
xadd(key, first_event)
expire(key, GENEROUS_TTL)

# producer finishes: bring it in
expire(key, RECONNECT_TTL)

Każda tura użytkownika ma swoją Redis Stream (lub wzorzec stream+grupa konsumentów), której kluczem jest turn_id. Agent publikuje zdarzenia tokenów/narzędzi; brama łączy się z ostatnim id klienta. Ponowne połączenie kontynuuje się od tego wskaźnika. Procesy mogą zostać zakończone; strumień przechowuje wystarczającą ilość historii dla danej tury.

Detale, które miały znaczenie

Dyscyplina wskaźnika — klienci potwierdzają ostatni widziany ID strumienia; nigdy nie rozpoczynają od 0 po pierwszym połączeniu.

Zdarzenia rytmu serca — zapobieganie zamknięciu połączeń nieaktywnych przez pośredników podczas oczekiwania narzędzia.

Cyklu życia operacji — wyraźne sygnały turn_started / turn_paused / turn_completed / turn_failed.

TTL — wygaśnięcie strumieni po zakończeniu operacji, aby Redis nie stał się nieskończonym archiwum.

Oprawa autoryzacyjna — turn_id nie stanowi uprawnień; operacje muszą być powiązane z autoryzowaną sesją.

Sytuacja, która to rozwiązała: kilkuminutowa pauza

Narzędzie przekazało zadanie innemu procesowi, który odpowiedział kilka minut później. Bezpośredni streaming HTTP przestał działać. Dzięki Redis Streams agent opublikował zdarzenie pauzy, klient pokazywał status „nadal w pracy”, a później tokeny wznowiły działanie po ponownym połączeniu bez konieczności ponownego uruchamiania całego planu.

To, co wysyłamy do klienta

Zdarzenia wpisywane ręcznie: tokeny, podsumowania tool_start i tool_result (nigdy tajemnice), błędy oraz zakończenie operacji. Zachowuj małe rozmiary danych przekazywanych; przechowuj duże pliki w systemie przechowywania obiektów i wysyłaj do nich odniesienia.

Koszty, ograniczenia, kwestie do uwagi

Bądź uważny na ilość pamięci używanej przez Redis, maksymalną długość strumienia oraz rozprzestrzenianie się danych, jeśli wiele bramek obsługuje jedną transakcję. Ustal ograniczenia liczby równoczesnych transakcji na użytkownika. Przeprowadź testy obciążeniowe w celu sprawdzenia sytuacji po ponownym podłączeniu po aktualizacji bramy.

Końcowy stan

Brama jest odczytującym elementem świadomym kontekstu; agent jest elementem zapisującym; Redis Streams stanowi miejsce spotkań. Interfejs użytkownika w czasie rzeczywistym przetrwa te nudne awarie, które niszczą architektury demonstracyjne.

Wskazówka operacyjna: przechowuj ostatni ID strumienia w ciasteczku sesji lub pamięci klienta, a także po stronie serwera w celu umożliwienia odtworzenia transmisji przez obsługę klienta. Gdy użytkownik zgłosi, że „wszystko zamarło”, obsługa powinna odtworzyć daną rozmowę na podstawie strumienia, bez konieczności proszenia go o ponowne czekanie dziesięć minut na urządzenie. Dodaj panele kontrolne pokazujące wskaźnik wznowień rozmów, liczbę porzuconych rozmów oraz średnią długość przerw, aby zrozumieć, czy problemem są wolni pracownicy, czy niestabilne sieci.

Wskazówka operacyjna: przechowuj ostatni ID strumienia w ciasteczku sesji lub pamięci klienta, a także po stronie serwera w celu umożliwienia odtworzenia transmisji przez obsługę klienta. Gdy użytkownik zgłosi, że „wszystko zamarło”, obsługa powinna odtworzyć daną rozmowę na podstawie strumienia, bez konieczności proszenia go o ponowne czekanie dziesięć minut na urządzenie. Dodaj panele kontrolne pokazujące wskaźnik wznowień rozmów, liczbę porzuconych rozmów oraz średni czas przerwy, aby zrozumieć, czy problemem są wolni pracownicy, czy niestabilne sieci.

Literatura pokrewna

Self-Hosting LangGraph Agent Server z Postgres i Redis — Dowiedz się, jak Langhost zastępuje warstwę trwałości LangGraph przez Postgres i Redis, umożliwiając zespołom samodzielne hostowanie niezmodyfikowanego Agent Servera pod licencją MIT.
LiteLLM w szczegółach: kontrolna warstwa bramy AI typu multi-tenant — Abstrakcja dostawcy, hierarchia użytkowników, budżety, polityka modeli, konta usługowe, routowanie, możliwości obserwacji, agenci/MCP oraz powód, dla którego brama stanowi kluczowy element.

Wszystkie artykuły Wyślij brief