Skip to content

流输出

流式 API

LangGraph 图表暴露了 .stream()(同步)和 .astream()(异步)方法,用于以迭代器的形式生成流式输出。

基本使用示例:

for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)
async for chunk in graph.astream(inputs, stream_mode="updates"):
    print(chunk)
扩展示例:流式更新
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()
)

for chunk in graph.stream( # (1)!
    {"topic": "ice cream"},
    stream_mode="updates", # (2)!
):
    print(chunk)
  1. stream() 方法返回一个迭代器,用于生成流式输出。
  2. 设置 stream_mode="updates" 只流式传输每个节点之后对图状态的更新。还有其他流式模式可用。有关详细信息,请参阅 支持的流式模式
{'refine_topic': {'topic': 'ice cream and cats'}}
{'generate_joke': {'joke': 'This is a joke about ice cream and cats'}}

支持的流式模式

模式 描述
values 在图表的每一步后流式传输状态的完整值。
updates 在图表的每一步后流式传输对状态的更新。如果在同一个步骤中进行了多次更新(例如,运行了多个节点),这些更新将分别流式传输。
custom 从您的图表节点内部流式传输自定义数据。
messages 从任何调用了 LLM 的图表节点中流式传输 2 元组(LLM 标记,元数据)。
debug 在图表执行过程中尽可能多地流式传输信息。

流式传输多种模式

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

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

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

流式图状态

使用流式模式 updatesvalues 来在图执行过程中流式传输图的状态。

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

API Reference: StateGraph | START | END

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

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

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="updates",
):
    print(chunk)

使用此选项在每一步后流式传输图的 完整状态

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="values",
):
    print(chunk)

子图

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

输出将以元组 (namespace, data) 的形式进行流式传输,其中 namespace 是一个元组,表示调用子图的节点路径,例如 ("parent_node:<task_id>", "child_node:<task_id>")

for chunk in graph.stream(
    {"foo": "foo"},
    subgraphs=True, # (1)!
    stream_mode="updates",
):
    print(chunk)
  1. 设置 subgraphs=True 以从子图中流式传输输出。
扩展示例:从子图中流式传输
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()

for chunk in graph.stream(
    {"foo": "foo"},
    stream_mode="updates",
    subgraphs=True, # (1)!
):
    print(chunk)
  1. 设置 subgraphs=True 以从子图中流式传输输出。
