langchain deep-agents ai-agents streaming python

Streaming von LangChain Deep Agents: Echtzeit-Pläne, Tool-Aufrufe und Tokens

Ersetzen Sie einen blockierenden Aufruf durch einen Stream, damit Nutzer die Arbeit eines Deep Agents sehen können: die drei Stream-Modi, Namensräume für Subagenten sowie das Filtern und Zusammenfassen von Tokens.

Verlag: REACTAPP.TOPAuf dieser Website hinzugefügt 28. Okt. 202610 Min. Lesezeit2188 Wörter

Quellen: LangChain Deep Agents: Your AI Agent Is Not Frozen. It Just Cannot Tell You It Is Working. II

Sehr wenige Dinge untergraben das Vertrauen schneller als ein eingefrorenes Bildschirmbild. Wenn ein Benutzer nicht erkennen kann, ob ein Agent nachdenkt, stecken geblieben ist oder bereits „tot“ ist, geht er vom Schlimmsten aus. Streaming ersetzt diese Stille durch eine Echtzeit-Überwachung: Der Plan, die Aufrufe von Tools sowie die gesprochenen Worte werden genau in dem Moment angezeigt, in dem sie stattfinden, sodass die Nutzer sehen können, dass etwas vor sich geht.

Warum `invoke` einen tiefen Agenten wie „tot“ erscheinen lässt

Probieren Sie es aus: Führen Sie einen tiefen Agenten bei einer echten Aufgabe aus, während jemand zuschaut. Der Aufruf erfolgt über invoke, danach bleibt der Bildschirm vierzig Sekunden lang unbewegt. Im Hintergrund liest der Agent Dateien, entwirft einen Plan, ruft Tools auf und durchdenkt die Situation. Der Benutzer sieht nur eine bewegungslose Box sowie einen Ladebalken – was sowohl auf anstrengende Arbeit als auch auf einen stillen Absturz hindeuten könnte. Da beides identisch aussieht, laden die Nutzer die Seite neu und melden einen Fehler, weil der Agent hängen bleibt.

Nichts wurde angezeigt. Der Agent hatte einfach keinen Kanal, um sich selbst zu beschreiben. invoke-Blöcke sind aus Gründen der Konzeption so gestaltet, dass sie den gesamten Loop im Verborgenen ausführen und anschließend eine einzige, fertige Antwort zurückgeben – was genau richtig ist für Skripte oder Cron-Jobs, die von niemandem beobachtet werden. Es ist jedoch die falsche Wahl, sobald ein Mensch auf der anderen Seite sitzt.

Die Lösung lässt den Loop unberührt. Sie ersetzt den blockierenden Aufruf durch einen Stream, sodass Pläne, Tool-Aufrufe und Tokens in Echtzeit angezeigt werden.

Von `invoke` zu `stream`

Streaming steht in engem Zusammenhang mit invoke. Anstatt auf einen Endzustand zu warten, wird agent.stream(...) durchlaufen, und jeweils ein Teilinhalt wird bereitgestellt, sobald der Loop etwas ausführt. Ein asynchroner Pendant, agent.astream(...), hat die gleiche Struktur und wird von einem Webserver aufgerufen.

Das mentale Modell ist einfach gestrickt. Das robuste LangGraph unter einem Deep Agent gibt bei jedem Schritt ein Ereignis ab, und der Streaming-Prozess leitet diese Ereignisse weiter, sobald sie erzeugt werden. Es handelt sich dabei nicht um einen anderen Agenten oder eine andere Schleife, sondern lediglich um einen Einblick in das bereits bekannte Modell.

Ein Detail muss zunächst geklärt werden, da jedes Beispiel davon abhängt: das v2-Stream-Format, das mit version="v2" aktiviert wird. Jeder Datenblock ist ein kleines Dictionary mit drei Schlüsseln:

type gibt an, um welche Art von Ereignis es sich handelt.
ns, der Namespace, angibt, wer das Ereignis erzeugt hat.
data enthält den Payload.

Diese drei Schlüssel bleiben unverändert, egal was gestreamt wird, wodurch der Code lesbar bleibt und nicht in einer Menge von Tuple-Entpackungen endet.

