流式输出实战让Agent边想边说提升用户体验上一篇聊回调的时候提到过on_llm_new_token说流式输出会用到它这篇就接着往下聊。做Agent应用有个很现实的体验问题用户点一下发送然后盯着空白屏幕干等长一点的回答要等十几秒耐心差的直接以为程序卡死了。换成流式输出模型一边生成一边往外吐字前端打字机效果跑起来用户至少知道活儿在干等待感一下子弱很多。这篇就把流式输出从头到尾捋一遍。为什么要流式输出先把账算明白。模型生成长文本整段返回和逐字返回总耗时其实差不多模型该算的token一个都没少。差别全在用户感知上。人等东西有个心理阈值大概两三秒。两秒内有反馈觉得流畅。超过五秒没动静就开始怀疑是不是崩了。整段返回就是让用户干等到最后那一刻才看见东西前面全是空白焦虑。流式输出把等待切成一小段一小段第一个token几百毫秒就到后面持续有字往外蹦大脑就知道活儿还在干体感上快得多。我做过一个客服Agent回答动辄三四百字整段返回平均八秒。用户投诉说卡后台一看其实没卡就是返回太慢用户以为卡了。换成流式首字一秒内出来投诉立马没了。流式输出解决的是感知问题把干等变成看得见的进度。LangChain的stream方法LangChain里所有Runnable都带stream方法上一篇讲LCEL的时候提过一句。调起来很简单把invoke换成stream返回的是一个迭代器逐个吐出输出块。fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser modelChatOpenAI(modelgpt-4o-mini)promptChatPromptTemplate.from_template(解释{concept})chainprompt|model|StrOutputParser()forchunkinchain.stream({concept:向量检索}):print(chunk,end,flushTrue)跑起来你会看到字一个一个往外冒这就是流式效果。chunk是模型每次吐出来的一小段文本可能一个字可能几个字取决于模型。flushTrue很关键不加的话终端会攒一批才刷新流式的感觉就没了。这里有个坑我踩过。流式输出时StrOutputParser吐的是字符串片段但如果你不接解析器直接遍历模型拿到的是AIMessageChunk对象得取它的content属性才是文本。新手第一次跑流式看到屏幕上一堆对象repr打印出来就是这个原因。astream异步流式同步stream在脚本里够用放到Web框架里就会卡住事件循环。FastAPI这类异步框架里得用astream异步版本的流式。importasynciofromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser modelChatOpenAI(modelgpt-4o-mini)promptChatPromptTemplate.from_template(解释{concept})chainprompt|model|StrOutputParser()asyncdefmain():asyncforchunkinchain.astream({concept:向量检索}):print(chunk,end,flushTrue)asyncio.run(main())async for遍历异步迭代器每个chunk到了就处理不阻塞。凡是跑在异步框架里的流式都得用astream用同步stream会把整个事件循环堵死别的请求全排队等它。token级别流式上面的stream拿到的是链最终输出的片段。有时候你想更细拿到模型原始的token流比如想在前端给思考过程和最终回答分开渲染。这时候得直接在模型上调stream或者用回调里的on_llm_new_token。直接在模型上流式最直接。modelChatOpenAI(modelgpt-4o-mini,streamingTrue)forchunkinmodel.stream(写一首关于秋天的诗):print(chunk.content,end,flushTrue)注意这里取的是chunk.content因为模型直接流式返回的是AIMessageChunk。LangChain新版本里streaming参数可以不显式传调stream方法会自动开启流式但写上更清楚免得后面看代码的人犯嘀咕。上一篇回调里埋的on_llm_new_token这里也能接上。把一个回调处理器塞进去每个token生成时触发一次想在token级别做点统计或者日志都走这个口子。还有个更现代的口子叫astream_events。链里混了工具调用、检索这些步骤时它能区分每个步骤的事件工具开始、工具结束、模型生成token分得清清楚楚。做Agent流式时这个特别有用你想把工具调用过程也流式展示给用户看靠astream_events最省心。结合FastAPI做SSE流式接口光在终端里流式没意思真正要落地得做个接口给前端调。Web上做流式推送的标准做法是SSE服务器发送事件。FastAPI原生支持StreamingResponse配合astream很顺手。fromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponsefromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser appFastAPI()modelChatOpenAI(modelgpt-4o-mini)promptChatPromptTemplate.from_template(回答问题{question})chainprompt|model|StrOutputParser()app.get(/chat)defchat(question:str):asyncdefevent_stream():asyncforchunkinchain.astream({question:question}):yieldfdata:{chunk}\n\nreturnStreamingResponse(event_stream(),media_typetext/event-stream)SSE的格式有讲究每条消息以data打头结尾两个换行。前端用EventSource接浏览器自动按这个格式解析。返回类型设成text/event-stream浏览器才知道这是SSE流。这里有个细节容易漏。chain.astream是异步的yield拼字符串是同步操作混在async函数里没问题。但如果你在流式过程中还要查数据库记得用异步查询方法别在async函数里调同步阻塞IO不然流式效果会一卡一卡地出字。还有一点流式连接断了得处理。用户关掉页面或者网络抖动前端EventSource会触发onerror后端这边异步生成器会收到取消信号。记得在event_stream里用try finally清理资源别留下半截没跑完的状态。前端展示方案后端接口有了前端怎么接。SSE用浏览器原生的EventSource最省事。constsourcenewEventSource(/chat?question什么是Agent);source.onmessage(event){document.getElementById(output).textContentevent.data;};source.onerror(){source.close();};每收到一个chunk拼到页面上打字机效果就有了。EventSource的坑在于只支持GET请求没法带请求体。要传大段参数或者走POST得用fetch配合ReadableStream手动解析或者上第三方库。简单场景EventSource够用复杂场景再换fetch。还有一个体验细节前端最好加个光标动画。字往外蹦的时候末尾跟个闪烁光标用户一眼就知道还在生成生成完了把光标去掉。这种小细节对体感提升挺明显花不了几行代码。小结这篇把流式输出从原理到落地过了一遍。为什么流式解决的是用户感知问题LangChain的stream同步流式astream异步流式token级别的细粒度流式FastAPI的SSE接口前端EventSource展示。串起来就是一个完整方案Agent从干等结果变成边想边说体验上了一个台阶。不过你会发现astream用了asyncFastAPI也全是async。异步这东西在Agent里到处都是但很多人用着用着就踩坑同步异步混着调事件循环堵死并发没提起来反而更慢。下一篇就聊异步与并发把asyncio在Agent里的门道理清楚。