流式 API

LangGraph SDK 让你可以通过多种模式从 LangSmith 部署 API 流式传输输出，从每个步骤后的完整状态快照到逐令牌的 LLM 输出。线程流式传输还支持可恢复性：如果连接中断，可以使用上一个事件 ID 重新连接，从上次中断处继续。

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

基本用法

基本用法示例：

Python
JavaScript
cURL

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)

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <API_KEY> });

// 使用名为 "agent" 的已部署图
const assistantID = "agent";

// 创建一个线程
const thread = await client.threads.create();
const threadID = thread["thread_id"];

// 创建一个流式运行
const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input,
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

创建一个线程：

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

创建一个流式运行：

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--header 'x-api-key: <API_KEY>'
--data "{
  \"assistant_id\": \"agent\",
  \"input\": <inputs>,
  \"stream_mode\": \"updates\"
}"

扩展示例：流式更新

这是一个你可以在 Agent Server 中运行的示例图。更多详情请参阅 LangSmith 快速入门。

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

一旦你的 Agent Server 运行起来，你就可以使用 LangGraph SDK 与其交互

Python
JavaScript
cURL

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() 方法返回一个迭代器，用于产出流式输出。 2. 设置 stream_mode="updates" 以仅流式传输每个节点后对图状态的更新。也支持其他流式模式。详情请参见支持的流式模式。

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// 使用名为 "agent" 的已部署图
const assistantID = "agent";

// 创建一个线程
const thread = await client.threads.create();
const threadID = thread["thread_id"];

// 创建一个流式运行
const streamResponse = client.runs.stream(  // (1)!
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "updates"  // (2)!
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

client.runs.stream() 方法返回一个迭代器，用于产出流式输出。
设置 streamMode: "updates" 以仅流式传输每个节点后对图状态的更新。也支持其他流式模式。详情请参见支持的流式模式。

创建一个线程：

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

创建一个流式运行：

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"updates\"
}"

