【LangChain】流式处理实战：实时响应优化

【LangChain】流式处理实战：实时响应优化

开篇：为什么你的Agent需要"实时感"？

想象这样一个场景：你走进一家餐厅，点完菜后服务员就消失了。你坐在那儿干等15分钟，完全不知道厨房是在备菜、炒菜，还是已经把你的订单搞丢了。这种信息黑洞让人焦虑，对吧？

这就是大多数AI Agent给用户的感觉——输入问题后，屏幕一片空白，只有一个小转圈在暗示"我还在活着"。

流式处理（Streaming）就是解决这个问题的银弹。它让用户实时看到Agent的"思考过程"：模型正在生成什么内容、调用了什么工具、进度如何。就像餐厅里的开放式厨房，你能看到厨师切菜、炒菜的全过程，等待也变得可以忍受了。

LangChain的新架构提供了三种核心流式模式，配合Token级输出和自定义进度反馈，能让你的Agent体验从"黑盒等待"跃升到"透明协作"。

一、三种stream_mode：选择你的"信息粒度"

LangChain的流式系统通过stream_mode参数控制信息输出的粒度。理解这三种模式的区别，是掌握流式处理的第一步。

1.updates模式：Agent步骤追踪

这是最适合监控Agent执行流程的模式。它会在每个Agent步骤完成后，推送状态更新。

fromlangchain.agentsimportcreate_agentdefget_weather(city:str)->str:returnf"It's always sunny in{city}!"agent=create_agent(model="gpt-5-nano",tools=[get_weather],)# 使用 updates 模式流式追踪forchunkinagent.stream({"messages":[{"role":"user","content":"What is the weather in SF?"}]},stream_mode="updates",version="v2",# 推荐使用v2统一格式):ifchunk["type"]=="updates":forstep,datainchunk["data"].items():print(f"🔄 步骤:{step}")print(f"📦 内容:{data['messages'][-1].content_blocks}")

输出示例：

🔄 步骤: model 📦 内容: [{'type': 'tool_call', 'name': 'get_weather', 'args': {'city': 'San Francisco'}}] 🔄 步骤: tools 📦 内容: [{'type': 'text', 'text': "It's always sunny in San Francisco!"}] 🔄 步骤: model 📦 内容: [{'type': 'text', 'text': 'The weather in San Francisco is sunny!'}]

适用场景：

• ✅ 需要展示"Agent正在做什么"的进度条
• ✅ 调试Agent执行流程
• ✅ 记录每个步骤的输入输出用于审计

2.messages模式：Token级实时输出

这是最接近ChatGPT用户体验的模式。它流式返回LLM生成的每个Token，配合元数据让你知道消息来自哪个节点。

forchunkinagent.stream({"messages":[{"role":"user","content":"What is the weather in SF?"}]},stream_mode="messages",version="v2",):ifchunk["type"]=="messages":token,metadata=chunk["data"]# 实时打印每个tokeniftoken.content:print(token.content,end="",flush=True)# 追踪工具调用片段iftoken.tool_call_chunks:print(f"\n🔧 工具调用:{token.tool_call_chunks}")

关键特性：

• 返回的是(token, metadata)元组
• metadata['langgraph_node']告诉你消息来自哪个节点
• token.tool_call_chunks展示工具调用的增量JSON构建过程

输出示例（工具调用片段）：

🔧 工具调用: [{'name': 'get_weather', 'args': '', 'id': 'call_xxx'}] 🔧 工具调用: [{'name': None, 'args': '{"', 'id': None}] 🔧 工具调用: [{'name': None, 'args': 'city', 'id': None}] 🔧 工具调用: [{'name': None, 'args': '":"', 'id': None}] 🔧 工具调用: [{'name': None, 'args': 'San', 'id': None}] 🔧 工具调用: [{'name': None, 'args': ' Francisco', 'id': None}]

看着工具参数像打字机一样逐字出现，用户能直观感受到模型正在"思考"如何调用工具。

3.custom模式：自定义进度信号

这是最有创造力的模式。通过get_stream_writer()，你可以在工具或节点内部发送任意自定义数据到前端。

fromlanggraph.configimportget_stream_writerdefget_weather(city:str)->str:"""获取城市天气，带进度反馈"""writer=get_stream_writer()# 发送自定义进度事件writer({"status":"searching","city":city,"progress":0})# ... 执行搜索 ...writer({"status":"analyzing","progress":50})# ... 分析数据 ...writer({"status":"complete","progress":100,"result":"sunny"})returnf"It's always sunny in{city}!"# 流式接收自定义事件forchunkinagent.stream({"messages":[{"role":"user","content":"What is the weather in SF?"}]},stream_mode="custom",version="v2",):ifchunk["type"]=="custom":print(f"📊 进度更新:{chunk['data']}")

输出示例：

