如何流式传输¶

先决条件

本指南假定您熟悉以下内容

流式传输对于增强构建在 LLM 上的应用程序的响应性至关重要。通过逐步显示输出，即使在完整响应准备好之前，流式传输也能显着改善用户体验 (UX)，尤其是在处理 LLM 的延迟时。

LangGraph 构建时就优先考虑对流式传输的支持。有几种不同的方法可以从图运行中流式传回输出

"values"：在每个步骤后发出状态中的所有值。
"updates"：在每个步骤后仅发出节点名称和节点返回的更新。如果在同一步骤中进行多次更新（例如，运行多个节点），则这些更新将分别发出。
"custom"：使用 StreamWriter 从节点内部发出自定义数据。
"messages"：将 LLM 消息逐个令牌地与节点内部任何 LLM 调用相关的元数据一起发出。
"debug"：为每个步骤发出尽可能多的调试事件信息。

您可以使用 graph.stream(..., stream_mode=<stream_mode>) 方法从图中流式传输输出，例如

同步异步

for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)

async for chunk in graph.astream(inputs, stream_mode="updates"):
    print(chunk)

您还可以通过向 stream_mode 参数提供列表来组合多种流式传输模式

同步异步

for chunk in graph.stream(inputs, stream_mode=["updates", "custom"]):
    print(chunk)

async for chunk in graph.astream(inputs, stream_mode=["updates", "custom"]):
    print(chunk)

设置¶

%%capture --no-stderr
%pip install --quiet -U langgraph langchain_openai

import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("OPENAI_API_KEY")

OPENAI_API_KEY:  ········

设置 LangSmith 以进行 LangGraph 开发

注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。 LangSmith 使您可以使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序 — 阅读此处以了解更多关于如何开始的信息。

让我们定义一个带有两个节点的简单图

定义图¶

from typing import TypedDict
from langgraph.graph import StateGraph, START


class State(TypedDict):
    topic: str
    joke: str


def refine_topic(state: State):
    return {"topic": state["topic"] + " and cats"}


def generate_joke(state: State):
    return {"joke": f"This is a joke about {state['topic']}"}


graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .compile()
)

API 参考：StateGraph | START

流式传输状态中的所有值 (stream_mode="values")¶

使用此方法流式传输每个步骤后状态中的所有值。

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="values",
):
    print(chunk)

{'topic': 'ice cream'}
{'topic': 'ice cream and cats'}
{'topic': 'ice cream and cats', 'joke': 'This is a joke about ice cream and cats'}

从节点流式传输状态更新 (stream_mode="updates")¶

使用此方法仅流式传输每个步骤后节点返回的状态更新。流式传输的输出包括节点名称以及更新。

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="updates",
):
    print(chunk)

{'refine_topic': {'topic': 'ice cream and cats'}}
{'generate_joke': {'joke': 'This is a joke about ice cream and cats'}}

流式传输调试事件 (stream_mode="debug")¶

使用此方法流式传输每个步骤的调试事件，其中包含尽可能多的信息。包括有关计划执行的任务以及任务执行结果的信息。

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="debug",
):
    print(chunk)

{'type': 'task', 'timestamp': '2025-01-28T22:06:34.789803+00:00', 'step': 1, 'payload': {'id': 'eb305d74-3460-9510-d516-beed71a63414', 'name': 'refine_topic', 'input': {'topic': 'ice cream'}, 'triggers': ['start:refine_topic']}}
{'type': 'task_result', 'timestamp': '2025-01-28T22:06:34.790013+00:00', 'step': 1, 'payload': {'id': 'eb305d74-3460-9510-d516-beed71a63414', 'name': 'refine_topic', 'error': None, 'result': [('topic', 'ice cream and cats')], 'interrupts': []}}
{'type': 'task', 'timestamp': '2025-01-28T22:06:34.790165+00:00', 'step': 2, 'payload': {'id': '74355cb8-6284-25e0-579f-430493c1bdab', 'name': 'generate_joke', 'input': {'topic': 'ice cream and cats'}, 'triggers': ['refine_topic']}}
{'type': 'task_result', 'timestamp': '2025-01-28T22:06:34.790337+00:00', 'step': 2, 'payload': {'id': '74355cb8-6284-25e0-579f-430493c1bdab', 'name': 'generate_joke', 'error': None, 'result': [('joke', 'This is a joke about ice cream and cats')], 'interrupts': []}}

流式传输 LLM 令牌 (stream_mode="messages")¶

使用此方法流式传输 LLM 消息，逐个令牌，以及节点或任务内部任何 LLM 调用的元数据。让我们修改上面的示例以包含 LLM 调用

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")


def generate_joke(state: State):
    llm_response = llm.invoke(
        [
            {"role": "user", "content": f"Generate a joke about {state['topic']}"}
        ]
    )
    return {"joke": llm_response.content}


graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .compile()
)

API 参考：ChatOpenAI

for message_chunk, metadata in graph.stream(
    {"topic": "ice cream"},
    stream_mode="messages",
):
    if message_chunk.content:
        print(message_chunk.content, end="|", flush=True)

Why| did| the| cat| sit| on| the| ice| cream| cone|?

|Because| it| wanted| to| be| a| "|p|urr|-f|ect|"| scoop|!| 🍦|🐱|

metadata

{'langgraph_step': 2,
 'langgraph_node': 'generate_joke',
 'langgraph_triggers': ['refine_topic'],
 'langgraph_path': ('__pregel_pull', 'generate_joke'),
 'langgraph_checkpoint_ns': 'generate_joke:568879bc-8800-2b0d-a5b5-059526a4bebf',
 'checkpoint_ns': 'generate_joke:568879bc-8800-2b0d-a5b5-059526a4bebf',
 'ls_provider': 'openai',
 'ls_model_name': 'gpt-4o-mini',
 'ls_model_type': 'chat',
 'ls_temperature': 0.7}

流式传输自定义数据 (stream_mode="custom")¶

使用此方法使用 StreamWriter 从节点内部流式传输自定义数据。

from langgraph.types import StreamWriter


def generate_joke(state: State, writer: StreamWriter):
    writer({"custom_key": "Writing custom data while generating a joke"})
    return {"joke": f"This is a joke about {state['topic']}"}


graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .compile()
)

API 参考：StreamWriter

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="custom",
):
    print(chunk)

{'custom_key': 'Writing custom data while generating a joke'}

配置多种流式传输模式¶

使用此方法组合多种流式传输模式。输出以元组 (stream_mode, streamed_output) 的形式流式传输。

for stream_mode, chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["updates", "custom"],
):
    print(f"Stream mode: {stream_mode}")
    print(chunk)
    print("\n")

Stream mode: updates
{'refine_topic': {'topic': 'ice cream and cats'}}


Stream mode: custom
{'custom_key': 'Writing custom data while generating a joke'}


Stream mode: updates
{'generate_joke': {'joke': 'This is a joke about ice cream and cats'}}