LangGraph 流式输出 Streaming 实战:四种 stream_mode 怎么选
点外卖时,你最焦虑的不是等 30 分钟,而是不知道外卖到哪了。所以平台会显示商家接单 → 骑手到店 → 已取餐 → 距你 500 米。等待时间没变,但焦虑没了——因为你看到了过程。前六集写的图都是黑盒执行graph.invoke(input),等几秒甚至几十秒,然后蹦出一个最终结果。这几十秒里用户完全不知道发生了什么,只能盯着一个转圈的 loading。graph.stream(...)不改变图做什么,只是把中间过程一条条交给你节点什么时候跑、LLM 吐了哪个 token、进度到哪了。这一集就是学会按需求挑看哪种过程。先看全景六种 stream_mode 推送的是完全不同的东西stream()用一个stream_mode参数决定推送什么模式何时推送推送什么裸数据直接拿典型用途values每个超级步结束完整 State一个 dict状态面板、节点高亮updates每个超级步结束{节点名: 该节点的返回}省带宽的增量更新messagesLLM 每个 token(消息块, metadata)元组打字机效果custom节点里调writer()你推的任意数据进度条、业务通知tasks每个节点起 / 止任务 dict名字、耗时、error执行监控、错误捕获一句话先记住稳定版stream(stream_mode...)直接 yield 裸数据。values 就是 State,updates 就是{节点:输出},custom 就是你推的东西——不要去chunk[data]里找,那是另一套实验 API 的信封,这里用不到。values vs updates全量快照还是增量变化同一张最小演示图——两个并行节点(规划大纲、收集资料)汇总生成内容,分别用两种模式跑一遍。values每步推一次完整 Statefor state in writer_graph.stream(demo_input, stream_modevalues): filled {k: v for k, v in state.items() if v} print(f已填字段: {list(filled)}) plaintext 第 0 帧已填字段: [topic]第 1 帧已填字段: [topic, outline, sources, log]第 2 帧已填字段: [topic, outline, sources, content, log]updates只推这一步哪个节点、返回了什么增量for update in writer_graph.stream(demo_input, stream_modeupdates): for node, payload in update.items(): print(f节点 {node} 更新了: {list(payload)}) plaintext 节点 gather_sources 更新了: [sources]节点 plan_outline 更新了: [outline]节点 generate_content 更新了: [content]区别很直接values每帧都是整盘 State,State 一大(比如消息很长)带宽就不省;updates只给变化,前端拿到后自己 merge。做状态面板、节点高亮用values;只关心哪个节点刚完成、返回了啥用updates。messages打字机效果也是最容易踩坑的一种这是打字机效果的核心LLM 每生成一个 token,立刻推给你。stream(stream_modemessages)yield 的是(消息块, metadata)元组——直接解包,不是chunk[data]。full_text for token, metadata in chat_graph.stream( {messages: [HumanMessage(用三句话介绍 LangGraph 的流式输出有什么用。)]}, stream_modemessages,): print(token.content, end, flushTrue) full_text token.content # 关键追加不是替换真实跑一次,逐字蹦出1. 流式输出允许在复杂工作流执行过程中实时返回中间结果显著提升用户交互体验和响应速度。2. 它能逐步展示模型生成或节点计算的进展便于开发者动态监控、调试和优化多步骤流程。3. 通过早期输出可在满足条件时提前终止后续计算从而节省资源并支持灵活的决策逻辑。这些 token 来自节点: writer第 1 步两个必须记住的点token 是增量,前端要追加不是替换full_text token.content。写成full_text token.content就只剩最后一个字——这是最常见的 bug,因为语法完全合法,不会报错,只会安静地丢数据。只有走 LLM 的输出才有 token节点里直接return 字符串(不经过 ChatModel)不会触发messages事件。custom节点里主动推任意进度前面几种都是 LangGraph 自动推的。custom是旁路在节点里拿到writer,想推什么推什么——进度百分比、“正在检索…”、中间结果。from langgraph.config import get_stream_writerdef search_with_progress(state: SearchState) - dict: writer get_stream_writer() for i, source in enumerate(sources, 1): writer({stage: 搜索, source: source, percent: i / len(sources) * 100}) return {progress: 100} plaintext stream_modecustom —— event 就是你推的 dict 本身[搜索] 财经数据 25%[搜索] 新闻 50%[搜索] 竞品 75%[搜索] 专家观点 100%writer必须在节点函数内部调用,而且只有stream_mode含custom时,这些推送才会流出来。tasks不用你手动推的节点生命周期不用手动埋点,LangGraph 在每个节点 start 和 finish 各推一次。区分开始/结束就看有没有result键,两次的时间差就是节点耗时,error字段在失败时自动带上——天然适合做监控和错误捕获▶ 开始 gather_sources▶ 开始 plan_outline✓ 完成 gather_sources (耗时 0.9 ms)✓ 完成 plan_outline (耗时 1.2 ms)▶ 开始 generate_content✓ 完成 generate_content (耗时 0.4 ms)组合模式和子图嵌套两个进阶开关真实 UI 常常既要打字机(messages)又要进度条(custom)。把stream_mode传一个列表就行,代价是每条事件变成(mode, chunk)元组,要先解包再按 mode 分流for mode, chunk in combo_graph.stream({done: False}, stream_mode[custom, tasks]): if mode custom: print(f [custom] 进度 {chunk[percent]}%) elif mode tasks: phase 完成 if result in chunk else 开始 print(f [tasks] {chunk[name]} {phase})第 5、6 集用过create_supervisor或子图当节点,图是嵌套的。默认stream()只给顶层事件,子图内部是黑盒。加subgraphsTrue,子图事件也会流出来——代价是每条事件前面多一截 namespace,变成(ns, chunk)元组for ns, chunk in parent_graph.stream({x: 1}, stream_modeupdates, subgraphsTrue): where 顶层图 if ns () else 子图 /.join(n.split(:)[0] for n in ns) print(f [{where}] ns{ns} {chunk}) plaintext 默认子图内部是黑盒只看到顶层 {parent_step: {x: 10}} {child: {x: 11}}subgraphsTrue每条变 (ns, chunk)子图事件也出来了 [顶层图] ns() {parent_step: {x: 10}} [子图 child] ns(child:4891c339-...,) {child_step: {x: 11}} [顶层图] ns() {child: {x: 11}}ns就是前端决定这条放主面板还是放子图详情面板的依据。工程化判断什么该流什么不该流流式不是越多越好。一个简单原则就能定大方向人读的东西流式机器读的东西非流式。LLM 生成的自然语言是给人看的,逐字蹦能消除等待焦虑;工具调用参数、错误码、审批请求是给程序 / 决策用的,片段没有意义,要一次性完整给出。输出流式为什么LLM 自然语言内容✅ 流消除等待焦虑提升体感速度工具调用参数 / 结果✗ 不流结构化数据碎片无意义错误 / 中断审批✗ 不流要立即、完整呈现供用户决策进度通知✗ 不流一次推送状态即可用custom学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】