我要投稿

深度解析：大模型应用实战中的 Stream 流式输出

发布日期：2025-08-07 08:09:21 浏览次数： 1987

作者：AI悠悠

微信搜一搜，关注“AI悠悠”

在传统的 RAG 流程中，我们通常会等待整个生成过程完成后，再将完整的内容返回给用户。但对于许多应用场景，尤其是需要实时交互的聊天机器人或问答系统，这种等待时间可能会导致糟糕的用户体验。流式输出则很好地解决了这个问题，它允许语言模型在生成内容的同时，将每个词或每个 Token 实时地返回给用户，就像我们看到别人打字一样。

一、为什么需要流式输出？

提升用户体验：用户无需漫长等待，可以立即看到内容逐字逐句地生成，大大减少了等待的焦虑感，使得交互更加流畅自然。
实时性：对于需要快速响应的应用至关重要，例如客服系统或实时聊天。
内存优化：完整生成大段文本会占用较多内存，而流式输出可以边生成边释放，有助于降低内存消耗。

二、 FastAPI 中实现流式输出

在 FastAPI 中实现流式输出，主要有两种常见方式：

1、StreamingResponse 直接流式输出

这是最基础、通用的方案，适合文件传输、日志、模拟分段输出等用途。

from fastapi import FastAPIfrom fastapi.responses import StreamingResponseimport asyncio
app = FastAPI()
async def fake_stream():    for i in range(5):        yield f"chunk {i}\n".encode("utf‑8")        await asyncio.sleep(1)
@app.get("/stream")async def stream():    return StreamingResponse(fake_stream(), media_type="text/plain")

在这个例子中，/stream 接口会返回一个流式响应，每秒发送一个数据块（模拟的“Chunk”），客户端在每次接收到数据时就能立即处理，避免等待所有数据传输完毕。

2、SSE 协议流式推送数据

使用 SSE（Server-Sent Events）协议流式推送数据,适合实时通知、聊天系统、前端长连接监听场景，前端通过 EventSource 或相似库消费消息。

from fastapi import FastAPIfrom sse_starlette.sse import EventSourceResponseimport asyncioimport json
app = FastAPI()
async def event_generator():    for i in range(10):        yield {"event": "message", "data": json.dumps(f"chunk {i}")}        await asyncio.sleep(1)
@app.get("/sse")async def sse_stream():    return EventSourceResponse(event_generator())

响应头自动设置 Content-Type: text/event-stream 和 Cache-Control: no-cache；
前端通过 JavaScript 的 new EventSource('/sse') 可接收每条 data: 消息 ;
可用于实时推送 ChatGPT 或 LLM 模型输出等应用；

3、 OpenAI 或 LLM 接口流式输出

结合 OpenAI 的 API stream=True 参数，将大语言模型 (LLM) 的令牌逐步传回客户端，样例（简化）：

from fastapi import FastAPIfrom fastapi.responses import StreamingResponseimport openaiimport asyncio
app = FastAPI()
async def proxy_openai(req_messages):    # openai.api_key 已设定    stream = openai.ChatCompletion.create(        model="gpt‑3.5‑turbo", messages=req_messages, stream=True    )    async for chunk in stream:        if "content" in chunk["choices"][0]["delta"]:            yield chunk["choices"][0]["delta"]["content"]        await asyncio.sleep(0)
@app.post("/chat_stream")async def chat_stream(req: dict):    req_messages = req.get("messages", [])    return StreamingResponse(proxy_openai(req_messages), media_type="text/plain")

根据以上信息，我们初步掌握了流式输出的基本原理和方法，接下来我们来看下在开发RAG或Agent等大模型应用中，如何使用流式输出！

三、RAG实现流式输出的核心逻辑

开发RAG或Agent，一般选择 LangChain（LangGraph）或 LlamaIndex 这两种框架。我们采用LlamaIndex来实现。

1、先来看下非流式输出

LlamaIndex内置了多种ChatEngine对话引擎，这里使用CondenseQuestionChatEngine+CitationQueryEngine，这种引擎特点是可以追溯来源，定位知识库中的元数据，这特点在开发RAG为主的应用中尤为常用。调用chat_engine.achat就可以进行多轮对话的查询了。核心的代码如下：


# 长期和缓存记忆	memory = await self.muxue_memory.get_session_memory(req.session_id)

# 知识库索引Index	kbm_index = await self.muxue_vector_store.init_kbm_vector_index()
# 先构造查询引擎	citation_query_engine = CitationQueryEngine.from_args(		kbm_index,		similarity_top_k=3,		citation_chunk_size=512)
# 再构造对话引擎	chat_engine = CondenseQuestionChatEngine.from_defaults(		query_engine=citation_query_engine,		condense_question_prompt=custom_prompt,		memory=memory,		verbose=True,	)
	resp =await  chat_engine.achat(req.query)  #多轮对话
# 溯源：知识库元数据	sources = [		{"id": getattr(n.node, "node_id", None),"text": n.node.get_content()[:200],"metadata": getattr(n.node, "metadata", {}),		}for n in resp.source_nodes	]
# 返回的数据封装	result=ChatQueryResponse(answer=resp.response,sources=sources)

使用memory组件，可以将历史信息保存到数据库和缓存中；memory组件的使用方法点击这里！
知识库的索引kbm_index，需事先将文档Embedding到知识库，然后创建索引Index;
查询引擎使用CitationQueryEngine，该引擎的特点是可溯源；
对话引擎使用CondenseQuestionChatEngine，初始化时需传入查询引擎、提示词、memory组件等，想看详细日志可以verbose=True;
多轮对话方法是chat_engine.achat；
AI回答的内容，需要溯源知识库元数据 sources；

