test(saa): M3 openai 流式静默挂死最小复现(诊断脚手架)
门控 -Dsaa.stream.repro=1 + NEWAPI_KEY,不破普通构建。逐验:stream() 是否真流式 / thinking 开时 content 帧是否到达 / blockLast 是否返回。全程显式超时不挂 900s。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
lili
parent
d342de088e
commit
4ed9d2a90f
@ -0,0 +1,177 @@
|
||||
package com.wanxiang.huijing.game.module.aigc.saa;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
|
||||
import org.springframework.ai.chat.messages.UserMessage;
|
||||
import org.springframework.ai.chat.model.ChatModel;
|
||||
import org.springframework.ai.chat.model.ChatResponse;
|
||||
import org.springframework.ai.chat.prompt.Prompt;
|
||||
import org.springframework.ai.openai.OpenAiChatModel;
|
||||
import org.springframework.ai.openai.api.OpenAiApi;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
/**
|
||||
* SAA openai+流式 跑 M3「静默挂死」最小复现（诊断脚手架，非产线门）。
|
||||
*
|
||||
* <p>逐一验诊断假设：① stream() 是否真流式（块逐步到达 vs 整体缓冲）；② thinking 开时 content 帧是否到达
|
||||
* （还是全进 reasoning_content → MessageAggregator 等不到 content 永阻塞）；③ blockLast() 是否返回。
|
||||
* 全程显式超时（不挂 900s），证据直打 stdout。
|
||||
*
|
||||
* <p>门控 {@code -Dsaa.stream.repro=1} + NEWAPI_KEY；不破普通构建。
|
||||
*/
|
||||
@EnabledIfEnvironmentVariable(named = "NEWAPI_KEY", matches = ".+")
|
||||
class SaaStreamHangReproTest {
|
||||
|
||||
private static final String BASE = System.getenv().getOrDefault("NEWAPI_BASE_URL", "http://100.64.0.8:3000");
|
||||
private static final String MODEL = "MiniMax-M3";
|
||||
|
||||
@Test
|
||||
void repro_stream_hang() throws Exception {
|
||||
if (!"1".equals(System.getProperty("saa.stream.repro"))) {
|
||||
System.out.println("[skip] 加 -Dsaa.stream.repro=1 才跑本诊断");
|
||||
return;
|
||||
}
|
||||
String key = System.getenv("NEWAPI_KEY");
|
||||
String base = SaaStudioGraph.stripV1Suffix(BASE);
|
||||
System.out.println("[repro] base=" + base + " model=" + MODEL + " keyLen=" + (key == null ? 0 : key.length()));
|
||||
|
||||
// 生产同款：NO_PROXY OpenAiApi（读超时 600s）+ model() 对 M3 注入 thinking:adaptive+reasoning_split+48K
|
||||
OpenAiApi api = SaaStudioGraph.openAiApiNoProxy(base, key, 600_000);
|
||||
OpenAiChatModel m3 = SaaStudioGraph.model(api, MODEL, 0.0, 16000); // maxTokens≥8000 → 生成角色配置（开 thinking）
|
||||
// 真实规模 prompt：强制大输出（完整游戏源码），逼近 code 角色真实负载（48K maxTokens + 长 thinking + 长 content），
|
||||
// 复现 e2e 大请求下的挂死（小 prompt 短答案不复现）。
|
||||
String bigSystem = "你是一个 H5 小游戏代码生成器。产出一个完整、可运行的原生 JS 打砖块游戏单文件模块，"
|
||||
+ "包含完整的挡板物理、小球反弹、砖块矩阵、碰撞检测、计分、胜负判定、渲染循环。"
|
||||
+ "代码要详尽完整、注释充分、不省略任何函数体。只输出一个 ```js 代码块。";
|
||||
Prompt prompt = new Prompt(List.of(
|
||||
new org.springframework.ai.chat.messages.SystemMessage(bigSystem),
|
||||
new UserMessage("实现打砖块：挡板接球把上方砖块全部消除即胜，掉球则负。要求代码完整详尽，不少于 300 行。")));
|
||||
|
||||
// ===== 实验1：直接订阅 stream()，逐帧打印到达时刻 + content/reasoning 长度（判真流式 + content 是否到达）=====
|
||||
System.out.println("\n===== 实验1：OpenAiChatModel.stream() 逐帧 =====");
|
||||
long t0 = System.currentTimeMillis();
|
||||
AtomicInteger frames = new AtomicInteger();
|
||||
AtomicInteger contentFrames = new AtomicInteger();
|
||||
AtomicLong contentChars = new AtomicLong();
|
||||
AtomicReference<Long> firstAt = new AtomicReference<>(null);
|
||||
AtomicLong lastFrameAt = new AtomicLong(0);
|
||||
AtomicLong maxGap = new AtomicLong(0);
|
||||
try {
|
||||
Flux<ChatResponse> flux = m3.stream(prompt);
|
||||
flux.doOnNext(cr -> {
|
||||
long at = System.currentTimeMillis() - t0;
|
||||
long gap = at - lastFrameAt.get();
|
||||
if (lastFrameAt.get() > 0 && gap > maxGap.get()) {
|
||||
maxGap.set(gap);
|
||||
}
|
||||
lastFrameAt.set(at);
|
||||
if (firstAt.get() == null) {
|
||||
firstAt.set(at);
|
||||
System.out.println(" [首帧 @" + at + "ms]");
|
||||
}
|
||||
int idx = frames.incrementAndGet();
|
||||
String txt = "";
|
||||
try {
|
||||
if (cr.getResult() != null && cr.getResult().getOutput() != null) {
|
||||
txt = cr.getResult().getOutput().getText();
|
||||
}
|
||||
} catch (Exception ignore) {
|
||||
}
|
||||
if (txt != null && !txt.isEmpty()) {
|
||||
contentFrames.incrementAndGet();
|
||||
contentChars.addAndGet(txt.length());
|
||||
}
|
||||
if (idx <= 6 || (txt != null && !txt.isEmpty() && contentFrames.get() <= 3)) {
|
||||
System.out.println(" 帧#" + idx + " @" + at + "ms content="
|
||||
+ (txt == null || txt.isEmpty() ? "无" : "有[" + txt.length() + "]"));
|
||||
}
|
||||
}).blockLast(Duration.ofSeconds(300));
|
||||
System.out.println(" [stream 完成 @" + (System.currentTimeMillis() - t0) + "ms] 总帧=" + frames.get()
|
||||
+ " content帧=" + contentFrames.get() + "(" + contentChars.get() + "字) 首帧=" + firstAt.get()
|
||||
+ "ms 最大帧间隔=" + maxGap.get() + "ms");
|
||||
} catch (Exception e) {
|
||||
System.out.println(" [stream 异常 @" + (System.currentTimeMillis() - t0) + "ms] "
|
||||
+ e.getClass().getName() + ": " + e.getMessage()
|
||||
+ " 已收帧=" + frames.get() + " content帧=" + contentFrames.get());
|
||||
}
|
||||
|
||||
// ===== 实验2：生产真链复刻 = StreamingCallChatModel.call() 套「180s 单次超时 + cancel(true) + ≤3 重试」 =====
|
||||
// 这是 SaaStudioNodes.callWithTimeoutAndRetry 的等价复刻（生产 generate 节点真用）。
|
||||
// 假设：M3 thinking 阶段 >180s（实验1 已证 content 首帧 @~203s）→ 每次 get(180s) 必 TimeoutException →
|
||||
// cancel(true) 中断 blockLast 线程，但底层 JDK HttpClient SSE 连接不被中断取消（连接泄漏）→ 反复重试 → 静默卡死。
|
||||
System.out.println("\n===== 实验2：生产真链复刻（StreamingCallChatModel.call + 180s超时×3重试 + cancel）=====");
|
||||
ChatModel wrapped = SaaStudioGraph.openAiFactory(api).create(MODEL, 0.0, 16000);
|
||||
long t1 = System.currentTimeMillis();
|
||||
AtomicReference<ChatResponse> got = new AtomicReference<>();
|
||||
AtomicReference<Throwable> err = new AtomicReference<>();
|
||||
// 整体放后台线程 + join 上限 620s（3×180s + 退避 + 余量），看是否在合理时间内「返回 / 抛超时 / 仍卡死」。
|
||||
Thread th = new Thread(() -> {
|
||||
try {
|
||||
got.set(prodLikeCallWithTimeoutAndRetry(wrapped, prompt));
|
||||
} catch (Throwable t) {
|
||||
err.set(t);
|
||||
}
|
||||
}, "repro-prodlike");
|
||||
th.setDaemon(true);
|
||||
th.start();
|
||||
th.join(620_000);
|
||||
long dt = System.currentTimeMillis() - t1;
|
||||
if (th.isAlive()) {
|
||||
System.out.println(" [!!! 生产真链 620s 未返回 = 静默卡死复现 @" + dt + "ms] 主线程仍 alive");
|
||||
for (StackTraceElement ste : th.getStackTrace()) {
|
||||
System.out.println(" at " + ste);
|
||||
}
|
||||
// 同时 dump 进程内所有 saa-llm-call 线程，看泄漏的 blockLast 线程是否堆积
|
||||
int leaked = 0;
|
||||
for (Thread t : Thread.getAllStackTraces().keySet()) {
|
||||
if (t.getName().contains("repro-llm") || t.getName().contains("saa-llm")) {
|
||||
leaked++;
|
||||
}
|
||||
}
|
||||
System.out.println(" 泄漏的 LLM 调用线程数=" + leaked + "（>1 = cancel 未真停底层流，连接泄漏）");
|
||||
} else if (err.get() != null) {
|
||||
System.out.println(" [生产真链 抛异常 @" + dt + "ms（约=N×180s）] " + err.get().getClass().getName()
|
||||
+ ": " + err.get().getMessage() + " → 生产把它归 llm_error，但耗时已逼近/超 900s 预算");
|
||||
} else {
|
||||
ChatResponse cr = got.get();
|
||||
String txt = (cr == null || cr.getResult() == null) ? null : cr.getResult().getOutput().getText();
|
||||
System.out.println(" [生产真链 返回 @" + dt + "ms] contentLen=" + (txt == null ? 0 : txt.length()));
|
||||
}
|
||||
System.out.println("\n===== 复现结束 =====");
|
||||
}
|
||||
|
||||
/** 生产 SaaStudioNodes.callWithTimeoutAndRetry 的等价复刻（180s 单次超时 + cancel(true) + ≤3 重试）。 */
|
||||
private static ChatResponse prodLikeCallWithTimeoutAndRetry(ChatModel model, Prompt prompt) throws Exception {
|
||||
java.util.concurrent.ExecutorService pool = java.util.concurrent.Executors.newCachedThreadPool(r -> {
|
||||
Thread t = new Thread(r, "repro-llm-call");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
});
|
||||
Exception lastEx = null;
|
||||
for (int attempt = 0; attempt < 3; attempt++) {
|
||||
long a0 = System.currentTimeMillis();
|
||||
java.util.concurrent.Future<ChatResponse> f = pool.submit(() -> model.call(prompt));
|
||||
try {
|
||||
ChatResponse r = f.get(180, java.util.concurrent.TimeUnit.SECONDS);
|
||||
System.out.println(" 尝试#" + (attempt + 1) + " 成功 @" + (System.currentTimeMillis() - a0) + "ms");
|
||||
return r;
|
||||
} catch (java.util.concurrent.TimeoutException te) {
|
||||
boolean cancelled = f.cancel(true);
|
||||
System.out.println(" 尝试#" + (attempt + 1) + " 180s 超时 → cancel(true) 返回=" + cancelled
|
||||
+ "（cancel=true 仅意味发了中断信号，不等于底层 HTTP 流真停）");
|
||||
lastEx = new RuntimeException("LLM 调用超时(180s)", te);
|
||||
} catch (java.util.concurrent.ExecutionException ee) {
|
||||
System.out.println(" 尝试#" + (attempt + 1) + " 异常 @" + (System.currentTimeMillis() - a0) + "ms: " + ee.getCause());
|
||||
lastEx = (ee.getCause() instanceof Exception) ? (Exception) ee.getCause() : new RuntimeException(ee.getCause());
|
||||
}
|
||||
if (attempt < 2) Thread.sleep(2000);
|
||||
}
|
||||
throw lastEx != null ? lastEx : new RuntimeException("重试 3 次后失败");
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user