LangChain 官方中文文档

LangGraph SDK 允许您从 LangSmith 部署 API 流式传输输出。

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

基本用法

基本用法示例：

Python

JavaScript

cURL

python

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

# 使用名为 "agent" 的部署图
assistant_id = "agent"

# 创建一个线程
thread = await client.threads.create()
thread_id = thread["thread_id"]

# 创建一个流式运行
async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input=inputs,
    stream_mode="updates"
):
    print(chunk.data)

扩展示例：流式传输更新

这是一个您可以在 Agent Server 中运行的示例图。有关更多详细信息，请参阅 LangSmith 快速入门。

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"这是一个关于 {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()
)

一旦您运行了 Agent Server，您就可以使用 LangGraph SDK 与其交互

Python

JavaScript

cURL

python

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

# 使用名为 "agent" 的部署图
assistant_id = "agent"

# 创建一个线程
thread = await client.threads.create()
thread_id = thread["thread_id"]

# 创建一个流式运行
async for chunk in client.runs.stream(  # (1)!
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="updates"  # (2)!
):
    print(chunk.data)

client.runs.stream() 方法返回一个迭代流式输出的迭代器。
设置 stream_mode="updates" 以仅流式传输每个节点后图状态的更新。其他流模式也可用。有关详细信息，请参阅支持的流模式。

python

{'run_id': '1f02c2b3-3cef-68de-b720-eec2a4a8e920', 'attempt': 1}
{'refine_topic': {'topic': 'ice cream and cats'}}
{'generate_joke': {'joke': '这是一个关于 ice cream and cats 的笑话'}}

支持的流模式

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

流式传输多种模式

您可以传递一个列表作为 stream_mode 参数来同时流式传输多种模式。

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

Python

JavaScript

cURL

python

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

流式传输图状态

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

updates 流式传输图每个步骤后状态的更新。
values 流式传输图每个步骤后状态的完整值。

示例图

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"这是一个关于 {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()
)

有状态运行 以下示例假设您希望在检查点 (checkpointer) 数据库中持久化流式运行的输出，并已创建了一个线程。要创建一个线程：

Python

JavaScript

cURL

python

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

# 使用名为 "agent" 的部署图
assistant_id = "agent"
# 创建一个线程
thread = await client.threads.create()
thread_id = thread["thread_id"]

如果您不需要持久化运行的输出，可以在流式传输时传递 None 而不是 thread_id。

流模式：`updates`

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

Python

JavaScript

cURL

python

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

流模式：`values`

使用此模式流式传输每个步骤后图的完整状态。

Python

JavaScript

cURL

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 以流式传输来自子图的输出。

扩展示例：从子图流式传输

这是一个您可以在 Agent Server 中运行的示例图。有关更多详细信息，请参阅 LangSmith 快速入门。

python

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

# 定义子图
class SubgraphState(TypedDict):
    foo: str  # 注意，此键与父图状态共享
    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()

# 定义父图
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "嗨！" + 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()

一旦您运行了 Agent Server，您就可以使用 LangGraph SDK 与其交互

Python

JavaScript

cURL

python

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

# 使用名为 "agent" 的部署图
assistant_id = "agent"

# 创建一个线程
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)

设置 stream_subgraphs=True 以流式传输来自子图的输出。

请注意，我们不仅收到了节点更新，还收到了命名空间，它们告诉我们正在从哪个图（或子图）进行流式传输。

调试

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

Python

JavaScript

cURL

python

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

LLM 令牌

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

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

message_chunk：来自 LLM 的令牌或消息段。
metadata：包含图节点和 LLM 调用详情的字典。

示例图

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-4o-mini")

def call_model(state: MyState):
    """调用 LLM 为某个主题生成一个笑话"""
    model_response = model.invoke( # (1)!
        [
            {"role": "user", "content": f"为 {state.topic} 生成一个笑话"}
        ]
    )
    return {"joke": model_response.content}

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

请注意，即使 LLM 是使用 invoke 而非 stream 运行的，消息事件也会发出。

Python

JavaScript

cURL

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)

1. "messages-tuple" 流模式返回元组 `(message_chunk, metadata)` 的迭代器，其中 `message_chunk` 是 LLM 流式传输的令牌，而 `metadata` 是包含有关调用 LLM 的图节点及其他信息的字典。

过滤 LLM 令牌

要通过 LLM 调用过滤流式令牌，您可以将 tags 与 LLM 调用关联。
要仅从特定节点流式传输令牌，请使用 stream_mode="messages" 并通过 langgraph_node 字段过滤输出（包含在流式元数据中）。

流式传输自定义数据

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

Python

JavaScript

cURL

python

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

流式传输事件

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

Python

JavaScript

cURL

python

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

无状态运行

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

Python

JavaScript

cURL

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)

1. 我们传递的是 `None` 而非 `thread_id` UUID。

加入并流式传输

LangSmith 允许您加入一个正在进行的后台运行并流式传输其输出。为此，您可以使用 LangGraph SDK 的 client.runs.join_stream 方法：

Python

JavaScript

cURL

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)

1. 这是您想要加入的现有运行的 `run_id`。

输出未缓冲 当您使用 .join_stream 时，输出不会被缓冲，因此在加入之前生成的任何输出都将无法接收。

API 参考

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

助手 (Assistants)

运行记录 (Runs)

双重消息并发处理 (Double-texting)

流式 API

基本用法

支持的流模式

流式传输多种模式

流式传输图状态

流模式：`updates`

流模式：`values`

子图

调试

LLM 令牌

过滤 LLM 令牌

流式传输自定义数据

流式传输事件

无状态运行

加入并流式传输

API 参考

流式 API

基本用法 ​

支持的流模式 ​

流式传输多种模式 ​

流式传输图状态 ​

流模式：updates ​

流模式：values ​

子图 ​

调试 ​

LLM 令牌 ​

过滤 LLM 令牌 ​

流式传输自定义数据 ​

流式传输事件 ​

无状态运行 ​

加入并流式传输 ​

API 参考 ​

基本用法

支持的流模式

流式传输多种模式

流式传输图状态

流模式：`updates`

流模式：`values`

子图

调试

LLM 令牌

过滤 LLM 令牌

流式传输自定义数据

流式传输事件

无状态运行

加入并流式传输

API 参考