流式 API - 以编程方式使用文档

该 LangGraph SDK 允许您从 LangSmith Deployment API 以多种模式流式输出结果，从每个步骤后的完整状态快照到逐 token 的 LLM 输出。线程流式传输还支持可恢复性：如果连接断开，使用上一个事件 ID 重新连接即可从断点继续。

基本用法

基本用法示例：

Python

    from langgraph_sdk import get_client
    client = get_client(url=, 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)

JavaScript

    const client = new Client({ apiUrl: , apiKey:  });

    // Using the graph deployed with the name "agent"
    const assistantID = "agent";

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

    // create a streaming run
    const streamResponse = client.runs.stream(
      threadID,
      assistantID,
      {
        input,
        streamMode: "updates"
      }
    );
    for await (const chunk of streamResponse) {
      console.log(chunk.data);
    }

cURL

创建线程：

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

创建流式运行：

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

Extended example: streaming 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

      from langgraph_sdk import get_client
      client = get_client(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)

1. 该 client.runs.stream() 方法返回一个迭代器，该迭代器产生流式输出。 2. 设置 stream_mode="updates" 仅流式传输每个节点之后的状态更新。其他流模式也可使用。请参阅支持的流模式了解更多详情。

JavaScript

      const client = new Client({ apiUrl:  });

      // Using the graph deployed with the name "agent"
      const assistantID = "agent";

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

      // create a streaming run
      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);
      }

1. 该 client.runs.stream() 方法返回一个迭代器，该迭代器产生流式输出。 2. 设置 streamMode: "updates" 仅流式传输每个节点之后的状态更新。其他流模式也可使用。请参阅支持的流模式了解更多详情。

cURL

创建线程：

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

创建流式运行：

      curl --request POST \
      --url /threads//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`	在每个 super-step.	`.stream()` / `.astream()` 后流式传输完整图状态 `stream_mode="values"`
`updates`	在图的每个步骤之后流式传输状态更新。如果在同一步骤中进行了多个更新（例如运行了多个节点），则这些更新会分别流式传输。	`.stream()` / `.astream()` 配合 `stream_mode="updates"`
`messages-tuple`	流式传输 LLM token 以及调用 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

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

JavaScript

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

cURL

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

流式传输图状态

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

* updates 流式传输 **更新** 到图的每一步之后的状态。
* values 流式传输 **完整值** 在图的每一步之后。

Example graph

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

提示

有状态运行 以下示例假设你希望 **持久化输出** 流式运行的结果到 checkpointer 数据库，并已创建了线程。创建线程：

Python

from langgraph_sdk import get_client
client = get_client(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"]

JavaScript

const client = new Client({ apiUrl:  });

// Using the graph deployed with the name "agent"
const assistantID = "agent";
// create a thread
const thread = await client.threads.create();
const threadID = thread["thread_id"]

cURL

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

如果不需要持久化运行的输出，可以传递 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)

JavaScript

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

cURL

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

流式模式： `values`

使用此模式流式传输 **完整状态** 在图的每一步之后。

Python

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

JavaScript

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

cURL

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

子图

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

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

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

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

一旦你有了正在运行的 Agent Server，可以使用 LangGraph SDK

Python

      from langgraph_sdk import get_client
      client = get_client(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)

1. 设置 stream_subgraphs=True 来流式传输子图的输出。

JavaScript

      const client = new Client({ apiUrl:  });

      // Using the graph deployed with the name "agent"
      const assistantID = "agent";

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

      // create a streaming run
      const streamResponse = client.runs.stream(
        threadID,
        assistantID,
        {
          input: { foo: "foo" },
          streamSubgraphs: true,  // (1)!
          streamMode: "updates"
        }
      );
      for await (const chunk of streamResponse) {
        console.log(chunk);
      }

1. 设置 streamSubgraphs: true 来流式传输子图的输出。

cURL

创建线程：

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

创建流式运行：

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

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

<a id="debug"></a> ## 调试

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

Python

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

JavaScript

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

cURL

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

<a id="messages"></a> ## LLM 令牌

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

从 messages-tuple 模式返回的流式输出是一个元组 (message_chunk, metadata) where:

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

Example graph

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

1. 请注意，即使使用 invoke 而不是 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)

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

JavaScript

    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)!
    }

cURL

    curl --request POST \
    --url /threads//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

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

JavaScript

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

cURL

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

流式传输事件

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

Python

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

JavaScript

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

cURL

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

无状态运行

如果您不想 **持久化输出** 流式传输运行的检查点 DB 中，您可以创建无状态运行而不创建线程：

Python

    from langgraph_sdk import get_client
    client = get_client(url=, 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.

JavaScript

    const client = new Client({ apiUrl: , apiKey:  });

    // create a streaming run
    const streamResponse = client.runs.stream(
      null,  // (1)!
      assistantID,
      {
        input,
        streamMode: "updates"
      }
    );
    for await (const chunk of streamResponse) {
      console.log(chunk.data);
    }

1. 我们传入 None 而不是 thread_id UUID.

cURL

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

加入并流式传输

LangSmith 允许您加入活跃的后台运行并从中流式输出。要实现这一点，你可以使用 LangGraph SDK 的 client.runs.join_stream method:

Python

    from langgraph_sdk import get_client
    client = get_client(url=, api_key=)

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

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

JavaScript

    const client = new Client({ apiUrl: , apiKey:  });

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

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

cURL

    curl --request GET \
    --url /threads//runs//stream \
    --header 'Content-Type: application/json' \
    --header 'x-api-key: '

流式传输线程

线程流式传输为线程打开一个长连接，并从 **每个运行** 中流式输出。这允许你从单个连接监控线程上的所有活动，例如，在聊天界面中，多个运行可能通过后续消息、 human-in-the-loop 恢复，或后台运行触发。要通过 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=, api_key=)

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

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

JavaScript

    const client = new Client({ apiUrl: , apiKey:  });

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

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

cURL

    curl --request GET \
    --url /threads//stream \
    --header 'x-api-key: '

线程流模式

线程流式传输支持三种流模式，用于控制返回哪些事件。通过 stream_mode parameter.

模式	描述
`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)

JavaScript

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

cURL

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

从最后一个事件恢复

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

Python

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

JavaScript

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

cURL

    curl --request GET \
    --url /threads//stream \
    --header 'x-api-key: ' \
    --header 'Last-Event-ID: '

API 参考

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

基本用法

Python

JavaScript

cURL

Extended example: streaming updates

Python

JavaScript

cURL

支持的流模式

流式传输多种模式

Python

JavaScript

cURL

流式传输图状态

Example graph

流式模式： updates

Python

JavaScript

cURL

流式模式： values

Python

JavaScript

cURL

子图

Extended example: streaming from subgraphs

Python

JavaScript

cURL

Python

JavaScript

cURL

Example graph

Python

JavaScript

cURL

过滤 LLM 令牌

流式传输自定义数据

Python

JavaScript

cURL

流式传输事件

Python

JavaScript

cURL

无状态运行

Python

JavaScript

cURL

加入并流式传输

Python

JavaScript

cURL

流式传输线程

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

基本用法

Python

JavaScript

cURL

线程流模式

Python

JavaScript

cURL

从最后一个事件恢复

Python

JavaScript

cURL

API 参考

流式模式： `updates`

流式模式： `values`