3.2 流式控制与实时反馈

本节导读：通过本节学习，你将掌握LangGraph的流式控制技术，实现智能体的实时输出和状态监控，提升用户体验和调试效率。

学习目标

• 理解流式输出的核心概念和优势
• 掌握不同流式模式的使用方法
• 学会实现实时反馈和进度跟踪
• 了解流式输出的性能优化策略
• 构建支持流式处理的智能体系统

核心概念

LangGraph的流式控制技术是区别于其他AI框架的关键特性之一。它允许智能体在执行过程中实时输出中间结果，而不是等待整个流程完成后再返回最终结果。

流式模式类型

1. values模式（流式状态）

流式输出所有状态变化，适合监控完整的执行过程：

# values模式：流式输出所有状态
for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "分析数据"}]}, 
    stream_mode="values"
):
    print("当前状态:", chunk)
​

2. messages模式（流式消息）

专门流式输出消息内容：

# messages模式：流式输出消息
for chunk in graph.stream(
    {"messages": user_input}, 
    stream_mode="messages"
):
    print(chunk.content, end="", flush=True)
​

3. custom模式

自定义流式逻辑，适合复杂场景：

# custom模式：自定义流式处理
for chunk in graph.stream(
    input_state, 
    stream_mode="custom",
    config={"recursion_limit": 10}
):
    process_custom_chunk(chunk)
​

环境准备 / 前置知识

依赖安装

# 基础流式支持
pip install -U langgraph langchain-openai

# 高级流式功能
pip install -U aiofiles
​

前置要求

• 掌握基本的StateGraph概念
• 了解异步编程基础
• 熟悉消息类型和状态管理

分步实战

步骤 1：基础流式输出实现

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, Sequence
import time

class StreamingState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    current_step: str
    progress: float

def analysis_with_progress(state: StreamingState):
    """带进度输出的分析节点"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    user_query = state["messages"][-1]["content"]
    
    # 模拟逐步分析过程
    steps = [
        "开始分析数据...",
        "识别关键模式...",
        "生成洞察报告...",
        "完成分析"
    ]
    
    messages = []
    for i, step in enumerate(steps):
        print(f"进度 {i+1}/{len(steps)}: {step}")
        
        # 更新进度状态
        progress = (i + 1) / len(steps)
        state["progress"] = progress
        
        response = llm.invoke([
            {"role": "system", "content": "你是数据分析专家，逐步报告你的分析过程"},
            {"role": "user", "content": f"{step}: {user_query}"}
        ])
        messages.append(response)
        
        # 模拟处理时间
        time.sleep(1)
    
    return {
        "messages": messages,
        "current_step": "analysis_complete",
        "progress": 1.0
    }

# 构建流式图
graph = StateGraph(StreamingState)
graph.add_node("analysis", analysis_with_progress)
graph.add_edge(START, "analysis")
graph.add_edge("analysis", END)

checkpointer = MemorySaver()
compiled_graph = graph.compile(checkpointer=checkpointer)

# 使用values模式流式执行
print("=== 使用values模式 ===")
for chunk in compiled_graph.stream(
    {"messages": [{"role": "user", "content": "分析销售趋势"}], 
     "current_step": "start",
     "progress": 0.0},
    stream_mode="values"
):
    print(f"当前状态: 进度 {chunk.get('progress', 0):.1%}")

print("\n=== 使用messages模式 ===")
for chunk in compiled_graph.stream(
    {"messages": [{"role": "user", "content": "分析销售趋势"}], 
     "current_step": "start",
     "progress": 0.0},
    stream_mode="messages"
):
    if hasattr(chunk, 'content'):
        print(chunk.content, end="", flush=True)
​

步骤 2：实时聊天智能体

class ChatState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    conversation_id: str
    response_buffer: str

def streaming_chatbot(state: ChatState):
    """流式聊天机器人"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    user_message = state["messages"][-1]["content"]
    
    # 构建对话上下文
    conversation = []
    for msg in state["messages"]:
        conversation.append(f"{msg['role']}: {msg['content']}")
    
    prompt = f"""
    对话历史：
    {'\n'.join(conversation)}
    
    用户最新消息：{user_message}
    
    请提供实时回复：
    """
    
    response = llm.invoke([
        {"role": "system", "content": "你是一个友好的聊天助手，提供详细且有用的回复"},
        {"role": "user", "content": prompt}
    ])
    
    return {
        "messages": [response],
        "conversation_id": state["conversation_id"],
        "response_buffer": response.content
    }

# 构建聊天流式图
chat_graph = StateGraph(ChatState)
chat_graph.add_node("chatbot", streaming_chatbot)
chat_graph.add_edge(START, "chatbot")
chat_graph.add_edge("chatbot", END)

chat_checkpointer = MemorySaver()
compiled_chat_graph = chat_graph.compile(checkpointer=chat_checkpointer)

# 模拟实时聊天体验
print("=== 实时聊天体验 ===")
conversation_id = f"chat_{int(time.time())}"

for chunk in compiled_chat_graph.stream(
    {"messages": [{"role": "user", "content": "你能帮我分析一下市场趋势吗？"}],
     "conversation_id": conversation_id,
     "response_buffer": ""},
    stream_mode="messages"
):
    if hasattr(chunk, 'content'):
        print(chunk.content, end="", flush=True)
        
        # 模拟打字效果
        import time
        time.sleep(0.05)
​

步骤 3：多模态流式处理

from typing import Any

class MultiModalState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    processing_stage: str
    intermediate_results: dict
    final_output: Any

