LLM 应用开发实战:实现流式响应与高效落地
一、背景
在智能客服产品的实际交付中,客户往往对回答的准确率表示认可,但普遍反馈机器人回复速度较慢。提升响应体验是优化重点之一,而流式响应(Streaming Response)是解决等待感过强的有效方案。
二、原理
在即时通讯场景中,长文本一次性发送会造成明显的等待延迟。若采用逐句或逐字发送的方式,用户的感知等待时间会显著降低。大语言模型(LLM)基于 Token 生成机制,天然支持流式输出,无需像人类写作那样反复检查修改。
实现完整的流式响应链路,需要三个环节均支持流式传输:
- LLM 生成端:模型服务需支持流式返回。例如百度文心一言接口,请求体中设置
stream=True即可开启。 - 服务器通道:应用服务器与 LLM 服务器之间的 HTTP 连接需支持流式读取。Python 的
requests库中,post方法可设置stream=True参数。 - 前端展示端:应用服务器返回给前端的通道及前端组件需支持流式渲染。例如 Gradio 的
ChatInterface支持接收生成器对象进行实时回显。
三、实践
1. Gradio 基础流式实现
Gradio 的回调函数若返回字符串,则一次性显示;若返回生成器(Generator),则支持流式输出。以下是一个简单的 Echo Demo,演示如何通过生成器实现逐字回显:
import time
import gradio as gr
def echo_response(message, history):
# 模拟流式生成,每次 yield 一部分内容
for i in range(len(message)):
time.sleep(0.3)
yield "你输入:" + message[:i+1]
gr.ChatInterface(echo_response).launch()
运行后,消息将一个字一个字地显示出来,极大降低了等待焦虑。
2. 接入 LLM 流式接口
将上述逻辑与 LLM 接口结合,核心在于将 LLM 的流式响应转换为 Gradio 可识别的生成器。以百度文心一言为例,流程如下:
- 调用 LLM API 时开启
stream=True。 - 使用
requests获取流式响应对象。 - 遍历响应中的 chunk,提取
result字段并 yield。
完整集成代码示例:
import requests
import gradio as gr
# 假设已配置好 API Key
API_KEY = "YOUR_API_KEY"
SECRET_KEY =
MODEL_NAME =
():
url =
headers = {: }
token_url =
params = {: , : API_KEY, : SECRET_KEY}
response = requests.post(token_url, params=params)
access_token = response.json().get()
payload = {
: [{: , : prompt}],
:
}
req = requests.post(url + access_token, headers=headers, json=payload, stream=)
line req.iter_lines():
line:
:
data = (line.decode().replace(, ))
result = data.get(, )
result:
result
Exception:
():
full_response =
chunk call_ernie_stream(message):
full_response += chunk
full_response
gr.ChatInterface(chat_interface).launch()


