流式传输

TIP

对于新应用，我们推荐事件流式传输——这是 LangGraph v1.2 中引入的类型化投影 API。事件流式传输为每个投影（消息、值、子图、输出）提供单独的迭代器，因此你可以独立地消费它们，而不必根据 stream_mode 块进行分支判断。

本页涵盖 LangGraph 的流模式 API。它通过 updates、values、messages、custom、checkpoints、tasks 和 debug 等流模式暴露图的执行。当你需要直接访问图运行时事件或特定的流模式输出时，请使用它。

开始使用

基本用法

LangGraph 图暴露了 stream（同步）和 astream（异步）方法，以迭代器的形式产出流式输出。传入一个或多个流模式来控制你接收的数据。

python

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["updates", "custom"],  
    version="v2",  
):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node {node_name} updated: {state}")
    elif chunk["type"] == "custom":
        print(f"Status: {chunk['data']['status']}")

shell

Status: thinking of a joke...
Node generate_joke updated: {'joke': 'Why did the ice cream go to school? To get a sundae education!'}

完整示例

python

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

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

def generate_joke(state: State):
    writer = get_stream_writer()
    writer({"status": "thinking of a joke..."})
    return {"joke": f"Why did the {state['topic']} go to school? To get a sundae education!"}

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

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["updates", "custom"],
    version="v2",
):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node {node_name} updated: {state}")
    elif chunk["type"] == "custom":
        print(f"Status: {chunk['data']['status']}")

shell

Status: thinking of a joke...
Node generate_joke updated: {'joke': 'Why did the ice cream go to school? To get a sundae education!'}

TIP

使用 LangSmith 调试流式事件、逐 token 检查 LLM 输出并监控延迟。按照追踪快速入门进行设置。

流输出格式 (v2)

NOTE

需要 LangGraph 1.1 或更高版本。本页所有示例都使用 version="v2"。

向 stream() 或 astream() 传入 version="v2" 可获得统一的输出格式。每个块都是一个形状一致的 StreamPart 字典——无论流模式、模式数量或子图设置如何：

python

{
    "type": "values" | "updates" | "messages" | "custom" | "checkpoints" | "tasks" | "debug",
    "ns": (),           # namespace tuple, populated for subgraph events
    "data": ...,        # the actual payload (type varies by stream mode)
}

每个流模式都有一个对应的 TypedDict，其中包含 ValuesStreamPart、UpdatesStreamPart、MessagesStreamPart、CustomStreamPart、CheckpointStreamPart、TasksStreamPart、DebugStreamPart。你可以从 langgraph.types 导入这些类型。联合类型 StreamPart 是基于 part["type"] 的可辨识联合，支持在编辑器和类型检查器中实现完整的类型收窄。

在 v1（默认）中，输出格式会根据你的流式传输选项而变化（单模式返回原始数据，多模式返回 (mode, data) 元组，子图返回 (namespace, data) 元组）。在 v2 中，格式始终相同：

v2（新增）v1（当前默认）

python

for chunk in graph.stream(inputs, stream_mode="updates", version="v2"):
    print(chunk["type"])  # "updates"
    print(chunk["ns"])    # ()
    print(chunk["data"])  # {"node_name": {"key": "value"}}

python

for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)  # {"node_name": {"key": "value"}}

v2 格式还支持类型收窄，这意味着你可以按 chunk["type"] 过滤块并获得正确的载荷类型。每个分支会将 part["data"] 收窄为该模式的具体类型：

python

for part in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["values", "updates", "messages", "custom"],
    version="v2",
):
    if part["type"] == "values":
        # ValuesStreamPart — full state snapshot after each step
        print(f"State: topic={part['data']['topic']}")
    elif part["type"] == "updates":
        # UpdatesStreamPart — only the changed keys from each node
        for node_name, state in part["data"].items():
            print(f"Node `{node_name}` updated: {state}")
    elif part["type"] == "messages":
        # MessagesStreamPart — (message_chunk, metadata) from LLM calls
        msg, metadata = part["data"]
        print(msg.content, end="", flush=True)
    elif part["type"] == "custom":
        # CustomStreamPart — arbitrary data from get_stream_writer()
        print(f"Progress: {part['data']['progress']}%")

流模式

将以下一种或多种流模式以列表形式传给 stream 或 astream 方法：