📊 进度更新: {'status': 'searching', 'city': 'San Francisco', 'progress': 0} 📊 进度更新: {'status': 'analyzing', 'progress': 50} 📊 进度更新: {'status': 'complete', 'progress': 100, 'result': 'sunny'}

实战技巧：配合前端进度条组件，你可以实现"正在连接天气服务…"、"正在解析数据…"等精细化反馈。

二、组合模式：多维度实时反馈

单一模式往往不够用。LangChain允许你同时启用多种模式，获得全方位的流式信息。

forchunkinagent.stream({"messages":[{"role":"user","content":"What is the weather in SF?"}]},stream_mode=["updates","messages","custom"],# 三管齐下！version="v2",):# 统一格式：每个chunk都有 type, ns, datastream_type=chunk["type"]data=chunk["data"]ifstream_type=="messages":token,metadata=dataprint(token.content,end="")# 实时打字机效果elifstream_type=="updates":fornode_name,state_updateindata.items():print(f"\n🔄 节点{node_name}完成")elifstream_type=="custom":print(f"\n📊 自定义事件:{data}")

v2格式的统一结构：

{"type":"updates"|"messages"|"custom"|...,"ns":(),# 命名空间（用于子Agent）"data":...,# 实际数据（类型取决于mode）}

最佳实践：

• 开发调试：["updates", "debug"]—— 看流程+看细节
• 生产UI：["messages", "custom"]—— 打字机效果+进度条
• 复杂Agent：["updates", "messages", "custom"]—— 全方位监控

三、思考/推理Token流式展示

现代模型（如Claude 3.7 Sonnet、o1系列）具备推理能力——它们会在回答前进行内部思考。LangChain支持将这些"思维过程"实时展示给用户。

fromlangchain_anthropicimportChatAnthropicfromlangchain.messagesimportAIMessageChunk# 启用推理模式model=ChatAnthropic(model_name="claude-sonnet-4-6",thinking={"type":"enabled","budget_tokens":5000},)agent=create_agent(model=model,tools=[get_weather])forchunkinagent.stream({"messages":[{"role":"user","content":"What is the weather in SF?"}]},stream_mode="messages",):ifnotisinstance(token:=chunk["data"][0],AIMessageChunk):continue# 分离推理内容和正式回答content_blocks=token.content_blocks# 提取推理块reasoning=[bforbincontent_blocksifb["type"]=="reasoning"]text=[bforbincontent_blocksifb["type"]=="text"]ifreasoning:print(f"🤔 [思考中]{reasoning[0]['reasoning']}",end="")iftext:print(f"💬{text[0]['text']}",end="")

输出效果：

🤔 [思考中] 用户询问旧金山天气，我需要调用天气工具 🤔 [思考中] 城市参数应该是"San Francisco" 💬 旧金山天气晴朗，气温22°C...

UI设计建议：

• 用灰色/斜体展示推理内容，与正式回答区分
• 提供"显示/隐藏思考过程"的开关
• 推理Token通常消耗额外成本，注意预算控制

四、工具调用进度实时反馈

工具执行往往是耗时最长的环节。用户需要知道：哪个工具正在运行？进度如何？是否卡住了？

基础方案：工具执行状态

defanalyze_data(query:str)->str:"""分析数据，带详细进度反馈"""writer=get_stream_writer()# 阶段1：数据获取writer({"tool":"analyze_data","phase":"fetching","message":"正在从数据库获取原始数据...","progress":10})time.sleep(1)# 阶段2：数据清洗writer({"tool":"analyze_data","phase":"cleaning","message":"清洗异常值和缺失数据...","progress":40})time.sleep(1)# 阶段3：分析计算writer({"tool":"analyze_data","phase":"analyzing","message":"执行统计分析和模式识别...","progress":70})time.sleep(1)# 完成writer({"tool":"analyze_data","phase":"complete","message":"分析完成！","progress":100})return"分析结果：数据趋势呈上升态势"

高级方案：多工具协调展示

当Agent并行调用多个工具时，你需要区分不同工具的执行状态：

fromtypingimportAnydef_render_tool_progress(chunk:dict)->None:"""渲染工具执行进度"""ifchunk["type"]=="custom":data=chunk["data"]# 工具进度条渲染if"tool"indataand"progress"indata:bar="█"*(data["progress"]//5)+"░"*(20-data["progress"]//5)print(f"\r🔧{data['tool']}: [{bar}]{data['progress']}% -{data['message']}",end="")# 工具完成ifdata.get("phase")=="complete":print(f"\n✅{data['tool']}执行完成")# 流式循环中调用forchunkinagent.stream(input_data,stream_mode=["messages","custom"]):_render_tool_progress(chunk)# ... 其他处理

前端集成思路：

• 使用stream_mode=["messages", "custom"]
• 为每个工具调用创建独立的进度卡片
• 通过tool_call_id关联进度更新与具体工具实例

五、多Agent系统的流式溯源

在多Agent架构中（主Agent调用子Agent），你需要知道每条消息来自哪个Agent。

fromlangchain.agentsimportcreate_agent# 创建子Agent（天气专家）weather_agent=create_agent(model="gpt-5-nano",tools=[get_weather],name="weather_specialist",# 关键：给Agent命名！)# 主Agent调用子Agentdefcall_weather_agent(query:str)->str:result=weather_agent.invoke({"messages":[{"role":"user","content":query}]})returnresult["messages"][-1].text supervisor=create_agent(model="gpt-5-nano",tools=[call_weather_agent],name="supervisor",)# 流式追踪，启用子图流式current_agent=Noneforchunkinsupervisor.stream({"messages":[{"role":"user","content":"北京天气如何？"}]},stream_mode=["messages","updates"],subgraphs=True,# 关键：启用子Agent流式！version="v2",):ifchunk["type"]=="messages":token,metadata=chunk["data"]# 通过 lc_agent_name 识别消息来源agent_name=metadata.get("lc_agent_name","unknown")ifagent_name!=current_agent:print(f"\n🤖 [{agent_name}]:")current_agent=agent_nameiftoken.content:print(token.content,end="")

输出示例：

🤖 [supervisor]: 我需要查询天气信息，让我调用天气专家。 🤖 [weather_specialist]: 正在获取北京天气数据... 北京今天晴朗，气温25°C。 🤖 [supervisor]: 根据天气专家的信息，北京今天天气很好，适合出门！

关键配置：

1. 给每个Agent设置name参数
2. 流式时启用subgraphs=True
3. 从metadata["lc_agent_name"]读取来源

六、实战模式：构建生产级流式UI

完整示例：带进度条的Agent

importasynciofromlangchain.agentsimportcreate_agentfromlanggraph.configimportget_stream_writerfromlangchain.messagesimportAIMessageChunk# 1. 定义带进度反馈的工具defresearch_topic(topic:str)->str:"""研究主题，实时反馈进度"""writer=get_stream_writer()steps=[("搜索相关资料",20),("阅读关键文章",40),("提取核心观点",60),("整理研究结论",80),("生成最终报告",100)]forstep_name,progressinsteps:writer({"type":"progress","tool":"research_topic","step":step_name,"progress":progress,"topic":topic})time.sleep(0.5)# 模拟耗时操作returnf"关于'{topic}'的研究完成：这是一个热门话题，近期有显著增长趋势。"# 2. 创建Agentagent=create_agent(model="gpt-5-nano",tools=[research_topic],)# 3. 流式处理函数asyncdefstream_with_ui(input_text:str):"""生产级流式处理，分离不同类型的更新"""# 状态管理current_tool=Nonemessage_buffer=""asyncforchunkinagent.astream({"messages":[{"role":"user","content":input_text}]},stream_mode=["messages","updates","custom"],version="v2",):stream_type=chunk["type"]data=chunk["data"]# 处理Token流（打字机效果）ifstream_type=="messages":token,metadata=dataifisinstance(token,AIMessageChunk):iftoken.text:message_buffer+=token.text# 实时更新UI文本yield{"type":"text","content":token.text}# 工具调用片段iftoken.tool_call_chunks:yield{"type":"tool_start","data":token.tool_call_chunks}# 处理状态更新elifstream_type=="updates":fornode_name,updateindata.items():ifnode_name=="tools":yield{"type":"tool_complete","data":update}elifnode_name=="model":yield{"type":"model_complete","data":update}# 处理自定义进度elifstream_type=="custom":ifdata.get("type")=="progress":yield{"type":"progress","tool":data["tool"],"step":data["step"],"progress":data["progress"]}# 4. 前端消费示例（伪代码）""" async for event in stream_with_ui("研究一下AI Agent发展趋势"): if event["type"] == "text": append_to_chat(event["content"]) # 打字机效果 elif event["type"] == "progress": update_progress_bar( tool=event["tool"], progress=event["progress"], label=event["step"] ) elif event["type"] == "tool_complete": show_tool_result(event["data"]) """

总结：流式处理最佳实践清单

核心要点回顾：

1. 始终使用version="v2"—— 统一格式，简化处理逻辑
2. 合理选择粒度—— 不是所有场景都需要Token级流式
3. 善用custom模式—— 它是构建高级UI的秘密武器
4. 给Agent命名—— 多Agent场景下溯源的关键
5. 分离关注点—— 用chunk["type"]路由不同类型的更新

流式处理不仅仅是技术优化，更是用户体验的革命。当你的Agent开始"边想边说"，用户从焦虑等待转为参与观察，信任感自然建立。

下一篇，我们将深入探讨中间件开发，学习如何在Agent执行链路中插入自定义逻辑，实现日志、监控、安全护栏等高级功能。

参考资源：

• LangChain Streaming官方文档
• LangGraph Streaming指南

关注公众号【dev派】，发送 “agent” 获取全部源码和模板