LangChain Streaming-Ausgabe

Die Streaming-Ausgabe bewirkt, dass die Antworten der KI Zeichen für Zeichen wie beim Tippen angezeigt werden, was das Nutzererlebnis erheblich verbessert. Der Agent von LangChain bietet integrierte umfassende Unterstützung für Streaming-Ausgabe.

Warum Streaming-Ausgabe benötigt wird

Wenn invoke() verwendet wird, müssen Nutzer warten, bis der Agent alle Schritte abgeschlossen hat (mehrere Modellaufrufe + Tool-Ausführung), bevor sie das Ergebnis sehen. Bei komplexen Aufgaben kann dies mehr als zehn Sekunden oder sogar länger dauern.

Die Streaming-Ausgabe löst dieses Problem: Jedes generierte Token wird sofort zurückgegeben, sodass Nutzer den Fortschritt in Echtzeit sehen können.

MethodeBenutzererfahrungGeeignete Szenarien
invoke()Warten → Das vollständige Ergebnis auf einmal sehenSkripte, API, Batchverarbeitung
stream()Jedes Token in Echtzeit sehenChat-Oberflächen, Echtzeitanzeige

stream_mode="messages" — Token-für-Token-Streaming

Dies ist der feingranularste Streaming-Modus, bei dem jeder Chunk einem Token entspricht:

Beispiel

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("deepseek:deepseek-v4-flash")
agent = create_agent(
    model=model,
    system_prompt="Du bist der Assistent von EXAMPLE Tutorial.",
)

# stream_mode="messages" gibt Token einzeln zurück
print("Echtzeit-Streaming-Ausgabe:")
for msg_chunk, metadata in agent.stream(
    {"messages": [HumanMessage(content="Stelle Python in einem Satz vor")]},
    stream_mode="messages",
):
    # msg_chunk ist ein AIMessageChunk
    # jeder Chunk enthält nur einen kleinen Teil des Inhalts
    if msg_chunk.content:
        print(msg_chunk.content, end="", flush=True)
print()  # Zeilenumbruch am Ende

Run result:

Live streaming output:
Example is a free online learning platform for programming beginners, offering rich technical tutorials and hands-on examples.

Metadaten verstehen

Metadaten enthalten Quellinformationen über diesen Chunk:

Beispiel

print("Metadateninformationen anzeigen:"\n")
for msg_chunk, metadata in agent.stream(
    {"messages": [HumanMessage(content="Hallo, stelle dich vor")]},
    stream_mode="messages",
):
    if msg_chunk.content and len(msg_chunk.content) > 5:
        print(f"Inhalt: {msg_chunk.content}")
        print(f"Quellknoten: {metadata.get('langgraph_node')}")
        print(f"Nachrichtentyp: {type(msg_chunk).__name__}")
        break  # nur den ersten aussagekräftigen Abschnitt anzeigen

Run result:

Inspect metadata:

Content: Hello! I am an AI assistant
Source node: model
Message type: AIMessageChunk

stream_mode="updates" — Den Ausführungsprozess des Agenten Schritt für Schritt anzeigen

Dieser Modus ist sehr nützlich beim Erstellen von Schnittstellen, die den "Denkprozess" anzeigen müssen:

Beispiel

from langchain.tools import tool


@tool
def search_course(keyword: str) -> str:
    """Suche nach Kursen auf EXAMPLE"""
    courses = {
        "python": "Python3-Basistutorial (30 Kapitel, 20 Stunden)",
        "html": "HTML-Basistutorial (25 Kapitel, 15 Stunden)",
    }
    return courses.get(keyword.lower(), "Keine verwandten Kurse gefunden")


agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
    tools=[search_course],
    system_prompt="Sie sind ein Kursberater bei EXAMPLE.",
)

