流式传输与 LangSmith 观测stream / astream / SSE 原理用普通的invoke调用模型必须等模型把整段回答全部生成完才能看到结果。如果模型思考 20 秒用户就干等 20 秒体验很差。流式传输让模型像 ChatGPT 官网一样「边想边吐字」显著改善体验。本篇讲清楚同步 stream、异步 astream、底层 SSE 协议与源码流程最后用 LangSmith 把整个调用过程可视化。目录为什么需要流式传输stream() 同步流式astream() 异步流式异步前置知识协程与事件循环配合 StrOutputParser 流式输出自定义流式输出解析器底层原理SSE 协议LangChain 流式流程源码分析用 LangSmith 跟踪 LLM 应用本篇小结1. 为什么需要流式传输流式处理对于基于 LLM 的应用响应最终用户至关重要通过逐步显示输出甚至在完整响应准备就绪之前流式传输可以显著改善用户体验。非流式invoke模型直接返回全量内容思考时间越长等待越久。流式stream模型一个字一个字地往外吐用户几乎立刻能看到反馈。2. stream() 同步流式聊天模型的.stream()方法返回一个迭代器在生成输出的同时逐步产生「消息块」用for循环实时处理每个块fromlangchain_openaiimportChatOpenAI modelChatOpenAI(modelgpt-4o-mini)chunks[]forchunkinmodel.stream(讲一个50字的笑话):chunks.append(chunk)print(chunk.content,end|,flushTrue)打印效果每个块之间用|分隔有|一天|兔|子|和|乌|龟|比赛|跑|步|。...2.1 chunk 是什么通过调试可以看到每个 chunk 是一个AIMessageChunk对象它代表 AIMessage 的一部分消息块。消息块可以直接相加拼接# 把前 5 个块拼起来得到一个完整的 AIMessageChunkwholechunks[0]chunks[1]chunks[2]chunks[3]chunks[4]print(whole)# content有一天兔 ...大白话AIMessageChunk就是「一句话的碎片」攒齐了拼起来就是完整回答。3. astream() 异步流式流式传输通常配合异步使用用.astream()方法 async for实现非阻塞的实时输出fromlangchain_openaiimportChatOpenAI modelChatOpenAI(modelgpt-4o-mini)asyncdefasync_stream():print( 异步调用 )asyncforchunkinmodel.astream(讲一个50字的笑话):print(chunk.content,end|,flushTrue)importasyncio asyncio.run(async_stream())4. 异步前置知识协程与事件循环4.1 同步方式阻塞想象你要煮一壶水等 5 秒还要发一条短信等 2 秒。同步方式必须一件一件来importtimedefboil_water():print(开始煮水...)time.sleep(5)# 模拟阻塞 5 秒print(水开了!)defsend_message():print(开始发短信...)time.sleep(2)# 模拟阻塞 2 秒print(短信发送成功!)defmain():boil_water()# 先花 5 秒煮水期间什么也做不了send_message()# 再花 2 秒发短信main()# 总耗时7 秒问题煮水等待的 5 秒里 CPU 完全空闲却不能去发短信效率低。4.2 异步方式协程 事件循环协程一种轻量级并发模型可看作用户态的「轻量级线程」。它用async def定义在需要暂停的地方用await调度完全由程序控制省去了线程切换的开销。事件循环asyncio 的核心像一个总调度员。它不断检查任务列表任务在等 I/O 时就暂停它、去执行下一个就绪任务等待结束后再恢复执行。importasyncioasyncdefboil_water_async():print(开始煮水...)awaitasyncio.sleep(5)# await等待时让出控制权print(水开了!)asyncdefsend_message_async():print(开始发短信...)awaitasyncio.sleep(2)print(短信发送成功!)asyncdefmain():# 创建两个任务交给事件循环调度task1asyncio.create_task(boil_water_async())task2asyncio.create_task(send_message_async())awaittask1awaittask2 asyncio.run(main())# 总耗时5 秒两个任务的等待时间是并发的总结协程用async def定义、用await暂停asyncio.run()会创建事件循环并运行协程。在网络请求、I/O 读写等「等待多、计算少」的场景协程并发效率很高。5. 配合 StrOutputParser 流式输出流式处理并不是聊天模型独有的能力而是所有 Runnable 实例都具备的能力Runnable 接口里的 Streamed。但并非所有组件都支持流式例如检索器 Retriever 就不提供流式处理。重点.stream()产生的块类型取决于正在流式输出的组件。聊天模型返回 AIMessageChunk如果链里加了输出解析器块类型就会变化。用 LCEL 构建「模型 → 字符串解析器」的链再流式输出fromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportStrOutputParser modelChatOpenAI(modelgpt-4o-mini)parserStrOutputParser()chainmodel|parserforchunkinchain.stream(写一段关于爱情的歌词需要5句话):print(chunk,end|,flushTrue)StrOutputParser会从 AIMessageChunk 中提取 content 字段所以此时流出来的每个块就是纯字符串|在星空下许下心愿||你的笑容如晨光|温暖||...6. 自定义流式输出解析器如果想改变输出样式比如「一句话一句话」地输出同时保留流式能力可以在链中使用生成器函数即可完成自定义流式。生成器签名同步是Iterator[Input] - Iterator[Output]异步是AsyncIterator[Input] - AsyncIterator[Output]。下面是一个「按句号切分成列表」的自定义解析器fromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportStrOutputParserfromtypingimportIterator,List modelChatOpenAI(modelgpt-4o-mini)parserStrOutputParser()# 输入是字符串流输出是「一句话」组成的列表流defsplit_into_list(input:Iterator[str])-Iterator[List[str]]:bufferforchunkininput:bufferchunkwhile。inbuffer:# 缓冲区内包含句号找到第一个句号位置stop_indexbuffer.index(。)# 把句号之前的内容去掉首尾空格作为一个句子产出yield[buffer[:stop_index].strip()]# 保留句号之后的内容bufferbuffer[stop_index1:]yield[buffer.strip()]chainmodel|parser|split_into_listforchunkinchain.stream(写一份关于爱情的歌词需要5句话每句话用句号分割):print(chunk,end|,flushTrue)打印效果[在星空下许下承诺的誓言]|[你的笑容如同晨曦温暖了我的心]|[无论时光如何流转我愿与你携手共行]|...7. 底层原理SSE 协议7.1 什么是 SSEHTTP 本身是无状态的请求-响应模式严格说服务器无法主动推送消息。但通过SSEServer-Sent Events服务器发送事件技术可以实现流式传输服务器向客户端声明「接下来发送的是流消息」客户端不关闭连接一直等待新的数据流。SSE 是一种基于 HTTP 的轻量级实时通信协议浏览器通过内置的 EventSource API 接收事件。核心特点基于 HTTP复用标准 HTTP/HTTPS无需额外端口兼容性好、易部署。单向通信只支持服务器向客户端单向推送客户端不能通过同一连接回发数据。自动重连连接中断时浏览器自动尝试重连可用retry字段指定间隔。自定义消息类型响应头设置Content-Type: text/event-stream标识事件流。7.2 数据格式服务端发送 SSE 数据需要设置 HTTP 头Content-Type: text/event-stream;charsetutf-8 Connection: keep-alive每条消息由若干行组成每行格式为[field]: value字段取值data必需数据内容。event可选自定义事件类型默认是 message。id可选数据标识符相当于每条数据的编号。retry可选指定浏览器重连的时间间隔。以冒号:开头的行表示注释。数据示例event: foo data: a foo event data: an unnamed event event: end data: a bar event消息之间用空行\n\n分隔。8. LangChain 流式流程源码分析LangChain 本身并不「创造」网络协议而是依赖底层模型供应商如 OpenAI和 Web 框架如 FastAPI的协议。模型供应商提供流式能力LangChain 调用后把数据处理成一个个 AIMessageChunk。以 OpenAI 为例流式请求通过BaseChatOpenAI的_stream()方法发起关键流程分五步def_stream(self,messages,stopNone,run_managerNone,stream_usageNone,**kwargs):# 1. 流式配置强制启用流式模式kwargs[stream]True# 2. 请求构建构造请求体并定义 AIMessageChunk 传输块payloadself._get_request_payload(messages,stopstop,**kwargs)default_chunk_classAIMessageChunk# 3. 发起调用走流式接口 client.chat.completions.stream(**payload)ifresponse_formatinpayload:response_streamself.root_client.beta.chat.completions.stream(**payload)else:responseself.client.create(**payload)# 4. 响应处理把 OpenAI 数据块转换为 AIMessageChunkwithcontext_managerasresponse:forchunkinresponse:generation_chunkself.convert_chunk_to_generation_chunk(chunk,default_chunk_class,base_generation_infoifis_first_chunkelse{})ifgeneration_chunkisNone:continue# 触发新 token 回调ifrun_manager:run_manager.on_llm_new_token(generation_chunk.text,...)# 5. 产出生成块yieldgeneration_chunk8.1 三个关键问题的答案问题1LangChain 请求 OpenAI 用什么协议LangChain 使用 OpenAI 官方的 Python SDK继承openai._base_client的 HTTP 客户端发起的是标准HTTP 调用。问题2如何支持流式传输在请求中设置streamTrue告诉 OpenAI 服务器以SSE 形式逐块发回数据HTTP 连接保持打开。设置streamTrue后收到的 OpenAI 事件块简化如下data: { id: chatcmpl-123, object: chat.completion.chunk, choices: [{index: 0, delta: {role: assistant, content: 你好}}] } data: { ... choices: [{delta: {content: }}] } data: { ... choices: [{delta: {content: 张三}}] }每个块通过delta增量字段携带新生成的一小段内容。问题3返回的块如何转换成 AIMessageChunk通过_convert_chunk_to_generation_chunk()→_convert_delta_to_message_chunk()完成提取 OpenAI 的delta数据根据roleuser/assistant/system/tool构造对应的消息块类型把增量内容填进 content。8.2 流程总结langchain-openai集成 OpenAI Python SDK提供 HTTP 客户端。LangChain 向 OpenAI API 发起请求。流式请求加入streamTrue说明用 SSE 协议流式返回。LangChain 接收 SSE 响应统一转换成自封装的 AIMessageChunk——这样可以用同一套方式处理来自不同供应商OpenAI、Anthropic 等的流式响应。9. 用 LangSmith 跟踪 LLM 应用用 LangChain 构建的应用往往包含多个步骤和多次 LLM 调用。应用越复杂越需要看清「链路内部到底发生了什么」。LangSmith就是干这个的——它是一个帮助构建生产级 LLM 应用的平台可以密切监控和评估应用与框架解耦可配合 langchain / langgraph 使用也可不用。平台地址https://smith.langchain.com/ 新用户需注册9.1 申请 API Key登录后点击 Settings → API Keys → Create API Key填写描述、选择 Personal Access Token、选择过期时间创建后保存好 Key。9.2 配置两个环境变量exportLANGSMITH_TRACINGtrueexportLANGSMITH_API_KEY你的 LangSmith API KeyWindows 在「环境变量」窗口中新建这两个系统变量即可。9.3 自动跟踪无需改代码配置好环境后任意执行你的 LangChain 代码比如上一篇的「工具 结构化输出」例子LangSmith 会在默认跟踪项目中自动生成这次调用的记录全程没有代码介入。点击最新一次调用可以看到系统状态Stats总步骤数、Token 消耗、耗时、错误率、P50/P90/P99 延迟等。瀑布流调用链以瀑布流形式展示完整步骤、每个步骤的详细信息和耗时。一次典型调用的内部结构RunnableSequence可运行序列就是我们说的「链」最外层。ChatOpenAI实际处理的第一步调用聊天模型生成结果。RunnableLambda实际处理的第二步把 Python 可调用对象包装成 Runnable例如把 AI 结果转换成结构化对象。点开每个节点还能看到该步骤的输入Input、输出Output、耗时、Token、状态以及原始数据Raw。价值调试 Agent、排查「模型为什么没调对工具 / 输出为什么不符合预期」、做性能分析和成本统计时LangSmith 是最直接的可视化工具。10. 本篇小结流式传输让模型边生成边返回显著降低用户等待感stream()同步、astream()异步。流式产生的是AIMessageChunk消息块块可以直接相加块类型取决于链中正在输出的组件。异步靠async def协程 await 事件循环适合 I/O 密集场景可并发执行多个等待任务。用生成器函数可以自定义流式解析如按句子输出。底层协议是SSE基于 HTTP、服务器单向推送、自动重连请求里设streamTrue开启OpenAI 用delta字段返回增量LangChain 统一转成 AIMessageChunk。LangSmith只需配置两个环境变量就能自动可视化跟踪每一次调用的步骤、耗时、Token 和错误。下一篇我们回到核心组件系统讲解消息Messages的管理与多轮对话包括消息裁剪、过滤与合并。