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:
@@ -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} 收进统一列表。
|
||||
*
|
||||
|
||||
Reference in New Issue
Block a user