diff --git a/src/main/java/com/gxwebsoft/ai/controller/AiController.java b/src/main/java/com/gxwebsoft/ai/controller/AiController.java index ef04980..2827a59 100644 --- a/src/main/java/com/gxwebsoft/ai/controller/AiController.java +++ b/src/main/java/com/gxwebsoft/ai/controller/AiController.java @@ -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 心跳间隔。 + * + *

图片/视频生成这类工作流，上游要跑一两分钟且中间**一个字节都不回**。 + * 生产链路在空闲 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; + } + + /** + * 带返回值的流式方法体：返回值会作为最后一帧发给端上。 + * + *

给「业务方法自己不推帧、只返回一个结果」的链路用 —— 创意助手的出图就是 + * 这种：中间没有任何推送，结果只在方法返回时才有。这类链路若仍用普通 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 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 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 finalFrame = new LinkedHashMap<>(); + finalFrame.put("type", SSE_FINAL_FRAME_TYPE); + finalFrame.put("result", payload); + return JSONUtil.toJSONString(finalFrame); + } + + /** 写一行 {@code data: } 帧；写入失败（端上断开）向上抛，由调用方决定是否吞掉。 */ + 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 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> competitionHistoryPreprocess( @RequestParam("files") List 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 秒即返回——即上游没有可用算力而立即拒绝，与内容无关。 * + *

2026-10-21 线上日志又多了两种同义失败，都发生在最前面的「问题分类器」 + * （LLM 节点，provider 为 SiliconFlow）： + *

+ * 两者都与内容无关，属于「上游此刻没算力」，应一并归到繁忙文案， + * 否则用户看到的是「生成失败」这种像是自己内容有问题的提示。 + * *

这只影响给用户的文案（{@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"); } /** * 把上游原文裁到可写进单行日志的长度。 + * + *

只留头部会把最关键的信息丢掉：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 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} 收进统一列表。 * diff --git a/src/test/java/com/gxwebsoft/ai/controller/AiCreativeBusyClassificationTest.java b/src/test/java/com/gxwebsoft/ai/controller/AiCreativeBusyClassificationTest.java new file mode 100644 index 0000000..95849f1 --- /dev/null +++ b/src/test/java/com/gxwebsoft/ai/controller/AiCreativeBusyClassificationTest.java @@ -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; + +/** + * 创意生成失败文案的归类。 + * + *

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}， + * 于是这两例都落到「生成失败，请稍后重试」——把上游繁忙说成用户内容的问题。 + * + *

这里直接复刻 {@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 data = new LinkedHashMap<>(); + data.put("status", "failed"); + data.put("error", upstreamError); + Map 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 responseData = new LinkedHashMap<>(); + responseData.put("data", new LinkedHashMap()); + + 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"), "日志行不能带换行"); + } +}