((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',), {'subgraph_node_1': {'bar': 'bar'}})
(('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',), {'subgraph_node_2': {'foo': 'hi! foobar'}})
((), {'node_2': {'foo': 'hi! foobar'}})

注意 我们不仅接收到节点更新,还接收到命名空间,这些命名空间告诉我们我们正在从哪个图(或子图)中进行流式传输。

调试

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

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="debug",
):
    print(chunk)

LLM tokens

使用 messages 流式模式从图中的任何部分(包括节点、工具、子图或任务)以 逐个 token 的方式流式输出大型语言模型(LLM)的输出。

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

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

如果你的 LLM 不是作为 LangChain 集成可用的,你可以改用 custom 模式来流式输出其结果。详情请参阅 与任意 LLM 一起使用

Python < 3.11 中需要手动配置异步

当使用 Python < 3.11 进行异步代码时,你必须显式地将 RunnableConfig 传递给 ainvoke() 以启用正确的流式处理。详情请参阅 Python < 3.11 的异步 或升级到 Python 3.11+。

API Reference: init_chat_model | StateGraph | START

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


llm = init_chat_model(model="openai:gpt-4o-mini")

def call_model(state: MyState):
    """调用 LLM 生成关于某个主题的笑话"""
    llm_response = llm.invoke( # (1)!
        [
            {"role": "user", "content": f"Generate a joke about {state.topic}"}
        ]
    )
    return {"joke": llm_response.content}

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

for message_chunk, metadata in graph.stream( # (2)!
    {"topic": "ice cream"},
    stream_mode="messages",
):
    if message_chunk.content:
        print(message_chunk.content, end="|", flush=True)
  1. 注意即使使用 .invoke 而不是 .stream 来运行 LLM,也会发出消息事件。
  2. "messages" 流式模式返回一个由元组 (message_chunk, metadata) 组成的迭代器,其中 message_chunk 是 LLM 流式的 token,而 metadata 是一个包含有关调用 LLM 的图节点和其他信息的字典。

按 LLM 调用过滤

你可以为 LLM 调用关联 tags,以便通过 LLM 调用来过滤流式 token。

API Reference: init_chat_model

from langchain.chat_models import init_chat_model

llm_1 = init_chat_model(model="openai:gpt-4o-mini", tags=['joke']) # (1)!
llm_2 = init_chat_model(model="openai:gpt-4o-mini", tags=['poem']) # (2)!

graph = ... # 定义一个使用这些 LLMs 的图

async for msg, metadata in graph.astream(  # (3)!
    {"topic": "cats"},
    stream_mode="messages",
):
    if metadata["tags"] == ["joke"]: # (4)!
        print(msg.content, end="|", flush=True)
  1. llm_1 标记为 "joke"。
  2. llm_2 标记为 "poem"。
  3. stream_mode 设置为 "messages" 以流式输出 LLM tokens。metadata 包含有关 LLM 调用的信息,包括标签。
  4. 通过 metadata 中的 tags 字段筛选流式 token,仅包括标记为 "joke" 的 LLM 调用的 token。
扩展示例:按标签进行筛选
from typing import TypedDict

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

joke_model = init_chat_model(model="openai:gpt-4o-mini", tags=["joke"]) # (1)!
poem_model = init_chat_model(model="openai:gpt-4o-mini", tags=["poem"]) # (2)!


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


async def call_model(state, config):
      topic = state["topic"]
      print("Writing joke...")
      # 注意:在 Python < 3.11 中,必须显式传递 config 以确保上下文变量正确传播
      # 因为在此之前未添加上下文变量支持:https://docs.python.org/3/library/asyncio-task.html#creating-tasks
      joke_response = await joke_model.ainvoke(
            [{"role": "user", "content": f"Write a joke about {topic}"}],
            config, # (3)!
      )
      print("\n\nWriting poem...")
      poem_response = await poem_model.ainvoke(
            [{"role": "user", "content": f"Write a short poem about {topic}"}],
            config, # (3)!
      )
      return {"joke": joke_response.content, "poem": poem_response.content}


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

async for msg, metadata in graph.astream(
      {"topic": "cats"},
      stream_mode="messages", # (4)!
):
    if metadata["tags"] == ["joke"]: # (4)!
        print(msg.content, end="|", flush=True)
  1. joke_model 被标记为 "joke"。
  2. poem_model 被标记为 "poem"。
  3. config 显式传递以确保上下文变量正确传播。这在使用异步代码时对于 Python < 3.11 是必需的。更多细节请参见 异步部分
  4. stream_mode 设置为 "messages" 以流式输出 LLM tokens。metadata 包含有关 LLM 调用的信息,包括标签。

按节点过滤

要仅从特定节点流式输出 token,请使用 stream_mode="messages" 并通过流式 metadata 中的 langgraph_node 字段筛选输出:

for msg, metadata in graph.stream( # (1)!
    inputs,
    stream_mode="messages",
):
    if msg.content and metadata["langgraph_node"] == "some_node_name": # (2)!
        ...
  1. "messages" 流式模式返回一个 (message_chunk, metadata) 元组,其中 message_chunk 是 LLM 流式的 token,而 metadata 是一个包含有关调用 LLM 的图节点和其他信息的字典。
  2. 通过流式 metadata 中的 langgraph_node 字段筛选流式 token,仅包括来自 write_poem 节点的 token。
扩展示例:从特定节点流式输出 LLM tokens
from typing import TypedDict
from langgraph.graph import START, StateGraph 
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-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)
      # 同时写笑话和诗
      .add_edge(START, "write_joke")
      .add_edge(START, "write_poem")
      .compile()
)

for msg, metadata in graph.stream( # (1)!
    {"topic": "cats"},
    stream_mode="messages",
):
    if msg.content and metadata["langgraph_node"] == "write_poem": # (2)!
        print(msg.content, end="|", flush=True)
  1. "messages" 流式模式返回一个 (message_chunk, metadata) 元组,其中 message_chunk 是 LLM 流式的 token,而 metadata 是一个包含有关调用 LLM 的图节点和其他信息的字典。
  2. 通过流式 metadata 中的 langgraph_node 字段筛选流式 token,仅包括来自 write_poem 节点的 token。

流式传输自定义数据

要在 LangGraph 节点或工具内部发送 自定义用户定义数据,请按照以下步骤操作:

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

Python < 3.11 的异步中没有 get_stream_writer()

在 Python < 3.11 上运行的异步代码中,get_stream_writer() 将无法工作。
相反,在你的节点或工具中添加一个 writer 参数,并手动传递它。
有关使用示例,请参见 Async with Python < 3.11

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):
    writer = get_stream_writer()  # (1)!
    writer({"custom_key": "Generating custom data inside node"}) # (2)!
    return {"answer": "some data"}

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

inputs = {"query": "example"}

# Usage
for chunk in graph.stream(inputs, stream_mode="custom"):  # (3)!
    print(chunk)
  1. 获取流式写入器以发送自定义数据。
  2. 发出一个自定义键值对(例如进度更新)。
  3. 设置 stream_mode="custom" 以在流中接收自定义数据。
from langchain_core.tools import tool
from langgraph.config import get_stream_writer

