流式传输

流式传输是构建响应式应用程序的关键。您可能需要流式传输以下几种类型的数据：

  1. 代理进度 — 在代理图中的每个节点执行后获取更新。
  2. LLM 令牌 — 在语言模型生成令牌时进行流式传输。
  3. 自定义更新 — 在执行过程中从工具发出自定义数据（例如，“已获取 10/100 条记录”）

您可以同时流式传输多种类型的数据。

image

等待是留给鸽子的。

代理进度

要流式传输代理进度，请使用带有 stream_mode="updates" 的 [stream()](https://langgraph.com.cn/reference/graphs/index.html#langgraph.graph.state.CompiledStateGraph.stream “ stream”) 或 [astream()](https://langgraph.com.cn/reference/graphs/index.html#langgraph.graph.state.CompiledStateGraph.astream “ astream async ”) 方法。这会在每个代理步骤后发出一个事件。

例如，如果您有一个代理调用工具一次，您应该会看到以下更新：

  • LLM 节点：带有工具调用请求的 AI 消息
  • 工具节点：带有执行结果的工具消息
  • LLM 节点：最终 AI 响应
<span id="__span-0-1">agent = create_react_agent(
<span id="__span-0-2">    model="anthropic:claude-3-7-sonnet-latest",
<span id="__span-0-3">    tools=[get_weather],
<span id="__span-0-4">)
<span id="__span-0-5">for chunk in agent.stream(
<span id="__span-0-6">    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
<span id="__span-0-7">    stream_mode="updates"
<span id="__span-0-8">):
<span id="__span-0-9">    print(chunk)
<span id="__span-0-10">    print("\n")

LLM 令牌

要流式传输 LLM 生成的令牌，请使用 stream_mode="messages"

<span id="__span-2-1">agent = create_react_agent(
<span id="__span-2-2">    model="anthropic:claude-3-7-sonnet-latest",
<span id="__span-2-3">    tools=[get_weather],
<span id="__span-2-4">)
<span id="__span-2-5">for token, metadata in agent.stream(
<span id="__span-2-6">    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
<span id="__span-2-7">    stream_mode="messages"
<span id="__span-2-8">):
<span id="__span-2-9">    print("Token", token)
<span id="__span-2-10">    print("Metadata", metadata)
<span id="__span-2-11">    print("\n")

要流式传输工具执行时的更新，您可以使用 [get_stream_writer](https://langgraph.com.cn/reference/config/index.html#langgraph.config.get_stream_writer “ get_stream_writer")。

<span id="__span-4-1">from langgraph.config import get_stream_writer
<span id="__span-4-2">
<span id="__span-4-3">def get_weather(city: str) -&gt; str:
<span id="__span-4-4">    """Get weather for a given city."""
<span id="__span-4-5">    writer = get_stream_writer()
<span id="__span-4-6">    # stream any arbitrary data
<span id="__span-4-7">    writer(f"Looking up data for city: {city}")
<span id="__span-4-8">    return f"It's always sunny in {city}!"
<span id="__span-4-9">
<span id="__span-4-10">agent = create_react_agent(
<span id="__span-4-11">    model="anthropic:claude-3-7-sonnet-latest",
<span id="__span-4-12">    tools=[get_weather],
<span id="__span-4-13">)
<span id="__span-4-14">
<span id="__span-4-15">for chunk in agent.stream(
<span id="__span-4-16">    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
<span id="__span-4-17">    stream_mode="custom"
<span id="__span-4-18">):
<span id="__span-4-19">    print(chunk)
<span id="__span-4-20">    print("\n")

注意

如果您在工具内部添加 get_stream_writer，您将无法在 LangGraph 执行上下文之外调用该工具。

流式传输多种模式

您可以通过将流模式作为列表传递来指定多种流式传输模式：stream_mode=["updates", "messages", "custom"]

<span id="__span-6-1">agent = create_react_agent(
<span id="__span-6-2">    model="anthropic:claude-3-7-sonnet-latest",
<span id="__span-6-3">    tools=[get_weather],
<span id="__span-6-4">)
<span id="__span-6-5">
<span id="__span-6-6">for stream_mode, chunk in agent.stream(
<span id="__span-6-7">    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
<span id="__span-6-8">    stream_mode=["updates", "messages", "custom"]
<span id="__span-6-9">):
<span id="__span-6-10">    print(chunk)
<span id="__span-6-11">    print("\n")

禁用流式传输

在某些应用程序中，您可能需要禁用给定模型的单个令牌流式传输。这在多代理系统中很有用，用于控制哪些代理流式传输其输出。

请参阅模型指南以了解如何禁用流式传输。

更多资源