Drei Modi, drei Ebenen

Ein Stream ist keine einzige Feuerwehrschlange. Die Einstellung stream_mode bestimmt die Art der Ereignisse, die Sie erhalten, wobei die drei gängigen Optionen mit drei Detailstufen übereinstimmen.

updates bietet eine schrittweise Ansicht. Jedes Mal, wenn ein Graphenknoten abgeschlossen ist, tritt ein Ereignis ein – in der Praxis bedeutet das, dass der Agent einen Schritt ausgeführt hat oder ein Tool gelaufen ist. Dies ist die Ebene für ein Fortschrittsprotokoll, das ausreicht, um zu zeigen, wie der Agent seine Aufgaben bearbeitet, ohne die Nutzer mit zu vielen Informationen zu überfluten, und stellt eine sinnvolle Standard-Einstellung für einen übersichtlichen „Hier macht der Agent gerade etwas“-Feed dar.

messages ist die Tokenansicht. Die Ausgabe des Modells kommt Stück für Stück, während sie generiert wird – genauso wie beim Tippen eines Chat-Assistenten. Verwenden Sie sie, wenn eine Person den Text des Agenten genau so liest, wie er erscheint. Auch Aufrufe von Streaming-Tools landen hier, sodass Sie „Aufruf von read_file“ genau in dem Moment ankündigen können, in dem das Modell den Aufruf ausführt, und nicht erst, nachdem er abgeschlossen ist.

values ist die Snapshots-Ansicht. Nach jedem Schritt erhalten Sie den gesamten Zustand des Agenten, nicht nur die Änderungen. Sie ist aufwändiger in der Verarbeitung und wird seltener benötigt, doch sie eignet sich, wenn das vollständige Bild zu jedem Zeitpunkt wichtig ist – beispielsweise um einen Echtzeit-Überblick über die To-Do-Liste oder Arbeitsdateien des Agenten anzuzeigen.

Durch das Übergeben einer Liste von Modi werden mehrere gleichzeitig zurückgegeben, so funktionieren in der Regel benutzerfreundliche Benutzeroberflächen: updates steuert einen Schrittanzeiger, während messages den sichtbaren Text liefert. Das grobe Beispiel sieht so aus, ein sauberer Fortschrittsverlauf von updates:

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}”)

Was ausgegeben wird, ist eine Liste der Graphenschritte, sobald jeder abgeschlossen ist – eine Zeile pro Modulumschritt und pro Toolaufruf. Ein zuvor als eingefroren erscheinendes Bildschirmbild bewegt sich nun sichtbar weiter. Wenn man stream_mode stattdessen auf "messages" setzt, erhält man den Text selbst, der direkt ausgegeben werden kann, während der Agent schreibt.

Unterscheidung zwischen dem Hauptagenten und seinen Unteragenten

Ein tiefer Agent arbeitet selten lange Zeit allein. Sobald er eine Unteraufgabe an einen Unteragenten weitergibt, erzeugen zwei oder mehr Agenten gleichzeitig Ausgaben. Wenn man alles in einen einzigen Stream sendet, entsteht ein Durcheinander: Ein Teil eines Satzes vom Koordinator, einige Token von einem Forscher, ein Toolaufruf aus unklarer Quelle. Das ns-Feld löst dieses Problem. Jeder Datenblock enthält ein solches Feld, wenn man die Untergraphen-Streaming-Funktion mit subgraphs=True aktiviert.

Die Regel ist einfach anzuwenden. Wenn der Namespace leer ist, hat der Hauptagent das Ereignis ausgelöst. Enthält er einen tools:-Abschnitt, dann hat ein Unteragent dies getan, wobei der Abschnitt den jeweiligen Unteragenten angibt. Die Ausgabe kann anschließend nach Ursprung sortiert werden: Der Koordinatortext hier, die Arbeit jedes Unteragents in einer eigenen Spalte.

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)  # ⑥

