当前位置：首页 > news >正文

流式输出实战，让Agent边想边说提升用户体验

news 2026/8/7 12:56:16

流式输出实战，让Agent边想边说提升用户体验

上一篇聊回调的时候提到过on_llm_new_token，说流式输出会用到它，这篇就接着往下聊。做Agent应用有个很现实的体验问题，用户点一下发送，然后盯着空白屏幕干等，长一点的回答要等十几秒，耐心差的直接以为程序卡死了。换成流式输出，模型一边生成一边往外吐字，前端打字机效果跑起来，用户至少知道活儿在干，等待感一下子弱很多。这篇就把流式输出从头到尾捋一遍。

为什么要流式输出

先把账算明白。模型生成长文本，整段返回和逐字返回，总耗时其实差不多，模型该算的token一个都没少。差别全在用户感知上。

人等东西有个心理阈值，大概两三秒。两秒内有反馈，觉得流畅。超过五秒没动静，就开始怀疑是不是崩了。整段返回就是让用户干等到最后那一刻才看见东西，前面全是空白焦虑。流式输出把等待切成一小段一小段，第一个token几百毫秒就到，后面持续有字往外蹦，大脑就知道活儿还在干，体感上快得多。

我做过一个客服Agent，回答动辄三四百字，整段返回平均八秒。用户投诉说卡，后台一看其实没卡，就是返回太慢用户以为卡了。换成流式，首字一秒内出来，投诉立马没了。流式输出解决的是感知问题，把干等变成看得见的进度。

LangChain的stream方法

LangChain里所有Runnable都带stream方法，上一篇讲LCEL的时候提过一句。调起来很简单，把invoke换成stream，返回的是一个迭代器，逐个吐出输出块。

fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser model=ChatOpenAI(model="gpt-4o-mini")prompt=ChatPromptTemplate.from_template("解释{concept}")chain=prompt|model|StrOutputParser()forchunkinchain.stream({"concept":"向量检索"}):print(chunk,end="",flush=True)

跑起来你会看到字一个一个往外冒，这就是流式效果。chunk是模型每次吐出来的一小段文本，可能一个字，可能几个字，取决于模型。flush=True很关键，不加的话终端会攒一批才刷新，流式的感觉就没了。

这里有个坑我踩过。流式输出时StrOutputParser吐的是字符串片段，但如果你不接解析器，直接遍历模型，拿到的是AIMessageChunk对象，得取它的content属性才是文本。新手第一次跑流式，看到屏幕上一堆对象repr打印出来，就是这个原因。

astream异步流式

同步stream在脚本里够用，放到Web框架里就会卡住事件循环。FastAPI这类异步框架里得用astream，异步版本的流式。

importasynciofromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser model=ChatOpenAI(model="gpt-4o-mini")prompt=ChatPromptTemplate.from_template("解释{concept}")chain=prompt|model|StrOutputParser()asyncdefmain():asyncforchunkinchain.astream({"concept":"向量检索"}):print(chunk,end="",flush=True)asyncio.run(main())

async for遍历异步迭代器，每个chunk到了就处理，不阻塞。凡是跑在异步框架里的流式，都得用astream，用同步stream会把整个事件循环堵死，别的请求全排队等它。

token级别流式

上面的stream拿到的是链最终输出的片段。有时候你想更细，拿到模型原始的token流，比如想在前端给思考过程和最终回答分开渲染。这时候得直接在模型上调stream，或者用回调里的on_llm_new_token。

直接在模型上流式最直接。

model=ChatOpenAI(model="gpt-4o-mini",streaming=True)forchunkinmodel.stream("写一首关于秋天的诗"):print(chunk.content,end="",flush=True)

注意这里取的是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 app=FastAPI()model=ChatOpenAI(model="gpt-4o-mini")prompt=ChatPromptTemplate.from_template("回答问题{question}")chain=prompt|model|StrOutputParser()@app.get("/chat")defchat(question:str):asyncdefevent_stream():asyncforchunkinchain.astream({"question":question}):yieldf"data:{chunk}\n\n"returnStreamingResponse(event_stream(),media_type="text/event-stream")

SSE的格式有讲究，每条消息以data打头，结尾两个换行。前端用EventSource接，浏览器自动按这个格式解析。返回类型设成text/event-stream，浏览器才知道这是SSE流。

这里有个细节容易漏。chain.astream是异步的，yield拼字符串是同步操作，混在async函数里没问题。但如果你在流式过程中还要查数据库，记得用异步查询方法，别在async函数里调同步阻塞IO，不然流式效果会一卡一卡地出字。

还有一点，流式连接断了得处理。用户关掉页面或者网络抖动，前端EventSource会触发onerror，后端这边异步生成器会收到取消信号。记得在event_stream里用try finally清理资源，别留下半截没跑完的状态。

前端展示方案

后端接口有了，前端怎么接。SSE用浏览器原生的EventSource最省事。

constsource=newEventSource("/chat?question=什么是Agent");source.onmessage=(event)=>{document.getElementById("output").textContent+=event.data;};source.onerror=()=>{source.close();};