模式	类型	描述
values	`ValuesStreamPart`	每个步骤之后的完整状态。
updates	`UpdatesStreamPart`	每个步骤之后的状态更新。同一步骤中的多个更新会分别流式输出。
messages	`MessagesStreamPart`	来自 LLM 调用的 (LLM token, metadata) 二元组。
custom	`CustomStreamPart`	节点通过 `get_stream_writer` 发出的自定义数据。
checkpoints	`CheckpointStreamPart`	检查点事件（与 `get_state()` 格式相同）。需要检查点器。
tasks	`TasksStreamPart`	带有结果和错误的任务开始/结束事件。需要检查点器。
debug	`DebugStreamPart`	所有可用信息——将 `checkpoints` 和 `tasks` 与额外元数据相结合。

图状态

使用流模式 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"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

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="updates",  
    version="v2",  
):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node `{node_name}` updated: {state}")

shell

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

LLM token

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

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

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

如果你的 LLM 无法作为 LangChain 集成使用，你可以改用 custom 模式来流式输出其结果。详情请参见与任何 LLM 配合使用。

WARNING

在 Python < 3.11 中使用异步时需要手动配置 在 Python < 3.11 上使用异步代码时，你必须显式地将 RunnableConfig 传给 ainvoke() 才能启用正确的流式传输。详情请参见 Python < 3.11 中的异步，或升级到 Python 3.11+。

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"""
    # Note that message events are emitted even when the LLM is run using .invoke rather than .stream
    model_response = model.invoke(  
        [
            {"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()
)

# The "messages" stream mode streams LLM tokens with metadata
# Use version="v2" for a unified StreamPart format
for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        message_chunk, metadata = chunk["data"]
        if message_chunk.content:
            print(message_chunk.content, end="|", flush=True)

按 LLM 调用过滤

你可以将 tags 与 LLM 调用关联，以便按 LLM 调用对流式输出的 token 进行过滤。

python

from langchain.chat_models import init_chat_model

# model_1 is tagged with "joke"
model_1 = init_chat_model(model="gpt-5.4-mini", tags=['joke'])
# model_2 is tagged with "poem"
model_2 = init_chat_model(model="gpt-5.4-mini", tags=['poem'])

graph = ... # define a graph that uses these LLMs

# The stream_mode is set to "messages" to stream LLM tokens
# The metadata contains information about the LLM invocation, including the tags
async for chunk in graph.astream(
    {"topic": "cats"},
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        # Filter the streamed tokens by the tags field in the metadata to only include
        # the tokens from the LLM invocation with the "joke" tag
        if metadata["tags"] == ["joke"]:
            print(msg.content, end="|", flush=True)

扩展示例：按标签过滤

python

from typing import TypedDict

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

# The joke_model is tagged with "joke"
joke_model = init_chat_model(model="gpt-5.4-mini", tags=["joke"])
# The poem_model is tagged with "poem"
poem_model = init_chat_model(model="gpt-5.4-mini", tags=["poem"])

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

async def call_model(state, config):
      topic = state["topic"]
      print("Writing joke...")
      # Note: Passing the config through explicitly is required for python < 3.11
      # Since context var support wasn't added before then: https://docs.python.org/3/library/asyncio-task.html#creating-tasks
      # The config is passed through explicitly to ensure the context vars are propagated correctly
      # This is required for Python < 3.11 when using async code. Please see the async section for more details
      joke_response = await joke_model.ainvoke(
            [{"role": "user", "content": f"Write a joke about {topic}"}],
            config,
      )
      print("\n\nWriting poem...")
      poem_response = await poem_model.ainvoke(
            [{"role": "user", "content": f"Write a short poem about {topic}"}],
            config,
      )
      return {"joke": joke_response.content, "poem": poem_response.content}

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

# The stream_mode is set to "messages" to stream LLM tokens
# The metadata contains information about the LLM invocation, including the tags
async for chunk in graph.astream(
      {"topic": "cats"},
      stream_mode="messages",
      version="v2",
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        if metadata["tags"] == ["joke"]:
            print(msg.content, end="|", flush=True)

从流中省略消息

使用 nostream 标签可将 LLM 输出完全排除在流之外。带有 nostream 标签的调用仍会运行并产生输出；只是在 messages 模式下不会发出它们的 token。

这在以下情况中很有用：

你需要 LLM 输出用于内部处理（例如结构化输出），但不希望将其流式输出到客户端
你通过不同的通道（例如自定义 UI 消息）流式输出相同的内容，并希望避免在 messages 流中出现重复输出

python

from typing import Any, TypedDict

from langchain_anthropic import ChatAnthropic
from langgraph.graph import START, StateGraph

stream_model = ChatAnthropic(model_name="claude-haiku-4-5-20251001")
internal_model = ChatAnthropic(model_name="claude-haiku-4-5-20251001").with_config(
    {"tags": ["nostream"]}
)

class State(TypedDict):
    topic: str
    answer: str
    notes: str

def answer(state: State) -> dict[str, Any]:
    r = stream_model.invoke(
        [{"role": "user", "content": f"Reply briefly about {state['topic']}"}]
    )
    return {"answer": r.content}

def internal_notes(state: State) -> dict[str, Any]:
    # Tokens from this model are omitted from stream_mode="messages" because of nostream
    r = internal_model.invoke(
        [{"role": "user", "content": f"Private notes on {state['topic']}"}]
    )
    return {"notes": r.content}

graph = (
    StateGraph(State)
    .add_node("write_answer", answer)
    .add_node("internal_notes", internal_notes)
    .add_edge(START, "write_answer")
    .add_edge("write_answer", "internal_notes")
    .compile()
)

initial_state: State = {"topic": "AI", "answer": "", "notes": ""}
stream = graph.stream_events(initial_state, version="v3")

View example trace

打开本示例的一个公开 LangSmith 运行记录。

按节点过滤

若要仅从特定节点流式输出 token，请使用 stream_mode="messages"，并按流式元数据中的 langgraph_node 字段过滤输出：

python

# The "messages" stream mode streams LLM tokens with metadata
# Use version="v2" for a unified StreamPart format
for chunk in graph.stream(
    inputs,
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        # Filter the streamed tokens by the langgraph_node field in the metadata
        # to only include the tokens from the specified node
        if msg.content and metadata["langgraph_node"] == "some_node_name":
            ...

扩展示例：从特定节点流式输出 LLM token

python

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

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

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

def write_joke(state: State):
      topic = state["topic"]
      joke_response = model.invoke(
            [{"role": "user", "content": f"Write a joke about {topic}"}]
      )
      return {"joke": joke_response.content}

def write_poem(state: State):
      topic = state["topic"]
      poem_response = model.invoke(
            [{"role": "user", "content": f"Write a short poem about {topic}"}]
      )
      return {"poem": poem_response.content}

graph = (
      StateGraph(State)
      .add_node(write_joke)
      .add_node(write_poem)
      # write both the joke and the poem concurrently
      .add_edge(START, "write_joke")
      .add_edge(START, "write_poem")
      .compile()
)

# The "messages" stream mode streams LLM tokens with metadata
# Use version="v2" for a unified StreamPart format
for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        # Filter the streamed tokens by the langgraph_node field in the metadata
        # to only include the tokens from the write_poem node
        if msg.content and metadata["langgraph_node"] == "write_poem":
            print(msg.content, end="|", flush=True)

自定义数据

要从 LangGraph 节点或工具内部发送 自定义的用户定义数据，请按以下步骤操作：

使用 get_stream_writer 访问流写入器并发出自定义数据。
在调用 .stream() 或 .astream() 时设置 stream_mode="custom"，以在流中获取自定义数据。你可以组合多个模式（例如 ["updates", "custom"]），但至少有一个必须是 "custom"。

WARNING

在 Python < 3.11 的异步环境中无法使用 get_stream_writer 在 Python < 3.11 上运行的异步代码中，get_stream_writer 将无法工作。请改为向节点或工具添加 writer 参数，并手动传递它。使用示例请参见 Python < 3.11 中的异步。

python

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

class State(TypedDict):
    query: str
    answer: str

def node(state: State):
    # Get the stream writer to send custom data
    writer = get_stream_writer()
    # Emit a custom key-value pair (e.g., progress update)
    writer({"custom_key": "Generating custom data inside node"})
    return {"answer": "some data"}

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

inputs = {"query": "example"}

# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(inputs, stream_mode="custom", version="v2"):
    if chunk["type"] == "custom":
        print(f"Custom event: {chunk['data']['custom_key']}")

子图输出

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

输出将作为元组 (namespace, data) 流式输出，其中 namespace 是一个元组，包含调用子图的节点路径，例如 ("parent_node:<task_id>", "child_node:<task_id>")。

使用 version="v2" 时，子图事件使用相同的 StreamPart 格式。ns 字段标识来源：

python

for chunk in graph.stream(
    {"foo": "foo"},
    subgraphs=True,  
    stream_mode="updates",
    version="v2", 
):
    print(chunk["type"])  # "updates"
    print(chunk["ns"])    # () for root, ("node_name:<task_id>",) for subgraph
    print(chunk["data"])  # {"node_name": {"key": "value"}}

NOTE

这适用于每一种 stream_mode，包括 "messages"。像 create_agent 这样的智能体构建器会返回编译后的图，因此将它作为一个节点添加会把它变成子图。如果不设置 subgraphs=True，父图上的 stream_mode="messages" 将不会发出内部智能体 LLM 调用产生的 token 块。而直接调用 agent.stream(...) 则会发出，这就是为什么这个问题通常只在包装之后才出现。

python

from langchain.agents import create_agent
from langgraph.graph import END, START, StateGraph

graph = (
    StateGraph(State)
    .add_node("agent", create_agent(model, tools, state_schema=State))
    .add_edge(START, "agent")
    .add_edge("agent", END)
    .compile()
)

for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "..."}]},
    stream_mode="messages",
    subgraphs=True,  
    version="v2",
):
    print(chunk["type"])  # "messages"
    print(chunk["ns"])    # () for root, ("agent:<task_id>",) for subgraph
    print(chunk["data"])  # (token, metadata)

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

python

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

for chunk in graph.stream(
    {"foo": "foo"},
    stream_mode="updates",
    # Set subgraphs=True to stream outputs from subgraphs
    subgraphs=True,  
    version="v2",  
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            print(f"Subgraph {chunk['ns']}: {chunk['data']}")
        else:
            print(f"Root: {chunk['data']}")

Root: {'node_1': {'foo': 'hi! foo'}}
Subgraph ('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',): {'subgraph_node_1': {'bar': 'bar'}}
Subgraph ('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',): {'subgraph_node_2': {'foo': 'hi! foobar'}}
Root: {'node_2': {'foo': 'hi! foobar'}}

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

检查点

使用 checkpoints 流式传输模式，在图执行过程中接收检查点事件。每个检查点事件的格式与 get_state() 的输出相同。需要一个检查点器。

python

from langgraph.checkpoint.memory import MemorySaver

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

config = {"configurable": {"thread_id": "1"}}

for chunk in graph.stream(
    {"topic": "ice cream"},
    config=config,
    stream_mode="checkpoints",  
    version="v2",  
):
    if chunk["type"] == "checkpoints":
        print(chunk["data"])

任务

使用 tasks 流式传输模式，在图执行过程中接收任务开始和结束事件。任务事件包含有关正在运行的节点、其结果以及任何错误的信息。需要一个检查点器。

python

from langgraph.checkpoint.memory import MemorySaver

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

config = {"configurable": {"thread_id": "1"}}

for chunk in graph.stream(
    {"topic": "ice cream"},
    config=config,
    stream_mode="tasks",  
    version="v2",  
):
    if chunk["type"] == "tasks":
        print(chunk["data"])

调试

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

python

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="debug",  
    version="v2",  
):
    if chunk["type"] == "debug":
        print(chunk["data"])

NOTE

debug 模式将 checkpoints 和 tasks 事件与额外的元数据组合在一起。如果你只需要调试信息的一部分，请直接使用 checkpoints 或 tasks。

同时使用多个模式

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

使用 version="v2" 时，每个块都是一个 StreamPart dict。使用 chunk["type"] 来区分不同的模式：

v2v1

python

for chunk in graph.stream(inputs, stream_mode=["updates", "custom"], version="v2"):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node `{node_name}` updated: {state}")
    elif chunk["type"] == "custom":
        print(f"Custom event: {chunk['data']}")

python

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

高级

与任何 LLM 配合使用

你可以使用 stream_mode="custom" 从 任何 LLM API 流式输出数据——即使该 API 没有实现 LangChain 聊天模型接口。

这让你能够集成原始的 LLM 客户端或提供自身流式接口的外部服务，使 LangGraph 对自定义配置具有高度的灵活性。

python

from langgraph.config import get_stream_writer

def call_arbitrary_model(state):
    """Example node that calls an arbitrary model and streams the output"""
    # Get the stream writer to send custom data
    writer = get_stream_writer()  
    # Assume you have a streaming client that yields chunks
    # Generate LLM tokens using your custom streaming client
    for chunk in your_custom_streaming_client(state["topic"]):
        # Use the writer to send custom data to the stream
        writer({"custom_llm_chunk": chunk})  
    return {"result": "completed"}

graph = (
    StateGraph(State)
    .add_node(call_arbitrary_model)
    # Add other nodes and edges as needed
    .compile()
)
# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="custom",  
    version="v2",  
):
    if chunk["type"] == "custom":
        # The chunk data will contain the custom data streamed from the llm
        print(chunk["data"])

扩展示例：流式输出任意聊天模型

python

import operator
import json

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

from openai import AsyncOpenAI

openai_client = AsyncOpenAI()
model_name = "gpt-5.4-mini"

async def stream_tokens(model_name: str, messages: list[dict]):
    response = await openai_client.chat.completions.create(
        messages=messages, model=model_name, stream=True
    )
    role = None
    async for chunk in response:
        delta = chunk.choices[0].delta

        if delta.role is not None:
            role = delta.role

        if delta.content:
            yield {"role": role, "content": delta.content}

# this is our tool
async def get_items(place: str) -> str:
    """Use this tool to list items one might find in a place you're asked about."""
    writer = get_stream_writer()
    response = ""
    async for msg_chunk in stream_tokens(
        model_name,
        [
            {
                "role": "user",
                "content": (
                    "Can you tell me what kind of items "
                    f"i might find in the following place: '{place}'. "
                    "List at least 3 such items separating them by a comma. "
                    "And include a brief description of each item."
                ),
            }
        ],
    ):
        response += msg_chunk["content"]
        writer(msg_chunk)

    return response

