Praktische Hinweise: Streaming-Responses in LangGraph – 3 praktische Muster, die jeder benötigt
Schritt-für-Schritt-Anleitung zu den Praktischen Hinweisen: Streaming-Responses in LangGraph – 3 praktische Muster, die jeder benötigt: Verträge, Überprüfungen sowie Code-Blöcke für Teams, die dieses Muster einsetzen.
Dieser Leitfaden zeigt Schritt für Schritt den Weg von Rohstoffen bis zu einem funktionsfähigen System für: Streaming Responses in LangGraph: 3 Praktische Muster, die jeder Agent-Entwickler kennen sollte. Der Fokus liegt auf ausführbaren Schritten, expliziten Überprüfungen sowie Code, den man ohne Rätseln über die Absicht direkt in ein Repository einfügen kann. In der Übersichtsphase sollten Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien definieren, bevor Sie Code ändern. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholungsversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.
Warum Streaming wichtiger ist, als viele denken
Wenn Sie die Phase „Warum Streaming wichtiger ist“ durchgehen, notieren Sie zunächst den Vertrag: erforderliche Eingaben, Erfolgsignal sowie was bei einem teilweisen Versagen passiert. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Legen Sie nach teuren Schritten einen Checkpoint an. Das Resume sollte bei einem Neversuch eines späteren Knotens nicht denselben LLM-Aufruf erneut berechnen.
Das Beispiel, an dem wir arbeiten
Beim Arbeiten an der Stufe „Das Beispiel, das wir durchführen“, sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Stufe als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Kontrollpunkte nach aufwändigen Schritten. Das Wiederaufnehmen des Vorgangs sollte keine erneute Abrechnung für denselben LLM-Aufruf veranlassen, wenn ein Operator einen späteren Knoten erneut versucht.
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())
Schritt 1: Verstehen Sie zunächst die Zustandsarchitektur
Beim Bearbeiten von Schritt 1 „Verständnis der Phase“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsam genutzte Umgebungen übergeht. Legen Sie nach aufwändigen Schritten einen Kontrollpunkt an. Das Wiederaufnehmen des Vorgangs sollte keine erneute Gebühr für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut versucht. Beim Bearbeiten von Schritt 1 „Verständnis der Phase“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholte Versuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen.
class State(TypedDict):
messages: Annotated[list[BaseMessage], operator.add]
{"messages": [some_new_message]}
Warum das für Streaming wichtig ist
„Warum das für Streaming wichtig ist“ funktioniert am besten, wenn es als messbare Struktur behandelt wird. Erfassen Sie eine gelungene Transkription, einen Fehlerfall sowie die Notizen zum Rollback, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren. Halten Sie den Zustand des Graphen einfach und typisiert – verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Problemen bei der Wiederaufnahme nach Unterbrechungen.
Schritt 2: Der Graph selbst ist absichtlich einfach
Die Phase „Schritt 2: Das Diagramm“ funktioniert am besten, wenn sie als messbare Oberfläche betrachtet wird. Erfassen Sie ein gelungenes Beispiel, einen Fehlfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Arbeiten ab. Halten Sie den Zustand des Diagramms einfach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Arbeit.
chatbot_node
Die Chatbotnode-Phase funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Erfassen Sie außerdem die Laufzeiten sowie die Kosten pro Token oder Abfrage zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsam genutzte Umgebungen übergeht. Halten Sie den Zustand des Graphen einfach und typisiert – verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen des Ablaufs. Die Chatbotnode-Phase funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Dokumentieren Sie den erfolgreichen Ablauf sowie den Wiederherstellungsprozess gemeinsam. Wiederholversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.
def chatbot_node(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
dummy_node
Zur dummynode-Phase sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor der Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Bevorzugen Sie kleine, testbare Einheiten vor umfangreichen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Setzen Sie menschliche Freigabe für Schritte ein, die Geld ausgeben oder Produktionsdaten ändern. Kompilierzeitbezogene Verbindungen bedeuten noch nicht vollständige Geschäftsabläufe.
def dummy_node(state: State) -> State:
return state
START → chatbot → dummy → END
Method 1: stream_mode="updates"
Zur Phase der Aktualisierungen im Streammode nach Methode 1 sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor der Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Setzen Sie menschliche Freigabe für Schritte ein, die Geld ausgeben oder Produktionsdaten ändern. Eine Kompilierzeit-Verkabelung entspricht nicht der Geschäftsabschlussfähigkeit.
for event in graph.stream(..., stream_mode="updates"):
Was es bewirkt
In der Phase „Was es tut“ sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor der Code geändert wird. Die Bediener sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Erhalten Sie neben den funktionalen Ergebnissen auch Aufzeichnungen der Laufzeiten sowie der Kosten für Token oder Abfragen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo-Umgebung in eine gemeinsam genutzte Umgebung wechselt. Setzen Sie menschliche Freigabevorgänge bei Schritten ein, die Geld kosten oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet noch nicht vollständige Geschäftsabdeckung. In der Phase „Was es tut“ sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor der Code geändert wird. Die Bediener sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallablauf. Wiederholungsversuche, menschliche Kontrollpunkte und die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.
{
"chatbot": {
"messages": [...]
}
}
{
"dummy": {
"messages": [...]
}
}
Warum dieser Modus nützlich ist
Beim Arbeiten an der Phase „Warum dieser Modus nützlich ist“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Erstellen Sie Zwischenkontrollpunkte nach aufwändigen Schritten. Das Wiederaufnehmen des Vorgangs sollte keine erneuten Aufrufe derselben LLM-Funktion veranlassen, wenn ein Operator einen späteren Knoten erneut versucht.
Beste Anwendungsfälle
Während der Phase der Best-Use-Cases-Erstellung sollte man zunächst einen Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikatoren sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Kontrollpunkte nach aufwändigen Schritten. Das System sollte bei einer Neuanfangsversuche eines Operators für einen späteren Knoten nicht denselben LLM-Aufruf erneut berechnen.
Reales Beispiel
Während der Phase mit dem Real-World-Beispiel sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsam genutzte Umgebungen übergeht. Legen Sie nach aufwändigen Schritten einen Kontrollpunkt an. Die Wiederaufnahme des Vorgangs sollte keine erneuten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut versucht. Während der Phase mit dem Real-World-Beispiel sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.
Methode 2: stream_mode="values"
Die Werte des Streammode-Modus „Method 2“ funktionieren am besten, wenn sie als messbare Oberfläche betrachtet werden. Erfassen Sie ein erfolgreiches Beispiel, einen Fehlerfall sowie die Notizen zum Rollback, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf. Halten Sie den Zustand der Graphen einfach und typisiert – verschachtelte Datenstrukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Problemen bei der Wiederaufnahme nach Unterbrechungen.
for snapshot in graph.stream(..., stream_mode="values"):
Was es bewirkt
Die Phase „Was es tut“ funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie ein gelungenes Beispiel, einen Fehlerfall sowie die Notizen zur Rücksetzung, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Halten Sie den Zustand der Graphen strukturiert und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.
print(f" State has {len(snapshot['messages'])} message(s) now")
print(snapshot["messages"])
Warum das wichtig ist
Die Phase „Warum das wichtig ist“ funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsam genutzte Umgebungen übergeht. Halten Sie den Zustand der Grafiken einfach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen des Ablaufs. Die Phase „Warum das wichtig ist“ funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Wiederherstellungsprozess. Wiederholversuche, menschliche Kontrollen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.
Beste Anwendungsfälle
In der Phase der besten Anwendungsfallgestaltung sollten Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Bevorzugen Sie kleine, testbare Einheiten vor umfangreichen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Setzen Sie menschliche Freigabe bei Schritten ein, die Geld ausgeben oder Produktionsdaten ändern. Kompilierzeitbezogene Verbindungen bedeuten noch nicht vollständige Geschäftsabläufe.
Reales Beispiel
In der Phase des Realweltbeispiels sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor der Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Setzen Sie menschliche Freigabe für Schritte ein, die Geld ausgeben oder Produktionsdaten ändern. Eine Kompilierzeit-Verkabelung entspricht nicht der Geschäftsabschlussfähigkeit.
Methode 3: astream_events() für Token-für-Token-Streaming
Für die Method-3-Astreamevents für die Phase sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Erhalten Sie neben den funktionalen Ergebnissen auch Aufzeichnungen der Laufzeiten sowie der Kosten für Token oder Abfragen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo-Umgebung in eine gemeinsam genutzte Umgebung wechselt. Wählen Sie bei dem nächsten Schritt, der ein Codeabschnitt oder einen Toolaufruf ist, strukturierte Ausgaben mit Schema-Validierung statt freier Prosa. Für die Method-3-Astreamevents für die Phase sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallplan. Wiederholungsversuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.
async for event in graph.astream_events(..., version="v2"):
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"].content
Was es bewirkt
Während der Phase „Was es bewirkt“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Legen Sie nach teuren Schritten Kontrollpunkte an. Das Wiederaufnehmen des Vorgangs sollte keine erneute Abrechnung für denselben LLM-Aufruf veranlassen, wenn ein Operator einen späteren Knoten erneut ausführt.
Warum das wichtig ist
Während der Phase „Warum ist das wichtig?“ sollten Sie zunächst einen Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikatoren sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Kontrollpunkte nach aufwändigen Schritten. Das Resume sollte bei einer Neuprobe eines späteren Knotens nicht denselben LLM-Aufruf erneut berechnen.
Warum astream_events() asynchron ist
Beim Bearbeiten der Phase „Warum sind astreamevents asynchron“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo-Umgebung in gemeinsam genutzte Umgebungen übergeht. Legen Sie nach aufwändigen Schritten einen Kontrollpunkt an. Die Wiederaufnahme des Prozesses sollte keine erneuten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt. Beim Bearbeiten der Phase „Warum sind astreamevents asynchron“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholte Versuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen.
asyncio.run(token_stream())
Verständnis des Ereignisfilters
Die Phase „Verständnis des Ereignisfilters“ funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie zunächst ein Beispiel für einen erfolgreichen Ablauf, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich verweisen und nicht auf einen verworrenen Ablaufprozess. Halten Sie den Zustand der Graphen einfach und typisiert. Verschachtelte Datenstrukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen dazu, dass der Ablauf nach Unterbrechungen nicht fortgesetzt werden kann.
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"].content
Die 3 Streaming-Modi in einfachen Worten
Die Phase der 3 Streaming-Modi funktioniert am besten, wenn sie als messbare Oberfläche betrachtet wird. Erfassen Sie einen erfolgreichen Fall, einen Fehlerfall sowie die Notizen zum Rollback, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Halten Sie den Zustand des Graphen flach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.
updates
Die Update-Phase funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsam genutzte Umgebungen übergeht. Halten Sie den Zustand der Graphen einfach und typisiert – verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen des Ablaufs. Die Update-Phase funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Wiederherstellungsprozess. Wiederholversuche, menschliche Kontrollen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.
values
Zur Wertebearbeitungsphase sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor der Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Zur Vorzugsbehandlung kommen kleine, testbare Einheiten vor umfangreichen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Menschliche Freigabe sollte bei Vorgängen erforderlich sein, die Geld ausgeben oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung.
astream_events
Für die Astreamevents-Phase sollten Sie vor dem Ändern des Codes die Eingaben, den Verantwortlichen für den jeweiligen Schritt sowie die Abbruchkriterien definieren. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Setzen Sie menschliche Freigabe für Schritte ein, die Geld ausgeben oder Produktionsdaten ändern. Eine Kompilierzeit-Verkabelung entspricht nicht der Geschäftsabschlussfähigkeit.
Welchen Streaming-Modus sollten Sie verwenden?
Für den Schritt „Welcher Streaming-Modus soll genutzt werden?“ sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erhalten Sie Aufzeichnungen der Laufzeiten sowie der Kosten für Token oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo-Umgebung in eine gemeinsam genutzte Umgebung wechselt. Setzen Sie menschliche Freigabevorgänge bei Schritten ein, die Geld kosten oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung. Für den Schritt „Welcher Streaming-Modus soll genutzt werden?“ sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallablauf. Wiederholungsversuche, menschliche Freigabestufen und die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.
.Verwenden Sie updates, wenn:
Wenn Sie die Phase „Use updates when“ durchgehen, notieren Sie zunächst den Vertrag: erforderliche Eingaben, Erfolgsignal sowie was bei einem teilweisen Versagen geschieht. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Erstellen Sie Kontrollpunkte nach kostspieligen Schritten. Das Wiederaufnehmen des Vorgangs sollte keine doppelte Abrechnung für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt.
Verwenden Sie values, wenn:
Beim Arbeiten an „Use values when stage“ sollten Sie zunächst einen Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Kontrollpunkte nach aufwändigen Schritten. Das Resume sollte bei einem Neversuch eines Operators an einem späteren Knoten nicht denselben LLM-Aufruf erneut berechnen.
Nutzen Sie astream_events, wenn:
Beim Arbeiten an dem Abschnitt „Use astreamevents when stage“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo-Umgebung in gemeinsam genutzte Umgebungen übergeht. Legen Sie nach aufwändigen Schritten einen Kontrollpunkt an. Das Wiederaufnehmen des Vorgangs sollte keine erneuten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt. Beim Arbeiten an dem Abschnitt „Use astreamevents when stage“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.
Ein besseres mentales Modell: Streaming für Benutzer versus Streaming für Entwickler
Die Phase „Ein besseres mentales Modell“ funktioniert am besten, wenn sie als messbarer Ansatz betrachtet wird. Erfassen Sie einen gelungenen Ablauf, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern.
Streaming für Benutzer
Streaming für Entwickler
Ein Produktionsmuster, das Sie wahrscheinlich benötigen werden
Häufige Fehler beim Streamen von LangGraph-Antworten
1. Verwendung des Token-Streamings, obwohl Sie eigentlich Node-Updates benötigen
2. Erwartung, dass values sich wie Token-Streaming verhalten
3. Vergessen, dass astream_events() asynchron ist
4. Fehlende Filterung der Ereignistypen
5>Erstellung von Graphen, die gestreamt werden, aber keinen sinnvollen Zustand offenbaren
Fortgeschrittter Tipp: Streaming wird in mehrschrittigen Agenten-Graphen deutlich mächtiger
updates
values
astream_events
Eine kleine Verbesserung, die Sie hinzufügen könnten
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)