第8章 响应模块开发与升级：提升聊天机器人响应体验
等AI回复要10秒？用户早就关了页面。流式响应才是正解。
我是怕浪猫，这章解决一个用户体验的硬伤——响应速度。不是让LLM变快，而是让用户感觉快。流式响应+打字机效果，感知延迟从10秒降到0.3秒。
8.1 LLMOps 长短期记忆功能模块开发
长短期记忆架构
短期记忆（内存/Redis）：当前对话的最近消息
↓ 摘要
长期记忆（数据库）：对话摘要 + 向量化存储短期记忆实现
python
# services/memory_service.py 扩展
class MemoryService:
def __init__(self, redis_client=None):
self.redis = redis_client
self.short_term_limit = 10 # 最近10条消息
def get_short_term(self, conv_id):
"""获取短期记忆（Redis缓存）"""
if self.redis:
cached = self.redis.get(f"conv:{conv_id}:short_term")
if cached:
return json.loads(cached)
# Redis没有，从数据库加载
messages = Message.query.filter_by(
conversation_id=conv_id
).order_by(Message.created_at.desc()).limit(self.short_term_limit).all()
messages.reverse()
result = [{"role": m.role, "content": m.content} for m in messages]
# 缓存到Redis
if self.redis:
self.redis.setex(
f"conv:{conv_id}:short_term",
3600, # 1小时过期
json.dumps(result)
)
return result
def add_to_short_term(self, conv_id, role, content):
"""添加到短期记忆"""
if self.redis:
key = f"conv:{conv_id}:short_term"
cached = self.redis.get(key)
messages = json.loads(cached) if cached else []
messages.append({"role": role, "content": content})
# 只保留最近N条
if len(messages) > self.short_term_limit:
messages = messages[-self.short_term_limit:]
self.redis.setex(key, 3600, json.dumps(messages))
def get_combined_memory(self, conv_id):
"""获取组合记忆（摘要+短期）"""
conv = Conversation.query.get(conv_id)
messages = []
if conv.summary:
messages.append({"role": "system", "content": f"对话摘要：{conv.summary}"})
messages.extend(self.get_short_term(conv_id))
return messages8.2 流式响应 vs 非流式响应：基础与应用场景
对比
| 维度 | 非流式 | 流式 |
|---|---|---|
| 用户体验 | 等待全部生成才显示 | 逐字显示，感知快 |
| 首字时间 | 等全部生成完 | 生成第一个token即显示 |
| 实现复杂度 | 简单 | 较复杂 |
| 上下文处理 | 完整内容一次返回 | 需要拼接chunks |
| 适用场景 | 短回答、批量处理 | 聊天、长文本生成 |
非流式实现
python
# 普通API调用
response = client.chat.completions.create(
model="gpt-4",
messages=messages
)
return response.choices[0].message.content # 一次性返回流式实现
python
# 流式API调用
stream = client.chat.completions.create(
model="gpt-4",
messages=messages,
stream=True # 关键参数
)
for chunk in stream:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content8.3 LangChain 流式响应实现
LangChain流式API
python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4", streaming=True)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有帮助的助手"),
("user", "{input}")
])
chain = prompt | llm
# 同步流式
for chunk in chain.stream({"input": "解释RAG"}):
print(chunk.content, end="", flush=True)
# 异步流式
async def stream_response():
async for chunk in chain.astream({"input": "解释RAG"}):
yield chunk.contentFlask集成流式响应
python
from flask import Response, stream_with_context
@chat_bp.route('/completions/stream', methods=['POST'])
def stream_completions():
data = request.json
def generate():
messages = build_messages(data['conversation_id'], data['message'])
stream = client.chat.completions.create(
model=data.get('model', 'gpt-4'),
messages=messages,
stream=True
)
full_content = ""
for chunk in stream:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_content += content
# SSE格式输出
yield f"data: {json.dumps({'content': content})}\n\n"
# 保存完整回复
save_message(data['conversation_id'], 'assistant', full_content)
yield f"data: {json.dumps({'done': True})}\n\n"
return Response(
stream_with_context(generate()),
mimetype='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'X-Accel-Buffering': 'no' # Nginx不缓冲
}
)8.4 后端：队列 + 协程突破 LangGraph 限制实现流式传输
LangGraph的流式限制
LangGraph默认不直接支持SSE流式输出，需要特殊处理。
解决方案：队列桥接
python
import asyncio
from queue import Queue
from threading import Thread
class StreamBridge:
"""将LangGraph的异步流式输出桥接到Flask的SSE"""
def __init__(self):
self.queue = Queue()
self.done = False
def put(self, data):
self.queue.put(data)
def finish(self):
self.done = True
self.queue.put(None) # 结束信号
def __iter__(self):
while not self.done or not self.queue.empty():
item = self.queue.get(timeout=1)
if item is None:
break
yield item
def run_agent_stream(query, bridge):
"""在独立线程中运行Agent"""
try:
for chunk in agent_executor.stream({"input": query}):
if isinstance(chunk, dict) and 'output' in chunk:
bridge.put(json.dumps({'content': chunk['output']}))
else:
bridge.put(json.dumps({'content': str(chunk)}))
except Exception as e:
bridge.put(json.dumps({'error': str(e)}))
finally:
bridge.finish()
@chat_bp.route('/agent/stream', methods=['POST'])
def stream_agent():
data = request.json
bridge = StreamBridge()
# 在独立线程运行Agent
thread = Thread(target=run_agent_stream, args=(data['message'], bridge))
thread.start()
return Response(
stream_with_context(
f"data: {item}\n\n" for item in bridge
),
mimetype='text/event-stream'
)8.5 前端：fetch 获取流式事件数据 + 打字机效果
fetch流式读取
javascript
// api/chat.js 新增
export const streamChat = async (data, onChunk, onDone) => {
const response = await fetch('/api/v1/chat/completions/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(data)
})
const reader = response.body.getReader()
const decoder = new TextDecoder()
while (true) {
const { done, value } = await reader.read()
if (done) break
const text = decoder.decode(value)
const lines = text.split('\n')
for (const line of lines) {
if (line.startsWith('data: ')) {
const data = JSON.parse(line.slice(6))
if (data.done) {
onDone()
} else if (data.content) {
onChunk(data.content)
}
}
}
}
}打字机效果组件
vue
<!-- components/StreamingMessage.vue -->
<script setup>
import { ref, watch } from 'vue'
const props = defineProps({
content: { type: String, default: '' },
streaming: { type: Boolean, default: false }
})
const displayContent = ref('')
const cursorVisible = ref(true)
// 打字机效果：逐字显示
watch(() => props.content, (newVal) => {
if (!props.streaming) {
displayContent.value = newVal
return
}
const target = newVal
const current = displayContent.value
if (target.length > current.length) {
// 新增内容，直接追加（不做逐字动画，否则跟不上流式速度）
displayContent.value = target
}
})
// 光标闪烁
let cursorTimer = null
if (props.streaming) {
cursorTimer = setInterval(() => {
cursorVisible.value = !cursorVisible.value
}, 500)
}
</script>
<template>
<div class="whitespace-pre-wrap">
{{ displayContent }}
<span v-if="streaming && cursorVisible" class="inline-block w-2 h-4 bg-gray-300 ml-0.5">|</span>
</div>
</template>Pinia集成流式
javascript
// stores/chat.js 升级
const useStreaming = ref(false)
const sendMessage = async (content) => {
loading.value = true
messages.value.push({ role: 'user', content })
messages.value.push({ role: 'assistant', content: '', streaming: true })
const assistantIndex = messages.value.length - 1
if (useStreaming.value) {
await chatAPI.streamChat({
message: content,
conversation_id: currentConvId.value
},
// onChunk
(chunk) => {
messages.value[assistantIndex].content += chunk
},
// onDone
() => {
messages.value[assistantIndex].streaming = false
loading.value = false
})
} else {
// 普通模式
const res = await chatAPI.sendMessage({ message: content })
messages.value[assistantIndex].content = res.data.message
messages.value[assistantIndex].streaming = false
loading.value = false
}
}8.6 流式响应下 Token 计数与中断功能
Token计数
python
# 流式响应中统计Token
@chat_bp.route('/completions/stream', methods=['POST'])
def stream_completions():
data = request.json
def generate():
full_content = ""
prompt_tokens = 0
completion_tokens = 0
stream = client.chat.completions.create(
model=data.get('model', 'gpt-4'),
messages=messages,
stream=True,
stream_options={"include_usage": True} # 包含Token统计
)
for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_content += content
yield f"data: {json.dumps({'content': content})}\n\n"
# 最后一个chunk包含usage
if hasattr(chunk, 'usage') and chunk.usage:
prompt_tokens = chunk.usage.prompt_tokens
completion_tokens = chunk.usage.completion_tokens
# 发送统计信息
yield f"data: {json.dumps({
'done': True,
'usage': {
'prompt_tokens': prompt_tokens,
'completion_tokens': completion_tokens,
'total_tokens': prompt_tokens + completion_tokens
}
})}\n\n"
# 保存完整消息
save_message(data['conversation_id'], 'assistant', full_content,
completion_tokens)
return Response(generate(), mimetype='text/event-stream')中断功能
python
# 中断机制：使用Redis标记
@chat_bp.route('/completions/abort/<conversation_id>', methods=['POST'])
def abort_completions(conversation_id):
"""中断正在进行的流式响应"""
redis_client.setex(f"abort:{conversation_id}", 60, "1")
return success()
# 在generate中检查中断
def generate():
for chunk in stream:
# 检查中断标记
if redis_client.get(f"abort:{conv_id}"):
yield f"data: {json.dumps({'aborted': True})}\n\n"
break
# ... 正常处理前端中断按钮
javascript
const abortRequest = async () => {
if (currentConvId.value) {
await chatAPI.abortCompletions(currentConvId.value)
loading.value = false
}
}8.7 前后端接口对接与测试完善
联调检查清单
| 检查项 | 预期 | 验证方式 |
|---|---|---|
| 非流式响应 | 正常返回完整内容 | curl测试 |
| 流式响应 | SSE逐chunk返回 | 浏览器DevTools |
| 打字机效果 | 逐字显示+光标 | 前端观察 |
| Token统计 | 正确计数 | 日志验证 |
| 中断功能 | 点击中断立即停止 | 手动测试 |
| 长文本 | 流式不中断 | 5000字+测试 |
| 并发 | 多用户同时流式 | 压测 |
流式响应测试脚本
python
# tests/test_stream.py
import requests
def test_stream_response():
response = requests.post(
'http://localhost:5000/api/v1/chat/completions/stream',
json={'message': '写一首关于编程的诗'},
stream=True
)
chunks = []
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: '):
data = json.loads(line[6:])
if data.get('content'):
chunks.append(data['content'])
if data.get('done'):
break
full_content = ''.join(chunks)
assert len(full_content) > 0
print(f"收到 {len(chunks)} 个chunk，总内容长度 {len(full_content)}")流式响应的实现不难，难的是在各种场景下稳定——网络抖动、用户中断、Agent多轮工具调用。把这些边界情况处理好，才算生产级。
本章小结
| 主题 | 核心要点 |
|---|---|
| 长短期记忆 | Redis缓存短期 + DB摘要长期 |
| 流式vs非流式 | 流式感知延迟低，实现复杂 |
| LangChain流式 | stream=True + SSE |
| 队列桥接 | 线程+Queue突破LangGraph限制 |
| 前端打字机 | fetch流式读取 + 逐字显示 |
| Token计数 | stream_options包含usage |
| 中断功能 | Redis标记 + 前端中断按钮 |
觉得有用？收藏起来，下次直接照抄。
你的流式响应实现遇到过什么坑？评论区聊聊。
关注怕浪猫，下期我们搞安全——JWT认证、GitHub第三方登录、路由守卫，让LLMOps平台有安全感。
系列进度 8/23
下章预告： 第9章授权认证模块——JWT原理、GitHub OAuth、Flask-Login、前端令牌管理、路由守卫，给你的平台加上安全锁。