class State(TypedDict):
    messages: Annotated[list[dict], operator.add]

# this is the tool-calling graph node
async def call_tool(state: State):
    ai_message = state["messages"][-1]
    tool_call = ai_message["tool_calls"][-1]

    function_name = tool_call["function"]["name"]
    if function_name != "get_items":
        raise ValueError(f"Tool {function_name} not supported")

    function_arguments = tool_call["function"]["arguments"]
    arguments = json.loads(function_arguments)

    function_response = await get_items(**arguments)
    tool_message = {
        "tool_call_id": tool_call["id"],
        "role": "tool",
        "name": function_name,
        "content": function_response,
    }
    return {"messages": [tool_message]}

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

让我们用一个包含工具调用的 AIMessage 来调用该图：

python

inputs = {
    "messages": [
        {
            "content": None,
            "role": "assistant",
            "tool_calls": [
                {
                    "id": "1",
                    "function": {
                        "arguments": '{"place":"bedroom"}',
                        "name": "get_items",
                    },
                    "type": "function",
                }
            ],
        }
    ]
}

async for chunk in graph.astream(
    inputs,
    stream_mode="custom",
    version="v2",
):
    if chunk["type"] == "custom":
        print(chunk["data"]["content"], end="|", flush=True)

为特定聊天模型禁用流式传输