def text_analysis_node(state: MultiModalState):
    """文本分析节点"""
    print("开始文本分析...")
    
    # 模拟文本处理
    text = state["messages"][-1]["content"]
    
    # 逐步处理文本
    steps = [
        ("文本预处理", "完成文本清理和标准化"),
        ("关键词提取", "识别关键概念和术语"),
        ("语义分析", "理解文本含义和关系"),
        ("情感分析", "判断文本的情感倾向")
    ]
    
    results = {}
    for step_name, step_desc in steps:
        print(f"正在处理: {step_name}")
        time.sleep(0.5)
        results[step_name] = {
            "status": "completed",
            "description": step_desc,
            "timestamp": time.time()
        }
    
    return {
        "messages": [{"role": "assistant", "content": "文本分析完成"}],
        "processing_stage": "text_analysis_complete",
        "intermediate_results": results,
        "final_output": None
    }

def image_analysis_node(state: MultiModalState):
    """图像分析节点"""
    print("开始图像分析...")
    
    # 模拟图像处理
    steps = [
        ("图像加载", "成功读取图像文件"),
        ("预处理", "调整尺寸和格式"),
        ("特征提取", "识别图像中的对象"),
        ("场景理解", "分析图像内容和上下文")
    ]
    
    results = {}
    for step_name, step_desc in steps:
        print(f"正在处理: {step_name}")
        time.sleep(0.8)
        results[step_name] = {
            "status": "completed",
            "description": step_desc,
            "progress": len(results) / len(steps)
        }
    
    return {
        "messages": [{"role": "assistant", "content": "图像分析完成"}],
        "processing_stage": "image_analysis_complete",
        "intermediate_results": results,
        "final_output": None
    }

def fusion_node(state: MultiModalState):
    """多模态融合节点"""
    print("开始多模态融合...")
    
    text_results = state["intermediate_results"].get("文本分析", {})
    image_results = state["intermediate_results"].get("图像分析", {})
    
    # 融合分析结果
    final_analysis = {
        "综合理解": "基于文本和图像的完整内容理解",
        "关键发现": "识别出的重要信息点",
        "建议": "基于分析结果的行动建议"
    }
    
    for key, value in final_analysis.items():
        print(f"生成: {key}")
        time.sleep(0.3)
    
    return {
        "messages": [{"role": "assistant", "content": str(final_analysis)}],
        "processing_stage": "fusion_complete",
        "intermediate_results": state["intermediate_results"],
        "final_output": final_analysis
    }

# 构建多模态流式图
multimodal_graph = StateGraph(MultiModalState)
multimodal_graph.add_node("text_analysis", text_analysis_node)
multimodal_graph.add_node("image_analysis", image_analysis_node)
multimodal_graph.add_node("fusion", fusion_node)

multimodal_graph.add_edge(START, "text_analysis")
multimodal_graph.add_edge("text_analysis", "image_analysis")
multimodal_graph.add_edge("image_analysis", "fusion")
multimodal_graph.add_edge("fusion", END)

multimodal_checkpointer = MemorySaver()
compiled_multimodal_graph = multimodal_graph.compile(checkpointer=multimodal_checkpointer)

# 执行多模态流式处理
print("=== 多模态流式处理 ===")
for chunk in compiled_multimodal_graph.stream(
    {"messages": [{"role": "user", "content": "分析产品宣传材料和用户评论"}],
     "processing_stage": "start",
     "intermediate_results": {},
     "final_output": None},
    stream_mode="values"
):
    print(f"当前状态: {chunk['processing_stage']}")
    
    # 显示中间结果
    if chunk.get("intermediate_results"):
        print(f"已完成步骤: {list(chunk['intermediate_results'].keys())}")
​

常见问题 FAQ

Q1：流式输出与批量输出的区别是什么？

A：流式输出与批量输出有显著区别：

• 批量输出：等待整个流程完成后返回结果，适合确定性的短任务
• 流式输出：实时输出中间结果，适合长时间运行的任务和需要用户反馈的场景

流式输出更适合：

• 长时间运行的分析任务
• 需要实时反馈的用户交互
• 复杂的多步骤处理流程
• 调试和监控执行过程

Q2：如何处理流式输出中的错误？

A：流式输出中的错误处理需要特别注意：

1. 错误捕获：在流式处理过程中捕获异常
2. 错误恢复：提供错误恢复机制
3. 错误通知：及时通知用户或系统管理员
4. 日志记录：记录错误信息用于后续分析

def safe_streaming_handler(chunk):
    try:
        process_chunk(chunk)
    except Exception as e:
        print(f"处理流式数据时发生错误: {e}")
        # 记录错误
        log_error(e)
        # 可以选择跳过此chunk或中断流
        continue_streaming = decide_whether_to_continue(e)
        return continue_streaming
    return True
​

Q3：如何优化流式性能？

A：流式性能优化包括：

1. 缓冲管理：合理设置缓冲区大小
2. 并发处理：使用异步处理提高并发性能
3. 资源控制：限制并发数量防止资源耗尽
4. 压缩传输：对大数据进行压缩减少传输开销

本节小结

通过本节学习，我们掌握了LangGraph的流式控制技术，包括values模式、messages模式和custom模式的使用，以及如何实现实时反馈和多模态流式处理。流式控制技术使得智能体能够提供更好的用户体验和调试体验。

下一节我们将探讨LangGraph的性能优化与并发控制技术。

延伸阅读

• LangGraph官方文档-流式输出
• 异步编程最佳实践

关键词：流式控制, 实时反馈, values模式, messages模式, 多模态处理
难度：进阶
预计阅读：30分钟