流式 API

LangGraph SDK 让你能够以多种模式从 LangSmith Deployment API 流式传输输出，从每步之后的完整状态快照到逐 token 的 LLM 输出。线程流式传输还支持可恢复性：如果连接中断，用最后一个事件 ID 重新连接，从中断处继续。

NOTE

LangGraph SDK 和 Agent Server 是 LangSmith 的一部分。

基本用法

基本用法示例：

python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>, api_key=<API_KEY>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

# create a streaming run
async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input=inputs,
    stream_mode="updates"
):
    print(chunk.data)

Extended example: streaming updates

This is an example graph you can run in the Agent Server. See LangSmith quickstart for more details.

python

# graph.py
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

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")
    .add_edge("generate_joke", END)
    .compile()
)

Once you have a running Agent Server, you can interact with it using LangGraph SDK

python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

# create a streaming run
async for chunk in client.runs.stream(  # (1)!
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="updates"  # (2)!
):
    print(chunk.data)

The client.runs.stream() method returns an iterator that yields streamed outputs. 2. Set stream_mode="updates" to stream only the updates to the graph state after each node. Other stream modes are also available. See supported stream modes for details.

python

{'run_id': '1f02c2b3-3cef-68de-b720-eec2a4a8e920', 'attempt': 1}
{'refine_topic': {'topic': 'ice cream and cats'}}
{'generate_joke': {'joke': 'This is a joke about ice cream and cats'}}

支持的流式模式

模式	描述	LangGraph Library 方法
`values`	在每个超步之后流式传输完整的图状态。	带 `stream_mode="values"` 的 `.stream()` / `.astream()`
`updates`	在图的每一步之后流式传输对状态的更新。如果在同一步骤中做了多个更新（例如运行了多个节点），这些更新会被分别流式传输。	带 `stream_mode="updates"` 的 `.stream()` / `.astream()`
`messages-tuple`	为调用 LLM 的那个图节点流式传输 LLM token 和元数据（对聊天应用有用）。	带 `stream_mode="messages"` 的 `.stream()` / `.astream()`
`debug`	在图的整个执行过程中流式传输尽可能多的信息。	带 `stream_mode="debug"` 的 `.stream()` / `.astream()`
`custom`	从你图内部流式传输自定义数据	带 `stream_mode="custom"` 的 `.stream()` / `.astream()`
`events`	流式传输所有事件（包括图的状态）；主要对迁移大型 LCEL 应用有用。	`.astream_events()`

流式传输多种模式

你可以传入一个列表作为 stream_mode 参数，以同时流式输出多个模式。

流式的输出是 (mode, chunk) 元组，其中 mode 是流式模式的名称，chunk 是该模式流式传输的数据。

python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input=inputs,
    stream_mode=["updates", "custom"]
):
    print(chunk)

流式传输图状态

使用流模式 updates 和 values 来流式输出图在执行时的状态。

updates 会在图的每一步之后流式输出对状态的更新。
values 会在图的每一步之后流式输出状态的 完整值。

Example graph

python

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

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")
  .add_edge("generate_joke", END)
  .compile()
)

NOTE

有状态运行 下面的示例假定你想在检查点器数据库中持久化流式运行的输出，并已创建了一个线程。要创建一个线程：

python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"
# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

如果你不需要持久化某次运行的输出，可以在流式传输时传 None 而非 thread_id。

流式模式：`updates`

用它来只流式传输各节点在每步之后返回的状态更新。流式的输出包含节点名以及该更新。

python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="updates"
):
    print(chunk.data)

流式模式：`values`

用它来在每步之后流式传输图的完整状态。

python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="values"
):
    print(chunk.data)

子图

要将子图的输出包含在流式输出中，你可以在父图的 .stream() 方法中设置 subgraphs=True。这将同时流式输出父图和所有子图的输出。

python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"foo": "foo"},
    stream_subgraphs=True, # (1)!
    stream_mode="updates",
):
    print(chunk)

设置 stream_subgraphs=True 以从子图流式传输输出。

Extended example: streaming from subgraphs

This is an example graph you can run in the Agent Server. See LangSmith quickstart for more details.

python

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

# Define subgraph
class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    bar: str

def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}

def subgraph_node_2(state: SubgraphState):
    return {"foo": state["foo"] + state["bar"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Define parent graph
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

Once you have a running Agent Server, you can interact with it using LangGraph SDK

python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"foo": "foo"},
    stream_subgraphs=True, # (1)!
    stream_mode="updates",
):
    print(chunk)

Set stream_subgraphs=True to stream outputs from subgraphs.

Note that we are receiving not just the node updates, but we also the namespaces which tell us what graph (or subgraph) we are streaming from.

调试

使用 debug 流式传输模式，在图执行的整个过程中流式输出尽可能多的信息。流式输出的内容包括节点名称以及完整状态。

python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="debug"
):
    print(chunk.data)

LLM token

