流式传输

流式传输是构建响应式应用程序的关键。有几种类型的数据是你想要进行流式传输的

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

你可以一次流式传输多种类型的数据。

等待是给鸽子的。

代理进度

要流式传输代理进度，请使用 stream() 或 astream() 方法，并配合使用 stream_mode="updates"。这会在每个代理步骤后发出一个事件。

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

• LLM 节点：带有工具调用请求的 AI 消息
• 工具节点：带有执行结果的工具消息
• LLM 节点：最终 AI 响应

同步异步

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="updates"
):
    print(chunk)
    print("\n")

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
async for chunk in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="updates"
):
    print(chunk)
    print("\n")

LLM 令牌

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

同步异步

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="messages"
):
    print("Token", token)
    print("Metadata", metadata)
    print("\n")

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
async for token, metadata in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="messages"
):
    print("Token", token)
    print("Metadata", metadata)
    print("\n")

工具更新

要流式传输工具执行时产生的更新，你可以使用 get_stream_writer。

同步异步

from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    writer = get_stream_writer()
    # stream any arbitrary data
    writer(f"Looking up data for city: {city}")
    return f"It's always sunny in {city}!"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="custom"
):
    print(chunk)
    print("\n")

from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    writer = get_stream_writer()
    # stream any arbitrary data
    writer(f"Looking up data for city: {city}")
    return f"It's always sunny in {city}!"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

async for chunk in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="custom"
):
    print(chunk)
    print("\n")

注意

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

流式传输多种模式

你可以通过传递一个列表来指定多种流式传输模式：stream_mode=["updates", "messages", "custom"]

同步异步

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

for stream_mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode=["updates", "messages", "custom"]
):
    print(chunk)
    print("\n")

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

async for stream_mode, chunk in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode=["updates", "messages", "custom"]
):
    print(chunk)
    print("\n")

禁用流式传输

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

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

附加资源

• LangGraph 中的流式传输

评论