在传统的 RAG 流程中，我们通常会等待整个生成过程完成后，再将完整的内容返回给用户。但对于许多应用场景，尤其是需要实时交互的聊天机器人或问答系统，这种等待时间可能会导致糟糕的用户体验。流式输出则很好地解决了这个问题，它允许语言模型在生成内容的同时，将每个词或每个 Token 实时地返回给用户，就像我们看到别人打字一样。
一、为什么需要流式输出？
-
提升用户体验：用户无需漫长等待，可以立即看到内容逐字逐句地生成，大大减少了等待的焦虑感，使得交互更加流畅自然。
-
实时性：对于需要快速响应的应用至关重要，例如客服系统或实时聊天。
-
内存优化：完整生成大段文本会占用较多内存，而流式输出可以边生成边释放，有助于降低内存消耗。
二、 FastAPI 中实现流式输出
在 FastAPI 中实现流式输出，主要有两种常见方式：
1、StreamingResponse 直接流式输出
这是最基础、通用的方案，适合文件传输、日志、模拟分段输出等用途。
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import 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 FastAPI
from sse_starlette.sse import EventSourceResponse
import asyncio
import 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 FastAPI
from fastapi.responses import StreamingResponse
import openai
import 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的多轮对话底层方法
@step
async def chat_step(self,ctx: Context, ev: ChatEvent) -> StopEvent:
req=ev.chat_req
print(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回答和溯源信息。流式输出。")
@inject
async 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回答和溯源信息。流式输出。")
@inject
async 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",
}
)
至此已经将流式输出的所有功能都讲完了。