feat(ai): 后台 AI 链路改走 HTTP 流式,WS 地址与心跳收敛到 utils

写作 / 图片 / 大纲 / 稿件四条链路用普通 POST 时,生成期间链路上零字节,
会被 60s 空闲超时掐成 504(出图实测 58s 断开)。后端已给每条链路开
/stream(SSE,每 15s 一帧心跳),这里改成消费它。

- 新增 utils/ai-stream.ts:SSE 帧解析 + 末帧 {type:'final', result} 拆包;
  runCreativeAssistantStream / runManuscriptGenStream 对外仍返回原来的 data,
  页面渲染逻辑不动
- 新增 utils/chat-socket.ts:buildWebSocketUrl 统一拼 /api/chat/{userId}
  (网关只放行 /api 前缀),attachHeartbeat 每 25s 发 ping 顶住代理空闲超时
- chatgpt 对话先试 WebSocket,握手失败落 /ai/chat/message/stream,
  帧内容与 WS 完全一致,处理函数直接复用
- 四个页面删掉各自重复的 buildWebSocketUrl,统一从 utils 引入
This commit is contained in:
2026-10-21 19:18:08 +08:00
parent 038e789cb0
commit 6f7f2f0d9a
7 changed files with 354 additions and 85 deletions
+228
View File
@@ -0,0 +1,228 @@
import { API_BASE_URL, TOKEN_HEADER_NAME } from '@/config/setting';
import { getSign, getToken } from '@/utils/token-util';
/**
* 后台的 HTTP 流式通道(SSE):WebSocket 连不上时的退路。
*
* 生产网关在 `/api` 前面就把 WebSocket 的 `Upgrade` 头丢了,握手永远到不了后端,
* 页面就卡在「正在建立 WebSocket 连接...」。后端因此给每条 AI 链路开了一份
* `/stream` 端点:请求体与原接口一模一样,响应是 `text/event-stream`,
* 每个 `data:` 行就是原来 WebSocket 上的一帧(同一段 JSON)。
*
* 所以页面只要把帧喂给原来的消息处理函数,落哪条通道都无感。
*/
const TENANT_ID = 10049;
export interface StreamOptions {
/** 相对 `/api` 的路径,如 `/ai/activityOutlineGen/stream` */
path: string;
data?: Record<string, unknown> | null;
/** 每帧回调,内容与 WebSocket 帧一致 */
onMessage: (raw: string) => void;
/** 中止生成(页面上的「停止」按钮) */
signal?: AbortSignal;
}
/** 发一条流式请求,把每个 SSE 帧原样交给 onMessage。 */
export async function streamAiRequest(options: StreamOptions): Promise<void> {
const payload: Record<string, unknown> = {
tenantId: TENANT_ID,
...(options.data || {})
};
// 与 axios 拦截器保持一致:后端校验 sign
payload.sign = getSign(payload);
const token = getToken();
const headers: Record<string, string> = {
'Content-Type': 'application/json'
};
if (token) {
headers[TOKEN_HEADER_NAME] = token;
}
const response = await fetch(`${API_BASE_URL}${options.path}`, {
method: 'POST',
headers,
body: JSON.stringify(payload),
signal: options.signal
});
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 decoder = new TextDecoder('utf-8');
let buffer = '';
const drain = () => {
let boundary = buffer.indexOf('\n\n');
while (boundary !== -1) {
const frame = buffer.slice(0, boundary);
buffer = buffer.slice(boundary + 2);
boundary = buffer.indexOf('\n\n');
if (frame.startsWith(':')) {
// SSE 注释行(后端用它把响应头先顶出去)
continue;
}
const data = frame
.split('\n')
.filter((line) => line.startsWith('data:'))
.map((line) => line.slice(5).replace(/^ /, ''))
.join('');
if (data) {
options.onMessage(data);
}
}
};
for (;;) {
const { done, value } = await reader.read();
if (done) {
break;
}
buffer += decoder.decode(value, { stream: true });
drain();
}
buffer += decoder.decode();
drain();
}
/** 对话(HTTP 流式) */
export function streamChatMessage(
data: Record<string, unknown>,
onMessage: (raw: string) => void,
signal?: AbortSignal
): Promise<void> {
return streamAiRequest({
path: '/ai/chat/message/stream',
data,
onMessage,
signal
});
}
/** 稿件生成(HTTP 流式) */
export function streamManuscriptGen(
data: Record<string, unknown>,
onMessage: (raw: string) => void,
signal?: AbortSignal
): Promise<void> {
return streamAiRequest({
path: '/ai/manuscriptGen/stream',
data,
onMessage,
signal
});
}
/** 活动策划大纲(HTTP 流式) */
export function streamActivityOutlineGen(
data: Record<string, unknown>,
onMessage: (raw: string) => void,
signal?: AbortSignal
): Promise<void> {
return streamAiRequest({
path: '/ai/activityOutlineGen/stream',
data,
onMessage,
signal
});
}
/** 创意助手(HTTP 流式) */
export function streamCreativeAssistant(
data: Record<string, unknown>,
onMessage: (raw: string) => void,
signal?: AbortSignal
): Promise<void> {
return streamAiRequest({
path: '/ai/creativeAssistant/stream',
data,
onMessage,
signal
});
}
/**
* 创意助手(出图/出视频):跑完才有结果,中途一个字节都不推。
*
* 直接用普通 POST 的话,出图那一两分钟链路上完全空闲,会在 60s 左右被网关
* 掐成 504 Gateway Time-out(实测 58s 断开)。走流式后后端每 15s 发一帧心跳
* 续命,真正的结果放在末帧 `{type:'final', result}` 里,这里拆包后原样返回
* 普通接口的 data 字段,页面的渲染逻辑不用改。
*/
export async function runCreativeAssistantStream(
data: Record<string, unknown>,
signal?: AbortSignal
): Promise<Record<string, any> | null> {
let finalResult: any = null;
await streamCreativeAssistant(
data,
(raw) => {
try {
const frame = JSON.parse(raw);
if (frame && frame.type === 'final') {
finalResult = frame.result;
}
} catch {
// 业务帧解析失败就忽略:真正的失败原因由末帧的 message 带出来
}
},
signal
);
if (!finalResult) {
throw new Error('创意助手没有返回结果,请稍后重试');
}
if (Number(finalResult.code) !== 0) {
throw new Error(finalResult.message || '创意助手处理失败');
}
return finalResult.data ?? null;
}
/**
* 稿件生成:正文逐段推帧,末帧 `{type:'final', result}` 带结构化结果。
*
* 自由创作(不选文物体裁)在后端会转「AI 创意助手」,整段**不推正文**,
* 只有末帧有内容;模板模式则是正文边推、末帧给标题等结构。
* 两种都按同一条路走:帧回调照常转发,末帧拆出来当普通接口的返回值。
*
* @param onMessage 业务帧回调(正文片段),可不传
* @returns 结构化结果(普通接口 data 字段);失败返回 null
*/
export async function runManuscriptGenStream(
data: Record<string, unknown>,
signal?: AbortSignal,
onMessage?: (raw: string) => void
): Promise<Record<string, any> | null> {
let finalResult: any = null;
await streamManuscriptGen(
data,
(raw) => {
try {
const frame = JSON.parse(raw);
if (frame && frame.type === 'final') {
finalResult = frame.result;
return;
}
} catch {
// 非 JSON 帧按业务帧处理
}
if (onMessage) {
onMessage(raw);
}
},
signal
);
if (!finalResult) {
return null;
}
if (Number(finalResult.code) !== 0) {
throw new Error(finalResult.message || '稿件生成失败');
}
return finalResult.data ?? null;
}
+53
View File
@@ -0,0 +1,53 @@
import { API_BASE_URL } from '@/config/setting';
/**
* 后台与大屏端(Vue)的聊天 WebSocket 细节,集中在这里。
*
* 四个页面(创意助手、活动策划大纲、稿件生成、公文校验)此前各自抄了一份
* 「拼 URL + 连 + 断」,路径改一次要改四处;现在统一走这里。
*
* 路径是 **`/api/chat/{userId}`**,不是 `/chat/{userId}`:生产网关只放行 `/api` 前缀,
* 老路径的握手请求到不了后端(实测 404)。后端同时保留了根路径端点给老版本小程序兜底,
* 见 `tuanwei-java` 的 `WebSocketConfig`。
*/
/** 心跳间隔。nginx / 网关的空闲超时普遍在 60s 上下,25s 留够余量。 */
const HEARTBEAT_INTERVAL = 25000;
export function buildWebSocketUrl(userId: number | string): string {
if (/^https?:\/\//.test(API_BASE_URL)) {
const url = new URL(API_BASE_URL);
url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
url.pathname = url.pathname.replace(/\/api\/?$/, '');
return `${url.toString().replace(/\/$/, '')}/api/chat/${userId}`;
}
// 相对路径(同源反代):后台站点自己的域名下同样挂 /api/chat
const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
return `${protocol}//${window.location.host}/api/chat/${userId}`;
}
/**
* 每 25s 发一个 `ping`,顶住反向代理与网关的空闲断开。
*
* 只发不收:页面把收到的每条消息当流式片段解析,服务端若回 pong 会被拼进正文。
* 服务端的 `@OnMessage` 收到文本直接丢弃。
*
* 连接关闭后定时器会自行结束(下次触发时发现 readyState 已是 CLOSED),
* 调用方不需要单独清理。
*/
export function attachHeartbeat(socket: WebSocket): () => void {
const timer = window.setInterval(() => {
if (socket.readyState === WebSocket.OPEN) {
try {
socket.send('ping');
} catch {
// 发送失败说明连接正在断,交给下一次 tick 或 onclose 处理
}
return;
}
if (socket.readyState === WebSocket.CLOSED) {
window.clearInterval(timer);
}
}, HEARTBEAT_INTERVAL);
return () => window.clearInterval(timer);
}
@@ -146,7 +146,7 @@
import ByteMdViewer from '@/components/ByteMdViewer/index.vue';
import { reviewWxxzxProjectBookQuality } from '@/api/gxmu/wxxzx-form';
import { stopMessage } from '@/api/oa/chatgpt';
import { API_BASE_URL } from '@/config/setting';
import { attachHeartbeat, buildWebSocketUrl } from '@/utils/chat-socket';
import { useUserStore } from '@/store/modules/user';
import { copyText } from '@/utils/common';
import type { WxxzxProjectQualityReviewResult } from '@/api/gxmu/wxxzx-form/model';
@@ -249,17 +249,6 @@
});
};
const buildWebSocketUrl = (userId: number) => {
if (/^https?:\/\//.test(API_BASE_URL)) {
const url = new URL(API_BASE_URL);
url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
url.pathname = url.pathname.replace(/\/api\/?$/, '');
return `${url.toString().replace(/\/$/, '')}/chat/${userId}`;
}
const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
return `${protocol}//${window.location.host}/chat/${userId}`;
};
const clearStreamResolver = () => {
streamResolverRef.value = null;
};
@@ -402,6 +391,10 @@
const connectPromise = new Promise<void>((resolve, reject) => {
let settled = false;
const socket = new WebSocket(buildWebSocketUrl(userId));
socket.onopen = () => {
// 握手成功后再挂心跳:连接没建起来时发出去只会失败
attachHeartbeat(socket);
};
cleanupPending = () => {
if (wsConnectPromiseRef.value === connectPromise) {
wsConnectPromiseRef.value = null;
+19 -27
View File
@@ -198,11 +198,12 @@
import { asBlob } from 'html-docx-js-typescript';
import ByteMdEditor from '@/components/ByteMdEditor/index.vue';
import { activityOutlineGen } from '@/api/ai/activityOutlineGen';
import { streamActivityOutlineGen } from '@/utils/ai-stream';
import type {
ActivityOutlineGenParams,
ActivityOutlineGenResponse
} from '@/api/ai/activityOutlineGen/model';
import { API_BASE_URL } from '@/config/setting';
import { attachHeartbeat, buildWebSocketUrl } from '@/utils/chat-socket';
import { useUserStore } from '@/store/modules/user';
import { copyText } from '@/utils/common';
import { resolveDisplayText, isFailedPayload, pickDisplayable, FAILED_TEXT } from '@/utils/ai-result';
@@ -422,17 +423,6 @@
return parts.join(' | ');
};
const buildWebSocketUrl = (userId: number) => {
if (/^https?:\/\//.test(API_BASE_URL)) {
const url = new URL(API_BASE_URL);
url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
url.pathname = url.pathname.replace(/\/api\/?$/, '');
return `${url.toString().replace(/\/$/, '')}/chat/${userId}`;
}
const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
return `${protocol}//${window.location.host}/chat/${userId}`;
};
const clearStreamResolver = () => {
streamResolverRef.value = null;
};
@@ -559,6 +549,10 @@
const connectPromise = new Promise<void>((resolve, reject) => {
let settled = false;
const socket = new WebSocket(buildWebSocketUrl(userId));
socket.onopen = () => {
// 握手成功后再挂心跳:连接没建起来时发出去只会失败
attachHeartbeat(socket);
};
cleanupPending = () => {
if (wsConnectPromiseRef.value === connectPromise) {
wsConnectPromiseRef.value = null;
@@ -637,23 +631,21 @@
}
loading.value = true;
clearResult();
resultSummary.value = '正在建立 WebSocket 连接...';
resultSummary.value = '正在提交生成任务...';
try {
await connectWebSocket();
resultSummary.value = '连接成功,已提交生成任务,等待流式返回...';
// 一律走 HTTP 流式(/stream),不再先连 WebSocket:
// 触发生成的那个 POST 在整段生成期间是零字节的,会被链路上的 60s 空闲
// 超时掐成 504 Gateway Time-out。/stream 每 15s 有后端心跳续命。
const payload = buildPayload();
resultSummary.value = '已提交生成任务,等待流式返回...';
const waitStream = createStreamWaiter();
const data = await activityOutlineGen(buildPayload());
result.value = data || null;
if (!streamStarted.value && data && parseOutline(data)) {
outlineMarkdown.value = parseOutline(data);
syncEditableOutline();
resultSummary.value = buildSummary() || '接口直接返回生成结果';
loading.value = false;
resolveStream();
disconnectWebSocket();
} else {
await waitStream;
}
await streamActivityOutlineGen(
payload as unknown as Record<string, unknown>,
(raw) => handleSocketMessage(raw)
);
// 兜底:流结束但没收到 __END__ 时也别把页面卡在加载中
resolveStream();
await waitStream;
if (!visibleOutlineMarkdown.value) {
message.warning('接口已返回,但未解析到正文内容');
return;
+3 -2
View File
@@ -189,7 +189,7 @@
import { computed, ref } from 'vue';
import { message } from 'ant-design-vue';
import ByteMdViewer from '@/components/ByteMdViewer/index.vue';
import { creativeAssistant } from '@/api/ai/creativeAssistant';
import { runCreativeAssistantStream } from '@/utils/ai-stream';
import type { FileRecord } from '@/api/system/file/model';
import type {
CreativeAssistantOutputs,
@@ -404,7 +404,8 @@
}
loading.value = true;
try {
const result = await creativeAssistant(buildRequestPayload());
// 出图中途零字节,普通 POST 会被网关按 60s 空闲超时掐成 504,改走流式
const result = await runCreativeAssistantStream(buildRequestPayload());
workflowResponse.value = result || null;
workflowOutputs.value = normalizeOutputPayload(result || null);
if (
+23 -23
View File
@@ -229,13 +229,14 @@
import { asBlob } from 'html-docx-js-typescript';
import { getProcessor } from 'bytemd';
import { manuscriptGen } from '@/api/ai/manuscriptGen';
import { runManuscriptGenStream } from '@/utils/ai-stream';
import type {
ManuscriptGenParams,
ManuscriptGenResult
} from '@/api/ai/manuscriptGen/model';
import { uploadFile } from '@/api/system/file';
import type { FileRecord } from '@/api/system/file/model';
import { API_BASE_URL } from '@/config/setting';
import { attachHeartbeat, buildWebSocketUrl } from '@/utils/chat-socket';
import { useUserStore } from '@/store/modules/user';
import { copyText } from '@/utils/common';
import { resolveDisplayText, isFailedPayload, pickDisplayable, FAILED_TEXT } from '@/utils/ai-result';
@@ -560,17 +561,6 @@
return typeof payload === 'object';
};
const buildWebSocketUrl = (userId: number) => {
if (/^https?:\/\//.test(API_BASE_URL)) {
const url = new URL(API_BASE_URL);
url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
url.pathname = url.pathname.replace(/\/api\/?$/, '');
return `${url.toString().replace(/\/$/, '')}/chat/${userId}`;
}
const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
return `${protocol}//${window.location.host}/chat/${userId}`;
};
const clearStreamResolver = () => {
streamResolverRef.value = null;
};
@@ -696,6 +686,7 @@
}
};
socket.onopen = () => {
attachHeartbeat(socket);
wsConnected.value = false;
};
socket.onmessage = (event) => {
@@ -859,25 +850,34 @@
loading.value = true;
clearGenerated();
generatedTitle.value = form.theme.trim();
responseSummary.value = '正在建立 WebSocket 连接...';
responseSummary.value = '正在提交生成任务...';
try {
await connectWebSocket();
// 一律走 HTTP 流式(/stream),不再先连 WebSocket:
// 触发生成的那个 POST 在整段生成期间是零字节的,会被链路上的 60s 空闲
// 超时掐成 504 Gateway Time-out(自由创作、不选文体时最先撞上)。
// /stream 每 15s 有后端心跳续命,正文逐段喂给原来的帧处理,
// 结构化结果从末帧 {type:'final'} 取。
const payload = buildPayload();
responseSummary.value = templateMode.value
? '连接成功,已提交生成任务,等待流式返回...'
? '已提交生成任务,等待流式返回...'
: '自由创作已提交给 AI 创意助手,生成完成后一次性返回...';
const waitStream = createStreamWaiter();
const result = await manuscriptGen(buildPayload());
if (!streamStarted.value && shouldTreatAsDirectResult(result)) {
const result = await runManuscriptGenStream(
payload as unknown as Record<string, unknown>,
undefined,
(raw) => handleSocketMessage(raw)
);
if (result) {
applyParsedResult(
typeof result === 'string' ? parseSerializedResult(result) : (result as ManuscriptGenResult),
typeof result === 'string'
? parseSerializedResult(result)
: (result as ManuscriptGenResult),
'接口直接返回生成结果'
);
loading.value = false;
resolveStream();
disconnectWebSocket();
} else {
await waitStream;
}
// 兜底:流结束但没收到 __END__ 时也别把页面卡在加载中
resolveStream();
await waitStream;
message.success('内容生成完成');
} catch (e: any) {
disconnectWebSocket();
+23 -21
View File
@@ -389,15 +389,14 @@ import type { AiChatHistory } from "@/api/ai/aiChatHistory/model";
import { listAiChatList, removeAiChatList } from "@/api/ai/aiChatList";
import type { AiChatList } from "@/api/ai/aiChatList/model";
import { sendMessage, stopMessage } from "@/api/oa/chatgpt";
import { manuscriptGen } from "@/api/ai/manuscriptGen";
import { runCreativeAssistantStream, runManuscriptGenStream, streamChatMessage } from "@/utils/ai-stream";
import {
WRITING_DOC_TYPE_FREE,
WRITING_DOC_TYPE_OPTIONS,
WRITING_EVENT_TYPE_OPTIONS,
WRITING_WORD_COUNT_OPTIONS
} from "./writing-params";
import { creativeAssistant } from "@/api/ai/creativeAssistant";
import { API_BASE_URL } from "@/config/setting";
import { attachHeartbeat, buildWebSocketUrl } from "@/utils/chat-socket";
import type { Menu } from "@/api/system/menu/model";
import { useUserStore } from "@/store/modules/user";
import { logout } from "@/utils/page-tab-util";
@@ -785,17 +784,6 @@ const findFirstChildMenuPath = (menus: AuthorityTreeItem[]) => {
return "";
};
const buildWebSocketUrl = (userId: number) => {
if (/^https?:\/\//.test(API_BASE_URL)) {
const url = new URL(API_BASE_URL);
url.protocol = url.protocol === "https:" ? "wss:" : "ws:";
url.pathname = url.pathname.replace(/\/api\/?$/, "");
return `${url.toString().replace(/\/$/, "")}/chat/${userId}`;
}
const protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
return `${protocol}//${window.location.host}/chat/${userId}`;
};
const formatHistoryMessages = (list: AiChatHistory[]) => {
const result: ChatMessageItem[] = [];
[...list]
@@ -972,6 +960,7 @@ const connectWebSocket = async () => {
}
};
socket.onopen = () => {
attachHeartbeat(socket);
wsConnected.value = false;
};
socket.onmessage = (event) => {
@@ -1335,7 +1324,8 @@ const sendModeMessage = async (
}
};
if (activeChatMode.value === "writing") {
const result = await manuscriptGen(buildWritingPayload(query), { signal });
// 稿件生成同样会被链路的 60s 空闲超时掐掉,改走流式(正文边推边显示)
const result = await runManuscriptGenStream(buildWritingPayload(query), signal);
if (typeof (result as any)?.taskId === "string") {
currentTaskId.value = (result as any).taskId;
}
@@ -1343,7 +1333,8 @@ const sendModeMessage = async (
return;
}
if (activeChatMode.value === "image") {
const result = await creativeAssistant(buildImagePayload(query), { signal });
// 出图中途零字节,普通 POST 会被网关按 60s 空闲超时掐成 504,改走流式
const result = await runCreativeAssistantStream(buildImagePayload(query), signal);
if (typeof (result as any)?.taskId === "string") {
currentTaskId.value = (result as any).taskId;
}
@@ -1471,13 +1462,24 @@ const onSend = async () => {
content.value = "";
scrollToBottom();
if (activeChatMode.value === "chat") {
await connectWebSocket();
await sendMessage({
// 先试 WebSocket;生产网关在 /api 前丢 Upgrade 头,握手必然失败,
// 这时落到 HTTP 流式端点,帧内容与 WS 完全一致,解析逻辑不用改
const payload = {
query,
conversationId: activeConversationId.value
}, {
signal: abortController.signal
});
};
const viaSocket = await connectWebSocket()
.then(() => true)
.catch(() => false);
if (viaSocket) {
await sendMessage(payload, { signal: abortController.signal });
} else {
await streamChatMessage(
payload as Record<string, unknown>,
(raw) => void handleSocketMessage(raw),
abortController.signal
);
}
} else {
await sendModeMessage(query, assistantId, abortController.signal);
streamingMessageId.value = undefined;