feat(ai): 小程序 AI 链路改走 HTTP 流式,WS 细节收敛到 utils/chat-socket

与 web 端同因:生成期间普通 POST 零字节,会被 60s 空闲超时掐成 504。

- 新增 utils/chat-socket.js:地址统一拼 /api/chat/{userId}(网关只放行 /api),
  attachHeartbeat 每 25s 发 ping;三个页面不再各抄一份
- chat-service 新增 streamAiRequest:H5 走 fetch + ReadableStream,小程序走
  enableChunked + onChunkReceived,并手写 UTF-8 分块解码,避免中文被切半
- 写作 / 图片 / 大纲 / 稿件改走 /stream,结构化结果从末帧 {type:'final'} 取;
  流断但有正文时保留已生成部分,不整块覆盖成错误文案
- 对话仍先试 WebSocket,连不上回退 /ai/chat/message/stream,帧处理复用
  handleSocketMessage
This commit is contained in:
2026-10-21 19:18:13 +08:00
parent 5ce66be025
commit d449abe966
7 changed files with 456 additions and 53 deletions
+25 -16
View File
@@ -158,8 +158,10 @@
<script>
import {
activityOutlineGen,
attachHeartbeat,
buildWebSocketUrl,
getCurrentUser
getCurrentUser,
streamActivityOutlineGen
} from './service'
import {
isFailedPayload,
@@ -195,6 +197,7 @@
activeMessageId: '',
socketTask: null,
socketConnectPromise: null,
socketHeartbeatStop: null,
streamStarted: false,
streamFailed: false,
streamResolver: null,
@@ -364,6 +367,10 @@
})
},
disconnectWebSocket() {
if (this.socketHeartbeatStop) {
this.socketHeartbeatStop()
this.socketHeartbeatStop = null
}
this.socketConnectPromise = null
const socketTask = this.socketTask
this.socketTask = null
@@ -471,6 +478,10 @@
}
}
socketTask.onOpen(() => {
// 握手成功后再挂心跳:连接没建起来时发出去只会失败
this.socketHeartbeatStop = attachHeartbeat(socketTask)
})
socketTask.onMessage((event) => {
const raw = String((event && event.data) || '')
if (raw === SOCKET_CONNECTED_TEXT) {
@@ -539,23 +550,21 @@
}
this.loading = true
this.clearResult()
this.resultSummary = '正在建立 WebSocket 连接...'
this.resultSummary = '正在提交生成任务...'
try {
await this.connectWebSocket()
this.resultSummary = '连接成功,已提交生成任务,等待流式返回...'
// 一律走 HTTP 流式(/stream),不再先连 WebSocket:
// 触发生成的那个 POST 在整段生成期间是零字节的,会被链路上的
// 60s 空闲超时掐成 504 Gateway Time-out。/stream 每 15s 有后端
// 心跳续命,帧内容与 WS 完全一致。
const payload = this.buildPayload()
this.resultSummary = '已提交生成任务,等待流式返回...'
const waitStream = this.createStreamWaiter()
const data = await activityOutlineGen(this.buildPayload())
this.result = data || null
if (!this.streamStarted && data && this.parseOutline(data)) {
this.outlineMarkdown = this.parseOutline(data)
this.syncEditableOutline()
this.resultSummary = this.buildSummary() || '接口直接返回生成结果'
this.loading = false
this.resolveStream()
this.disconnectWebSocket()
} else {
await waitStream
}
await streamActivityOutlineGen(payload, {
onMessage: (raw) => this.handleSocketMessage(raw)
}).promise
// 兜底:流结束但没收到 __END__ 时也别把页面卡在加载中
this.resolveStream()
await waitStream
if (this.streamFailed) {
// 后端判定工作流失败,正文区已显示失败文案,这里同步提示
this.showToast('生成失败,请稍后重试')
+4 -2
View File
@@ -1,11 +1,13 @@
import {
API_BASE_URL,
apiRequest,
attachHeartbeat,
buildWebSocketUrl,
getCurrentUser
getCurrentUser,
streamActivityOutlineGen
} from '../assistant/chat-service'
export { buildWebSocketUrl, getCurrentUser }
export { attachHeartbeat, buildWebSocketUrl, getCurrentUser, streamActivityOutlineGen }
export function activityOutlineGen(data) {
return apiRequest({
+214 -9
View File
@@ -486,14 +486,219 @@ export function uploadAiFile(file, app) {
})
}
export function buildWebSocketUrl(userId) {
const url = String(API_BASE_URL || '')
if (/^https?:\/\//.test(url)) {
return `${url
.replace(/^https:/, 'wss:')
.replace(/^http:/, 'ws:')
.replace(/\/api\/?$/, '')
.replace(/\/$/, '')}/chat/${userId}`
// WebSocket 的地址拼接与心跳统一在 utils/chat-socket.js,
// 这里继续转发,三个页面的 import 不用动
export { buildWebSocketUrl, attachHeartbeat } from '../../utils/chat-socket'
// ---------------------------------------------------------------------------
// HTTP 流式(SSE)通道:WebSocket 握手被网关挡掉时的退路
//
// 生产网关在 /api 前面丢 Upgrade 头,wss 建不起来(网关侧不让改)。后端因此
// 把每条 AI 链路都开了一份 /stream 端点:请求体与原接口一字不差,响应是
// text/event-stream,每个 data: 行就是原来 WebSocket 上的一帧(同一段 JSON)。
// 端上先试 WebSocket,连不上就落到这里,页面处理逻辑完全复用 handleSocketMessage。
// ---------------------------------------------------------------------------
/** 把字节流切成字符串,并返回「还没凑齐的多字节尾部」留给下一块。 */
function decodeUtf8Chunk(bytes) {
let text = ''
let index = 0
while (index < bytes.length) {
const first = bytes[index]
if (first < 0x80) {
text += String.fromCharCode(first)
index += 1
continue
}
let extra = 0
if ((first & 0xe0) === 0xc0) {
extra = 1
} else if ((first & 0xf0) === 0xe0) {
extra = 2
} else if ((first & 0xf8) === 0xf0) {
extra = 3
} else {
// 非法首字节,跳过,避免整条流卡死
index += 1
continue
}
if (index + extra >= bytes.length) {
// 中文字符被切在两块之间:整段留给下一次拼
break
}
let codePoint = first & (0x3f >> extra)
for (let step = 1; step <= extra; step += 1) {
codePoint = (codePoint << 6) | (bytes[index + step] & 0x3f)
}
text += String.fromCodePoint(codePoint)
index += extra + 1
}
return `${url.replace(/\/api\/?$/, '')}/chat/${userId}`
return { text, consumed: index }
}
/**
* 发一条流式请求,把每个 SSE 帧原样交给 onMessage。
*
* @param {object} options
* @param {string} options.url 后端 `/stream` 端点
* @param {object} options.data 与普通接口相同的请求体
* @param {(raw: string) => void} options.onMessage 每帧回调,内容与 WebSocket 帧一致
* @param {() => void} [options.onDone] 流正常结束
* @param {(error: Error) => void} [options.onError] 流异常结束
* @returns {{ abort: () => void, promise: Promise<void> }}
*/
export function streamAiRequest({ url, data, onMessage, onDone, onError, timeout = AI_REQUEST_TIMEOUT }) {
const requestData = data ? { ...data } : {}
requestData.tenantId = DEFAULT_TENANT_ID
addSignature(requestData)
const token = getToken()
const header = { 'Content-Type': 'application/json' }
if (token) {
header[TOKEN_HEADER_NAME] = token
}
let pending = new Uint8Array(0)
let textBuffer = ''
let settled = false
let abortHandle = () => undefined
const finish = (error) => {
if (settled) {
return
}
settled = true
if (error) {
onError && onError(error)
return
}
onDone && onDone()
}
const drainFrames = () => {
let boundary = textBuffer.indexOf('\n\n')
while (boundary !== -1) {
const frame = textBuffer.slice(0, boundary)
textBuffer = textBuffer.slice(boundary + 2)
boundary = textBuffer.indexOf('\n\n')
if (frame.startsWith(':')) {
// SSE 注释行(后端用它把响应头先顶出去)
continue
}
const payload = frame
.split('\n')
.filter((line) => line.indexOf('data:') === 0)
.map((line) => line.slice(5).replace(/^ /, ''))
.join('')
if (payload && onMessage) {
onMessage(payload)
}
}
}
const handleBytes = (arrayBuffer) => {
const incoming = new Uint8Array(arrayBuffer)
const merged = new Uint8Array(pending.length + incoming.length)
merged.set(pending, 0)
merged.set(incoming, pending.length)
const decoded = decodeUtf8Chunk(merged)
pending = merged.slice(decoded.consumed)
if (decoded.text) {
textBuffer += decoded.text
drainFrames()
}
}
const promise = new Promise((resolve, reject) => {
const fail = (error) => {
finish(error)
reject(error)
}
// #ifdef H5
const controller = typeof AbortController === 'function' ? new AbortController() : null
abortHandle = () => controller && controller.abort()
fetch(url, {
method: 'POST',
headers: header,
body: JSON.stringify(requestData),
signal: controller ? controller.signal : undefined
})
.then((response) => {
if (!response.ok) {
throw new Error(`流式请求失败(HTTP ${response.status})`)
}
if (!response.body || typeof response.body.getReader !== 'function') {
throw new Error('当前环境不支持流式读取')
}
const reader = response.body.getReader()
const pump = () =>
reader.read().then(({ done, value }) => {
if (done) {
finish()
resolve()
return
}
handleBytes(value)
return pump()
})
return pump()
})
.catch(fail)
// #endif
// #ifndef H5
const task = uni.request({
url,
method: 'POST',
data: requestData,
header,
enableChunked: true,
timeout,
responseType: 'arraybuffer',
success: () => {
// 分块模式下 success 只在流结束时触发,正常完成以这里为准
finish()
resolve()
},
fail: (error) => {
fail(new Error((error && (error.errMsg || error.message)) || '流式请求失败'))
}
})
if (task && typeof task.onChunkReceived === 'function') {
task.onChunkReceived((chunk) => handleBytes(chunk.data))
abortHandle = () => task.abort && task.abort()
} else {
// 老基础库没有分块能力:直接报错,让页面提示升级微信
fail(new Error('当前微信版本不支持流式请求,请升级微信后重试'))
}
// #endif
})
return { abort: () => abortHandle(), promise }
}
/** 对话(HTTP 流式) */
export function streamChatMessage(data, handlers) {
return streamAiRequest({ url: `${API_BASE_URL}/ai/chat/message/stream`, data, ...handlers })
}
/** 稿件生成(HTTP 流式) */
export function streamManuscriptGen(data, handlers) {
return streamAiRequest({ url: `${API_BASE_URL}/ai/manuscriptGen/stream`, data, ...handlers })
}
/** 创意助手(HTTP 流式) */
export function streamCreativeAssistant(data, handlers) {
return streamAiRequest({ url: `${API_BASE_URL}/ai/creativeAssistant/stream`, data, ...handlers })
}
/** 活动策划大纲(HTTP 流式) */
export function streamActivityOutlineGen(data, handlers) {
return streamAiRequest({ url: `${API_BASE_URL}/ai/activityOutlineGen/stream`, data, ...handlers })
}
/** 智能文档处理(HTTP 流式) */
export function streamDocumentProcess(data, handlers) {
return streamAiRequest({ url: `${API_BASE_URL}/ai/documentProcess/stream`, data, ...handlers })
}
+95 -9
View File
@@ -303,13 +303,15 @@
<script>
import dayjs from '../../uni_modules/uv-ui-tools/libs/util/dayjs.js'
import {
attachHeartbeat,
buildWebSocketUrl,
creativeAssistant,
streamCreativeAssistant,
listAiChatHistory,
listAiChatList,
manuscriptGen,
streamManuscriptGen,
removeAiChatList,
sendMessage as sendChatMessage,
streamChatMessage,
} from './chat-service'
import {
WRITING_DOC_TYPE_FREE,
@@ -387,6 +389,7 @@
},
socketTask: null,
socketConnectPromise: null,
socketHeartbeatStop: null,
initialized: false
}
},
@@ -820,6 +823,10 @@
this.syncActiveChat()
},
disconnectWebSocket() {
if (this.socketHeartbeatStop) {
this.socketHeartbeatStop()
this.socketHeartbeatStop = null
}
this.socketConnectPromise = null
this.wsConnected = false
const socketTask = this.socketTask
@@ -887,6 +894,8 @@
socketTask.onOpen(() => {
this.wsConnected = false
// 握手成功后再挂心跳:连接没建起来时发出去只会失败
this.socketHeartbeatStop = attachHeartbeat(socketTask)
})
socketTask.onMessage((event) => {
const raw = String((event && event.data) || '')
@@ -1278,13 +1287,78 @@
}
}
if (this.activeChatMode === 'writing') {
const result = await manuscriptGen(this.buildWritingPayload(query))
setResult(this.formatWritingResult(result))
// 稿件生成同样要挂几十秒到一两分钟,普通 POST 期间链路上零字节,
// 会被 60s 空闲超时掐成 504。走 /stream 后后端每 15s 发心跳续命,
// 正文边推边显示,结构化的最终结果(标题等)放在末帧里。
let finalPayload = null
await streamManuscriptGen(this.buildWritingPayload(query), {
onMessage: (raw) => {
let frame = null
try {
frame = JSON.parse(raw)
} catch (error) {
return
}
if (!frame) {
return
}
if (frame.type === 'final') {
finalPayload = frame.result
return
}
if (frame.answer === '__END__') {
return
}
const chunk = frame.notice || frame.answer
if (chunk && target) {
const base = target.content === ASSISTANT_PENDING_TEXT ? '' : target.content
target.content = base + chunk
}
}
}).promise
if (finalPayload && Number(finalPayload.code) === 0) {
setResult(this.formatWritingResult(finalPayload.data))
return
}
if (finalPayload) {
setResult(finalPayload.message || FAILED_TEXT)
return
}
// 流断了但已经收到正文:保留已生成的部分,不要把整块覆盖成错误文案
if (target && target.content && target.content !== ASSISTANT_PENDING_TEXT) {
setResult(target.content)
return
}
setResult('生成超时了,请稍后重试。')
return
}
if (this.activeChatMode === 'image') {
const result = await creativeAssistant(this.buildImagePayload(query))
setResult(this.formatCreativeResult(result))
// 出图要一两分钟,这期间后端一个字节都不会推。普通 POST 在这段时间里
// 链路上完全空闲,会在 58s 左右被网关掐成 504(实测),所以改走 /stream:
// 后端每 15s 发一帧心跳续命,结果放在末帧 {type:'final', result} 里。
let finalPayload = null
await streamCreativeAssistant(this.buildImagePayload(query), {
onMessage: (raw) => {
let frame = null
try {
frame = JSON.parse(raw)
} catch (error) {
return
}
if (frame && frame.type === 'final') {
finalPayload = frame.result
}
}
}).promise
if (!finalPayload) {
setResult('出图超时了,请稍后重试。')
return
}
if (Number(finalPayload.code) !== 0) {
setResult(finalPayload.message || FAILED_TEXT)
return
}
setResult(this.formatCreativeResult(finalPayload.data))
}
},
openHistoryPanel() {
@@ -1384,11 +1458,23 @@
this.draft = ''
this.scrollToBottom()
if (this.activeChatMode === 'chat') {
await this.connectWebSocket()
await sendChatMessage({
// 先试 WebSocket。生产网关在 /api 前面把 Upgrade 头丢了,握手必然失败;
// 连不上就落到后端的 /stream 端点,帧内容与 WS 完全一致,
// 所以解析逻辑仍然复用 handleSocketMessage。
const viaSocket = await this.connectWebSocket()
.then(() => true)
.catch(() => false)
const payload = {
query,
conversationId: this.activeConversationId || undefined
})
}
if (viaSocket) {
await sendChatMessage(payload)
return
}
await streamChatMessage(payload, {
onMessage: (raw) => this.handleSocketMessage(raw)
}).promise
return
}
await this.sendModeMessage(query, assistantId)
+50 -15
View File
@@ -212,9 +212,11 @@
<script>
import {
attachHeartbeat,
buildWebSocketUrl,
getCurrentUser,
manuscriptGen,
streamManuscriptGen,
uploadTempFile
} from './service'
import {
@@ -261,6 +263,7 @@
activeConversationId: '',
socketTask: null,
socketConnectPromise: null,
socketHeartbeatStop: null,
streamStarted: false,
streamFailed: false,
streamResolver: null,
@@ -521,6 +524,10 @@
})
},
disconnectWebSocket() {
if (this.socketHeartbeatStop) {
this.socketHeartbeatStop()
this.socketHeartbeatStop = null
}
this.socketConnectPromise = null
this.wsConnected = false
const socketTask = this.socketTask
@@ -621,6 +628,8 @@
socketTask.onOpen(() => {
this.wsConnected = false
// 握手成功后再挂心跳:连接没建起来时发出去只会失败
this.socketHeartbeatStop = attachHeartbeat(socketTask)
})
socketTask.onMessage((event) => {
const raw = String((event && event.data) || '')
@@ -784,25 +793,23 @@
this.loading = true
this.clearGenerated()
this.generatedTitle = String(this.form.theme || '').trim()
this.responseSummary = '正在建立 WebSocket 连接...'
this.responseSummary = '正在提交生成任务...'
try {
await this.connectWebSocket()
// 一律走 HTTP 流式(/stream),不再先连 WebSocket:
// 触发生成的那个 POST 在整段生成期间是零字节的,会被链路上的
// 60s 空闲超时掐成 504 Gateway Time-out(自由创作、不选文体时最先撞上)。
// /stream 每 15s 有后端心跳续命,帧内容与 WS 完全一致。
const payload = this.buildPayload()
this.responseSummary = this.freeMode
? '自由创作已提交给 AI 创意助手,生成完成后一次性返回...'
: '连接成功,已提交生成任务,等待流式返回...'
: '已提交生成任务,等待流式返回...'
const waitStream = this.createStreamWaiter()
const result = await manuscriptGen(this.buildPayload())
if (!this.streamStarted && this.shouldTreatAsDirectResult(result)) {
this.applyParsedResult(
typeof result === 'string' ? this.parseSerializedResult(result) : result,
'接口直接返回生成结果'
)
this.loading = false
this.resolveStream()
this.disconnectWebSocket()
} else {
await waitStream
}
await streamManuscriptGen(payload, {
onMessage: (raw) => this.handleStreamFrame(raw)
}).promise
// 兜底:流结束但没收到 __END__ 时也别把页面卡在加载中
this.resolveStream()
await waitStream
if (this.streamFailed) {
// 后端判定工作流失败,正文区已显示失败文案,这里同步提示
this.showToast('生成失败,请稍后重试')
@@ -816,6 +823,34 @@
this.showToast((error && error.message) || '内容生成失败')
}
},
/**
* SSE 帧分发:末帧带着接口的返回值(自由创作模式整段都不推正文,只能靠它),
* 其余帧与 WebSocket 完全一致,继续交给 handleSocketMessage。
*/
handleStreamFrame(raw) {
let frame = null
try {
frame = JSON.parse(raw)
} catch (error) {
this.handleSocketMessage(raw)
return
}
if (!frame || frame.type !== 'final') {
this.handleSocketMessage(raw)
return
}
const result = frame.result
if (result && Number(result.code) === 0 && result.data) {
this.applyParsedResult(
typeof result.data === 'string' ? this.parseSerializedResult(result.data) : result.data,
'接口直接返回生成结果'
)
return
}
if (result && result.message) {
this.showToast(result.message)
}
},
copyGenerated() {
if (!this.markdownContent) {
this.showToast('暂无可复制的正文内容')
+4 -2
View File
@@ -1,12 +1,14 @@
import {
API_BASE_URL,
apiRequest,
attachHeartbeat,
buildWebSocketUrl,
getCurrentUser,
getToken
getToken,
streamManuscriptGen
} from '../assistant/chat-service'
export { buildWebSocketUrl, getCurrentUser }
export { attachHeartbeat, buildWebSocketUrl, getCurrentUser, streamManuscriptGen }
export function manuscriptGen(data) {
return apiRequest({
+64
View File
@@ -0,0 +1,64 @@
import { API_BASE_URL } from './request'
/**
* 端与后端 WebSocket 的连接细节,集中在这里。
*
* 三条链路(创意助手 / 稿件生成 / 活动策划大纲)此前各自散落着一份一模一样的
* 「拼 URL + 连 + 断」,改一次路径要改三处;现在统一走这里。
*/
/** 心跳间隔。nginx 的 proxy_read_timeout / 网关空闲超时普遍在 60s 上下,25s 留够余量。 */
const HEARTBEAT_INTERVAL = 25000
/**
* 拼接聊天 WebSocket 地址。
*
* 路径是 **`/api/chat/{userId}`**,不是 `/chat/{userId}`:
* 生产网关只放行 `/api` 前缀,老路径的握手请求到不了后端(实测 404),
* 这正是「生产连不上、本地却正常」的根因。后端同时保留了根路径端点,
* 给还没更新的老版本小程序兜底,见 `WebSocketConfig`。
*/
export function buildWebSocketUrl(userId) {
const url = String(API_BASE_URL || '')
if (/^https?:\/\//.test(url)) {
return `${url
.replace(/^https:/, 'wss:')
.replace(/^http:/, 'ws:')
.replace(/\/api\/?$/, '')
.replace(/\/$/, '')}/api/chat/${userId}`
}
return `${url.replace(/\/api\/?$/, '')}/api/chat/${userId}`
}
/**
* 给连接挂一个心跳,返回停止函数。
*
* 只发不收:页面把收到的每条消息都当作流式片段解析,服务端回 pong 反而会被
* 拼进正文。心跳的目的只是让「客户端 → 服务端」方向持续有帧,
* 顶住 nginx 的 proxy_send_timeout 与各层空闲探测。
* 服务端的 `@OnMessage` 收到文本直接丢弃(见 `WebSocketServer.onMessage`)。
*
* @param {object} socketTask uni.connectSocket 返回的 SocketTask
* @returns {() => void} 停止心跳
*/
export function attachHeartbeat(socketTask) {
if (!socketTask || typeof socketTask.send !== 'function') {
return () => undefined
}
let stopped = false
const stop = () => {
if (stopped) {
return
}
stopped = true
clearInterval(timer)
}
const timer = setInterval(() => {
socketTask.send({
data: 'ping',
// 连接已经断了:下一次静默失败时顺手把定时器收掉,免得一直空转
fail: () => stop()
})
}, HEARTBEAT_INTERVAL)
return stop
}