摘要：本文围绕 LangGraph 中多模式流式输出的融合使用展开，重点说明如何通过给 stream_mode 传入多个输出类型，同时获取 messages、updates 和 values 三类数据。文章还介绍了当大模型本身不支持流式输出时，如何关闭流式功能，并通过状态控制把 stream 调用改为 invoke 调用，最后结合一段完整的 Python 示例演示写作与评分节点的混合流式处理流程。

内容参考于：图灵AI大模型全栈

融合多模式就是下图红框的代码，通过给stream_mode传递多个输出类型

如果一个大模型它本身并不支持流式输出，那么就需要写下图红框的代码来关闭流式输出功能，然后最好通过状态，让代码中调用流式输出的stream函数改成调用invoke函数

代码

from typing import TypedDict
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
import os

load_dotenv()

# ==========================================
# 1. 定义状态和工具
# ==========================================

class EditorState(TypedDict):
    topic: str  # 主题
    content: str  # 生成的内容
    score: int  # 评分
    status: str  # 当前状态描述


llm = init_chat_model(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model_provider="openai",
    model='qwen3.8-flash',
    # 如果模型不支持流式输出，那么必须要写下方的代码，
    # 显示关闭流失输出
    disable_streaming=True
)


# ==========================================
# 2. 定义节点逻辑
# ==========================================

def write_article(state: EditorState):
    """节点1：负责写文章（耗时操作，会有流式输出）"""
    topic = state["topic"]
    # 这里我们用 invoke，依靠 stream_mode="messages" 来捕获流
    response = llm.invoke(f"请写一段关于'{topic}'的短文，50字左右。")
    return {
        "content": response.content,
        "status": "写作完成"
    }


def review_article(state: EditorState):
    """节点2：负责打分（逻辑操作，瞬间完成）"""
    # 简单模拟打分逻辑
    content_len = len(state["content"])
    score = min(100, content_len * 2)
    return {
        "score": score,
        "status": "评分完成"
    }


# ==========================================
# 3. 构建图
# ==========================================

workflow = StateGraph(EditorState)

workflow.add_node("writer", write_article)
workflow.add_node("reviewer", review_article)

workflow.add_edge(START, "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_edge("reviewer", END)

app = workflow.compile()


# ==========================================
# 4. 核心：融合流式输出处理
# ==========================================

def run_mixed_mode_demo():
    inputs = {
        "topic": "人工智能的未来",
        "content": "",
        "score": 0,
        "status": "开始任务"
    }

    print(f"任务启动：主题 - {inputs['topic']}\n")
    print("-" * 50)

    # 关键点：传入一个列表 ["messages", "updates", "values"]
    # 这样 app.stream 会返回一个元组：(mode, chunk)
    for mode, chunk in app.stream(inputs, stream_mode=["messages", "updates", "values"]):

        # --- 模式 A: Messages (处理打字机效果) ---
        if mode == "messages":
            # chunk 结构是 (message, metadata)
            message, metadata = chunk
            # 只显示 AI 生成的内容，过滤掉系统消息等
            if message.content:
                print(message.content, end="", flush=True)

        # --- 模式 B: Updates (处理节点完成通知) ---
        elif mode == "updates":
            # chunk 是该节点刚刚更新的字段
            # 这里的 chunk 类似于Key-Value：{'writer': {'content': '...', 'status': '...'}}
            node_name = list(chunk.keys())[0]
            updates = chunk[node_name]
            print(f"\n\n[节点完成] {node_name} -> 状态: {updates.get('status')}")
            if "score" in updates:
                print(f"[评分结果] 得分: {updates['score']}")
            print("-" * 50)  # 分割线

        # --- 模式 C: Values (处理全局状态快照) ---
        elif mode == "values":
            # chunk 是当前的完整 State
            print(f"\n📦 [全量状态快照] {chunk}")

    print("\n流程结束！")


if __name__ == "__main__":
    run_mixed_mode_demo()