* 核心原则：模型流驱动渠道流，State 只保存最终聚合结果。 *
* 所有面向用户的 LLM 节点（ReasoningNode、StepExecutionNode、PlanSummaryNode 等） * 统一使用此 helper，而不是各自散落 {@code chatModel.call()}。 * * @author MateClaw Team */ @Slf4j public class NodeStreamingChatHelper { private final ChatStreamTracker streamTracker; /** * Ordered fallback chain tried after the primary model exhausts retries. * Each entry is attempted once (no retry); the first successful response * wins. Empty list disables fallover entirely. See RFC-009. * *
Stored as {@link vip.mate.llm.failover.FallbackEntry} (providerId + * ChatModel) so the chain walker can consult {@link vip.mate.llm.failover.ProviderHealthTracker} * — cooldown state is keyed by providerId, not by ChatModel instance.
*/ private final List* 用于 PlanGenerationNode 等返回结构化 JSON 的节点 —— LLM 输出不应直接展示给用户， * 需要后续解析后再决定是否广播。 * * @param chatModel LLM 模型 * @param prompt 完整 prompt * @param conversationId 会话 ID（仅用于日志，不广播） * @param phase 阶段标识 * @return 聚合结果 */ public StreamResult streamCallSilent(ChatModel chatModel, Prompt prompt, String conversationId, String phase) { return streamCallInternal(chatModel, prompt, conversationId, phase, false); } /** * 广播文本内容到前端（用于 silent 调用后手动推送 direct_answer 等） */ public void broadcastContent(String conversationId, String content) { if (content != null && !content.isEmpty()) { broadcastDelta(conversationId, "content_delta", content); } } /** * Broadcast a lightweight progress event so the frontend shows activity * during silent LLM calls (e.g. triage). Sent as a "progress" SSE event. */ public void broadcastProgress(String conversationId, String message) { if (streamTracker == null || conversationId == null || conversationId.isEmpty()) { return; } streamTracker.broadcastObject(conversationId, "progress", Map.of("message", message != null ? message : "")); } // ==================== 重试配置 ==================== /** * Soft upper bound on per-call thinking ({@code reasoning_content}) chars * with zero visible content and zero tool calls. Beyond this the helper * disposes the upstream subscription and returns a partial result so the * graph can advance instead of streaming thinking forever. Calibrated * against typical Claude/DeepSeek extended-thinking budgets — well above * normal long-form reasoning, low enough to bound a runaway loop in under * ~30 seconds of wall clock. */ private static final int THINKING_ONLY_HARD_CAP_CHARS = 32768; /** * Narrow content-repetition guard — fires when the buffer ends with * the same period-sized chunk repeated {@link * #CONTENT_REPEAT_MAX_OCCURRENCES}+ times in a row. Picked to catch * the specific failure mode where reasoning-mode models (qwen3.6, * deepseek-r1) get into a "Wait, I should X. → 写答案 → Wait, I * should Y. → 写同一份答案 → …" self-arguing loop and emit the same * final-answer paragraph dozens of times until {@code max_tokens} * runs out. * *
Tests probe sizes from {@link #CONTENT_REPEAT_MIN_PERIOD} up
* to {@link #CONTENT_REPEAT_MAX_PERIOD}; the smallest period that
* yields the required consecutive copies trips the guard. 4
* verbatim consecutive copies of any 24+ char unit is a near-
* impossible coincidence in real text, so false positives are very
* rare. Not as exhaustive as the previous {@code RepetitionDetector}
* (removed at 42d406ff for being brittle on legitimate long-form
* content), just the cheap specific check that catches this loop.
*/
public static final int CONTENT_REPEAT_MIN_PERIOD = 24;
public static final int CONTENT_REPEAT_MAX_PERIOD = 240;
private static final int CONTENT_REPEAT_MAX_OCCURRENCES = 4;
/**
* Re-scan every N chars of new content. Smaller = faster reaction,
* larger = less CPU. The probe loop is O(period_range × occurrences)
* char comparisons per scan — cheap even at 400-char intervals.
*/
private static final int CONTENT_REPEAT_CHECK_INTERVAL = 200;
private static final int MAX_RETRIES = 5;
// RATE_LIMIT: fail fast to failover chain — staying on the same
// provider during a rate-limit window wastes time without recovery.
// SERVER_ERROR keeps MAX_RETRIES (upstream flaps often self-heal).
private static final int MAX_RETRIES_RATE_LIMIT = 2;
private static final long BACKOFF_BASE_MS = 3000;
private static final long BACKOFF_CAP_MS = 60_000;
private static final ObjectMapper TOOL_ARG_JSON_MAPPER = new ObjectMapper();
/**
* 判断错误是否可重试（基于状态码/异常类型）
*/
private static boolean isRetryable(Throwable error) {
String msg = extractFullErrorChain(error);
// Kimi engine_overloaded / 标准 HTTP 错误 / 速率限制
return msg.contains("engine_overloaded")
|| msg.contains("rate_limit") || msg.contains("RateLimitError")
|| msg.contains("429") || msg.contains("Too Many Requests")
|| msg.contains("500") || msg.contains("502") || msg.contains("503") || msg.contains("504")
|| msg.contains("APITimeoutError") || msg.contains("APIConnectionError")
|| msg.contains("Connection reset") || msg.contains("Connection refused");
}
/**
* 分类错误类型（用于分级重试和上层 Node 决策）
*/
private static ErrorType classifyError(Throwable error) {
String msg = extractFullErrorChain(error);
// PTL: prompt too long / context length exceeded
if (msg.contains("prompt is too long")
|| msg.contains("context_length_exceeded")
|| msg.contains("context length exceeded")
|| msg.contains("maximum context length")
|| msg.contains("token limit")
|| msg.contains("This model's maximum context length")
|| msg.contains("请求体中的 input tokens 总数超出了模型允许")) {
return ErrorType.PROMPT_TOO_LONG;
}
// Auth errors
if (msg.contains("401") || msg.contains("Unauthorized") || msg.contains("Invalid API Key")
|| msg.contains("authentication") || msg.contains("AuthenticationError")) {
return ErrorType.AUTH_ERROR;
}
// Rate limit
if (msg.contains("429") || msg.contains("rate_limit") || msg.contains("RateLimitError")
|| msg.contains("Too Many Requests") || msg.contains("engine_overloaded")) {
return ErrorType.RATE_LIMIT;
}
// Thinking block errors (Anthropic: old thinking blocks cannot be modified)
if (msg.contains("thinking blocks cannot be modified")
|| msg.contains("thinking content is not allowed")
|| msg.contains("thinking block")) {
return ErrorType.THINKING_BLOCK_ERROR;
}
// RFC-009 P3.2: BILLING — payment / quota exhausted. Distinct from AUTH because
// a different provider may have credits, so we should fall back instead of
// terminating the call. Both OpenAI ("insufficient_quota") and Anthropic
// ("credit balance is too low") use these phrases in 402-class responses.
if (msg.contains("402") || msg.contains("insufficient_quota")
|| msg.contains("credit balance is too low")
|| msg.contains("billing_error") || msg.contains("billing_hard_limit_reached")
|| msg.contains("You exceeded your current quota")
|| msg.contains("quota exceeded") || msg.contains("Quota exceeded")) {
return ErrorType.BILLING;
}
// RFC-009 P3.2: MODEL_NOT_FOUND — provider rejects the requested model id.
// Includes DashScope's "[InvalidParameter] url error, please check url"
// (https://help.aliyun.com/zh/model-studio/error-code#error-url) which despite
// the wording is the provider rejecting an unknown/unsupported model id on
// the native protocol. Splitting this out from CLIENT_ERROR lets us hand off
// to the fallback chain instead of terminating — a different provider may
// recognize the model name (or have an equivalent default).
if (msg.contains("Model not exist")
|| msg.contains("model_not_found")
|| msg.contains("Model not found")
|| msg.contains("does not exist")
|| msg.contains("[InvalidParameter]")
|| msg.contains("InvalidParameter")
|| msg.contains("url error")
// Volcano Ark: model exists but the user's account hasn't opened it,
// or the id isn't valid for this region. Both are hard failures —
// retrying won't help, and a different provider may serve the model.
|| msg.contains("ModelNotOpen")
|| msg.contains("InvalidEndpointOrModel")) {
return ErrorType.MODEL_NOT_FOUND;
}
// Client errors (400 Bad Request — unsupported format, invalid params, etc.) — NOT retryable
if (msg.contains("400") || msg.contains("Bad Request")
|| msg.contains("invalid_request_error") || msg.contains("unsupported")) {
return ErrorType.CLIENT_ERROR;
}
// Server errors and transient TLS / socket-level network hiccups.
// Without the TLS-specific patterns, a single SSL fatal alert
// (e.g. bad_record_mac during long-running streams) falls through to
// UNKNOWN — non-retryable — so one transient handshake glitch surfaces
// to the user as "LLM 调用失败" with no recovery attempt. These are
// network-layer transients that almost always succeed on retry, so
// they belong in the same retryable bucket as 5xx/timeouts.
if (msg.contains("500") || msg.contains("502") || msg.contains("503") || msg.contains("504")
|| msg.contains("APITimeoutError") || msg.contains("APIConnectionError")
|| msg.contains("Connection reset") || msg.contains("Connection refused")
|| msg.contains("timeout") || msg.contains("Timeout")
// TLS-layer transients: bad_record_mac (RFC 5246 §7.2.2 fatal
// alert 20), aborted handshakes, mid-stream protocol errors.
|| msg.contains("SSLException") || msg.contains("SSLHandshakeException")
|| msg.contains("SSLProtocolException") || msg.contains("bad_record_mac")
// Socket-level transients: a peer closing the TCP connection
// mid-response, or the OS reporting a half-closed pipe.
|| msg.contains("SocketException") || msg.contains("Broken pipe")
|| msg.contains("Premature close") || msg.contains("PrematureCloseException")
|| msg.contains("Connection prematurely closed")
|| msg.contains("Connection closed prematurely")
// Reactor Netty wraps the raw socket cause in WebClientRequestException;
// surface that wrapper too so retries fire even when the cause chain
// string is "WebClientRequestException ...; nested ... SSLException".
|| msg.contains("WebClientRequestException")) {
return ErrorType.SERVER_ERROR;
}
return ErrorType.UNKNOWN;
}
/** 提取完整异常链信息用于关键字匹配 */
private static String extractFullErrorChain(Throwable error) {
StringBuilder sb = new StringBuilder();
Throwable cur = error;
while (cur != null) {
if (cur.getMessage() != null) {
sb.append(cur.getMessage()).append(" | ");
}
sb.append(cur.getClass().getSimpleName()).append(" | ");
// Include the HTTP response body for WebClient errors. Many providers
// (Volcano Ark, Ollama, …) put the actionable error code only in the
// body, while the surface message is just "404 Not Found from POST X".
// Without this, classifyError() can never see codes like ModelNotOpen.
if (cur instanceof WebClientResponseException wre) {
try {
String body = wre.getResponseBodyAsString();
if (body != null && !body.isEmpty()) {
sb.append(body.length() > 1024 ? body.substring(0, 1024) : body)
.append(" | ");
}
} catch (Exception ignored) {
}
}
cur = cur.getCause();
}
return sb.toString();
}
private StreamResult streamCallInternal(ChatModel chatModel, Prompt prompt,
String conversationId, String phase,
boolean broadcast) {
// 在开始 LLM 调用前检查停止标志
if (streamTracker.isStopRequested(conversationId)) {
log.info("[{}] Stop requested before LLM call, aborting: conversationId={}", phase, conversationId);
throw new CancellationException("Stream stopped by user");
}
// RFC-009 P3.1 + Phase 4: short-circuit the primary retry loop in two cases.
// (a) primary is in cooldown (P3.3) — soft, transient
// (b) primary was HARD-removed from the pool (Phase 4) — auth/billing/missing model
// Either way, retrying the same model wastes seconds; head straight to fallback.
boolean primaryInCooldown = primaryProviderId != null
&& healthTracker != null
&& healthTracker.isInCooldown(primaryProviderId);
boolean primaryOutOfPool = primaryProviderId != null && !inPool(primaryProviderId);
boolean primarySkipped = primaryInCooldown || primaryOutOfPool;
if (primarySkipped) {
String reason = primaryOutOfPool ? "removed from pool" : "in cooldown";
log.warn("[{}] Primary provider={} {} — skipping straight to fallback chain",
phase, primaryProviderId, reason);
if (broadcast) {
broadcastDelta(conversationId, "warning",
buildDeltaJson("主模型暂时不可用（" + (primaryOutOfPool ? "已下线" : "冷却中")
+ "），直接尝试备选模型..."));
}
}
// D-6: performance counters
int retryCount = 0;
long totalBackoffMs = 0;
int failoverCount = 0;
int llmCallCount = 0;
long callStartMs = System.currentTimeMillis();
// 主模型重试循环
StreamResult lastResult = null;
if (!primarySkipped) for (int attempt = 0; attempt <= MAX_RETRIES; attempt++) {
llmCallCount++;
if (attempt > 0) retryCount++;
lastResult = doStreamCall(chatModel, prompt, conversationId, phase, broadcast, attempt);
if (lastResult != null) {
// PTL: 不重试，直接返回给上层 Node 处理
if (lastResult.errorType() == ErrorType.PROMPT_TOO_LONG) {
return lastResult;
}
// AUTH: primary key 失效不会自愈，跳过同模型重试，交给 fallback chain
// — 其它 provider 的 key 可能仍然可用（与 BILLING / MODEL_NOT_FOUND 同策略）。
// recordPrimary(false) 仍记一次失败用于 healthTracker 冷却累计。
// 若 fallback chain 全部 401，walker 末尾会把最后一次 AUTH_ERROR 透出，
// 不会静默吞错。
if (lastResult.errorType() == ErrorType.AUTH_ERROR) {
log.warn("[{}] Primary auth failed — skipping same-model retries, handing off to fallback chain", phase);
recordPrimary(false);
removeFromPool(primaryProviderId, ErrorType.AUTH_ERROR, lastResult.errorMessage());
break;
}
// RFC-009 P3.2: BILLING / MODEL_NOT_FOUND — provider-side hard failures
// that won't change on retry. Skip to fallback chain (a different
// provider may have credits, or the model name may be valid there).
if (lastResult.errorType() == ErrorType.BILLING
|| lastResult.errorType() == ErrorType.MODEL_NOT_FOUND) {
log.warn("[{}] Primary error={} — skipping same-model retries, handing off to fallback chain",
phase, lastResult.errorType());
recordPrimary(false);
removeFromPool(primaryProviderId, lastResult.errorType(), lastResult.errorMessage());
break;
}
// CLIENT_ERROR (400 Bad Request): 不重试（参数/格式错误重试也不会变）
if (lastResult.errorType() == ErrorType.CLIENT_ERROR) {
return lastResult;
}
// THINKING_BLOCK_ERROR: 剥离旧 thinking 块后单次重试
if (lastResult.errorType() == ErrorType.THINKING_BLOCK_ERROR && attempt == 0) {
log.warn("[{}] Thinking block error detected, stripping old thinking and retrying once", phase);
prompt = stripThinkingFromPrompt(prompt);
continue; // 重试一次
}
if (lastResult.errorType() == ErrorType.THINKING_BLOCK_ERROR) {
return lastResult; // 已经重试过了
}
// RFC-009: EMPTY_RESPONSE — break the primary-retry loop and fall through to
// the fallback chain. Retrying the same model that returned nothing is rarely
// productive; a different provider has a better chance of succeeding.
if (lastResult.errorType() == ErrorType.EMPTY_RESPONSE) {
log.warn("[{}] Primary returned empty response — skipping same-model retries, handing off to fallback chain", phase);
recordPrimary(false);
break;
}
// 成功
if (lastResult.errorMessage() == null || lastResult.errorType() == ErrorType.NONE) {
recordPrimary(true);
addToPool(primaryProviderId);
logPerfSummary(phase, conversationId, callStartMs, llmCallCount, retryCount, failoverCount);
return lastResult;
}
// Any other non-null errored result with a classified type that doStreamCall
// chose NOT to retry (i.e. UNKNOWN, or RATE_LIMIT/SERVER_ERROR past MAX_RETRIES)
// must exit — otherwise we silently spin through attempts and waste seconds
// per turn on unrecoverable errors like DashScope's "url error" / unknown model.
recordPrimary(false);
logPerfSummary(phase, conversationId, callStartMs, llmCallCount, retryCount, failoverCount);
return lastResult;
}
// lastResult == null 表示需要重试
}
// If we exhausted the retry loop without a verdict, primary effectively failed.
if (!primarySkipped && lastResult != null && lastResult.errorType() != ErrorType.NONE) {
recordPrimary(false);
}
// Primary exhausted retries — walk the fallback chain in priority order.
// Each fallback gets a single shot (no retry); first successful result wins.
// Same-instance entries (e.g., primary accidentally included in the chain)
// are skipped so we don't re-try the exact model that just failed.
// Providers in cooldown (RFC-009 P3.3) are also skipped so a known-bad
// provider doesn't add latency to every conversation turn.
for (int i = 0; i < fallbackChain.size(); i++) {
vip.mate.llm.failover.FallbackEntry entry = fallbackChain.get(i);
ChatModel fallback = entry.chatModel();
if (fallback == chatModel) continue;
// RFC-009 Phase 4 — pool gate (the real runtime fence). A provider
// HARD-removed earlier (or by another conversation) must not even
// be attempted here. Build-time filtering is best-effort; this is
// the one that matters when pool state changes mid-conversation.
if (!inPool(entry.providerId())) {
log.info("[{}] Skipping fallback {}/{} provider={} — not in pool",
phase, i + 1, fallbackChain.size(), entry.providerId());
continue;
}
if (healthTracker != null && healthTracker.isInCooldown(entry.providerId())) {
log.info("[{}] Skipping fallback {}/{} provider={} — in cooldown",
phase, i + 1, fallbackChain.size(), entry.providerId());
continue;
}
log.warn("[{}] Primary exhausted, trying fallback {}/{} provider={} ({}) for conversation {}",
phase, i + 1, fallbackChain.size(), entry.providerId(),
fallback.getClass().getSimpleName(), conversationId);
if (broadcast) {
broadcastDelta(conversationId, "warning",
buildDeltaJson("主模型不可用，正在切换到备选模型 (" + (i + 1) + "/" + fallbackChain.size() + ")..."));
}
failoverCount++;
llmCallCount++;
StreamResult fallbackResult = doStreamCall(fallback, prompt, conversationId,
phase + "_fallback_" + (i + 1), broadcast, 0);
// Accept only fully successful fallbacks. Non-successful results (auth
// error, client error, still-rate-limited) propagate to the next
// fallback instead of being surfaced as the final result.
if (fallbackResult != null
&& fallbackResult.errorType() == ErrorType.NONE
&& fallbackResult.errorMessage() == null) {
if (healthTracker != null) healthTracker.recordSuccess(entry.providerId());
addToPool(entry.providerId());
logPerfSummary(phase, conversationId, callStartMs, llmCallCount, retryCount, failoverCount);
return fallbackResult;
}
if (healthTracker != null) healthTracker.recordFailure(entry.providerId());
if (fallbackResult != null) {
// RFC-009 Phase 4: HARD errors evict from the pool so later
// walks skip this provider outright. SOFT errors keep it
// in-pool and let the tracker's cooldown absorb the blip.
removeFromPool(entry.providerId(), fallbackResult.errorType(), fallbackResult.errorMessage());
lastResult = fallbackResult; // remember most recent to report if the whole chain fails
}
}
logPerfSummary(phase, conversationId, callStartMs, llmCallCount, retryCount, failoverCount);
return lastResult != null ? lastResult
: buildErrorResult("LLM 调用失败，已达最大重试次数", conversationId, phase);
}
/** D-6: log a structured performance summary for the LLM call phase. */
private void logPerfSummary(String phase, String conversationId, long startMs,
int llmCallCount, int retryCount, int failoverCount) {
long totalMs = System.currentTimeMillis() - startMs;
log.info("[{}] perf_summary: conversationId={} total_ms={} llm_call_count={} retry_count={} failover_count={}",
phase, conversationId, totalMs, llmCallCount, retryCount, failoverCount);
}
/**
* 单次流式调用尝试。
* @return StreamResult 如果成功/降级/不可重试；null 如果应该重试
*/
private StreamResult doStreamCall(ChatModel chatModel, Prompt prompt,
String conversationId, String phase,
boolean broadcast, int attempt) {
// PR-2 L4 (RFC-049 §2.4.2): normalize as a pre-egress step (not only on retry).
// Strip reasoning_content from prior-turn AssistantMessages (i <= lastUserIdx),
// preserving in-turn thinking (i > lastUserIdx) so DeepSeek's contract holds.
// The returned Prompt shares `options` by reference with the input prompt.
Prompt outbound = stripThinkingFromPrompt(prompt);
// RFC-049 follow-up (2026-04-27): trim trailing AssistantMessage from the
// outbound prompt. Triggered in practice by the summarizing→reasoning
// graph transition: the summarizer emits an in-turn AssistantMessage,
// graph state ends with it, reasoning's next LLM call sends history
// ending with assistant. Anthropic Claude returns 400 "does not
// support assistant message prefill"; some DeepSeek model variants 400
// similarly. The dropped assistant is summarizer scaffolding, not
// user-relevant content, so removing it before egress is safe.
outbound = dropTrailingAssistant(outbound);
// PR-2 L3 (RFC-049 §2.3.2): producer-side relay stash. Extract per-assistant
// thinking from the normalized prompt (cross-turn positions are already "" due
// to strip), stash with the caller's original `user` field, and overwrite
// `options.user` with the relay token. The consumer in
// AgentGraphBuilder.patchReasoningContent restores the original user when
// rebuilding the outbound ChatCompletionRequest; the token never reaches the
// provider. We only activate relay on OpenAiChatOptions paths — Anthropic has
// its own thinking mechanism (extended thinking via AnthropicChatOptions.thinking).
String relayToken = null;
String originalUser = null;
org.springframework.ai.openai.OpenAiChatOptions oaiOptsForRelay = null;
if (outbound.getOptions() instanceof org.springframework.ai.openai.OpenAiChatOptions oaiOpts) {
List This is the linchpin of the structural fix: without writing thinking back into
* the AssistantMessage that enters the next ReAct round's state, the outbound
* request's {@code reasoning_content} is lost (Spring AI 1.1.4's
* {@code OpenAiChatModel.lambda$createRequest$20} hardcodes {@code null} on the
* outbound conversion, so the relay in {@code AssistantThinkingRelay} is the only
* way back — see RFC-049 §2.3 L3).
*
* Note the Spring AI naming asymmetry: the builder method is
* {@code .properties(Map)} but the reader is {@code getMetadata()} (see
* {@link #stripThinkingFromPrompt} L937).
*/
private static AssistantMessage buildAssistantMessageWithThinking(
String fullContent, String fullThinking, List PR-2 L4 (RFC-049 §2.4.1): The old semantics "keep only the last AssistantMessage's
* thinking" broke DeepSeek's contract for multi-round tool-calls within a single user
* turn (DeepSeek requires all in-turn assistant thinking to be passed back on subsequent
* rounds). Now the boundary is the most recent {@link UserMessage}: AssistantMessages at
* index {@code <= lastUserIdx} are prior-turn history (their thinking must be stripped
* per DeepSeek's "reset across user turns" rule); AssistantMessages at {@code > lastUserIdx}
* are in-turn (their thinking must be preserved).
*
* PR-2 L4 (RFC-049 §2.4.2): This method is called as a normal pre-egress step from
* {@link #doStreamCall}, not only from the {@code THINKING_BLOCK_ERROR} retry path. The
* retry path still calls it too (idempotent), serving as defensive re-application.
*
* Note: {@code Prompt.getOptions()} is preserved by reference into the returned
* {@code Prompt} (this is existing behavior). Callers rely on that — mutations to
* {@code options.user} via {@link AssistantThinkingRelay} must stay visible after
* normalize.
*/
static Prompt stripThinkingFromPrompt(Prompt prompt) {
List
* Spring AI 1.1.3 的 OpenAiChatModel 在流式路径中会将 delta.reasoning_content
* 放入 properties 的 "reasoningContent" key。
*/
private String extractReasoningContent(AssistantMessage msg) {
Map Algorithm: find the smallest period in {@code [minPeriod,
* maxPeriod]} where the buffer ends with that unit repeated 2+
* times consecutively, then return everything up to (and including)
* the FIRST copy of that unit. Conservative — if no period yields
* 2+ consecutive matches, returns the buffer unchanged.
*
* Public for unit-testing alongside {@link #hasRepeatingSuffix}.
*/
public static String dedupTrailingRepeats(String content, int minPeriod, int maxPeriod) {
if (content == null || content.isEmpty()) return content;
int len = content.length();
if (minPeriod <= 0 || maxPeriod < minPeriod) return content;
int periodCap = Math.min(maxPeriod, len / 2);
for (int p = minPeriod; p <= periodCap; p++) {
int unitStart = len - p;
// Walk backward as far as the unit keeps matching.
int copies = 1;
int blockStart = unitStart - p;
while (blockStart >= 0
&& content.regionMatches(blockStart, content, unitStart, p)) {
copies++;
blockStart -= p;
}
if (copies >= 2) {
// Keep prefix + ONE copy. The first copy starts at
// (blockStart + p) since the loop walked back one step
// past the last match.
int firstCopyStart = blockStart + p;
int trimEnd = firstCopyStart + p;
return content.substring(0, trimEnd);
}
}
return content;
}
/**
* Detect whether {@code accum} ends with the same {@code period}-sized
* unit repeated at least {@code minOccurrences} times consecutively,
* for some {@code period} in {@code [minPeriod, maxPeriod]}. Returns
* true when the model is stuck in a "self-arguing" loop emitting the
* same final-answer chunk over and over.
*
* Algorithm: probe period sizes from small to large. For each
* candidate period {@code p}, take the last {@code p} chars as the
* unit and check whether the {@code minOccurrences-1} preceding
* blocks of length {@code p} are byte-identical. The smallest period
* that yields the required consecutive copies trips the guard. We
* iterate small→large because tighter periods are more specific:
* a 30-char unit repeated 4× is a stronger signal than a 200-char
* unit happening to appear once.
*
* Cost: O(periodRange × occurrences × period) char comparisons.
* For default thresholds (~200 × 4 × 100) that's ~80K comparisons
* per scan — microseconds against an LLM call. Throttled by the
* caller via {@code lastContentRepeatCheckLen} so the scan amortizes.
*
* Package-private + static for unit-testing the threshold without
* spinning up a full {@code StreamResult}.
*/
static boolean hasRepeatingSuffix(CharSequence accum, int minPeriod, int maxPeriod,
int minOccurrences) {
if (accum == null) return false;
int len = accum.length();
if (minPeriod <= 0 || minOccurrences <= 1 || maxPeriod < minPeriod) return false;
if (len < minPeriod * minOccurrences) return false;
String s = accum.toString();
int periodCap = Math.min(maxPeriod, len / minOccurrences);
for (int p = minPeriod; p <= periodCap; p++) {
// Unit = last p chars. Check prior (minOccurrences - 1)
// blocks of length p match the unit byte-for-byte.
int unitStart = len - p;
boolean allMatch = true;
for (int k = 2; k <= minOccurrences; k++) {
int blockStart = len - k * p;
if (blockStart < 0) { allMatch = false; break; }
if (!s.regionMatches(blockStart, s, unitStart, p)) {
allMatch = false;
break;
}
}
if (allMatch) return true;
}
return false;
}
/**
* Best-effort character count of the outbound prompt for the
* {@code context_prepared} event. Cheaper than tokenizing and only used
* for UI presentation, so an exact figure is unnecessary.
*/
private static int approximatePromptChars(Prompt prompt) {
if (prompt == null || prompt.getInstructions() == null) return 0;
int total = 0;
for (Message m : prompt.getInstructions()) {
String text = m.getText();
if (text != null) total += text.length();
}
return total;
}
/**
* Pick a stable model identifier from whatever {@link ChatModel}
* implementation we received — Spring AI doesn't expose a single accessor.
* We try the well-known fields by reflection so this stays decoupled from
* concrete provider classes (Anthropic / OpenAI / DashScope all expose
* {@code defaultOptions.model} or equivalent).
*/
private static String identifyModel(ChatModel chatModel) {
if (chatModel == null) return "";
try {
// Common Spring AI shape: getDefaultOptions().getModel()
java.lang.reflect.Method getDefaultOptions = chatModel.getClass().getMethod("getDefaultOptions");
Object opts = getDefaultOptions.invoke(chatModel);
if (opts != null) {
try {
java.lang.reflect.Method getModel = opts.getClass().getMethod("getModel");
Object model = getModel.invoke(opts);
if (model != null) return model.toString();
} catch (NoSuchMethodException ignored) {
// fall through
}
}
} catch (Exception ignored) {
// fall through to class-name fallback
}
return chatModel.getClass().getSimpleName();
}
/**
* 构建 {"delta":"..."} JSON
*/
private static String buildDeltaJson(String delta) {
StringBuilder sb = new StringBuilder("{\"delta\":\"");
for (int k = 0; k < delta.length(); k++) {
char c = delta.charAt(k);
if (c == '"') sb.append("\\\"");
else if (c == '\\') sb.append("\\\\");
else if (c == '\n') sb.append("\\n");
else if (c == '\t') sb.append("\\t");
else if (c == '\r') sb.append("\\r");
else sb.append(c);
}
sb.append("\"}");
return sb.toString();
}
/**
* 累积 tool call 分片。
*
* 流式模式下 tool calls 可能分多个 chunk 到来：
* - 第一个 chunk 携带 id、name 和部分 arguments
* - 后续 chunk 只有 arguments 增量
*
* 采用增量累积方式合并分片 tool_call。
*/
private void accumulateToolCalls(List
* Some providers (e.g. aliyun-codingplan) reject the entire follow-up
* request with HTTP 400 when the assistant message in history carries a
* tool call whose {@code arguments} is not parseable JSON. Streaming
* accumulation can produce such payloads when:
*
*
* Both cases are normalized to {@code "{}"} so the chat-completions
* round-trip stays valid. Tool execution downstream still re-validates
* arguments and surfaces a per-tool error if the empty payload is wrong
* for that tool.
*/
private static String sanitizeToolCallArguments(String toolName, String arguments) {
if (arguments == null || arguments.isBlank()) {
return "{}";
}
try {
TOOL_ARG_JSON_MAPPER.readTree(arguments);
return arguments;
} catch (Exception e) {
log.warn("Tool '{}' arguments are not valid JSON after stream aggregation "
+ "(len={}, head={}); replacing with empty object so the "
+ "follow-up chat-completions request stays well-formed. "
+ "Parse error: {}",
toolName,
arguments.length(),
arguments.substring(0, Math.min(80, arguments.length())),
e.getMessage());
return "{}";
}
}
private static class ToolCallAccumulator {
String id;
String type;
String name;
StringBuilder arguments = new StringBuilder();
}
// ====================