LangGraph篇-流式处理

SuifengFlying

公告

欢迎来到我的博客，这里记录生活与技术。

了解更多

1588 字

8 分钟

LangGraph篇-流式处理

2026-07-29

LangGraph

流式处理对于增强基于 LLM 的应用响应性至关重要。通过逐步显示输出，即使在完整响应尚未准备好之前，流式处理显著改善用户体验（UX），尤其是在处理 LLM 的延迟时。

官方文档地址：https://langchain-ai.github.io/langgraph/how-tos/#streaming

中文文档地址：https://www.aidoczh.com/langgraph/how-tos/#_3

流式完整状态#

LangGraph 支持多种流式模式。主要的有

values: 此流式模式流回图的值。这是每个节点调用后图的完整状态。
updates: 此流式模式流回图的更新。这是每个节点调用后图状态的更新。

stream_mode="values"

行为: 在 values 模式下，数据流会返回每个步骤的完整输出。这意味着每一次迭代都会得到一个完整的结果对象。
适用场景: 适用于需要每一步的完整结果进行进一步处理的场景。
示例: 如果你有一个复杂的计算过程，每一步都需要完整的上下文来进行下一步的计算，那么 values 模式是合适的选择。

stream_mode="updates"

行为: updates模式下，数据流会返回每个步骤的增量更新。这意味着每一次迭代只会得到自上一次迭代以来的变化部分。
适用场景: 适用于需要实时更新或增量更新的场景，比如实时显示处理进度或逐步输出的场景。
示例: 如果你在构建一个实时聊天应用程序，需要逐步显示对话内容，那么 updates 模式是合适的选择。

设置#

在本指南中，我们将使用一个简单的 ReAct 代理。

1
pip install -U langgraph langchain-openai

1
# 导入异步IO模块
2
import asyncio
3
# 从类型提示模块中导入Literal
4
from typing import Literal
5
# 导入工具装饰器
6
from langchain_core.tools import tool
7
# 导入OpenAI聊天模型
8
from langchain_openai import ChatOpenAI
9
# 导入创建react代理的函数
10
from langgraph.prebuilt import create_react_agent
11

12

13
# 定义一个工具函数
14
@tool
15
# 定义一个获取天气的函数，参数city只能是"nyc"或"sf"
16
def get_weather(city: Literal["nyc", "sf"]):
17
    # 函数的文档字符串，描述函数用途
18
    """Use this to get weather information."""
19
    # 如果城市是纽约
20
    if city == "nyc":
21
        # 返回纽约的天气信息
22
        return "It might be cloudy in nyc"
23
    # 如果城市是旧金山
24
    elif city == "sf":
25
        # 返回旧金山的天气信息
26
        return "It's always sunny in sf"
27
    # 如果城市不是nyc或sf
28
    else:
29
        # 抛出一个断言错误
30
        raise AssertionError("Unknown city")
31

32

33
# 将get_weather函数放入工具列表
34
tools = [get_weather]
35

36
# 创建一个OpenAI聊天模型实例，使用gpt-4o模型，温度设为0
37
model = ChatOpenAI(model_name="gpt-4o", temperature=0)
38
# 使用模型和工具创建一个react代理
39
graph = create_react_agent(model, tools)

流值#

1
# 定义输入消息，询问旧金山的天气
2
inputs = {"messages": [("human", "what's the weather in sf")]}
3
# 异步迭代代理的输出流，模式为"values"
4
async for chunk in graph.astream(inputs, stream_mode="values"):
5
    # 打印最后一条消息的格式化内容
6
    chunk["messages"][-1].pretty_print()

1
================================ Human Message =================================
2
what's the weather in sf
3
================================== Ai Message ==================================
4
Tool Calls:
5
get_weather (call_61VvIzqVGtyxcXi0z6knZkjZ)
6
Call ID: call_61VvIzqVGtyxcXi0z6knZkjZ
7
Args:
8
city: sf
9
================================= Tool Message =================================
10
Name: get_weather
11

12
It's always sunny in sf
13
================================== Ai Message ==================================
14

15
The weather in San Francisco is currently sunny.

如果我们只想获取最终结果，我们可以使用相同的方法，并跟踪我们收到的最后一个值

1
# 再次定义输入消息，询问旧金山的天气
2
inputs = {"messages": [("human", "what's the weather in sf")]}
3
# 异步迭代代理的输出流，模式为"values"
4
async for chunk in graph.astream(inputs, stream_mode="values"):
5
    # 将最后一个chunk赋值给final_result
6
    final_result = chunk

1
final_result