Sechs Aspekte sind in diesem Auszug wichtig. Erstens verleiht subgraphs=True ns seine Bedeutung, da ohne dieses Parameter die Ereignisse von Subagenten keinen unterscheidbaren Namespace haben. Zweitens behält die Schleife nur messages-Ereignisse bei, da sie Tokens und nicht Schritte weiterleitet. Drittens wird die Nutzlast jedes Abschnitts in ein Token zusammen mit seinen Metadaten aufgeteilt. Viertens erledigt der Namespace-Test die eigentliche Arbeit: Ein tools:-Segment irgendwo in ns kennzeichnet das Token als stammend von einem Subagenten. Fünftens wird das resultierende Boolesche Wert in eine lesbare Quellbezeichnung umgewandelt. Sechstens führt das Ausgeben mit dieser Bezeichnung als Präfix dazu, dass ein vermischtes Stream in zwei benannte Kanäle umgewandelt wird.

Jeder Token wird vor dem Ausgeben mit seinem Ersteller versehen, sodass ein Koordinator, der an einen Forscher weiterleitet, als zwei benannte Ströme erscheint. Eine vollständige Web-Oberfläche erfordert mehr Arbeit, doch der Namespace in jedem Chunk ist derselbe Signal, den eine React-Frontend-Anwendung lesen würde.

Die störenden Tokens, die nicht vom Agenten stammen

Das überrascht die Nutzer bei ihrem ersten längeren Stream-Versuch. Um seinen eigenen Kontext zu verwalten, konsolidiert das System ältere Nachrichten, sobald das Gespräch die Grenzen des Modells erreicht. Der Konsolidierungsprozess ist ebenfalls ein Modellaufruf, wodurch Tokens erzeugt werden, die zusammen mit der eigentlichen Antwort des Agenten in den messages-Stream fließen. Wenn dieser Stream direkt an den Benutzer gesendet wird, sieht es so aus, als würde der Agent plötzlich eine Zusammenfassung seiner eigenen Historie vorlesen – was wie ein Fehler wirkt.

Zum Glück sind die Tokens gekennzeichnet. Jedes messages-Ereignis enthält Metadaten, und die Zusammenfassungstokens tragen einen lc_source-Wert von "summarization". Überspringen Sie diese und behalten Sie den Rest.

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

Eine Überprüfung der Metadaten reicht aus, um den Agenten, der mit dem Benutzer spricht, vom System zu unterscheiden, das mit sich selbst kommuniziert. Nur der erste wird angezeigt. Es handelt sich um eine geringe Aufwand, der jedoch zu einer deutlichen Verbesserung der Qualität führt.

Beobachtung eines Ablaufs versus Aufrechterhaltung einer Konversation

Streaming und Konversation werden oft verwechselt, daher sollte eine klare Trennung erfolgen. Alles, was oben beschrieben wurde, bezieht sich auf die Beobachtung eines Ablaufs, während er stattfindet. Jeder stream-Aufruf, genauso wie jeder invoke-Aufruf, ist standardmäßig einmalig. Der Agent erledigt die Aufgabe, beendet sie und vergisst sie – daher beginnt ein nachfolgender Aufruf ohne Erinnerung an den vorherigen Austausch.

Ein separates Mechanismus sorgt dafür, dass ein Gespräch über mehrere Schritte hinweg aufrechterhalten wird, sodass der Satz „Beheben Sie auch den anderen Test“ funktioniert, weil der Agent sich an die erste Behebung erinnert. Dafür werden Threads sowie ein Checkpointer verwendet – die Persistenzschicht, in der das dauerhafte Graph seine Zustände speichert. Zu unterscheiden ist dabei: Streaming gewährleistet Sichtbarkeit innerhalb eines Laufs, während Threads Kontinuität zwischen den Läufen bieten. Beide sind unabhängig voneinander, und meist werden beide benötigt.

Eine kleine Terminal-Oberfläche, die die Arbeit anzeigt

Danach sollte alles zu einer Konsole zusammengefasst werden, die tatsächlich ausgeführt werden kann. Sie sollte pro Schritt eine Zeile ausgeben, während der Agent voranschreitet, und den Text des Agents in Echtzeit ausgeben, sobald er geschrieben wird. Dafür werden zwei Modi gleichzeitig angefordert, wobei je nach Wert des Feldes type unterschiedlich verfahren wird.

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)  # ⑥