@tool
def query_database(query: str) -> str:
    """Query the database."""
    writer = get_stream_writer() # (1)!
    writer({"data": "Retrieved 0/100 records", "type": "progress"}) # (2)!
    # perform query
    writer({"data": "Retrieved 100/100 records", "type": "progress"}) # (3)!
    return "some-answer" 


graph = ... # define a graph that uses this tool

for chunk in graph.stream(inputs, stream_mode="custom"): # (4)!
    print(chunk)
  1. 访问流式写入器以发送自定义数据。
  2. 发出一个自定义键值对(例如进度更新)。
  3. 发出另一个自定义键值对。
  4. 设置 stream_mode="custom" 以在流中接收自定义数据。

与任意 LLM 一起使用

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

这使得你可以集成原始的 LLM 客户端或提供自己流式接口的外部服务,从而使 LangGraph 在自定义设置中具有高度的灵活性。

API Reference: get_stream_writer

from langgraph.config import get_stream_writer

def call_arbitrary_model(state):
    """调用任意模型并流式传输输出的示例节点"""
    writer = get_stream_writer() # (1)!
    # 假设你有一个可以生成块的流式客户端
    for chunk in your_custom_streaming_client(state["topic"]): # (2)!
        writer({"custom_llm_chunk": chunk}) # (3)!
    return {"result": "completed"}

graph = (
    StateGraph(State)
    .add_node(call_arbitrary_model)
    # 根据需要添加其他节点和边
    .compile()
)

for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="custom", # (4)!
):
    # 块将包含从 LLM 流式传输的自定义数据
    print(chunk)
  1. 获取流写入器以发送自定义数据。
  2. 使用你的自定义流式客户端生成 LLM 令牌。
  3. 使用写入器将自定义数据发送到流中。
  4. 设置 stream_mode="custom" 以在流中接收自定义数据。
扩展示例:流式传输任意聊天模型
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-4o-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}


# 这是我们的工具
async def get_items(place: str) -> str:
    """使用此工具列出可能在询问的地方中找到的物品。"""
    writer = get_stream_writer()
    response = ""
    async for msg_chunk in stream_tokens(
        model_name,
        [
            {
                "role": "user",
                "content": (
                    "你能告诉我可能会在以下地方找到哪些类型的物品吗:'{place}'。 "
                    "列出至少3种这样的物品,并用逗号分隔它们。 "
                    "并为每种物品包括一个简短的描述。"
                ),
            }
        ],
    ):
        response += msg_chunk["content"]
        writer(msg_chunk)

    return response


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


# 这是调用工具的图节点
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"不支持工具 {function_name}")

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

让我们使用包含工具调用的 AI 消息来调用图:

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",
):
    print(chunk["content"], end="|", flush=True)

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

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

在初始化模型时,设置 disable_streaming=True

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "anthropic:claude-3-7-sonnet-latest",
    disable_streaming=True # (1)!
)
  1. 设置 disable_streaming=True 以禁用聊天模型的流式传输。
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="o1-preview", disable_streaming=True) # (1)!
  1. 设置 disable_streaming=True 以禁用聊天模型的流式传输。

使用 Python < 3.11 的异步功能

在 Python 版本 < 3.11 中,asyncio tasks 不支持 context 参数。
这限制了 LangGraph 自动传播上下文的能力,并影响 LangGraph 的流式处理机制的两个关键方面:

  1. 必须 显式地将 RunnableConfig 传递给异步 LLM 调用(例如 ainvoke()),因为回调不会自动传播。
  2. 不能 在异步节点或工具中使用 get_stream_writer() —— 你必须直接传递 writer 参数。
扩展示例:手动配置的异步 LLM 调用
from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain.chat_models import init_chat_model

llm = init_chat_model(model="openai:gpt-4o-mini")

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

async def call_model(state, config): # (1)!
    topic = state["topic"]
    print("Generating joke...")
    joke_response = await llm.ainvoke(
        [{"role": "user", "content": f"Write a joke about {topic}"}],
        config, # (2)!
    )
    return {"joke": joke_response.content}

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

async for chunk, metadata in graph.astream(
    {"topic": "ice cream"},
    stream_mode="messages", # (3)!
):
    if chunk.content:
        print(chunk.content, end="|", flush=True)
  1. 在异步节点函数中接受 config 作为参数。
  2. config 传递给 llm.ainvoke() 以确保正确的上下文传播。
  3. 设置 stream_mode="messages" 来流式传输 LLM 的 token。
扩展示例:自定义流式处理与流式写入器
from typing import TypedDict
from langgraph.types import StreamWriter

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

async def generate_joke(state: State, writer: StreamWriter): # (1)!
      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()
)

async for chunk in graph.astream(
      {"topic": "ice cream"},
      stream_mode="custom", # (2)!
):
      print(chunk)
  1. 在异步节点或工具的函数签名中添加 writer 作为参数。LangGraph 会自动将流式写入器传递给该函数。
  2. 设置 stream_mode="custom" 以接收自定义数据流。