微信小程序实现ChatGPT流式响应的Flask后端实战

a1b2c

309人浏览 · 2026-02-14 04:18:53

a1b2c · 2026-02-14 04:18:53 发布

1. 为什么需要流式响应？

想象一下这样的场景：你在微信小程序里问AI一个问题，如果等待5秒后突然看到完整答案，和等待1秒后开始逐字出现答案，哪种体验更自然？这就是流式响应的核心价值——消除等待焦虑。

传统API交互就像等快递：下单→等待→收货。而流式响应如同现场观看厨师做菜：切菜→翻炒→装盘，每个步骤都可见。在AI对话场景中，这种"逐字输出"的效果能带来三个关键优势：

心理层面：人类对200ms内的反馈感知为"即时响应"。流式输出让用户在300ms内就能看到首个字符，避免"是不是卡住了"的疑虑
性能层面：大语言模型生成20个token平均需要2秒，但生成首个token可能只需200ms。流式传输充分利用了这个特性
成本层面：微信小程序云开发按流量计费，流式响应可以边生成边传输，避免大块数据一次性传输的超时风险

2. Flask后端核心实现

2.1 基础环境搭建

先确保你的Python环境符合以下要求：

python -m pip install flask==2.3.2 openai==0.27.8

建议使用虚拟环境隔离依赖：

python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate.bat  # Windows

2.2 流式接口代码解析

创建app.py作为主入口文件：

from flask import Flask, request, Response, jsonify
import openai
import time

app = Flask(__name__)

# 配置Azure OpenAI参数
openai.api_type = "azure"
openai.api_base = "https://你的资源名称.openai.azure.com/"
openai.api_version = "2023-05-15"
openai.api_key = "你的API密钥"

@app.route('/chat', methods=['POST'])
def chat_stream():
    messages = request.json.get('messages', [])
    
    def generate():
        try:
            # 创建流式响应
            response = openai.ChatCompletion.create(
                engine="gpt-35-turbo",
                messages=messages,
                temperature=0.7,
                stream=True
            )
            
            # 逐块处理响应
            for chunk in response:
                if chunk.choices:
                    delta = chunk.choices[0].delta
                    if hasattr(delta, 'content'):
                        yield delta.content
                        time.sleep(0.02)  # 控制输出速度
                        
        except Exception as e:
            yield f"[ERROR] {str(e)}"

    return Response(generate(), mimetype='text/event-stream')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

关键点说明：

mimetype='text/event-stream'：声明这是SSE(Server-Sent Events)流
yield关键字：将函数变为生成器，实现分块输出
time.sleep(0.02)：人为降低输出速度，模拟更自然的打字效果

2.3 部署注意事项

本地测试时直接运行即可，但生产环境建议：

gunicorn -w 4 -b :5000 app:app

常见问题处理：

跨域问题：添加Flask-CORS扩展

超时设置：Nginx默认超时60秒，需调整：

proxy_read_timeout 300s;
proxy_connect_timeout 75s;

3. 微信小程序对接实战

3.1 基础请求封装

在utils/request.js中创建流式请求方法：

function streamRequest(options) {
  const { url, data, onMessage, onError, onComplete } = options
  
  const task = wx.request({
    url: url,
    method: 'POST',
    data: data,
    enableChunked: true,
    success(res) {
      if (res.statusCode !== 200) {
        onError?.(res.errMsg)
      }
    },
    fail: onError
  })
  
  task.onChunkReceived(res => {
    const array = new Uint8Array(res.data)
    const text = new TextDecoder('utf-8').decode(array)
    onMessage?.(text)
  })
  
  return {
    abort: () => task.abort()
  }
}

3.2 页面调用示例

在聊天页面中使用：

import { streamRequest } from '../../utils/request'

Page({
  data: {
    messages: [],
    currentReply: ''
  },
  
  sendMessage() {
    const newMsg = {role: 'user', content: '你好'}
    this.setData({messages: [...this.data.messages, newMsg]})
    
    this.stream = streamRequest({
      url: 'https://你的域名/chat',
      data: {messages: this.data.messages},
      onMessage: (text) => {
        this.setData({
          currentReply: this.data.currentReply + text
        })
      },
      onComplete: () => {
        this.setData({
          messages: [...this.data.messages, 
                    {role: 'assistant', content: this.data.currentReply}],
          currentReply: ''
        })
      }
    })
  },
  
  onUnload() {
    this.stream?.abort()
  }
})

3.3 性能优化技巧

防抖处理：频繁setData会导致性能问题

let buffer = ''
let timer = null

onMessage: (text) => {
  buffer += text
  if (!timer) {
    timer = setTimeout(() => {
      this.setData({currentReply: buffer})
      timer = null
    }, 100)
  }
}

错误重试：网络波动时自动重试

let retryCount = 0

onError: (err) => {
  if (retryCount < 3) {
    setTimeout(() => {
      this.sendMessage()
      retryCount++
    }, 1000 * retryCount)
  }
}

4. 常见问题解决方案

4.1 流式中断处理

现象：输出到一半突然停止 排查步骤：

检查服务端日志是否有异常
测试直接访问API端点是否稳定
在小程序开发工具中开启"不校验域名"选项临时测试

4.2 编码问题

特殊字符显示乱码时，需要统一编码：

# 服务端
yield chunk.encode('utf-8')

# 小程序端
const decoder = new TextDecoder('gbk')  // 根据实际情况调整

4.3 安卓兼容性问题

部分安卓机型需要额外配置：

wx.request({
  enableQuic: true,
  enableHttp2: true
})

5. 进阶优化方向

5.1 上下文管理

实现多轮对话记忆：

from collections import deque

MAX_HISTORY = 10
message_queue = deque(maxlen=MAX_HISTORY)

@app.route('/chat', methods=['POST'])
def chat_stream():
    user_msg = request.json.get('message')
    message_queue.append({"role": "user", "content": user_msg})
    
    response = openai.ChatCompletion.create(
        engine="gpt-35-turbo",
        messages=list(message_queue),
        stream=True
    )

5.2 速率限制

防止API滥用：

from flask_limiter import Limiter

limiter = Limiter(
    app=app,
    key_func=lambda: request.remote_addr
)

@app.route('/chat')
@limiter.limit("5 per minute")
def chat_stream():
    # ...

5.3 监控接入

使用Prometheus监控API健康状态：

from prometheus_flask_exporter import PrometheusMetrics

metrics = PrometheusMetrics(app)
metrics.info('app_info', 'Chat Stream Service', version='1.0.0')

实际项目中，我在处理一个教育类小程序时发现，当并发用户超过50人时，流式响应延迟明显增加。后来通过以下优化将P99延迟从3.2秒降到800ms：

增加Redis缓存高频问题答案
使用Nginx负载均衡
将长响应拆分为多个短响应块

这种实现方式不仅适用于聊天场景，还能扩展到：

实时股票行情推送
长文章分页加载
教育类应用的答题反馈
游戏中的实时对话系统

调试时建议先用固定文本测试流式传输：

@app.route('/test')
def test_stream():
    def generate():
        text = "这是一段测试文本，用于验证流式传输效果。"
        for char in text:
            yield char
            time.sleep(0.1)
    
    return Response(generate(), mimetype='text/event-stream')