1
{'messages': [HumanMessage(content="what's the weather in sf", id='54b39b6f-054b-4306-980b-86905e48a6bc'),
2
              AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_avoKnK8reERzTUSxrN9cgFxY', 'function': {'arguments': '{"city":"sf"}', 'name': 'get_weather'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 57, 'total_tokens': 71}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_5e6c71d4a8', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-f2f43c89-2c96-45f4-975c-2d0f22d0d2d1-0', tool_calls=[{'name': 'get_weather', 'args': {'city': 'sf'}, 'id': 'call_avoKnK8reERzTUSxrN9cgFxY'}], usage_metadata={'input_tokens': 57, 'output_tokens': 14, 'total_tokens': 71}),
3
              ToolMessage(content="It's always sunny in sf", name='get_weather', id='fc18a798-c7b2-4f73-84fa-8ffdffb6ddcb', tool_call_id='call_avoKnK8reERzTUSxrN9cgFxY'),
4
              AIMessage(content='The weather in San Francisco is currently sunny. Enjoy the sunshine!', response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 84, 'total_tokens': 98}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_5e6c71d4a8', 'finish_reason': 'stop', 'logprobs': None}, id='run-21418147-da8e-4738-a076-239377397c40-0', usage_metadata={'input_tokens': 84, 'output_tokens': 14, 'total_tokens': 98})]}

1
# 打印final_result中最后一条消息的格式化内容
2
print(final_result["messages"][-1].pretty_print())

1
================================== Ai Message ==================================
2

3
The weather in San Francisco is currently sunny. Enjoy the sunshine!

流式状态更新#

LangGraph 支持多种流式模式。主要的有

values: 此流式模式会将图的值流式传输回来。这是每个节点被调用后图的完整状态。
updates: 此流式模式会将图的更新流式传输回来。这是每个节点被调用后图状态的更新。

本指南涵盖 stream_mode="updates"。

设置#

在本指南中，我们将使用一个简单的 ReAct 代理。

1
%pip install -U langgraph langchain-openai

1
import asyncio
2
from typing import Literal
3
from langchain_core.tools import tool
4
from langchain_openai import ChatOpenAI
5
from langgraph.prebuilt import create_react_agent
6

7

8
# 定义一个工具函数，用于获取天气信息
9
@tool
10
def get_weather(city: Literal["nyc", "sf"]):
11
    """Use this to get weather information."""
12
    if city == "nyc":
13
        return "It might be cloudy in nyc"
14
    elif city == "sf":
15
        return "It's always sunny in sf"
16
    else:
17
        raise AssertionError("Unknown city")
18

19

20
# 创建一个包含工具函数的列表
21
tools = [get_weather]
22

23
# 初始化一个OpenAI的聊天模型，使用gpt-4o模型，温度设为0（生成结果更确定）
24
model = ChatOpenAI(model_name="gpt-4o", temperature=0)
25

26

27
# 定义一个异步主函数
28
async def main():
29
    # 创建一个反应式代理，使用聊天模型和工具
30
    graph = create_react_agent(model, tools)
31
    # 定义输入消息，包含一个人类的消息询问sf的天气
32
    inputs = {"messages": [("human", "what's the weather in sf")]}
33
    # 异步迭代获取代理的更新流
34
    async for chunk in graph.astream(inputs, stream_mode="updates"):
35
        # 遍历每个更新块中的节点和值
36
        for node, values in chunk.items():
37
            # 打印接收到更新的节点名称
38
            print(f"Receiving update from node: '{node}'")
39
            # 打印节点的值
40
            print(values)
41
            print()
42

43
# 运行异步函数
44
asyncio.run(main())

流更新#

1
# 定义输入消息，包含一个人类的消息询问sf的天气
2
inputs = {"messages": [("human", "what's the weather in sf")]}
3
# 异步迭代获取代理的更新流
4
async for chunk in graph.astream(inputs, stream_mode="updates"):
5
    # 遍历每个更新块中的节点和值
6
    for node, values in chunk.items():
7
        # 打印接收到更新的节点名称
8
        print(f"Receiving update from node: '{node}'")
9
        # 打印节点的值
10
        print(values)
11
        print()

1
Receiving update from node: 'agent'
2
{'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_YXpVnZfYSkoN24AKqS2ntXoJ', 'function': {'arguments': '{"city":"sf"}', 'name': 'get_weather'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 57, 'total_tokens': 71}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_3aa7262c27', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-a03f2f71-174f-435e-9866-0b0d72c06868-0', tool_calls=[{'name': 'get_weather', 'args': {'city': 'sf'}, 'id': 'call_YXpVnZfYSkoN24AKqS2ntXoJ', 'type': 'tool_call'}], usage_metadata={'input_tokens': 57, 'output_tokens': 14, 'total_tokens': 71})]}
3

4
Receiving update from node: 'tools'
5
{'messages': [ToolMessage(content="It's always sunny in sf", name='get_weather', tool_call_id='call_YXpVnZfYSkoN24AKqS2ntXoJ')]}
6

7
Receiving update from node: 'agent'
8
{'messages': [AIMessage(content='The weather in San Francisco is currently sunny.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 10, 'prompt_tokens': 84, 'total_tokens': 94}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_3aa7262c27', 'finish_reason': 'stop', 'logprobs': None}, id='run-f56102b0-6de1-453c-96d0-b0be2d4cac32-0', usage_metadata={'input_tokens': 84, 'output_tokens': 10, 'total_tokens': 94})]}

LangGraph篇-流式处理

https://mizuki.mysqil.com/posts/langgraph篇-流式处理/

作者

SuifengFlying

发布于

2026-07-29

许可协议

CC BY-NC-SA 4.0

欢迎来到 SuifengFlying!

流式完整状态#

设置#

流值#

流式状态更新#

设置#

流更新#

Related articles