如果你的应用混合使用了支持流式传输的模型和不支持流式传输的模型，你可能需要为不支持流式传输的模型显式禁用流式传输。

在初始化模型时设置 streaming=False。

python

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "claude-sonnet-4-6",
    # Set streaming=False to disable streaming for the chat model
    streaming=False
)

NOTE

并非所有聊天模型集成都支持 streaming 参数。如果你的模型不支持，请改用 disable_streaming=True。此参数通过基类在所有聊天模型上可用。

迁移到 v2

v2 流式传输格式（本页通篇使用）提供了统一的输出格式。以下是主要差异以及如何迁移的摘要：

场景	v1（默认）	v2（`version="v2"`）
单一流模式	原始数据（dict）	带有 `type`、`ns`、`data` 的 `StreamPart` dict
多个流模式	`(mode, data)` 元组	相同的 `StreamPart` dict，按 `chunk["type"]` 过滤
子图流式传输	`(namespace, data)` 元组	相同的 `StreamPart` dict，检查 `chunk["ns"]`
多个模式 + 子图	`(namespace, mode, data)` 三元组	相同的 `StreamPart` dict
`invoke()` 返回类型	纯 dict（状态）	带有 `.value` 和 `.interrupts` 的 `GraphOutput`
中断位置（流）	状态 dict 中的 `__interrupt__` 键	`values` 流分部上的 `interrupts` 字段
中断位置（invoke）	结果 dict 中的 `__interrupt__` 键	`GraphOutput` 上的 `.interrupts` 属性
Pydantic/dataclass 输出	返回纯 dict	强制转换为模型/dataclass 实例