# updates-Modus verwenden, um jeden Schritt anzuzeigen
print("=== Agenten-Ausführungsprozess ==="\n")
for chunk in agent.stream(
    {"messages": [HumanMessage(content=Hilf mir, Python-Kurse zu prüfen)]},
    stream_mode="updates",
):
    for node_name, update in chunk.items():
        print(f"[{node_name}]", end=" ")
        if "messages" in update:
            for msg in update["messages"]:
                if msg.type == "ai":
                    if hasattr(msg, 'tool_calls') and msg.tool_calls:
                        calls = [tc['name'] for tc in msg.tool_calls]
                        print(fAnfrageaufrufe: {calls})
                    elif msg.content:
                        print(fAntwort: {msg.content[:80]})
                elif msg.type == "tool":
                    print(fTool gab zurück [{msg.name}]: {msg.content})

Run result:

=== Agent 执行过程 ===

[model] 请求调用: ['search_course']
[tools] 工具返回 [search_course]: Python3 基础教程（30章，20小时）
[model] Reply: Example has the Python3 basics tutorial, 30 chapters, about 20 hours of study, perfect for Python beginners.

stream_mode="custom" — Benutzerdefinierte Ereignisse senden

Über die runtime.stream_writer()-Methode der Middleware können Sie benutzerdefinierte Ereignisse an den Stream senden:

Beispiel

from langchain.agents.middleware import before_model, after_model


@before_model
def notify_before(state, runtime):
    Benutzerdefinierte Ereignisse vor dem Modellaufruf senden
    runtime.stream_writer({
        "type": "status",
        "message": "Denkt nach...",
    })
    return None


@after_model
def notify_after(state, runtime):
    Sende benutzerdefinierte Ereignisse nach dem Modellaufruf
    last_msg = state["messages"][-1] if state.get("messages") else None
    has_tools = hasattr(last_msg, 'tool_calls') and last_msg.tool_calls

    if has_tools:
        tool_names = [tc['name'] for tc in last_msg.tool_calls]
        runtime.stream_writer({
            "type": "status",
            "message": fRufe Tools auf: {', '.join(tool_names)}...,
        })
    else:
        runtime.stream_writer({
            "type": "status",
            "message": Antwort abgeschlossen,
        })
    return None


agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
    tools=[search_course],
    middleware=[notify_before, notify_after],
    system_prompt="Sie sind ein Kursberater bei EXAMPLE.",
)

# Verwende stream_mode=["updates", "custom"], um beide Ereignisse gleichzeitig zu empfangen
print(=== Gemischte Streaming-Ausgabe ===\n")
for mode, chunk in agent.stream(
    {"messages": [HumanMessage(content=Python-Kurse prüfen)]},
    stream_mode=["updates", "custom"],
):
    if mode == "custom":
        print(f[Benutzerdefiniertes Ereignis] Status: {chunk['message']})
    elif mode == "updates":
        for node_name, update in chunk.items():
            if "messages" in update:
                for msg in update["messages"]:
                    if msg.type == "ai" and msg.content:
                        print(f[Antwort] {msg.content})

Run result:

=== 混合流式输出 ===

[custom event] Status: thinking...
[custom event] Status: calling tool: search_course...
[reply] Example has the Python3 basics tutorial, 30 chapters, about 20 hours of study.
[custom event] Status: thinking...
[custom event] Status: answer completed

stream_mode-Modi können kombiniert werden, wie z. B. stream_mode=["updates", "custom", "messages"]. Allerdings erhöhen zu viele Modi die Anzahl der Ereignisse im Stream, daher wird empfohlen, sie nach Bedarf auszuwählen.

Asynchrone Streaming-Ausgabe

In Webdiensten kann die Verwendung von asynchronem Streaming das Blockieren der Ereignisschleife vermeiden:

Beispiel

import asyncio


async def stream_agent():
    Agent asynchron im Streaming-Modus ausführen
    agent = create_agent(
        model=init_chat_model("deepseek:deepseek-v4-flash"),
        system_prompt="Du bist der Assistent von EXAMPLE Tutorial.",
    )

    full_response = ""
    async for msg_chunk, metadata in agent.astream(
        {"messages": [HumanMessage(content=Stelle Python in einem Satz vor)]},
        stream_mode="messages",
    ):
        if msg_chunk.content:
            full_response += msg_chunk.content
            print(msg_chunk.content, end="", flush=True)

    print(f"\n\nVollständige Antwortlänge: {len(full_response)} Zeichen)


# Die asynchrone Funktion ausführen
asyncio.run(stream_agent())

FastAPI-Integrationsbeispiel

Beispiel

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

app = FastAPI()
agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash"),
    system_prompt="Du bist der Assistent von EXAMPLE Tutorial.",
)


@app.get("/chat")
async def chat(message: str):
    Chat-Oberfläche, gibt eine SSE-Streaming-Antwort zurück
    async def generate():
        async for msg_chunk, metadata in agent.astream(
            {"messages": [HumanMessage(content=message)]},
            stream_mode="messages",
        ):
            if msg_chunk.content:
                # SSE-Format: data: xxx
                yield f"data: {msg_chunk.content}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
    )

# Starten: uvicorn main:app --reload

In einer Produktionsumgebung wird empfohlen, die Agent-Instanz als globales Singleton zu erstellen, um zu vermeiden, dass sie bei jeder Anfrage neu erstellt wird. Die Kosten für die Erstellung eines Agenten sind gering (hauptsächlich das Kompilieren des Graphen), aber die Wiederverwendung der Instanz ist effizienter.

stream_mode Kurzreferenz

ModeGranularitätIterationsobjektTypische Verwendung
messagesToken-Ebene(AIMessageChunk, metadata)Tippeffekt, Echtzeit-Chat
updatesKnotenebene{node: state_update}Denkprozess anzeigen
valuesKnotenebene (vollständig)Vollständiger ZustandZustands-Snapshot, Debugging
customIndividuellBeliebiges dictFortschrittsbenachrichtigung, Status-Push
debugDetailliertDebug-InformationenFehlerbehebung während der Entwicklung
Weitere Erweiterungen