使用 messages-tuple 流式模式，从你图的任何部分（包括节点、工具、子图或任务）逐 token 流式传输大语言模型（LLM）的输出。

来自 messages-tuple 模式的流式输出是一个 (message_chunk, metadata) 元组，其中：

message_chunk：来自 LLM 的 token 或消息片段。
metadata：一个字典，包含有关图节点和 LLM 调用的详细信息。

Example graph

python

from dataclasses import dataclass

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START

@dataclass
class MyState:
    topic: str
    joke: str = ""

model = init_chat_model(model="gpt-5.4-mini")

def call_model(state: MyState):
    """Call the LLM to generate a joke about a topic"""
    model_response = model.invoke( # (1)!
        [
            {"role": "user", "content": f"Generate a joke about {state.topic}"}
        ]
    )
    return {"joke": model_response.content}

graph = (
    StateGraph(MyState)
    .add_node(call_model)
    .add_edge(START, "call_model")
    .compile()
)

Note that the message events are emitted even when the LLM is run using invoke rather than stream.

python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="messages-tuple",
):
    if chunk.event != "messages":
        continue

    message_chunk, metadata = chunk.data  # (1)!
    if message_chunk["content"]:
        print(message_chunk["content"], end="|", flush=True)

The "messages-tuple" stream mode returns an iterator of tuples (message_chunk, metadata) where message_chunk is the token streamed by the LLM and metadata is a dictionary with information about the graph node where the LLM was called and other information.

过滤 LLM token

要按 LLM 调用过滤流式 token，你可以将 tags 与 LLM 调用关联。
要只从特定节点流式传输 token，使用 stream_mode="messages" 并按流式元数据中的 langgraph_node 字段过滤输出。

流式传输自定义数据

要发送自定义用户定义数据：

python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"query": "example"},
    stream_mode="custom"
):
    print(chunk.data)

流式事件

要流式传输所有事件，包括图的状态：

python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="events"
):
    print(chunk.data)

无状态运行

如果你不想在检查点器数据库中持久化流式运行的输出，可以在不创建线程的情况下创建一个无状态运行：

python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>, api_key=<API_KEY>)

async for chunk in client.runs.stream(
    None,  # (1)!
    assistant_id,
    input=inputs,
    stream_mode="updates"
):
    print(chunk.data)

We are passing None instead of a thread_id UUID.

加入并流式传输

LangSmith 允许你加入一个活动的后台运行并从它流式传输输出。为此，你可以使用 LangGraph SDK 的 client.runs.join_stream 方法：

python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>, api_key=<API_KEY>)

async for chunk in client.runs.join_stream(
    thread_id,
    run_id,  # (1)!
):
    print(chunk)

This is the run_id of an existing run you want to join.

WARNING

输出不缓冲 当你使用 .join_stream 时，输出不会被缓冲，因此在加入之前产生的任何输出都不会被接收到。

流式传输一个线程

线程流式传输为一个线程打开一个长生命周期的连接，并从该线程上执行的每一次运行流式传输输出。这让你能够从单个连接监视线程上的所有活动，例如在一个聊天 UI 中，可能通过跟进消息、人在回路恢复或后台运行随时间触发多次运行。要按 ID 加入某个特定的现有运行，参阅加入并流式传输。

比较线程与运行流式传输

	线程流式传输	运行流式传输
SDK 方法	`client.threads.join_stream()`	`client.runs.stream()`
REST 端点	`GET /threads/{thread_id}/stream`	`POST /threads/{thread_id}/runs/stream`
范围	线程上的所有运行	单次运行
连接生命周期	无限期保持打开	在运行完成时关闭
创建运行	否	是
用例	监视进行中的线程活动	执行并流式传输单次交互

基本用法

python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>, api_key=<API_KEY>)

thread = await client.threads.create()
thread_id = thread["thread_id"]

async for chunk in client.threads.join_stream(thread_id):
    print(chunk)

线程流式模式

线程流式传输支持三种控制返回哪些事件的流式模式。通过 stream_mode 参数传入一个或多个模式。

模式	描述
`run_modes`（默认）	流式传输所有运行事件，等同于 `client.runs.stream()` 的输出。
`lifecycle`	只流式传输运行的开始和结束事件。用它来轻量地监视运行状态，而不带完整输出。
`state_update`	只流式传输状态更新事件，在每次运行完成后提供线程状态。

python

async for chunk in client.threads.join_stream(
    thread_id,
    stream_mode=["lifecycle", "state_update"],
):
    print(chunk.event, chunk.data)

从最后一个事件恢复

线程流式传输通过 Last-Event-ID 标头支持可恢复性。如果连接中断，传入你收到的最后一个事件的 ID 以在不丢失事件的情况下恢复。传入 "-" 以从头重放。

python

async for chunk in client.threads.join_stream(
    thread_id,
    last_event_id="<LAST_EVENT_ID>",
):
    print(chunk)

API 参考

有关 API 用法和实现，参阅 API 参考。

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织