Praktyczne wskazówki: Strumieniowanie odpowiedzi w LangGraph: 3 praktyczne wzorce, które powinny być stosowane przez wszystkich.
Szczegółowy przewodnik po praktycznych wskazówkach: Strumieniowanie odpowiedzi w LangGraph: 3 praktyczne wzorce, które powinny być stosowane przez zespoły tworzące takie rozwiązania, wraz z kontraktami, kontrolami oraz gotowymi fragmentami kodu.
To przewodnik pokazuje, jak przejść od surowców do działającego systemu w celu realizacji: Strumieniowanie odpowiedzi w LangGraph: 3 praktyczne wzorce, które powinien znać każdy deweloper agentów. Skupiamy się na krokach operacyjnych, wyraźnych sprawdzeniach oraz kodzie, który można bez problemu wdrożyć do repozytorium, bez konieczności domyślania się intencji. Na etapie przeglądu należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności domyślania się ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.
Dlaczego strumieniowanie jest ważniejsze, niż myślą ludzie
Gdy przechodzisz przez etap „Dlaczego streaming jest ważniejszy”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki od rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustaw punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
Przykład, nad którym pracujemy
Gdy przechodzisz przez etap „Przykład, który przygotowujemy”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć możliwość cichego, częściowego ukończenia zadania. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy element.
import asyncio
import operator
from typing import Annotated, TypedDict
from dotenv import load_dotenv
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
load_dotenv()
# --- 1. STATE & GRAPH ---
class State(TypedDict):
messages: Annotated[list[BaseMessage], operator.add]
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
def chatbot_node(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
def dummy_node(state: State) -> State:
return state
builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_node("dummy", dummy_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", "dummy")
builder.add_edge("dummy", END)
graph = builder.compile()
# --- 2A. stream_mode="updates" — one event per node ---
print("=== Method 1: stream_mode='updates' ===")
for event in graph.stream(
{
"messages": [HumanMessage("List 3 benefits of LangGraph in one line each")],
},
stream_mode="updates",
):
for node_name, output in event.items():
print(f" [{node_name}] {output['messages']}")
# --- 2B. stream_mode="values" — full state after each node ---
print("\n=== Method 2: stream_mode='values' ===")
for snapshot in graph.stream(
{
"messages": [HumanMessage("Say hello in 3 languages")],
},
stream_mode="values",
):
print(f" State has {len(snapshot['messages'])} message(s) now")
print(snapshot["messages"])
# --- 2C. astream_events — token-by-token (async) ---
async def token_stream():
print("\n=== Method 3: Token-by-token streaming ===")
print("🤖 Bot: ", end="", flush=True)
async for event in graph.astream_events(
{
"messages": [HumanMessage("Count from 1 to 5 slowly, one per line")],
},
version="v2",
):
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"].content
if chunk:
print(chunk, end="", flush=True)
print()
asyncio.run(token_stream())
Krok 1: Najpierw zrozum projekt stanu
Gdy przechodzisz przez etap 1 – Zrozumienie, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czas trwania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Ustal punkty kontrolne po kosztownych krokach. System powrotu nie powinien ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł. Gdy przechodzisz przez etap 1 – Zrozumienie, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych wywołań, mechanizmy ludzkiej kontroli oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
class State(TypedDict):
messages: Annotated[list[BaseMessage], operator.add]
{"messages": [some_new_message]}
Dlaczego to ma znaczenie dla strumieniowania
Najlepiej traktować „Dlaczego to ma znaczenie dla scenicznych przedstawień” jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres analizy. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wtórne struktury danych ukrywają informację o tym, który węzeł zapisał dane do którego pola, co utrudnia kontynuację pracy po przerwach.
Krok 2: Sam graf jest celowo prosty
Etap 2, czyli faza Grafu, funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć przypadki częściowego ukończenia pracy bez informacji.
chatbot_node
Faza chatbotnode funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia do analizy. Zapisz jeden idealny zapis rozmowy, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Zapisuj czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwę w kontynuacji pracy po zakłóceniach. Faza chatbotnode funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia do analizy. Zapisz jeden idealny zapis rozmowy, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Dokumentuj zarówno optymalną ścieżkę działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później.
def chatbot_node(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
dummy_node
W fazie dummynode należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Lepiej używać małych, łatwych do przetestowania jednostek niż rozbudowanych skryptów. Gdy dany krok zawiedzie, powinno to wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Konieczna jest ludzka akceptacja w przypadkach, gdy dochodzi do wydawania pieniędzy lub modyfikacji danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się kompletności biznesowej.
def dummy_node(state: State) -> State:
return state
START → chatbot → dummy → END
Metoda 1: stream_mode="updates"
W fazie aktualizacji w trybie strumieniowym Metody 1 należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy artefaktom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Wymagaj ludzkiej aprobaty w przypadkach, gdy dochodzi do wydawania pieniędzy lub zmiany danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełności biznesowej.
for event in graph.stream(..., stream_mode="updates"):
Jak to działa
Na etapie „Co robi”, zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zapisz czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Wprowadź ludzką aprobatę dla przypadków, w których wydawane są pieniądze lub zmieniane są dane produkcyjne. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności rozwiązania biznesowego. Na etapie „Co robi”, zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych działań, ludzkie kontrolne punkty oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
{
"chatbot": {
"messages": [...]
}
}
{
"dummy": {
"messages": [...]
}
}
Dlaczego ten tryb jest przydatny
Podczas przechodzenia przez etap „Dlaczego ten tryb jest przydatny”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustaw punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
Najlepsze przypadki użycia
Gdy przechodzisz przez etap najlepszych przypadków użycia, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Ustal punkty kontrolne po kosztownych krokach. System powrotu nie powinien ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
Przykład z praktyki
Gdy przechodzisz przez etap przykładu z rzeczywistego świata, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czas wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł. Gdy przechodzisz przez etap przykładu z rzeczywistego świata, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych wywołań, kontrolne punkty ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodatkowej optymalizacji.
Metoda 2: stream_mode="values"
Wartości mode'u strumieniowego Metody 2 działają najlepiej, gdy traktuje się je jako mierzalną powierzchnię. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wtórne struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, i utrudniają kontynuację pracy po przerwach.
for snapshot in graph.stream(..., stream_mode="values"):
Jak to działa
Etap „Co robi” funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w prostej formie i określonej typowości. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po interwencjach.
print(f" State has {len(snapshot['messages'])} message(s) now")
print(snapshot["messages"])
Dlaczego to ma znaczenie
Etap „Dlaczego to ma znaczenie” funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia do analizy. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Zapisuj czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwę w kontynuacji pracy po zakłóceniach. Etap „Dlaczego to ma znaczenie” funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia do analizy. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Dokumentuj zarówno optymalną ścieżkę działania, jak i ścieżkę przywracania do normalnego stanu. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później.
Najlepsze przypadki użycia
W fazie najlepszych przypadków użycia należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Konieczne jest ludzkie zatwierdzenie w tych przypadkach, gdy dochodzi do wydatków lub zmian w danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się kompletności procesu biznesowego.
Przykład z praktyki
W fazie przykładu z rzeczywistego świata należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć przypadkowe, częściowe ukończenie zadań. Wymagaj ludzkiej aprobaty w przypadkach, gdy dochodzi do wydawania pieniędzy lub modyfikacji danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności procesu biznesowego.
Metoda 3: astream_events() do przesyłania danych token po tokenie
Dla astreameventów Metody 3 przeznaczonych do użytku na etapie realizacji projektu, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. W przypadku, gdy następnym krokiem jest kod lub wywołanie narzędzia, lepiej używać ustrukturyzowanych wyników z walidacją schematu niż prostego tekstu. Dla astreameeventów Metody 3 przeznaczonych do użytku na etapie realizacji projektu, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżki naprawcze. Próby ponownego wykonania, kontrole ludzkie oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
async for event in graph.astream_events(..., version="v2"):
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"].content
Jak to działa
Podczas przechodzenia przez etap „Jak to działa”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co się dzieje w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustalaj punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
Dlaczego to ma znaczenie
Gdy przechodzisz przez etap „Dlaczego to ma znaczenie”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć możliwość cichego, częściowego ukończenia zadania. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy element.
Dlaczego astream_events() jest asynchroniczny
Gdy przechodzisz przez etap Why astreamevents Is Async, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demo do środowisk współdzielonych. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł. Gdy przechodzisz przez etap Why astreamevents Is Async, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych wywołań, mechanizmy ludzkiej kontroli oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.
asyncio.run(token_stream())
Rozumienie filtru zdarzeń
Etap rozumienia filtru zdarzeń działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny zapis transkrypcji, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres badania. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co utrudnia kontynuację pracy po przerwach.
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"].content
3 tryby przepływu danych w prostym języku
Faza 3 trybów strumieniowania działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przepis, jeden przypadek niepowodzenia oraz notatkę o cofnięciu zmian, zanim rozszerzysz zakres. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy artefaktom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po przerwach.
updates
Faza aktualizacji działa najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane do którego pola, i powodują przerwę w kontynuacji działania po zakłóceniach. Faza aktualizacji działa najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Dokumentuj zarówno pomyślny przebieg działania, jak i ścieżkę odzyskiwania. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później.
values
Na etapie określania wartości należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Konieczna jest ludzka akceptacja w przypadkach, gdy dochodzi do wydawania pieniędzy lub modyfikacji danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się kompletności rozwiązania biznesowego.
astream_events
Dla etapu astreamevents należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy artefaktom, zdefiniuj sprawdzenia sukcesu i odrzuć ciche, częściowe ukończenie zadania. Wymagaj ludzkiej aprobaty w przypadkach, gdy dochodzi do wydawania pieniędzy lub zmiany danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności biznesowej.
Jaki tryb strumieniowania powinieneś wybrać?
Dla etapu „Which Streaming Mode Should” należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zapisuj czas trwania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Wprowadź ludzką aprobatę dla przypadków, w których wydawane są pieniądze lub zmieniane są dane produkcyjne. Połączenia skompilowane w czasie kompilacji nie równają się kompletności rozwiązania biznesowego. Dla etapu „Which Streaming Mode Should” należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych działań, ludzkie kontrolne punkty oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
.Używaj updates, gdy:
Gdy przechodzisz przez etap „Używaj aktualizacji”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustaw punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
Używaj values, gdy:
Gdy pracujesz nad rozdziałem „Używanie wartości w fazie”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć możliwość cichego, częściowego ukończenia zadania. Ustaw punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
Używaj astream_events, gdy:
Gdy pracujesz nad etapem „Use astreamevents when stage”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł. Gdy pracujesz nad etapem „Use astreamevents when stage”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych wywołań, mechanizmy ludzkiej kontroli oraz obsługa wiadomości błędowych są częścią produktu, a nie elementami dodatkowej optymalizacji.
Lepszy model mentalny: streaming dla użytkowników vs streaming dla programistów
Etap „Lepszy model mentalny” działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy.
Streaming skierowany do użytkowników
Streaming skierowany do programistów
Wzorzec produkcyjny, który prawdopodobnie będziesz chciał wykorzystać
Powszechne błędy przy streamowaniu odpowiedzi LangGraph
1. Używanie streamowania tokenów, gdy faktycznie potrzebne są aktualizacje węzłów
2. Oczekiwanie, że values będą zachowywać się jak streamowanie tokenów
3> Zapominanie, że astream_events() jest operacją asynchroniczną
4. Brak filtrowania typów zdarzeń
5. Tworzenie grafów, które streamują dane, ale nie ujawniają istotnego stanu
Wskazówka zaawansowana: streamowanie staje się znacznie potężniejsze w wieloetapowych grafach agentów
updates
values
astream_events
Niewielka poprawka, którą możesz wprowadzić
for event in graph.stream(
{
"messages": [HumanMessage("List 3 benefits of LangGraph in one line each")],
},
stream_mode="updates",
):
for node_name, output in event.items():
print(f"\nNode: {node_name}")
for message in output["messages"]:
print(message)