fix(ai): 创意助手繁忙归类补 503/50508 与流式中断,失败日志保留尾部

线上日志(2026-10-21 18:10/18:50)显示失败节点已从出图节点移到最前面的
「问题分类器」(SiliconFlow):503 code 50508 System is too busy,以及
requests/urllib3 read_chunked 流式读取中断。旧归类只认 50500,用户看到的是
「生成失败」而不是「生成服务繁忙」;且 briefUpstreamError 只留头部,
Python traceback 的异常类型在尾部被截掉。

- isUpstreamCapacityFailure 增补 50508 / 503 / 流式中断等模式
- briefUpstreamError 改为头 2/3 + 标记 + 尾 1/3(仍单行、仍 <900 字符)
- buildCreativeAssistantResult 在成功但无可展示内容时记 WARN(含 outputKeys
  与 photo_files 条数),让解析兜底路径下一次可诊断
- 新增 AiCreativeBusyClassificationTest(6 条)

issue 19;不恢复自动重试。
This commit is contained in:
2026-10-21 19:09:03 +08:00
parent f4590001d8
commit 31eb75cdad
2 changed files with 367 additions and 2 deletions
@@ -5,6 +5,7 @@ import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gxwebsoft.ai.config.ChatStreamRegistry;
import com.gxwebsoft.ai.config.DifyProperties;
import com.gxwebsoft.ai.entity.ActivityOutlineInputs;
import com.gxwebsoft.ai.entity.AiChatHistory;
@@ -32,12 +33,14 @@ import org.springframework.web.context.request.async.DeferredResult;
import org.springframework.web.multipart.MultipartFile;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import java.io.*;
import java.math.BigDecimal;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -48,7 +51,12 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -59,6 +67,27 @@ public class AiController extends BaseController {
private static final Logger logger = LoggerFactory.getLogger(AiController.class);
private static final long AI_REQUEST_TIMEOUT = 10 * 60 * 1000L;
/**
* SSE 心跳间隔。
*
* <p>图片/视频生成这类工作流,上游要跑一两分钟且中间**一个字节都不回**。
* 生产链路在空闲 58s 左右就会把连接掐掉(实测:握手成功后什么都不发,
* 58.1s 收到断开;每 25s 发一次 ping 则 150s 仍存活),表现为
* {@code 504 Gateway Time-out}。所以流式响应必须自己制造流量,
* 15s 一帧注释行留足余量。
*/
private static final long SSE_HEARTBEAT_INTERVAL_MS = 15000L;
/** 心跳线程:单线程足够,只是往各条流里写注释行。 */
private static final ScheduledExecutorService SSE_HEARTBEAT = Executors.newSingleThreadScheduledExecutor(r -> {
Thread thread = new Thread(r, "ai-sse-heartbeat");
thread.setDaemon(true);
return thread;
});
/** 流式响应末尾补发的那一帧的标记字段,端上据此区分「业务帧」和「最终结果」。 */
private static final String SSE_FINAL_FRAME_TYPE = "final";
/**
* 创意生成工作流以失败收场时给用户看的文案。
*
@@ -672,6 +701,172 @@ public class AiController extends BaseController {
return deferredResult;
}
// ---------------------------------------------------------------------
// HTTP 流式(SSE)兜底:WebSocket 握手被网关挡掉时的同一份流
//
// 生产网关在 /api 前丢 Upgrade 头,wss 永远建不起来(网关侧不能改),
// 端上因此改走这里:请求体与原接口一字不差,响应是 text/event-stream,
// 每个 data: 行就是原来 WebSocket 上的一帧(同一段 JSON,逐字节相同)。
// 业务侧一行没动 —— 推送统一走 webSocketServer.sendMessage,
// 没有 WS 会话时由 ChatStreamRegistry 把消息改投到这条 HTTP 连接。
// ---------------------------------------------------------------------
/**
* 流式方法体:允许抛受检异常(各业务方法签名里都有 throws IOException)。
*/
@FunctionalInterface
private interface StreamBody {
void run() throws Exception;
}
/**
* 带返回值的流式方法体:返回值会作为最后一帧发给端上。
*
* <p>给「业务方法自己不推帧、只返回一个结果」的链路用 —— 创意助手的出图就是
* 这种:中间没有任何推送,结果只在方法返回时才有。这类链路若仍用普通 POST,
* 出图那一两分钟里链路上零字节,必然被 60s 空闲超时掐成 504;包成流式后
* 有心跳续命,结果补一帧下发即可。
*/
@FunctionalInterface
private interface StreamTask {
Object run() throws Exception;
}
/**
* 把一次 AI 生成过程写进 HTTP 响应。
*
* @param body 与普通 POST 完全相同的业务调用;它内部所有 WS 推送会被改投到这里
*/
private void streamSse(HttpServletResponse response, StreamBody body) throws IOException {
streamSseTask(response, () -> {
body.run();
return null;
});
}
/**
* 同上,但业务方法的返回值会作为 {@code {"type":"final","result":...}} 帧补在流末尾。
*/
private void streamSseTask(HttpServletResponse response, StreamTask task) throws IOException {
response.setStatus(HttpServletResponse.SC_OK);
response.setContentType("text/event-stream;charset=UTF-8");
response.setCharacterEncoding("UTF-8");
response.setHeader("Cache-Control", "no-cache, no-store, max-age=0");
response.setHeader("Connection", "keep-alive");
// 关键:nginx 见到这个响应头就关掉自己的缓冲,不需要改 nginx 配置。
// 否则整段 AI 输出会被网关攒到响应结束才一次性下发,前端看起来像「卡住不动」。
response.setHeader("X-Accel-Buffering", "no");
final PrintWriter writer = response.getWriter();
// 心跳线程与业务线程会同时写同一个 writer,所有写入都必须串行化。
final Object writeLock = new Object();
final String userId = String.valueOf(getLoginUserId());
Consumer<String> sink = payload -> {
try {
writeSseFrame(writer, writeLock, payload);
} catch (RuntimeException e) {
throw new IllegalStateException("sse write failed", e);
}
};
// 心跳:出图/出视频时上游几十秒不回一帧,没有它链路会按空闲超时断开。
ScheduledFuture<?> heartbeat = SSE_HEARTBEAT.scheduleAtFixedRate(
() -> writeSseComment(writer, writeLock, "ping"),
SSE_HEARTBEAT_INTERVAL_MS,
SSE_HEARTBEAT_INTERVAL_MS,
TimeUnit.MILLISECONDS);
ChatStreamRegistry.bind(userId, sink);
try {
// 先吐一个 SSE 注释行,把响应头立刻推出去:端上要能立刻确认「流开了」,
// 也顺手把 nginx 的缓冲区逼掉(它必须开始转发,否则拿不到第一帧)。
writeSseComment(writer, writeLock, "connected");
Object finalPayload = task.run();
if (finalPayload != null) {
writeSseFrame(writer, writeLock, finalFrameJson(finalPayload));
}
} catch (Exception e) {
logger.error("sse stream failed, userId={}", userId, e);
try {
HashMap<String, String> failData = new HashMap<>();
failData.put("answer", "出错了,请晚点再来试~");
failData.put("taskId", "0");
writeSseFrame(writer, writeLock, JSONUtil.toJSONString(failData));
} catch (RuntimeException ignore) {
// 端上已断开,错误帧没人接了
}
} finally {
heartbeat.cancel(false);
ChatStreamRegistry.unbind(userId, sink);
}
}
/** 末帧 JSON:{@code {"type":"final","result":<业务方法返回值>}}。 */
static String finalFrameJson(Object payload) {
Map<String, Object> finalFrame = new LinkedHashMap<>();
finalFrame.put("type", SSE_FINAL_FRAME_TYPE);
finalFrame.put("result", payload);
return JSONUtil.toJSONString(finalFrame);
}
/** 写一行 {@code data: <payload>} 帧;写入失败(端上断开)向上抛,由调用方决定是否吞掉。 */
private static void writeSseFrame(PrintWriter writer, Object writeLock, String payload) {
synchronized (writeLock) {
writer.write("data: ");
writer.write(payload);
writer.write("\n\n");
writer.flush();
}
}
/** 写一行 SSE 注释({@code : connected} / {@code : ping}),端上会直接丢弃,只用来续命。 */
private static void writeSseComment(PrintWriter writer, Object writeLock, String comment) {
synchronized (writeLock) {
try {
writer.write(": ");
writer.write(comment);
writer.write("\n\n");
writer.flush();
} catch (RuntimeException ignored) {
// 端上已断开:业务线程收尾时会再试一次,这里不必打断心跳线程
}
}
}
@ApiOperation("对话(HTTP 流式)")
@PostMapping("/chat/message/stream")
public void messageStream(@RequestBody ChatMessage message, HttpServletResponse response) throws IOException {
streamSseTask(response, () -> message(message));
}
@ApiOperation("稿件生成(HTTP 流式)")
@PostMapping("/manuscriptGen/stream")
public void manuscriptGenStream(@RequestBody ManuscriptGenInputs message, HttpServletResponse response) throws IOException {
streamSseTask(response, () -> manuscriptGen(message));
}
@ApiOperation("创意助手(HTTP 流式)")
@PostMapping("/creativeAssistant/stream")
public void creativeAssistantStream(@RequestBody Map<String, Object> requestBody, HttpServletResponse response) throws IOException {
final Integer loginUserId = getLoginUserId();
// 直接调同步体 executeCreativeAssistant:它自己不推帧,结果只从返回值出来,
// 所以走 streamSseTask 让返回值补成末帧;包 DeferredResult 反而会把响应交到
// 另一个线程,SSE 就没法在线程内写完。
streamSseTask(response, () -> executeCreativeAssistant(requestBody, loginUserId));
}
@ApiOperation("活动策划大纲(HTTP 流式)")
@PostMapping("/activityOutlineGen/stream")
public void activityOutlineGenStream(@RequestBody ActivityOutlineInputs message, HttpServletResponse response) throws IOException {
streamSseTask(response, () -> activityOutlineGen(message));
}
@ApiOperation("智能文档处理(HTTP 流式)")
@PostMapping("/documentProcess/stream")
public void documentProcessStream(@RequestBody DocumentProcessInputs message, HttpServletResponse response) throws IOException {
streamSseTask(response, () -> documentProcess(message));
}
@PostMapping("/competition-history/preprocess")
public DeferredResult<ApiResult<?>> competitionHistoryPreprocess(
@RequestParam("files") List<MultipartFile> files,
@@ -1353,6 +1548,17 @@ public class AiController extends BaseController {
* {@code {"code":50500,...,"status_code":110001,"status_message":"no more instances to retry"}},
* 整节点仅耗时 3.1~3.5 秒即返回——即上游没有可用算力而立即拒绝,与内容无关。
*
* <p>2026-10-21 线上日志又多了两种同义失败,都发生在最前面的「问题分类器」
* (LLM 节点,provider 为 SiliconFlow):
* <ul>
* <li>{@code API request failed with status code 503:
* {"code":50508,"message":"System is too busy now. Please try again later."}}</li>
* <li>{@code requests}/{@code urllib3} 的 {@code read_chunked} 流式读取中断
* (上游把长连接掐了)。</li>
* </ul>
* 两者都与内容无关,属于「上游此刻没算力」,应一并归到繁忙文案,
* 否则用户看到的是「生成失败」这种像是自己内容有问题的提示。
*
* <p>这只影响给用户的文案({@link #CREATIVE_BUSY_MESSAGE}),不再触发自动重试。
*/
private static boolean isUpstreamCapacityFailure(String reason) {
@@ -1362,11 +1568,27 @@ public class AiController extends BaseController {
String lower = reason.toLowerCase(Locale.ROOT);
return lower.contains("no more instances to retry")
|| lower.contains("\"code\":50500")
|| lower.contains("code: 50500");
|| lower.contains("code: 50500")
|| lower.contains("\"code\":50508")
|| lower.contains("code: 50508")
|| lower.contains("system is too busy")
|| lower.contains("status code 503")
|| lower.contains("http 503")
|| lower.contains("503 service unavailable")
// provider 插件把流式响应读到一半被上游掐断(2026-10-21 18:10 那例)。
// 这类异常不是「内容有问题」,重试/稍后再试才有意义。
|| lower.contains("chunkedencodingerror")
|| lower.contains("read_chunked")
|| lower.contains("connection broken");
}
/**
* 把上游原文裁到可写进单行日志的长度。
*
* <p>只留头部会把最关键的信息丢掉:Python traceback 的异常类型与
* {@code raise ...} 那一行都在**尾部**(见 2026-10-21 日志里
* {@code read_chunked} 那例,截断后只剩调用栈开头,看不出是哪类异常)。
* 因此改成「头 + 标记 + 尾」,总量仍受 {@link #UPSTREAM_ERROR_LOG_MAX_LENGTH} 约束。
*/
private static String briefUpstreamError(String reason) {
if (reason == null) {
@@ -1375,7 +1597,11 @@ public class AiController extends BaseController {
if (reason.length() <= UPSTREAM_ERROR_LOG_MAX_LENGTH) {
return reason;
}
return reason.substring(0, UPSTREAM_ERROR_LOG_MAX_LENGTH) + "...(truncated)";
int headLength = UPSTREAM_ERROR_LOG_MAX_LENGTH * 2 / 3;
int tailLength = UPSTREAM_ERROR_LOG_MAX_LENGTH - headLength;
return reason.substring(0, headLength)
+ "...(truncated)..."
+ reason.substring(reason.length() - tailLength);
}
private Map<String, Object> parseInputs(String inputs) {
@@ -2098,6 +2324,15 @@ public class AiController extends BaseController {
}
if (answer != null) {
result.put("answer", answer);
} else {
// 成功却挑不出可展示内容:端上只会看到「接口已返回,但未解析到可展示的内容」,
// 而这条路径过去在日志里**没有任何痕迹**——上游形状一变,端上就悄悄退化,
// 排查时只能靠猜。这里把 outputs 的键与图片条数记下来,
// 下一次出现时能直接区分「真没内容」还是「又换了字段名」(issue 19)。
logger.warn("creative assistant returned no displayable answer, status={}, outputKeys={}, normalizedPhotoFiles={}",
status,
normalizedOutputs == null ? "<无 outputs>" : normalizedOutputs.keySet(),
normalizedOutputs == null ? 0 : photoFileCount(normalizedOutputs.get("photo_files")));
}
String workflowRunId = stringValue(responseData.get("workflow_run_id"));
@@ -2169,6 +2404,16 @@ public class AiController extends BaseController {
return normalized;
}
/**
* 数一下归一后的 {@code photo_files} 有几个元素(只用于日志)。
*/
private static int photoFileCount(Object photoFiles) {
if (photoFiles instanceof Collection) {
return ((Collection<?>) photoFiles).size();
}
return photoFiles == null ? 0 : 1;
}
/**
* 把 {@code photo_files} / {@code photo_file} 收进统一列表。
*
@@ -0,0 +1,120 @@
package com.gxwebsoft.ai.controller;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Method;
import java.util.LinkedHashMap;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* 创意生成失败文案的归类。
*
* <p>2026-10-21 线上日志(dev 实例 18:10 / 18:50 两例)显示:失败节点从
* 「即梦AI-图像生成」换成了最前面的「问题分类器」(LLM 节点,provider 为 SiliconFlow),
* 上游返回的是 503 / {@code code:50508 "System is too busy"} 与流式读取中断。
* 旧实现只认 {@code 50500} 与 {@code no more instances to retry},
* 于是这两例都落到「生成失败,请稍后重试」——把上游繁忙说成用户内容的问题。
*
* <p>这里直接复刻 {@code creativeFailureAnswer} 的入口,锁住「哪些原文算繁忙」。
*/
class AiCreativeBusyClassificationTest {
/** 期望的繁忙文案(与 {@code CREATIVE_BUSY_MESSAGE} 同值)。 */
private static final String BUSY = "生成服务繁忙,请稍后重试";
/** 期望的通用失败文案(与 {@code CREATIVE_FAILED_MESSAGE} 同值)。 */
private static final String FAILED = "生成失败,请稍后重试";
/**
* 复刻调用入口:把上游 error 放进 {@code data.error},交给私有的
* {@code creativeFailureAnswer(responseData)}。
*/
private String answerFor(String upstreamError) throws Exception {
Map<String, Object> data = new LinkedHashMap<>();
data.put("status", "failed");
data.put("error", upstreamError);
Map<String, Object> responseData = new LinkedHashMap<>();
responseData.put("event", "workflow_finished");
responseData.put("data", data);
AiController controller = new AiController();
Method method = AiController.class.getDeclaredMethod("creativeFailureAnswer", Map.class);
method.setAccessible(true);
return String.valueOf(method.invoke(controller, responseData));
}
/** 2026-10-21 18:50 的真实原文:SiliconFlow 503 / 50508。 */
@Test
void 硅基流动503繁忙_归为繁忙文案() throws Exception {
String reason = "req_id: 1cdf48dd63 PluginInvokeError: {\"args\":{\"description\":\"[models] "
+ "Error: API request failed with status code 503: {\"code\":50508,"
+ "\"message\":\"System is too busy now. Please try again later.\",\"data\":null}\"}}";
assertEquals(BUSY, answerFor(reason));
}
/** 2026-10-21 18:10 的真实原文:流式响应读到一半被掐断(尾部才是异常类型)。 */
@Test
void 流式读取中断_归为繁忙文案() throws Exception {
String reason = "req_id: b0f0928ba8 PluginInvokeError: {\"args\":{\"traceback\":\"Traceback (most recent call last):\\n"
+ " File \\\"/app/.../requests/models.py\\\", line 937, in generate\\n"
+ " yield from self.raw.stream(chunk_size, decode_content=True)\\n"
+ " File \\\"/app/.../urllib3/response.py\\\", line 1260, in stream\\n"
+ " yield from self.read_chunked(amt, decode_content=decode_content)\\n"
+ "requests.exceptions.ChunkedEncodingError: ('Connection broken: IncompleteRead(0 bytes read)')\"}";
assertEquals(BUSY, answerFor(reason));
}
/** 老规则仍然生效:图片工具没有可用实例。 */
@Test
void 即梦无可用实例_归为繁忙文案() throws Exception {
String reason = "{\"code\":50500,\"status_code\":110001,\"status_message\":\"no more instances to retry\"}";
assertEquals(BUSY, answerFor(reason));
}
/** 凭证/权限类失败不是繁忙,别把「去检查配置」说成「稍后再试」。 */
@Test
void 权限失败_仍为通用失败文案() throws Exception {
String reason = "req_id: dc688258de PluginInvokeError: [models] Error: API request failed with status code 403";
assertEquals(FAILED, answerFor(reason));
}
/** 无 error 字段时不能抛异常,按通用失败处理。 */
@Test
void 无error字段_按通用失败处理() throws Exception {
Map<String, Object> responseData = new LinkedHashMap<>();
responseData.put("data", new LinkedHashMap<String, Object>());
AiController controller = new AiController();
Method method = AiController.class.getDeclaredMethod("creativeFailureAnswer", Map.class);
method.setAccessible(true);
assertEquals(FAILED, String.valueOf(method.invoke(controller, responseData)));
}
/** 截断必须保留尾部:Python traceback 的异常类型只在最后一行。 */
@Test
void 超长上游原文_头尾都保留() throws Exception {
StringBuilder reason = new StringBuilder("HEAD_MARKER");
while (reason.length() < 900) {
reason.append(" middle-padding");
}
reason.append("ChunkedEncodingError: TAIL_MARKER");
AiController controller = new AiController();
Method method = AiController.class.getDeclaredMethod("briefUpstreamError", String.class);
method.setAccessible(true);
String brief = String.valueOf(method.invoke(null, reason.toString()));
org.junit.jupiter.api.Assertions.assertTrue(brief.startsWith("HEAD_MARKER"), brief);
org.junit.jupiter.api.Assertions.assertTrue(brief.endsWith("TAIL_MARKER"), brief);
org.junit.jupiter.api.Assertions.assertTrue(brief.contains("...(truncated)..."), brief);
org.junit.jupiter.api.Assertions.assertTrue(brief.length() < 900, "截断后仍应短:" + brief.length());
org.junit.jupiter.api.Assertions.assertFalse(brief.contains("\n"), "日志行不能带换行");
}
}