流输出¶
流式 API¶
LangGraph 图表暴露了 .stream()
(同步)和 .astream()
(异步)方法,用于以迭代器的形式生成流式输出。
基本使用示例:
扩展示例:流式更新
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)
stream()
方法返回一个迭代器,用于生成流式输出。- 设置
stream_mode="updates"
只流式传输每个节点之后对图状态的更新。还有其他流式模式可用。有关详细信息,请参阅 支持的流式模式。
支持的流式模式¶
模式 | 描述 |
---|---|
values |
在图表的每一步后流式传输状态的完整值。 |
updates |
在图表的每一步后流式传输对状态的更新。如果在同一个步骤中进行了多次更新(例如,运行了多个节点),这些更新将分别流式传输。 |
custom |
从您的图表节点内部流式传输自定义数据。 |
messages |
从任何调用了 LLM 的图表节点中流式传输 2 元组(LLM 标记,元数据)。 |
debug |
在图表执行过程中尽可能多地流式传输信息。 |
流式传输多种模式¶
您可以将列表作为 stream_mode
参数传递,以同时流式传输多种模式。
流式输出将为 (mode, chunk)
的元组,其中 mode
是流式模式的名称,chunk
是该模式流式传输的数据。
流式图状态¶
使用流式模式 updates
和 values
来在图执行过程中流式传输图的状态。
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()
)
子图¶
要将 子图 的输出包含在流式输出中,可以在父图的 .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)
- 设置
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)
- 设置
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
流式传输模式,在图执行过程中尽可能多地传输信息。流式输出包括节点名称以及完整状态。
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)
- 注意即使使用
.invoke
而不是.stream
来运行 LLM,也会发出消息事件。 - "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)
llm_1
标记为 "joke"。llm_2
标记为 "poem"。- 将
stream_mode
设置为 "messages" 以流式输出 LLM tokens。metadata
包含有关 LLM 调用的信息,包括标签。 - 通过
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)
joke_model
被标记为 "joke"。poem_model
被标记为 "poem"。config
显式传递以确保上下文变量正确传播。这在使用异步代码时对于 Python < 3.11 是必需的。更多细节请参见 异步部分。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)!
...
- "messages" 流式模式返回一个
(message_chunk, metadata)
元组,其中message_chunk
是 LLM 流式的 token,而metadata
是一个包含有关调用 LLM 的图节点和其他信息的字典。 - 通过流式 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)
- "messages" 流式模式返回一个
(message_chunk, metadata)
元组,其中message_chunk
是 LLM 流式的 token,而metadata
是一个包含有关调用 LLM 的图节点和其他信息的字典。 - 通过流式 metadata 中的
langgraph_node
字段筛选流式 token,仅包括来自write_poem
节点的 token。
流式传输自定义数据¶
要在 LangGraph 节点或工具内部发送 自定义用户定义数据,请按照以下步骤操作:
- 使用
get_stream_writer()
获取流式写入器并发出自定义数据。 - 在调用
.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)
- 获取流式写入器以发送自定义数据。
- 发出一个自定义键值对(例如进度更新)。
- 设置
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)
- 访问流式写入器以发送自定义数据。
- 发出一个自定义键值对(例如进度更新)。
- 发出另一个自定义键值对。
- 设置
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)
- 获取流写入器以发送自定义数据。
- 使用你的自定义流式客户端生成 LLM 令牌。
- 使用写入器将自定义数据发送到流中。
- 设置
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
。
使用 Python < 3.11 的异步功能¶
在 Python 版本 < 3.11 中,asyncio tasks 不支持 context
参数。
这限制了 LangGraph 自动传播上下文的能力,并影响 LangGraph 的流式处理机制的两个关键方面:
- 你 必须 显式地将
RunnableConfig
传递给异步 LLM 调用(例如ainvoke()
),因为回调不会自动传播。 - 你 不能 在异步节点或工具中使用
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)
- 在异步节点函数中接受
config
作为参数。 - 将
config
传递给llm.ainvoke()
以确保正确的上下文传播。 - 设置
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)
- 在异步节点或工具的函数签名中添加
writer
作为参数。LangGraph 会自动将流式写入器传递给该函数。 - 设置
stream_mode="custom"
以接收自定义数据流。