v2 invoke 格式

当你将 version="v2" 传给 invoke() 或 ainvoke() 时，它会返回一个带有 .value 和 .interrupts 属性的 GraphOutput 对象：

python

from langgraph.types import GraphOutput

result = graph.invoke(inputs, version="v2")

assert isinstance(result, GraphOutput)
result.value       # your output — dict, Pydantic model, or dataclass
result.interrupts  # tuple[Interrupt, ...], empty if none occurred

对于除默认 "values" 之外的任何流模式，invoke(..., stream_mode="updates", version="v2") 会返回 list[StreamPart] 而不是 list[tuple]。

WARNING

在 GraphOutput 上以字典风格访问（result["key"]、"key" in result、result["__interrupt__"]）出于向后兼容仍然可用，但已被弃用，并将在未来版本中移除。请迁移到 result.value 和 result.interrupts。

这样将状态与中断元数据分离。在 v1 中，中断被嵌入到返回的 dict 中，位于 __interrupt__ 下：

v2（新）v1（当前默认）

python

config = {"configurable": {"thread_id": "thread-1"}}
result = graph.invoke(inputs, config=config, version="v2")

if result.interrupts:
    print(result.interrupts[0].value)
    graph.invoke(Command(resume=True), config=config, version="v2")

python

config = {"configurable": {"thread_id": "thread-1"}}
result = graph.invoke(inputs, config=config)

