摘要在基于大模型构建的 Agent、工作流应用中流式输出是提升用户体验、降低延迟感知的核心技术。 本文基于 LangGraph 最新官方文档详细讲解 LangGraph Streaming v2 统一格式、7 种流模式、 状态流式、LLM 逐 Token 输出、自定义进度流、子图流式、异步兼容方案 附带大量可直接运行的代码示例适合 LLM 应用开发、Agent 工程化落地同学快速上手。一、前言在现在的 LLM 应用聊天机器人、智能 Agent、自动化工作流中流式输出Streaming已经不是加分项而是标配能力。它能做到不用等模型完整输出边生成边展示实时看到节点执行、状态变化、中间过程大幅提升用户体验降低延迟感知LangGraph 作为构建复杂 LLM 应用的主流框架提供了一套强大、统一、可扩展的流式系统。本文基于官方最新文档带你从零到一掌握v2 统一流式格式7 种 Stream Mode 完整用法状态流 / LLM 打字机流 / 自定义流子图流式、异步兼容、低版本 Python 避坑生产级最佳实践二、为什么一定要用 LangGraph Streaming降低用户等待焦虑LLM 响应慢流式输出直接解决体感问题执行过程透明化Agent 每一步在干嘛一目了然支持复杂交互进度条、状态提示、中间结果推送v2 格式彻底统一单模式、多模式、子图结构完全一致类型安全支持 IDE 自动提示减少 Bug三、核心v2 统一流式格式从LangGraph ≥ 1.1开始强烈推荐使用versionv2。所有流式返回结构统一为{type:values|updates|messages|custom|...,ns:(),# 子图命名空间data:...# 真实数据}v1 与 v2 对比v1单模式、多模式、子图返回结构不一样容易乱v2结构完全统一只用type区分代码更干净四、7 种 Stream ModeLangGraph 提供7 种流式模式覆盖几乎所有工程场景模式用途values每一步执行后的完整状态updates仅返回节点状态增量更新messagesLLM逐 Token 打字机输出custom自定义流进度、状态、事件推送checkpoints检查点持久化事件需记忆tasks任务开始/结束事件debug最全调试信息下面逐个实战。五、最常用流式模式实战1. updates状态增量流只返回节点修改的内容轻量、高效。forchunkingraph.stream({topic:ice cream},stream_modeupdates,versionv2):ifchunk[type]updates:fornode,stateinchunk[data].items():print(f节点{node}更新:{state})2. values完整状态流调试用每一步都返回全量状态。forchunkingraph.stream(...,stream_modevalues,versionv2):print(完整状态:,chunk[data])3. messagesLLM 逐字输出打字机效果聊天界面必备。forchunkingraph.stream(...,stream_modemessages,versionv2):msg,metachunk[data]print(msg.content,end,flushTrue)还能精准过滤按metadata[tags]过滤不同模型按metadata[langgraph_node]过滤指定节点六、自定义流 custom最强扩展能力你可以在任意节点、任意工具里主动推送进度条中间状态自定义日志第三方 API 流式结果使用步骤获取流写入器主动发送数据开启stream_modecustom示例fromlanggraph.configimportget_stream_writerdefgenerate_joke(state):writerget_stream_writer()writer({status:正在思考笑话...})return{joke:为什么冰淇淋去上学为了得到圣代学历}接收forchunkingraph.stream(...,stream_modecustom,versionv2):print(chunk[data][status])适用场景非 LangChain 模型接入前端进度条多步骤任务提示七、子图流式输出如果你用了子图Subgraph只需开启subgraphsTrue示例graph.stream(...,subgraphsTrue,stream_modeupdates,versionv2)通过chunk[ns]判断来源()主图(node_2:xxx,)子图八、同时开启多种流模式v2 格式支持多模式同时流式stream_mode[updates,custom]遍历判断type即可forchunkin...:ifchunk[type]updates:# 处理状态更新elifchunk[type]custom:# 处理自定义事件九、异步 Python ❤️.11 避坑重要Python 3.11 异步不支持上下文自动传递两个关键点异步调用 LLM 必须手动传 configasyncdefcall_model(state,config):awaitmodel.ainvoke(...,config)不能用get_stream_writer()改为参数注入fromlanggraph.typesimportStreamWriterasyncdefgenerate_joke(state,writer:StreamWriter):writer({msg:异步自定义流})场景推荐方案聊天机器人打字机 →messagesAgent 执行步骤展示 →updates进度条、中间提示 →custom复杂嵌套图 →subgraphsTrue调试、回溯、断点 →checkpoints/debug第三方模型流式 →custom封装
Langgraph基础(3)-流响应
摘要在基于大模型构建的 Agent、工作流应用中流式输出是提升用户体验、降低延迟感知的核心技术。 本文基于 LangGraph 最新官方文档详细讲解 LangGraph Streaming v2 统一格式、7 种流模式、 状态流式、LLM 逐 Token 输出、自定义进度流、子图流式、异步兼容方案 附带大量可直接运行的代码示例适合 LLM 应用开发、Agent 工程化落地同学快速上手。一、前言在现在的 LLM 应用聊天机器人、智能 Agent、自动化工作流中流式输出Streaming已经不是加分项而是标配能力。它能做到不用等模型完整输出边生成边展示实时看到节点执行、状态变化、中间过程大幅提升用户体验降低延迟感知LangGraph 作为构建复杂 LLM 应用的主流框架提供了一套强大、统一、可扩展的流式系统。本文基于官方最新文档带你从零到一掌握v2 统一流式格式7 种 Stream Mode 完整用法状态流 / LLM 打字机流 / 自定义流子图流式、异步兼容、低版本 Python 避坑生产级最佳实践二、为什么一定要用 LangGraph Streaming降低用户等待焦虑LLM 响应慢流式输出直接解决体感问题执行过程透明化Agent 每一步在干嘛一目了然支持复杂交互进度条、状态提示、中间结果推送v2 格式彻底统一单模式、多模式、子图结构完全一致类型安全支持 IDE 自动提示减少 Bug三、核心v2 统一流式格式从LangGraph ≥ 1.1开始强烈推荐使用versionv2。所有流式返回结构统一为{type:values|updates|messages|custom|...,ns:(),# 子图命名空间data:...# 真实数据}v1 与 v2 对比v1单模式、多模式、子图返回结构不一样容易乱v2结构完全统一只用type区分代码更干净四、7 种 Stream ModeLangGraph 提供7 种流式模式覆盖几乎所有工程场景模式用途values每一步执行后的完整状态updates仅返回节点状态增量更新messagesLLM逐 Token 打字机输出custom自定义流进度、状态、事件推送checkpoints检查点持久化事件需记忆tasks任务开始/结束事件debug最全调试信息下面逐个实战。五、最常用流式模式实战1. updates状态增量流只返回节点修改的内容轻量、高效。forchunkingraph.stream({topic:ice cream},stream_modeupdates,versionv2):ifchunk[type]updates:fornode,stateinchunk[data].items():print(f节点{node}更新:{state})2. values完整状态流调试用每一步都返回全量状态。forchunkingraph.stream(...,stream_modevalues,versionv2):print(完整状态:,chunk[data])3. messagesLLM 逐字输出打字机效果聊天界面必备。forchunkingraph.stream(...,stream_modemessages,versionv2):msg,metachunk[data]print(msg.content,end,flushTrue)还能精准过滤按metadata[tags]过滤不同模型按metadata[langgraph_node]过滤指定节点六、自定义流 custom最强扩展能力你可以在任意节点、任意工具里主动推送进度条中间状态自定义日志第三方 API 流式结果使用步骤获取流写入器主动发送数据开启stream_modecustom示例fromlanggraph.configimportget_stream_writerdefgenerate_joke(state):writerget_stream_writer()writer({status:正在思考笑话...})return{joke:为什么冰淇淋去上学为了得到圣代学历}接收forchunkingraph.stream(...,stream_modecustom,versionv2):print(chunk[data][status])适用场景非 LangChain 模型接入前端进度条多步骤任务提示七、子图流式输出如果你用了子图Subgraph只需开启subgraphsTrue示例graph.stream(...,subgraphsTrue,stream_modeupdates,versionv2)通过chunk[ns]判断来源()主图(node_2:xxx,)子图八、同时开启多种流模式v2 格式支持多模式同时流式stream_mode[updates,custom]遍历判断type即可forchunkin...:ifchunk[type]updates:# 处理状态更新elifchunk[type]custom:# 处理自定义事件九、异步 Python ❤️.11 避坑重要Python 3.11 异步不支持上下文自动传递两个关键点异步调用 LLM 必须手动传 configasyncdefcall_model(state,config):awaitmodel.ainvoke(...,config)不能用get_stream_writer()改为参数注入fromlanggraph.typesimportStreamWriterasyncdefgenerate_joke(state,writer:StreamWriter):writer({msg:异步自定义流})场景推荐方案聊天机器人打字机 →messagesAgent 执行步骤展示 →updates进度条、中间提示 →custom复杂嵌套图 →subgraphsTrue调试、回溯、断点 →checkpoints/debug第三方模型流式 →custom封装