Der Code ist kurz, aber jeder Teil hat eine bestimmte Aufgabe. Die Struktur bleibt unverändert, da create_deep_agent den Agenten genauso wie zuvor erstellt. Zwei Modi in einem Aufruf bedeuten, dass eine einzige Schleife sowohl einen Schrittindikator als auch einen Textstrom liefern kann. Wenn ein Datenblock ein updates-Ereignis ist, gibt die Schleife eine [step]-Zeile für den gerade abgeschlossenen Knoten aus. Bei einem messages-Ereignis wird der erzeugte Prosa-Text ausgegeben. Die Zusammenfassungsprüfung aus dem vorherigen Abschnitt befindet sich innerhalb dieser Schleife, sodass die internen Verarbeitungsschritte des Frameworks niemals das Endresultat erreichen, und leere Tokens werden vor dem Ausgeben übersprungen.

Das Ergebnis ist eine [step]-Zeile jedes Mal, wenn der Agent voranschreitet, die abwechselnd mit seinem Prosa-Ausdruck, der Zeichen für Zeichen eingefügt wird, erscheint – wobei bereits ein Filter vorhanden ist, damit nichts durchsickert. Es sieht nicht schön aus, ist aber ehrlich. Ein Benutzer kann erkennen, dass der Agent aktiv ist, und beobachten, wie er denkt.

Neue Versionen der Deep Agents bieten eine höherwertige API dafür an, nämlich agent.stream_events(...) unter version="v3", zusammen mit typisierten Projektionen wie stream.subagents. Die Dokumentation leitet neue Anwendungen in diese Richtung, da jede Projektion ihr eigenes strukturiertes Iterator ist und es nicht notwendig ist, auf stream_mode zu schalten oder Namensräume selbst zu entschlüsseln. Es lohnt sich dennoch, zunächst den stream_mode-Ansatz zu erlernen, da die Projektionen auf denselben Ereignissen basieren. Überprüfen Sie die aktuellen Dokumentationen, bevor Sie sich für eine umfangreiche Schnittstelle entscheiden.

Ein Checkliste zum sofortigen Anwenden

Beginnen Sie damit, jede invoke-Anweisung zu finden, auf die jemand wartet, und ersetzen Sie diese durch agent.stream(...) im updates-Modus. Ein grober Schritt-Log ist immer noch besser als ein leeres Bildschirm.
Betrachten Sie einen echten Datensatz. Führen Sie die Anweisung mit version="v2" aus, geben Sie den Ausgang aus und lesen Sie die Felder type, ns sowie data, damit die Struktur klar wird.
Bevor echte Benutzer den Ausgang sehen, entfernen Sie alle messages-Datensätze, deren Metadaten lc_source == "summarization" angeben.
Falls der Agent Aufgaben an Unteragenten weiterleitet, setzen Sie subgraphs=True und sortieren Sie den Ausgang nach Namespace – jeweils ein Abschnitt pro Agent.
Behalten Sie eine kleine Konsole wie die oben beschriebene im Repository. Dadurch können Sie beobachten, wie jede neue Funktionalität ausgeführt wird, anstatt nur aus der letzten Nachricht zu raten.

Was hat sich geändert und was nicht

Letztes Fazit: Ein blockierender Aufruf verwandelte sich in einen Live-Stream. stream_mode legt fest, wie viele Details angezeigt werden, ns gibt an, wer jedes Ereignis erzeugt hat, der Zusammenfassungsfilter entfernt das eigene Gerede des Tools – und es sollte nun klar sein, dass das Beobachten eines Ablaufs und das Merken desselben unterschiedliche Aufgaben sind.

Am Agenten oder seinem Laufzyklus hat sich nichts geändert. Die einzige Ergänzung war ein Fenster, und dieses Fenster markiert die Grenze zwischen einem Tool, dem die Menschen vertrauen, und einem, das sie in Panik neu laden. Wenn ein Agent seine Arbeit zeigt, glauben die Nutzer, dass es funktioniert.

Deshalb sollten Sie beim nächsten Mal, wenn jemand berichtet, dass der Agent „hängt“, auf den Profiler verzichten und zunächst einen Stream ausprobieren. Der Agent hat die ganze Zeit gesprochen – er brauchte einen Kanal.