嗨老友，聊聊 LangGraph 的 Stream 流式输出到底是怎么玩儿的

说实话，我刚把这玩意儿玩了几遍，心里一阵小激动——哈哈，真的有点像打开了新世界的大门。

先别急，我先给你掰扯掰扯概念。LangGraph 本身是个状态机图，每个节点douKe以往下走、往上回、甚至跳到别的分支。

关键来了：stream_mode。这玩意儿决定了你在图跑的过程中到底想kan什么。

values每一步dou把完整的状态对象抛出来适合想要全貌的童鞋。

updates只送本轮产生的增量geng新，轻量又直观。

custom自定义数据流，你Ke以在节点内部随时塞进任何信息。

还有个调试专属的 "debug" 模式，想要追根溯源？它会把每一次调用的细节全给你打印出来。

'''StreamMultipleModes.py
LangGraph 多模式流式传输演示'''
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 定义状态类型
class AtguiguState:
    question: str
    answer: str
    confidence: float   # 置信度分数
    steps: list
def think -> AtguiguState:
    """思考节点"""
    question = state
    # 模拟思考过程
    steps = 
    return {"steps": steps}
def respond -> AtguiguState:
    """回应节点"""
    question = state
    # 根据问题生成答案
    if "天气" in question:
        answer = "今天天气晴朗"
        confidence = 0.92
    elif "时间" in question:
        answer = "现在是上午10点"
        confidence = 0.88
    else:
        answer = "这是一个hen好的问题"
        confidence = 0.75
    return {
        "answer": answer,
        "confidence": confidence
    }
def reflect -> AtguiguState:
    """反思节点"""
    answer = state
    confidence = state
    steps = state.get
    steps.append
    steps.append
    if confidence> 0.9:
        conclusion = "高置信度答案"
    elif confidence> 0.8:
        conclusion = "中等置信度答案"
    else:
        conclusion = "低置信度答案"
    steps.append
    return {"steps": steps}
def main:
    # 构建图
    builder = StateGraph
    builder.add_node
    builder.add_node
    builder.add_node
    builder.add_edge
    builder.add_edge
    builder.add_edge

builder.addedge

上面那段代码我故意留了点小坑——不对不对，应该是 .add_edge 才对。哈哈，这种错误在调试时Zui容易被忽视，记得多跑几遍。

下面给你演示一下各种模式下跑图的差别：

print
print
for chunk in graph.stream:
     print
print
print
for chunk in graph.stream:
     print
print
for mode, chunk in graph.stream:
     print
print
try:
     for chunk in graph.stream:
          print
except Exception as e:
     print

运行结果会像下面这样：

--- 使用 stream_mode='values' 模式 ---
{'question': '今天天气怎么样?', 'answer': '', 'confidence': 0, 'steps': }
{'question': '今天天气怎么样?', 'answer': '', 'confidence': 0,
 'steps': }
{'question': '今天天气怎么样?', 'answer': '今天天气晴朗',
 'confidence': 0.92,
 'steps': }
--- 使用 stream_mode='updates' 模式 ---
{'think': {'steps': }}
{'respond': {'answer':'今天天气晴朗','confidence':0.92}}
{'reflect': {'steps':}}
--- 同时使用  多种流模式 ---
: {...}
: {...}
...

有时候我们不想只kan状态本身，而是想实时推送进度条、日志之类的信息。LangGraph 给了 get_stream_writer 一招，让你在节点里随手写入自定义数据块。

from langgraph.config import get_stream_writer
def node_with_custom_streaming:
     writer = get_stream_writer
     writer
     writer
     # ...业务逻辑...
     writer
     writer
     result = f"处理结果:{state.upper}"
     return {"answer":result}

然后在调用时把 stream_mode="custom" 带进去，就Neng收到这些自定义块啦：

for chunk in graph.stream:
      print

这个嘛，其实跟内容质量和站点结构dou有关系。要确保页面有明确的标题和 meta 描述；要让关键字自然出现，而不是硬塞进去；再者，站点必须Neng被搜索引擎爬虫正常抓取——robots.txt 不该阻止重要路径；Zui后Ru果页面加载太慢、JS 渲染太重，也会导致抓取失败。

简单来说就是“内容好、结构清晰、访问顺畅”。你懂的，把这些Zuo好后百度基本就会乖乖收录啦。

L​LM 在图里也Neng像流水线一样逐 token 输出，这叫 "messages". 每个 token dou伴随元数据，让你知道它来自哪颗节点、用了哪个模型。

for chunk, meta_data in graph.stream:
      print}")
      print   # 实际上是 Token 对象或字符串片段
      # meta_data 包含 node_name、model_name 等信息，可用于日志追踪

Ru果你要Zuo聊天机器人 UI，这套方案简直完美——边生成边渲染，不会卡顿。

好了我先把这些干货甩给你。记住：

#1 : 根据需求选对 "values"/"updates"/"custom"/"debug".

#2 : 用 .stream) 时一定要配好输入状态，否则跑不起来。

#3 : 想让前端实时感知进度？别忘了写 get_stream_writer.

#4 : 调试卡住？直接切到 "debug", 把所有调用栈dou抖出来kankan。

#5 : SEO 要兼顾可读性和技术深度，适当加入中文关键词和自然描述，就算不被百度立刻收录，也会慢慢爬上去。

Aha~ 那么你现在Ke以大胆去实验啦！遇到报错或者奇怪行为，记得先检查一下边界条件——比如有没有忘记返回完整的字典结构。呵呵，不对，我说的是“完整字典”。 多跑几遍，你会发现 Stream 就像是一条不断冒泡的河流，信息不停从节点冲出来让你的应用geng“活泼”。祝玩得开心，咱们下次再聊别的话题吧！👋