微信小程序使用websocket调用大模型返回逐字显示的结果

页面展示小程序代码aichat.js:const config require(../../utils/config.js); Page({ data: { messages: [], inputText: , isLoading: false, scrollToId: scroll-bottom, currentStreamingId: null, isConnecting: false, isReceiving: false, socketOpen: false, }, // 逐字显示相关的变量 typewriterTimer: null, fullResponseContent: , currentDisplayIndex: 0, // 新增数据缓冲相关变量 buffer: , // 数据缓冲区 incompleteLine: , // 不完整的行 onLoad() { this.addMessage(assistant, 您好我是AI助手有什么可以帮助您的吗); }, onInputChange(e) { this.setData({ inputText: e.detail.value }); }, addMessage(role, content, isStreaming false, messageId null) { const id messageId || Date.now().toString(); const newMessage { id: id, role: role, content: content, isStreaming: isStreaming, timestamp: new Date().getTime() }; const existingIndex this.data.messages.findIndex(msg msg.id id); if (existingIndex ! -1) { const messages [...this.data.messages]; messages[existingIndex] newMessage; this.setData({ messages }); } else { this.setData({ messages: [...this.data.messages, newMessage] }); } this.setData({ scrollToId: scroll-bottom }); return id; }, appendContent(newContent) { if (!newContent) return; this.fullResponseContent newContent; if (!this.typewriterTimer this.fullResponseContent.length 0) { this.startTypewriter(); } }, startTypewriter() { if (this.typewriterTimer) { clearInterval(this.typewriterTimer); this.typewriterTimer null; } const messageId this.data.currentStreamingId; if (!messageId) return; this.typewriterTimer setInterval(() { if (this.currentDisplayIndex this.fullResponseContent.length) { const step Math.min(2, this.fullResponseContent.length - this.currentDisplayIndex); this.currentDisplayIndex step; const displayContent this.fullResponseContent.substring(0, this.currentDisplayIndex); this.updateStreamingMessage(messageId, displayContent, false); } else { if (this.data.isReceiving false this.currentDisplayIndex this.fullResponseContent.length) { clearInterval(this.typewriterTimer); this.typewriterTimer null; this.completeStreaming(messageId); } } }, 30); }, updateStreamingMessage(messageId, content, isComplete false) { const messages this.data.messages.map(msg { if (msg.id messageId) { return { ...msg, content: content, isStreaming: !isComplete }; } return msg; }); this.setData({ messages, scrollToId: scroll-bottom }); }, async sendMessage() { const text this.data.inputText ? this.data.inputText.trim() : ; if (!text || this.data.isLoading) { if (!text) { wx.showToast({ title: 请输入内容, icon: none }); } return; } if (this.typewriterTimer) { clearInterval(this.typewriterTimer); this.typewriterTimer null; } this.fullResponseContent ; this.currentDisplayIndex 0; this.buffer ; // 重置缓冲区 this.incompleteLine ; // 重置不完整的行 this.addMessage(user, text); this.setData({ inputText: , isLoading: true, isReceiving: true }); const aiMessageId Date.now().toString(); this.addMessage(assistant, , true, aiMessageId); this.setData({ currentStreamingId: aiMessageId }); const chatHistory this.data.messages.filter(msg msg.id ! aiMessageId msg.content.trim() ! ); this.connectWebSocket(chatHistory, aiMessageId); }, // 处理接收到的数据解决分包问题 processReceivedData(data) { // 将新数据添加到缓冲区 this.buffer data; // 按行分割 const lines this.buffer.split(\n); // 最后一行可能不完整保存起来下次使用 this.buffer lines.pop() || ; for (const line of lines) { if (line.trim() ) continue; // 处理SSE格式的数据 if (line.startsWith(data: )) { const jsonStr line.slice(6).trim(); if (jsonStr [DONE]) { this.setData({ isReceiving: false }); if (this.currentDisplayIndex this.fullResponseContent.length) { if (this.typewriterTimer) { clearInterval(this.typewriterTimer); this.typewriterTimer null; } this.completeStreaming(this.data.currentStreamingId); } return; } // 尝试解析JSON try { const parsed JSON.parse(jsonStr); const content this.extractContentFromChunk(parsed); if (content) { console.log(成功解析内容:, content); this.appendContent(content); } } catch (e) { // JSON解析失败可能是分包导致的暂存到不完整行 console.warn(JSON解析失败可能是不完整数据:, jsonStr.substring(0, 50)); this.incompleteLine jsonStr; // 尝试解析累积的不完整数据 if (this.incompleteLine.length 0) { try { const parsed JSON.parse(this.incompleteLine); const content this.extractContentFromChunk(parsed); if (content) { console.log(成功解析不完整数据:, content); this.appendContent(content); this.incompleteLine ; // 解析成功清空 } } catch (e2) { // 仍然无法解析继续等待更多数据 // 如果积累太多可能是格式错误清空避免内存问题 if (this.incompleteLine.length 10000) { console.error(不完整数据过长清空缓冲区); this.incompleteLine ; } } } } } else if (line.trim()) { // 如果不是SSE格式可能是直接的JSON try { const parsed JSON.parse(line); const content this.extractContentFromChunk(parsed); if (content) { console.log(成功解析直接JSON:, content); this.appendContent(content); } } catch (e) { console.warn(无法解析的行:, line.substring(0, 100)); } } } }, // 从数据块中提取内容 extractContentFromChunk(chunk) { // 处理多种可能的格式 if (chunk.choices chunk.choices[0]) { const choice chunk.choices[0]; // OpenAI格式 if (choice.delta choice.delta.content) { return choice.delta.content; } // 其他可能的格式 if (choice.text) { return choice.text; } if (choice.message choice.message.content) { return choice.message.content; } } // 直接包含content字段 if (chunk.content) { return chunk.content; } // 包含data字段某些特殊格式 if (chunk.data chunk.data.content) { return chunk.data.content; } return null; }, connectWebSocket(messages, aiMessageId) { const that this; const socket wx.connectSocket({ url: config.WEBSOCKET_URL/building_ws, success: () { console.log(WebSocket连接中...); this.setData({ isConnecting: true }); }, fail: (err) { console.error(WebSocket连接失败, err); this.handleError(连接失败请重试, aiMessageId); } }); socket.onOpen(() { console.log(WebSocket连接已打开); that.setData({ socketOpen: true, isConnecting: false }); // 添加延迟确保连接完全稳定 setTimeout(() { const messageData { messages: messages, api_key: sk-vllmapi123456789 }; const messageStr JSON.stringify(messageData); console.log(准备发送消息:, messageStr); socket.send({ data: messageStr, success: () { console.log(消息已成功发送); }, fail: (err) { console.error(发送失败, err); that.handleError(发送失败请重试, aiMessageId); } }); }, 100); }); socket.onMessage((res) { console.log(收到原始数据:, res.data); // 处理收到的消息 that.processReceivedData(res.data); }); socket.onError((err) { console.error(WebSocket错误, err); that.handleError(连接出错请重试, aiMessageId); }); this.socket socket; }, completeStreaming(messageId) { if (this.currentDisplayIndex this.fullResponseContent.length) { this.updateStreamingMessage(messageId, this.fullResponseContent, false); } this.setData({ isLoading: false, isReceiving: false, currentStreamingId: null }); const messages this.data.messages.map(msg { if (msg.id messageId) { return { ...msg, isStreaming: false }; } return msg; }); this.setData({ messages }); this.fullResponseContent ; this.currentDisplayIndex 0; this.buffer ; this.incompleteLine ; }, handleError(errorMsg, aiMessageId) { this.setData({ isLoading: false, isReceiving: false, socketOpen: false, currentStreamingId: null }); if (this.typewriterTimer) { clearInterval(this.typewriterTimer); this.typewriterTimer null; } this.fullResponseContent ; this.currentDisplayIndex 0; this.buffer ; this.incompleteLine ; this.updateStreamingMessage(aiMessageId, 错误${errorMsg}, true); wx.showToast({ title: errorMsg, icon: none, duration: 2000 }); }, closeConnection() { if (this.socket) { this.socket.close(); this.setData({ isReceiving: false, isLoading: false }); } if (this.typewriterTimer) { clearInterval(this.typewriterTimer); this.typewriterTimer null; } }, onUnload() { this.closeConnection(); } });注意前往服务器web-ssl证书下的nginx配置# building websocket location /building_ws { proxy_pass http://www.***.com:8057; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Origin ; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; # 关键添加这些配置确保消息能正常传输 proxy_set_header Content-Type application/json; proxy_set_header Content-Length $content_length; proxy_connect_timeout 60s; proxy_send_timeout 3600s; proxy_read_timeout 3600s; proxy_buffering off; proxy_cache off; # 关键禁用请求体缓冲 proxy_request_buffering off; client_max_body_size 10m; }接下来是nodeJsexpress的服务器js:app.js:const express require(express); const http require(http); const app express(); const { db } require(./config/database); //连接数据库 // 创建HTTP服务器 const server http.createServer(app); // 引入websocket.js const { initWebSocket } require(./websocket); // 初始化WebSocket服务器 initWebSocket(server); // 小程序 const picMiniRouter require(./routes/miniprogram/pic); const userMiniRouter require(./routes/miniprogram/user); const hobbyMiniRouter require(./routes/miniprogram/hobby); // 后台Vue const userAdminRouter require(./routes/vue-element-admin/user); const roadAdminRouter require(./routes/vue-element-admin/road); const dakaAdminRouter require(./routes/vue-element-admin/daka); const dakaorderAdminRouter require(./routes/vue-element-admin/dakaorder); const dakadistanceAdminRouter require(./routes/vue-element-admin/dakadistance); const sheshiAdminRouter require(./routes/vue-element-admin/sheshi); const sheshiorderAdminRouter require(./routes/vue-element-admin/sheshiorder); const articleAdminRouter require(./routes/vue-element-admin/article); const deviceAdminRouter require(./routes/vue-element-admin/device); const deviceWsAdminRouter require(./routes/vue-element-admin/device_ws); // 添加这两行中间件 app.use(express.json()); // 解析application/json app.use(express.urlencoded({ extended: true })); // 解析表单数据 // 配置白名单不需要Token验证的接口 const publicPaths [ // 小程序 /api/miniprogram/user/login, /api/miniprogram/user/logouot, /api/miniprogram/pic/getpic, /api/miniprogram/pic/upload_daily, /api/miniprogram/hobby/*, // 后台Vue /api/vue-element-admin/user/login, /api/vue-element-admin/user/logout, /api/vue-element-admin/device/parse_tdms, /api/vue-element-admin/device/channel_data, ]; // Token验证中间件 async function verifyToken(req, res, next) { let token ; // 1. 从Authorization头获取令牌 // 获取小程序的token const authHeader req.headers[authorization] || req.headers[Authorization]; if (!authHeader) { // console.log(未提供Authorization头); // console.log(req.headers); // 获取后台Vue的Token const xToken req.headers[X-Token] || req.headers[x-token]; // console.log(X-Token:, xToken); if (!xToken) { return res.status(401).json({ code: 401, message: 未提供认证令牌, error: Missing Authorization header, }); } token xToken; } else { // console.log(提供Authorization头); // 小程序的token // 2. 检查是否为Bearer Token console.log(Authorization头:, authHeader); const parts authHeader.split( ); if (parts.length ! 2) { return res.status(401).json({ code: 401, message: 认证令牌错误请重新登录, error: Missing Authorization header, }); } else { if (parts[0] Bearer) { if (parts[1] ) { return res.status(401).json({ code: 401, message: 未提供认证令牌, error: Missing Authorization header, }); } } else { return res.status(401).json({ code: 401, message: 令牌格式错误应为 Bearer token, error: Invalid token format, }); } } token parts[1]; } // connection await db.getConnection(); const [userTokenRows] await db.execute( SELECT * FROM facility_person_token WHERE access_token ?, [token] ); if (userTokenRows.length 0) { return res.status(401).json({ success: false, message: 令牌过期请重新登录, }); } // 检查update_time是否在2小时内 const userToken userTokenRows[0]; const updateTime new Date(userToken.update_time); const currentTime new Date(); const timeDiff currentTime - updateTime; const twoHours 2 * 60 * 60 * 1000; // 2小时的毫秒数 if (timeDiff twoHours) { // 超过2小时删除该记录 使用delete语句 await db.execute( DELETE FROM facility_person_token WHERE access_token ?, [token] ); return res.status(401).json({ success: false, code:401, message: access_token已过期请重新登录, }); } else { // 小于2小时就更新update_time const update_time new Date().toLocaleString(); await db.execute( UPDATE facility_person_token SET update_time? WHERE access_token ?, [update_time, token] ); } const uid userToken.uid; // 从数据库中查询用户信息 const [userRows] await db.execute( SELECT * FROM facility_person WHERE id ?, [uid] ); if (userRows.length 0) { return res.status(401).json({ success: false, message: 用户不存在, }); } const user userRows[0]; req.user user; next(); } // 智能验证中间件核心逻辑 function smartAuthMiddleware(req, res, next) { // 检查当前路径是否在白名单中 const isPublicPath publicPaths.some((path) { // 精确匹配 if (req.path path) return true; // 前缀匹配支持 /api/daka/public/* 这种模式 if (path.endsWith(/*)) { const basePath path.slice(0, -2); return req.path.startsWith(basePath); } // 部分匹配比如 /api/daka/public 匹配 /api/daka/public/xxx if (req.path.startsWith(path /)) return true; return false; }); // 调试信息开发时使用 console.log( 请求路径: ${req.path}, 方法: ${req.method}, 是否公开: ${isPublicPath} ); if (isPublicPath) { // 公共接口跳过Token验证 console.log(跳过Token验证: ${req.path}); return next(); } // 需要Token验证的接口 console.log(需要Token验证: ${req.path}); return verifyToken(req, res, next); } // 应用中间件 // 注意smartAuthMiddleware 必须在路由之前应用 app.use(smartAuthMiddleware); // 路由定义 // 根路由在白名单中 // 注册打卡模块路由 // 小程序 app.use(/api/miniprogram/hobby, hobbyMiniRouter); app.use(/api/miniprogram/user, userMiniRouter); app.use(/api/miniprogram/pic, picMiniRouter); // 后台vue app.use(/api/vue-element-admin/user, userAdminRouter); app.use(/api/vue-element-admin/road, roadAdminRouter); app.use(/api/vue-element-admin/daka, dakaAdminRouter); app.use(/api/vue-element-admin/dakaorder, dakaorderAdminRouter); app.use(/api/vue-element-admin/dakadistance, dakadistanceAdminRouter); app.use(/api/vue-element-admin/sheshi, sheshiAdminRouter); app.use(/api/vue-element-admin/sheshiorder, sheshiorderAdminRouter); app.use(/api/vue-element-admin/article, articleAdminRouter); app.use(/api/vue-element-admin/device, deviceAdminRouter); app.use(/api/vue-element-admin/device_ws, deviceWsAdminRouter); // 404处理 app.use((req, res) { res.status(404).json({ code: 404, message: 接口 ${req.method} ${req.path} 不存在, error: Not Found, }); }); // 错误处理中间件 app.use((err, req, res, next) { console.error(服务器错误:, err.stack); res.status(500).json({ code: 500, message: 服务器内部错误, error: process.env.NODE_ENV development ? err.message : undefined, }); }); // 启动服务器 // const PORT process.env.PORT || 8060; // app.listen(PORT, () { // console.log(✅ Server running on port ${PORT}); // console.log( 白名单配置:); // publicPaths.forEach((path) { // console.log( - ${path}); // }); // }); server.listen(8057, () { console.log(✅ 服务器已启动http://localhost:8057); console.log(✅ WebSocket 已启用ws://localhost:8057); console.log( 白名单配置:); publicPaths.forEach((path) { console.log( - ${path}); }); }); // 导出WebSocket相关实例供路由文件使用websocket.js:const WebSocket require(ws); const axios require(axios); // 创建全局 WebSocket 服务器实例 let wss null; let vueClients new Set(); // 初始化 WebSocket 服务器 function initWebSocket(server) { if (!wss) { wss new WebSocket.Server({ server }); // WebSocket 连接处理 wss.on(connection, (ws) { console.log(Vue客户端已连接); vueClients.add(ws); console.log(当前连接的Vue客户端数量:, vueClients.size); ws.on(close, () { vueClients.delete(ws); }); ws.on(error, (error) { console.error(Vue客户端错误:, error); }); // 处理客户端消息 -- 流式API代理路由 ws.on(message, async (message) { const { messages, api_key } JSON.parse(message); const response await axios({ method: POST, url: https://www.***.com/bigmodel_api/v1/chat/completions, headers: { Content-Type: application/json, Authorization: Bearer ${api_key || sk-****} }, data: { model: Qwen/Qwen3-8B, messages: messages, stream: true, temperature: 0.7, max_tokens: 2048 }, responseType: stream }); response.data.on(data, (chunk) { // 解析并发送每个chunk ws.send(chunk.toString()); }); response.data.on(end, () { ws.send([DONE]); ws.close(); }); }); // 处理客户端消息 -- 流式API代理路由 end // 发送消息给客户端说注意了 使用json格式 ws.send(JSON.stringify({ type: notice, message: 我是从服务器发送的WebSocket信息注意了 })); }); } return wss; } // 获取 WebSocket 服务器实例 function getWebSocketServer() { return wss; } // 获取 Vue 客户端集合 function getVueClients() { return vueClients; } module.exports { initWebSocket, getWebSocketServer, getVueClients };