流式传输¶

流式传输是构建响应式应用的关键。你可能会想流式传输以下几种类型的数据：

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

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

Agent 进度¶

要流式传输 agent 的进度，请使用 stream() 或 astream() 方法，并将 stream_mode="updates" 作为参数传入。这将在每个 agent 步骤后发出一个事件。

例如，如果你的 agent 调用了一次工具，你应该会看到以下更新：

LLM node: 包含工具调用请求的人工智能消息
Tool node: 包含执行结果的工具消息
LLM node: 最终的人工智能响应

同步异步

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 tokens¶

要流式传输由LLM生成的tokens，请使用 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:
    """获取指定城市的天气信息。"""
    writer = get_stream_writer()
    # 流式传输任意数据
    writer(f"正在查找城市 {city} 的数据")
    return f"{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:
    """获取指定城市的天气信息。"""
    writer = get_stream_writer()
    # 流式传输任意数据
    writer(f"正在查找城市 {city} 的数据")
    return f"{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")

禁用流式传输¶

在某些应用中，你可能需要禁用某个模型的单个 token 流式传输。这对于多代理系统来说很有用，可以控制哪些代理流式传输其输出。

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

其他资源¶

LangGraph中的流式处理