当前位置：首页 > news >正文

LangGraph 深度解析：stream\_mode messages 与 values 核心区别（含HITL适配与代码实战）

news 2026/7/22 17:56:36

一、前言

在基于 LangGraph 构建 AI Agent、工具调用、人在回路（HITL）交互式应用时，流式输出（stream）是最常用的能力，主要用于实现模型实时打字效果、工具调用状态反馈、人工介入中断等交互场景。

LangGraph 提供多种流式输出模式，其中 stream_mode="messages" 和 stream_mode="values" 是开发中最常用的两种模式。多数开发者会出现认知误区：认为 values 模式是一次性输出、messages 模式是纯流式输出，或是认为两种模式的图执行逻辑存在差异。

本文将从底层原理、执行流程、数据输出规则、HITL 中断适配、代码实战对比五个维度，详细拆解两种模式的核心差异，解决 Agent 开发中流式展示与人工中断冲突的核心问题。

二、核心前置概念铺垫

2.1 HITL 人在回路机制

HITL（Human-in-the-loop，人在回路）是 LangGraph 提供的人工介入能力，核心作用是在 Agent 调用工具的关键节点暂停流程，等待人工确认、修改指令后，再继续执行任务。

关键底层认知：HITL 中断（interrupt）不是大模型的原生能力，也不是模型生成的消息内容。它是 LangGraph 框架底层通过 wrap_tool_call 拦截机制，在工具节点执行前后主动触发的引擎级暂停行为，属于图执行引擎的控制信号，而非模型输出的消息数据。

2.2 Stream 流式输出的本质

LangGraph 的流式输出分为两个完全独立的层级，这是区分两种模式的核心关键：

内部执行引擎层：负责模型调用、节点运行、路由跳转、状态更新、HITL 中断暂停，stream_mode 不会改变这一层的任何执行逻辑，两种模式下 Agent 的运行流程完全一致。
对外数据输出层：stream_mode 仅作为数据过滤器，决定图每一步执行完成后，向外暴露什么数据给开发者/前端。

2.3 两种模式基础定义

stream_mode="messages"：仅过滤并输出大模型、节点产生的消息对象（AIMessage、ToolCallMessage 等），只承载对话、工具调用类业务数据。
stream_mode="values"：输出图每一个节点执行完成后的完整全局状态快照（State），包含所有对话消息、状态参数、运行上下文，可关联图的执行状态。

三、两种模式核心底层差异（重点）

3.1 数据输出来源差异

3.1.1 messages 模式

数据源仅为模型/节点产出的消息片段。LLM 流式生成的每一个 token、每一段增量消息，都会被实时透传输出。该模式完全屏蔽 LangGraph 引擎的运行控制信号，只专注于对话内容输出。

由于 HITL 的 interrupt 中断是引擎控制信号，不属于消息对象，因此 messages 模式的数据流中完全不存在中断信息。流式循环结束后，无法直接区分流程是「正常执行完毕」还是「被 HITL 中断暂停」。

3.1.2 values 模式

数据源为节点执行完成后的完整 State 快照。LangGraph 的节点是原子执行单元，必须等待单个节点全部执行完毕、状态更新完成后，才会向外输出一次完整状态数据。

该模式不会暴露节点内部 LLM 逐 token 的中间生成过程，因此视觉上呈现「一次性输出完整内容」的效果，但本质是模型依旧流式生成，只是框架过滤了中间增量片段。核心优势是可以通过全局状态快照，配合 get_state() 方法捕获引擎层的 HITL 中断标记。

3.2 流式渲染能力差异

messages 模式：支持原生逐字流式渲染，可直接实现前端打字效果，无需额外处理，适合纯对话展示场景。
values 模式：原生不支持逐字输出，仅输出节点执行完成后的完整结果。如需流式渲染，需要手动对比前后状态的消息增量，自行封装流式逻辑。

3.3 HITL 中断适配能力差异

messages 模式致命缺陷：流式循环终止后，无任何状态标识区分正常结束和中断暂停。无法在流式执行过程中感知 HITL 触发，仅能事后手动查询状态，不适合需要实时弹窗人工确认的交互场景。
values 模式核心优势：每次输出完整状态快照，可全程监控 Agent 运行进度。流终止后，通过图状态快照的 __interrupt__ 属性，可精准判断是否触发人工中断，完美适配 HITL 人机交互场景。

3.4 核心差异汇总表