从代码量来看真实的RAG落地，其工程化的确需 Python功底和对LlamaIndex的各个组件的掌握的！流式输出会更加复杂；在开发RAG中，还会碰到其他的需求，我们一般在核心代码外部还需要包一层Workflow，扩展性和灵活性瞬间上升一个级别！

2、流式输出的核心代码

2.1 LlamaIndex的多轮对话底层方法

@stepasync def chat_step(self,ctx: Context, ev: ChatEvent) -> StopEvent:	req=ev.chat_reqprint(f"chat_step.chat_req={ev.chat_req}")
# 记忆组件      memory = await self.muxue_memory.get_session_memory(req.session_id)
# 知识库索引     kbm_index = await self.muxue_vector_store.init_kbm_vector_index()
# 先构造查询引擎，流式输出=True 	citation_query_engine = CitationQueryEngine.from_args(		kbm_index,		similarity_top_k=3,		citation_chunk_size=512,		streaming=True,)
# 再构造对话引擎	chat_engine = CondenseQuestionChatEngine.from_defaults(		query_engine=citation_query_engine,		condense_question_prompt=custom_prompt,		memory=memory,		verbose=True,)
#多轮对话，流式输出	resp :StreamingAgentChatResponse =await  chat_engine.astream_chat(req.query)  
async for token in resp.async_response_gen():		ctx.write_event_to_stream(StreamEvent(delta=token))
	sources = [		{"id": getattr(n.node, "node_id", None),"text": n.node.get_content()[:200],"metadata": getattr(n.node, "metadata", {}),		}for n in resp.sources[0].raw_output.source_nodes	]
	result=ChatQueryResponse(answer=resp.response,sources=sources)
return StopEvent(result=result)

大部分逻辑与上面的一致，只有以下几点需要调整！

构造查询引擎，流式输出 streaming=True ；
多轮对话流式输出 chat_engine.astream_chat(req.query) ；
大模型返回的一个一个数据块方法：
async for token in resp.async_response_gen()，
因为这里是使用workflow，所以需要将其保存到上下文的流里write_event_to_stream；
若不在workflow里，则直接使用 yield token；
溯源的Source数据可以放在最终的返回结果里；

2.2 Service层写法

async  def chat_stream(self, req: ChatQueryRequest)->ChatQueryResponse:"""  对话服务，返回固定的回答和来源,流式输出 """	handler = self.chat_agent_wf.run(chat_req=req, module="test_module")async for chunk in  handler.stream_events():if isinstance(chunk, StreamEvent):#print(f"chat_service.chat-chunk: {chunk.delta}")yield chunk.delta
	final_result :ChatQueryResponse = await handler# print("最终的完整的答案：", final_result)yield  final_result

之所以有services层，是为了对流数据统一管理，因为第一步中，source并没有放流里。（也可以在第一步中将source数据放流里）

接收流输出的写法依旧是 async for chunk in handler.stream_events() ；
最终的完整的答案需要使用await handler 来获取；

2.3 FastApi的WebApi接口层写法

@chat_router.post("/chat_stream",summary="多轮对话问答",    description="提交用户问题，返回AI回答和溯源信息。流式输出。")@injectasync def chat_stream(req: ChatQueryRequest,request: Request,    chat_service: ChatService = Depends(Provide[Container.chat_service]) )-> EventSourceResponse:
    async def event_stream():async for chunk in  chat_service.chat_stream(req):if isinstance(chunk, ChatQueryResponse):yield {"event": "source", "data": chunk.sources}else:yield {"event": "message", "data": chunk}
    return EventSourceResponse(        event_stream(),        media_type="text/event-stream",        headers={            "Cache-Control": "no-cache",            "Connection": "keep-alive",        }    )

使用yield返回一个一个数据块；
返回的是字典类型（对象），event 对应的值表示消息的类型，data就是消息内容；

效果如下：

2.4 前端停止后接口的处理

FastAPI 可以通过直接监听请求的 disconnect 事件来感知客户端断开连接，进而停止数据发送并释放资源。webapi层完整的代码如下：

# 活跃连接 Task ID 集合active_tasks: Set[int] = set()
@chat_router.post("/chat_stream",summary="多轮对话问答,流式输出",    description="提交用户问题，返回AI回答和溯源信息。流式输出。")@injectasync def chat_stream(req: ChatQueryRequest,request: Request,    chat_service: ChatService = Depends(Provide[Container.chat_service]) )-> EventSourceResponse:
    task = asyncio.current_task()    task_id = id(task) if task else None    if task_id:        active_tasks.add(task_id)        logger.info(f"新连接建立 (task_id={task_id})，当前活跃连接数：{len(active_tasks)}")
    async def event_stream():        try:            async for chunk in  chat_service.chat_stream(req):                # 检查客户端是否断开                if await request.is_disconnected():                    logger.info(f"客户端断开 (task_id={task_id})")                    break
                if isinstance(chunk, ChatQueryResponse):                    yield {"event": "source", "data": chunk.sources}                else:                    yield {"event": "message", "data": chunk}        finally:            # 清理任务            if task_id and task_id in active_tasks:                active_tasks.remove(task_id)                logger.info(f"连接关闭 (task_id={task_id})，剩余活跃连接数：{len(active_tasks)}")
    return EventSourceResponse(        event_stream(),        media_type="text/event-stream",        headers={            "Cache-Control": "no-cache",            "Connection": "keep-alive",        }    )

至此已经将流式输出的所有功能都讲完了。

53AI，企业落地大模型首选服务商

产品：场景落地咨询+大模型应用平台+行业解决方案

承诺：免费POC验证，效果达标后再合作。零风险落地应用大模型，已交付160+中大型企业