Strumieniowanie zaawansowanych agentów LangChain: plany w czasie rzeczywistym, wywołania narzędzi i tokeny.
Zastąp blokujące wywołania strumieniami, aby użytkownicy mogli obserwować pracę zaawansowanego agenta: trzy tryby strumieniowania, przestrzenie nazw dla podagentów oraz mechanizmy filtrowania i podsumowywania tokenów.
Źródła: LangChain Deep Agents: Your AI Agent Is Not Frozen. It Just Cannot Tell You It Is Working. II
Niewiele rzeczy tak szybko niszczy zaufanie jak zamarły ekran. Gdy użytkownik nie może stwierdzić, czy agent myśli, utknął czy już przestał funkcjonować, zakłada najgorsze. Strumieniowanie zastępuje tę ciszę rzeczywistym obrazem działania: plan, wywołania narzędzi oraz tekst pojawiają się w momencie ich realizacji, dzięki czemu ludzie widzą, że coś się dzieje.
Dlaczego invoke sprawia, że zaawansowany agent wygląda na nieaktywny
Spróbuj uruchomić zaawansowanego agenta przy wykonywaniu rzeczywistego zadania pod obserwacją kogoś innego. Wywołanie przechodzi przez invoke, a następnie wyświetlanie pozostaje nieruchome przez czterdzieści sekund. W środku agent czyta pliki, opracowuje plan, używa narzędzi i analizuje sytuację. Użytkownik widzi jedynie nieruchomy prostokąt i ikonkę spinająca, co może oznaczać intensywne działanie lub cichy awarię. Ponieważ oba te stany wyglądają identycznie, ludzie ładują stronę ponownie i składają zgłoszenie o jej zawieszeniu.
Nic nie zostało wyświetlone. Agent po prostu nie miał żadnego kanału do prezentowania swoich działań. Bloki invoke są tak zaprojektowane, że wykonują cały cykl poza zasięgiem wzroku i zwracają jedną, gotową odpowiedź, co jest dokładnie tym, czego potrzebuje skrypt lub zadań w cronie, którymi nikt nie zarządza. Jest to jednak zła decyzja, gdy po drugiej stronie znajduje się człowiek.
Rozwiązanie to pozostawia cykl nietknięty. Zastępuje on blokujące wywołanie strumieniem, dzięki czemu plany, wywołania narzędzi oraz tokeny pojawiają się w czasie rzeczywistym.
Od invoke do stream
Strumieniowanie jest ściśle powiązane z invoke. Zamiast czekać na końcowy stan, iteruje się nad agent.stream(...), a dane przychodzą w każdym momencie, gdy cykl wykona jakieś działanie. Asynchroniczna wersja, agent.astream(...), ma taką samą strukturę i jest to to, co wywołałby serwer internetowy.
Model mentalny jest prosty. Trwały LangGraph znajdujący się u podstaw zaawansowanego agenta emituje zdarzenie na każdym kroku, a mechanizm strumieniowania przekazuje te zdarzenia w miarę ich powstawania. Nie chodzi o innego agenta ani inny pętlę, lecz jedynie o widok na ten, który już znasz.
Najpierw trzeba ustalić jeden szczegół, od którego zależy każdy przykład: format strumienia v2, aktywowany za pomocą version="v2". Każdy fragment to mały słownik z trzema kluczami:
typewskazuje, o jakim rodzaju zdarzenia chodzi.ns, czyli przestrzeń nazw, informuje, kto je wygenerował.datazawiera treść zdarzenia.
Te trzy klucze pozostają niezmienne niezależnie od tego, co przekazujesz w formie strumienia, co sprawia, że kod jest czytelny, a nie plątaniną operacji rozpakowywania tuple.
Trzy tryby, trzy poziomy
Strumień to nie pojedyncza wąż strażacki. Ustawienie stream_mode określa rodzaj otrzymywanych zdarzeń, a trzy dostępne opcje odpowiadają trzem poziomom szczegółowości.
updates to widok krok po kroku. Pojawia się jedno zdarzenie za każdym razem, gdy węzeł grafu jest ukończony, co w praktyce oznacza, że model wykonał swój ruch lub uruchomiono narzędzie. To poziom odpowiedni do dziennika postępów – wystarczający, by pokazać, jak agent wykonuje swoje zadania, bez przytłaczania informacjami, oraz rozsądny domyślny wybór dla uporządkowanego przekazu typu „oto, co robi agent”.
messages to widok oparty na tokenach. Wyniki generowane przez model przychodzą po częściach w miarę ich tworzenia, tak jak widać podczas pisania przez asystenta czatowego. Użyj go wtedy, gdy ktoś chce czytać tekst generowany przez agenta w momencie jego powstawania. Tutaj trafiają również wywołania narzędzi strumieniowych, więc możesz ogłosić „wywołanie read_file” w chwili, gdy model zdecyduje się na to wywołanie, a nie dopiero po jego zakończeniu.
values to widok w postaci zrzutu stanu. Po każdym kroku otrzymujesz cały stan agenta, a nie tylko to, co się zmieniło. Jest to bardziej obciążające i wymagane rzadziej, ale jest właściwym wyborem, gdy ważny jest pełny obraz w każdym momencie, na przykład przy wyświetlaniu aktualnej listy zadań lub plików roboczych agenta.
Przekazanie listy trybów zwraca kilka z nich jednocześnie, co jest standardowym sposobem działania dopracowanych interfejsów użytkownika: updates steruje wskaźnikiem postępu, podczas gdy messages dostarcza widoczny tekst. Przykład prostego logu postępu generowanego przez updates wygląda tak:
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Summarize the repo’s failing tests”}]},
stream_mode=”updates”,
version=”v2”,
):
if chunk[”type”] == “updates”:
for node_name in chunk[”data”]:
print(f”step: {node_name}”)
Wydrukowywana jest lista kroków grafu w miarę ich ukończenia – po jednej linii na każdą zmianę modelu i każdą eksploatację narzędzia. Ekran, który wydawał się zamarły, teraz wyraźnie się porusza. Ustawienie stream_mode na "messages" zamiast tego dostarczyłoby bezpośrednio sam tekst, gotowy do wydruku w trakcie pisania przez agenta.
Rozróżnienie głównego agenta od jego podagentów
Agent głęboki rzadko pracuje sam przez długi czas. Gdy tylko przekazuje podzadanie podagentowi, od razu dwóch lub więcej agentów wytwarza wyniki. Jeśli wszystko wysłać do jednego strumienia, powstaje bałagan: fragment zdania od koordynatora, kilka tokenów od badacza, wywołanie narzędzia z nieznanej lokalizacji. Pole ns rozwiązuje ten problem. Każdy fragment zawiera takie oznaczenie, gdy włączysz strumieniowanie podgrafów za pomocą subgraphs=True.
Zasada ta jest łatwa do zastosowania. Jeśli przestrzeń nazw jest pusta, to główny agent wyemitował zdarzenie. Jeśli zawiera segment tools:, to zrobił to podagent, a ten segment wskazuje, który to podagent. Wyniki można następnie posortować według źródła: tekst koordynatora tutaj, a prace każdego podagenta w osobnej sekcji.
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Research and summarize quantum computing”}]},
stream_mode=”messages”,
subgraphs=True, # ①
version=”v2”,
):
if chunk[”type”] == “messages”: # ②
token, metadata = chunk[”data”] # ③
is_subagent = any(s.startswith(”tools:”) for s in chunk[”ns”]) # ④
source = “subagent” if is_subagent else “main” # ⑤
if token.content:
print(f”[{source}] {token.content}”, end=”“, flush=True) # ⑥
W tym fragmencie istotne są sześć szczegółów. Po pierwsze, to subgraphs=True nadaje znaczenie ns, ponieważ bez niego wydarzenia subagenta nie mają żadnego rozróżniającego przestrzeni nazw. Po drugie, gałąź przechowuje tylko wydarzenia messages, ponieważ ten pętla kieruje tokeny, a nie kroki. Po trzecie, zawartość każdego fragmentu jest rozpakowywana na token wraz z jego metadanymi. Po czwarte, to test przestrzeni nazw wykonuje główną pracę: segment tools: w dowolnym miejscu w ns oznacza, że token pochodzi od subagenta. Po piąte, uzyskany wartość logiczna staje się czytelnym etykietą źródła. Po szóste, wydrukowanie z tą etykietą jako prefiksem przekształca jeden pomieszany strumień w dwa oznaczone pasma.
Każdy token jest oznaczony swoim twórcą przed wydrukowaniem, więc koordynator przekazujący dane badaczowi pojawia się jako dwa nazwane strumienie. Pełna interfejs internetowa to większe zadanie, ale nazwa przestrzeni w każdym fragmencie to ten sam sygnał, który odczytałby frontend React.
Dodatkowe tokeny, które nie pochodzą od rozmawiającego agenta
To zaskakuje ludzi podczas pierwszego długiego streamowania. Aby zarządzać własnym kontekstem, narzędzie skraca starsze wiadomości, gdy rozmowa zbliża się do limitu modelu. Krok skracania to również wywołanie modelu, więc generuje tokeny, które trafiają do strumienia messages obok rzeczywistej odpowiedzi agenta. Jeśli ten strumień trafia bezpośrednio do użytkownika, ten zobaczy, jak agent nagle odczytuje streszczenie własnej historii, co wygląda jak błąd.
Na szczęście tokeny są oznaczone. Każde zdarzenie messages zawiera metadane, a tokeny podsumowania mają wartość lc_source równą "summarization". Pomiń je i zachowaj resztę.
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “...”}]},
stream_mode=”messages”,
version=”v2”,
):
token, metadata = chunk[”data”]
if metadata.get(”lc_source”) == “summarization”:
continue
# ... handle real agent tokens here
Jedna weryfikacja metadanych wystarczy, aby odróżnić agenta rozmawiającego z użytkownikiem od systemu komunikującego się samego ze sobą. Wyświetlany jest tylko pierwszy przypadek. Jest to prosta czynność, która przynosi znaczące korzyści pod względem jakości.
Obserwowanie przebiegu procesu versus prowadzenie rozmowy
Striming i rozmowy są często mylone, dlatego należy wyraźnie je rozróżnić. Wszystko, co zostało powiedziane powyżej, dotyczy obserwowania przebiegu procesu w czasie rzeczywistym. Każda funkcja stream, podobnie jak każda funkcja invoke, działa domyślnie jednorazowo. Agent wykonuje zadanie, kończy je i zapomina o nim, więc kolejna komunikacja rozpoczyna się bez żadnych informacji o wcześniejszej wymianie zdań.
Kontynuowanie rozmowy na przemian, tak aby fraza „napraw też drugi test” funkcjonowała, ponieważ agent pamięta pierwszą naprawę, to odrębny mechanizm. Opiera się on na wątkach oraz wskaźniku stanu, czyli warstwie trwałości, w której trwała struktura grafowa zapisuje swój stan. Należy pamiętać o różnicy: przepływ danych zapewnia widoczność w trakcie jednego wykonywania, natomiast wątki zapewniają ciągłość między poszczególnymi wykonywaniami. Są one niezależne, a często będzie się chciało ich obu.
Niewielki interfejs terminalowy pokazujący postępy
Następnie należy połączyć wszystko w konsolę, którą można faktycznie uruchomić. Powinna ona wyświetlać jedną linię na krok w miarę postępów agenta oraz przesyłać tekst agenta w momencie jego zapisywania, dlatego należy poprosić o oba tryby jednocześnie i rozgałęzić działanie w zależności od wartości pola type.
from deepagents import create_deep_agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
system_prompt="You are a code-maintenance assistant for a small Python repo.",
) # ①
prompt = "List the files you'd look at first to find a failing test, and explain why."
for chunk in agent.stream(
{"messages": [{"role": "user", "content": prompt}]},
stream_mode=["updates", "messages"], # ②
version="v2",
):
if chunk["type"] == "updates": # ③
for node_name in chunk["data"]:
print(f"\n[step] {node_name}")
elif chunk["type"] == "messages": # ④
token, metadata = chunk["data"]
if metadata.get("lc_source") == "summarization": # ⑤
continue
if token.content:
print(token.content, end="", flush=True) # ⑥
Kod jest krótki, ale każda jego część pełni określoną funkcję. Struktura pozostaje niezmieniona, ponieważ create_deep_agent tworzy agenta dokładnie tak samo jak wcześniej. Dwa tryby wymagane w jednej wywołaniu oznaczają, że jeden pętla może dostarczać zarówno wskaźnik kroku, jak i strumień tekstu. Gdy fragment to zdarzenie updates, pętla wyświetla wiersz [step] dla węzła, który właśnie został zakończony. Gdy jest to zdarzenie messages, pętla wyświetla wygenerowany tekst. Sprawdzenie podsumowania z poprzedniego rozdziału znajduje się wewnątrz tej gałęzi, dzięki czemu zadania utrzymaniowe narzędzia nigdy nie docierają do terminala, a puste tokeny są pomijane przed wyświetleniem.
W rezultacie za każdym razem, gdy agent robi krok do przodu, pojawia się linia [step], która jest przeplatana z tekstem przedstawianym znak po znaku; filtr jest już wdrożony, więc nic nie ucieka. Nie wygląda to elegancko, ale jest uczciwe. Użytkownik może zobaczyć, że agent jest aktywny, i obserwować, jak myśli.
Nowsze wersje Deep Agents oferują wyższościowy API do tego celu, agent.stream_events(...) w wersji version="v3", wraz z typowanymi projekcjami takimi jak stream.subagents. Dokumentacja skłania nowe aplikacje do korzystania z niego, ponieważ każda projekcja jest samodzielnym, uporządkowanym iteratorem i nie ma potrzeby tworzenia odgałęzień na podstawie stream_mode ani samodzielnego dekodowania nazw przestrzeni. Nadal warto najpierw poznać metodę stream_mode, ponieważ projekcje opierają się na tych samych zdarzeniach. Sprawdź aktualną dokumentację przed zdecydowaniem się na użycie złożonego interfejsu.
Lista kontrolna do zastosowania dziś
- Najpierw znajdź każde
invoke, na które czeka osoba, i zamień je naagent.stream(...)w trybieupdates. Nawet proste logi kroków są lepsze niż pusta ekran. - Przyjrzyj się jednemu rzeczywistemu fragmentowi. Uruchom z parametrem
version="v2", wydrukuj wynik i przeczytaj polatype,nsorazdata, aby mieć dokładny obraz jego struktury. - Zanim rzeczywiści użytkownicy zobaczą wyniki, usuń wszystkie fragmenty
messages, których metadane wskazująlc_source == "summarization". - Jeśli agent przekazuje zadania podagentom, ustaw
subgraphs=Truei uporządkuj wyniki według nazwy przestrzeni – po jednej grupie na agenta. - Zachowaj w repozytorium małą konsolę podobną do powyższej. Pozwala ona obserwować działanie każdej nowej funkcjonalności zamiast zgadywać na podstawie ostatniej wiadomości.
Co się zmieniło, a co nie
Wynik końcowy: połączenie blokujące przekształciło się w transmisję na żywo. stream_mode określa stopień szczegółowości informacji, ns wskazuje, kto stworzył każde zdarzenie, a filtr podsumowujący usuwa własne komunikaty narzędzia – teraz powinno być jasne, że obserwowanie procesu a zapamiętywanie go to różne problemy.
Nic się nie zmieniło w agencie ani w jego pętli. Jedyną dodatkową elementem było okno, które wyznacza granicę pomiędzy narzędziem, któremu ludzie ufają, a tym, które aktualizują w panice. Gdy agent pokazuje swoją pracę, użytkownicy wierzą, że funkcjonuje prawidłowo.
Dlatego następnym razem, gdy ktoś zgłosi, że agent „zawisa”, pomijaj narzędzie do analizy i najpierw spróbuj transmisji na żywo. Agent cały czas „rozmawiał” – potrzebował kanału komunikacji.
Literatura pokrewna
- Twoj system plików agenta to decyzja: wyjaśnienie backendów Deep Agents — Jak backendy LangChain Deep Agents decydują, co zostanie zachowane: pięć wbudowanych rozwiązań, mechanizm routingu CompositeBackend, pułapka nazw przestrzeni dla magazynów wielu użytkowników oraz wzorce sandbox.
- Beta zarządzanych Deep Agents: sposób oparty na katalogach do wdrażania agentów LangChain — Zarządzane Deep Agents od LangChain przekształcają folder projektu w hostowanego agenta. Co robi każdy plik, jak działa mda deploy oraz jakie są braki w wersji beta, które warto znać.