对比维度	stream_mode="messages"	stream_mode="values"
内部图执行逻辑	完整执行路由、中断、状态更新（无差异）	与 messages 模式完全一致
输出数据内容	仅模型/节点生成的消息片段	节点执行完成后的完整全局状态
逐字流式渲染	原生支持，开箱即用	原生不支持，需手动封装增量逻辑
HITL 中断捕获	无法实时捕获，无法区分结束状态	可精准捕获，适配人工介入场景
适用场景	纯对话流式展示、无人工中断需求	工具调用、HITL 人机交互、流程状态监控

四、完整代码实战与逐行解析

本节通过可运行代码，直观对比两种模式的输出差异、HITL 中断捕获效果，所有代码基于 LangGraph 最新稳定版本，可直接复制运行。

4.1 环境依赖安装

pip install langgraph langchain-openai python-dotenv

4.2 完整实战代码

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition# 加载环境变量（配置模型API密钥）
load_dotenv()# 1. 定义工具（模拟需要人工审核的工具调用场景）
def get_weather(city: str) -> str:"""查询指定城市天气信息"""return f"{city} 当前气温26℃，天气晴朗，无降水"# 注册工具列表
tools = [get_weather]# 初始化大模型并绑定工具，开启原生流式能力
llm = ChatOpenAI(model="gpt-4o", api_key=os.getenv("OPENAI_API_KEY"), streaming=True)
llm_with_tools = llm.bind_tools(tools)# 2. 定义Agent核心节点
def agent(state: MessagesState):"""Agent思考节点：调用模型生成回复/工具调用指令"""resp = llm_with_tools.invoke(state["messages"])return {"messages": [resp]}# 初始化工具执行节点
tool_node = ToolNode(tools)# 3. 构建LangGraph工作流
graph_builder = StateGraph(MessagesState)
# 添加核心节点
graph_builder.add_node("agent", agent)
graph_builder.add_node("tools", tool_node)
# 添加条件路由：Agent需要调用工具时，跳转至工具节点
graph_builder.add_conditional_edges("agent", tools_condition)
# 工具执行完成后，重新回到Agent节点
graph_builder.add_edge("tools", "agent")
# 设置图入口
graph_builder.set_entry_point("agent")# 核心配置：在工具节点执行前触发HITL中断，等待人工确认
graph = graph_builder.compile(interrupt_before=["tools"])if __name__ == "__main__":# 初始化用户请求user_input = {"messages": [("user", "帮我查询北京的实时天气")]}# ========== 测试1：stream_mode="messages" 模式 ==========print("===== 【messages模式】流式输出结果 =====")print("特点：仅输出消息片段，无法捕获HITL中断\n")stream_messages = graph.stream(user_input, stream_mode="messages")for chunk in stream_messages:# 打印流式消息片段print(f"消息片段：{chunk}")# 流结束后查询状态，无法实时感知中断snapshot_msg = graph.get_state()print(f"\nmessages模式-是否触发中断：{snapshot_msg.__interrupt__ if snapshot_msg.__interrupt__ else '否'}")print("-" * 80)# ========== 测试2：stream_mode="values" 模式 ==========print("===== 【values模式】流式输出结果 =====")print("特点：输出完整状态快照，可精准捕获HITL中断\n")stream_values = graph.stream(user_input, stream_mode="values")for state_snapshot in stream_values:# 打印每一步的完整状态消息latest_msg = state_snapshot["messages"][-1]print(f"最新消息内容：{latest_msg.content if latest_msg.content else '无文本内容（工具调用）'}")# 流结束后捕获中断信息snapshot_val = graph.get_state()print(f"\nvalues模式-是否触发中断：{snapshot_val.__interrupt__ if snapshot_val.__interrupt__ else '否'}")print(f"中断详情：{snapshot_val.__interrupt__}")

4.3 代码核心逻辑解析

HITL 中断配置：interrupt_before=["tools"] 表示在执行工具节点前强制暂停流程，触发人工中断，该行为由 LangGraph 引擎底层完成，与模型无关。
模型流式配置：模型开启 streaming=True，保证模型本身是逐 token 生成，排除模型输出方式的干扰。
messages 模式执行逻辑：仅透传模型生成的工具调用消息，流式循环结束后无任何中断提示，只能事后查询状态，无法实时交互。
values 模式执行逻辑：输出每一步完整状态，流程暂停后可通过 __interrupt__ 属性精准获取中断原因、待执行工具信息，支持前端实时弹窗确认。