if "__interrupt__" in result:
    print(result["__interrupt__"][0].value)
    graph.invoke(Command(resume=True), config=config)

Pydantic 与 dataclass 状态强制转换

当你的图状态是 Pydantic 模型或 dataclass 时，v2 的 values 模式会自动将输出强制转换为正确的类型：

python

from pydantic import BaseModel
from typing import Annotated
import operator

class MyState(BaseModel):
    value: str
    items: Annotated[list[str], operator.add]

# With version="v2", chunk["data"] is a MyState instance
for chunk in graph.stream(
    {"value": "x", "items": []}, stream_mode="values", version="v2"
):
    print(type(chunk["data"]))  # <class 'MyState'>

Python < 3.11 中的异步

在 Python 版本 < 3.11 中，asyncio 任务不支持 context 参数。这限制了 LangGraph 自动传播上下文的能力，并以两种关键方式影响 LangGraph 的流式传输机制：

你必须显式地将 RunnableConfig 传入异步 LLM 调用（例如 ainvoke()），因为回调不会自动传播。
你不能在异步节点或工具中使用 get_stream_writer——必须直接传入 writer 参数。

扩展示例：使用手动配置的异步 LLM 调用

python

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

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

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

# Accept config as an argument in the async node function
async def call_model(state, config):
    topic = state["topic"]
    print("Generating joke...")
    # Pass config to model.ainvoke() to ensure proper context propagation
    joke_response = await model.ainvoke(  
        [{"role": "user", "content": f"Write a joke about {topic}"}],
        config,
    )
    return {"joke": joke_response.content}

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

# Set stream_mode="messages" to stream LLM tokens
async for chunk in graph.astream(
    {"topic": "ice cream"},
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        message_chunk, metadata = chunk["data"]
        if message_chunk.content:
            print(message_chunk.content, end="|", flush=True)

扩展示例：使用 stream writer 的异步自定义流式传输

python

from typing import TypedDict
from langgraph.types import StreamWriter

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

# Add writer as an argument in the function signature of the async node or tool
# LangGraph will automatically pass the stream writer to the function
async def generate_joke(state: State, writer: StreamWriter):  
      writer({"custom_key": "Streaming custom data while generating a joke"})
      return {"joke": f"This is a joke about {state['topic']}"}

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

# Set stream_mode="custom" to receive the custom data in the stream  #
async for chunk in graph.astream(
      {"topic": "ice cream"},
      stream_mode="custom",
      version="v2",
):
      if chunk["type"] == "custom":
          print(chunk["data"])

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织