{'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 库方法
`values`	在每个超步骤之后流式传输完整的图状态。	`.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

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

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input,
    streamMode: ["updates", "custom"]
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk);
}

curl --request POST \
 --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
 --header 'Content-Type: application/json' \
 --data "{
   \"assistant_id\": \"agent\",
   \"input\": <inputs>,
   \"stream_mode\": [
     \"updates\"
     \"custom\"
   ]
 }"

流式传输图状态

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

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

示例图

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()
)

有状态运行 以下示例假设你希望将流式运行的输出持久化到检查点器数据库中，并且已经创建了一个线程。要创建一个线程：

Python
JavaScript
cURL

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"]

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// 使用名为 "agent" 的已部署图
const assistantID = "agent";
// 创建一个线程
const thread = await client.threads.create();
const threadID = thread["thread_id"]

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

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

流式模式：`updates`

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

Python
JavaScript
cURL

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

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"updates\"
}"

流式模式：`values`

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

Python
JavaScript
cURL

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

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "values"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"values\"
}"

子图

要在流式输出中包含来自子图的输出，你可以在父图的 .stream() 方法中设置 subgraphs=True。这将流式传输来自父图和任何子图的输出。

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 快速入门。

# 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": "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()

一旦你的 Agent Server 运行起来，你就可以使用 LangGraph SDK 与其交互

Python
JavaScript
cURL

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

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// 使用名为 "agent" 的已部署图
const assistantID = "agent";

// 创建一个线程
const thread = await client.threads.create();
const threadID = thread["thread_id"];

// 创建一个流式运行
const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { foo: "foo" },
    streamSubgraphs: true,  // (1)!
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk);
}

设置 streamSubgraphs: true 以流式传输来自子图的输出。

创建一个线程：

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

创建一个流式运行：

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"foo\": \"foo\"},
  \"stream_subgraphs\": true,
  \"stream_mode\": [
    \"updates\"
  ]
}"

注意：我们收到的不仅是节点更新，还有命名空间，它们告诉我们正在流式传输哪个图（或子图）。

调试

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

Python
JavaScript
cURL

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

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "debug"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"debug\"
}"

LLM 令牌

使用 messages-tuple 流式模式，从图的任何部分（包括节点、工具、子图或任务）逐令牌流式传输大语言模型 (LLM) 输出。来自 messages-tuple 模式的流式输出是一个元组 (message_chunk, metadata)，其中：

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

示例图

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):
    """调用 LLM 生成关于某个主题的笑话"""
    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()
)

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

Python
JavaScript
cURL

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)

“messages-tuple” 流式模式返回一个元组 (message_chunk, metadata) 的迭代器，其中 message_chunk 是 LLM 流式传输的令牌，metadata 是一个包含调用 LLM 的图节点信息和其他信息的字典。

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "messages-tuple"
  }
);
for await (const chunk of streamResponse) {
  if (chunk.event !== "messages") {
    continue;
  }
  console.log(chunk.data[0]["content"]);  // (1)!
}

“messages-tuple” 流式模式返回一个元组 (message_chunk, metadata) 的迭代器，其中 message_chunk 是 LLM 流式传输的令牌，metadata 是一个包含调用 LLM 的图节点信息和其他信息的字典。

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"messages-tuple\"
}"

过滤 LLM 令牌

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

流式传输自定义数据

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

Python
JavaScript
cURL

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

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { query: "example" },
    streamMode: "custom"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"query\": \"example\"},
  \"stream_mode\": \"custom\"
}"

流式传输事件

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

Python
JavaScript
cURL

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

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "events"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"events\"
}"

无状态运行

如果你不希望将流式运行的输出持久化到检查点器数据库中，可以创建一个无状态运行而无需创建线程：

Python
JavaScript
cURL

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)

我们传递的是 None 而不是 thread_id UUID。

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <API_KEY> });

// 创建一个流式运行
const streamResponse = client.runs.stream(
  null,  // (1)!
  assistantID,
  {
    input,
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

我们传递的是 None 而不是 thread_id UUID。

curl --request POST \
--url <DEPLOYMENT_URL>/runs/stream \
--header 'Content-Type: application/json' \
--header 'x-api-key: <API_KEY>'
--data "{
  \"assistant_id\": \"agent\",
  \"input\": <inputs>,
  \"stream_mode\": \"updates\"
}"

加入并流式传输

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

Python
JavaScript
cURL

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)

这是你想要加入的现有运行的 run_id。

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <API_KEY> });

const streamResponse = client.runs.joinStream(
  threadID,
  runId  // (1)!
);
for await (const chunk of streamResponse) {
  console.log(chunk);
}

这是你想要加入的现有运行的 run_id。

curl --request GET \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/<RUN_ID>/stream \
--header 'Content-Type: application/json' \
--header 'x-api-key: <API_KEY>'

输出未缓冲 当你使用 .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
JavaScript
cURL

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)

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <API_KEY> });

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

for await (const chunk of client.threads.joinStream(threadID)) {
  console.log(chunk);
}

curl --request GET \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/stream \
--header 'x-api-key: <API_KEY>'

线程流式模式

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

模式	描述
`run_modes` (默认)	流式传输所有运行事件，等同于 `client.runs.stream()` 输出。
`lifecycle`	仅流式传输运行开始和结束事件。使用此模式可以轻量级地监控运行状态，而无需完整输出。
`state_update`	仅流式传输状态更新事件，提供每次运行完成后的线程状态。

Python
JavaScript
cURL

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

for await (const chunk of client.threads.joinStream(threadID, {
  streamMode: ["lifecycle", "state_update"],
})) {
  console.log(chunk.event, chunk.data);
}

curl --request GET \
--url '<DEPLOYMENT_URL>/threads/<THREAD_ID>/stream?stream_modes=lifecycle&stream_modes=state_update' \
--header 'x-api-key: <API_KEY>'

从上一个事件恢复

线程流通过 Last-Event-ID 头部支持可恢复性。如果连接中断，传递你收到的最后一个事件的 ID 即可恢复，不会错过事件。传递 "-" 可以从头开始重放。

Python
JavaScript
cURL

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

for await (const chunk of client.threads.joinStream(threadID, {
  lastEventId: "<LAST_EVENT_ID>",
})) {
  console.log(chunk);
}

curl --request GET \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/stream \
--header 'x-api-key: <API_KEY>' \
--header 'Last-Event-ID: <LAST_EVENT_ID>'

API 参考

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

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Develop your application

Capabilities

How to build

Auth & access control

Server customization

基本用法

支持的流式模式

流式传输多个模式

流式传输图状态

流式模式：`updates`

流式模式：`values`

子图

调试

LLM 令牌

过滤 LLM 令牌

流式传输自定义数据

流式传输事件

无状态运行

加入并流式传输

流式传输线程

比较线程和运行流式传输

基本用法

线程流式模式

从上一个事件恢复

API 参考

​基本用法

​支持的流式模式

​流式传输多个模式

​流式传输图状态

​流式模式：updates

​流式模式：values

​子图

​调试

​LLM 令牌

​过滤 LLM 令牌

​流式传输自定义数据

​流式传输事件

​无状态运行

​加入并流式传输

​流式传输线程

​比较线程和运行流式传输

​基本用法

​线程流式模式

​从上一个事件恢复

​API 参考

基本用法

支持的流式模式

流式传输多个模式

流式传输图状态

流式模式：`updates`

流式模式：`values`

子图

调试

LLM 令牌

过滤 LLM 令牌

流式传输自定义数据

流式传输事件

无状态运行

加入并流式传输

流式传输线程

比较线程和运行流式传输

基本用法

线程流式模式

从上一个事件恢复

API 参考