diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java index 4ac2c382..8a8291fc 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java @@ -408,7 +408,14 @@ public class AgentGraphBuilder { primaryModelConfig != null ? primaryModelConfig.getProvider() : null, providerPool); ToolExecutionExecutor executor = new ToolExecutionExecutor(toolSet, toolGuardService, approvalService, streamTracker, toolTimeoutProperties, toolResultStorage, toolConcurrencyRegistry); - ReasoningNode reasoningNode = new ReasoningNode(chatModel, toolSet, reasoningEffort, streamingHelper, conversationWindowManager, streamTracker, 0, wikiContextService); + // PR-1.2 (RFC-049 L1-B): propagate the bound model's capability so ReasoningNode + // can gate the ThinkingLevelHolder override explicitly, rather than inferring + // capability from reasoningEffort == null. + boolean supportsReasoningEffort = primaryModelConfig != null + && ModelFamily.detect(primaryModelConfig.getModelName()).supportsReasoningEffort(); + ReasoningNode reasoningNode = new ReasoningNode(chatModel, toolSet, reasoningEffort, + supportsReasoningEffort, + streamingHelper, conversationWindowManager, streamTracker, 0, wikiContextService); ActionNode actionNode = new ActionNode(executor, streamTracker); ObservationProcessor observationProcessor = new ObservationProcessor(graphObservationProperties); ObservationNode observationNode = new ObservationNode(observationProcessor, streamTracker); @@ -1037,7 +1044,8 @@ public class AgentGraphBuilder { public org.springframework.http.ResponseEntity chatCompletionEntity( OpenAiApi.ChatCompletionRequest chatRequest, MultiValueMap additionalHttpHeader) { - chatRequest = patchReasoningContent(chatRequest); + chatRequest = sanitizeReasoningEffortForProvider(chatRequest, provider); + chatRequest = patchReasoningContent(chatRequest, provider); chatRequest = stripReasoningEffortIfIncompatible(chatRequest); chatRequest = patchVideoMediaContent(chatRequest); if (kimiSearchEnabled) { @@ -1056,7 +1064,8 @@ public class AgentGraphBuilder { public Flux chatCompletionStream( OpenAiApi.ChatCompletionRequest chatRequest, MultiValueMap additionalHttpHeader) { - chatRequest = patchReasoningContent(chatRequest); + chatRequest = sanitizeReasoningEffortForProvider(chatRequest, provider); + chatRequest = patchReasoningContent(chatRequest, provider); chatRequest = stripReasoningEffortIfIncompatible(chatRequest); chatRequest = patchVideoMediaContent(chatRequest); if (kimiSearchEnabled) { @@ -1108,13 +1117,28 @@ public class AgentGraphBuilder { } private String resolveReasoningEffort(String modelName, Map kwargs, ModelFamily family) { - // generateKwargs 显式覆盖始终优先 + // PR-1.1 (RFC-049 L1-A): Only families that actually accept reasoning_effort may receive + // it. Previously only the default-inject branch checked capability; the generateKwargs + // override branch did not, so a provider-level `reasoningEffort: "high"` would leak to + // deepseek-chat / kimi-k2 / deepseek-reasoner etc., triggering the incident documented + // in RFC-049 (DeepSeek "reasoning_content missing" 400). + if (!family.supportsReasoningEffort()) { + Object overridden = findOptionValue(kwargs, "reasoningEffort"); + if (overridden != null) { + log.warn("Dropping reasoningEffort='{}' from generateKwargs — model '{}' (family={}) " + + "does not accept reasoning_effort. For DeepSeek thinking use " + + "extra_body.thinking; for Kimi thinking the model activates it natively.", + overridden, modelName, family); + } + return null; + } + // generateKwargs 显式覆盖始终优先（仅在白名单族内） Object value = findOptionValue(kwargs, "reasoningEffort"); if (value instanceof String text && StringUtils.hasText(text)) { return text.trim(); } // 仅支持 reasoning_effort 的模型族才自动注入默认值 - if (family.isThinking() && family.supportsReasoningEffort()) { + if (family.isThinking()) { return "medium"; } return null; @@ -1376,62 +1400,135 @@ public class AgentGraphBuilder { } /** - * 修补 assistant 消息缺失的 reasoningContent 字段。 - *

- * Spring AI 1.1.3 在将 AssistantMessage 转回 ChatCompletionMessage 时不会设置 reasoningContent， - * 导致某些启用 thinking 模式的 API（如 Kimi K2.5）在多轮对话中报错： - * "thinking is enabled but reasoning_content is missing in assistant tool call message" - *

- * 触发条件（放宽）： - *

    - *
  • 条件 A：请求明确设置了 reasoningEffort
    • - *
  • 条件 B：消息历史中已有 assistant 消息携带 reasoningContent（说明模型天然启用了 thinking）
    • - *
- * 修复策略：为缺失 reasoningContent 的 assistant tool_call 消息注入空字符串 "" 以满足 API 校验。 - * 使用 record canonical constructor 重建 ChatCompletionRequest，避免反射修改不可变字段。 + * Consume the {@link AssistantThinkingRelay} entry and rebuild the outbound + * {@link OpenAiApi.ChatCompletionRequest} so that assistant tool-call / thinking + * messages carry the correct {@code reasoning_content}. + * + *

PR-2 (RFC-049 §2.3.2): This is the consumer side of the relay. + * {@code NodeStreamingChatHelper.doStreamCall} stashes per-assistant thinking + * keyed on a token embedded in {@code request.user()}. Here we: + *

    + *
  1. {@link AssistantThinkingRelay#take(String)} the entry and restore + * {@code request.user()} to {@code entry.originalUser()} (internal token + * never reaches the provider).
    2. + *
  2. Compute {@code lastUserIdx} (the boundary of the current user turn), + * symmetric to {@code stripThinkingFromPrompt}. Assistant messages at + * {@code i <= lastUserIdx} are prior-turn history: their + * {@code reasoning_content} must stay null. Only {@code i > lastUserIdx} + * messages are eligible for patching.
    3. + *
  3. Select a {@link FallbackPolicy} by {@code providerId}. When relay has + * a real value, we use it; when empty, the policy decides whether to + * inject {@code " "} (legacy tolerance: KIMI/OPENAI/DEFAULT) or leave + * {@code null} to surface an explicit provider error (DEEPSEEK).
    4. + *
+ * + *

The relay iterator advances for every assistant message (including + * prior-turn ones) to stay positionally aligned with the producer's extraction + * in {@code NodeStreamingChatHelper.extractAssistantThinkings}. */ - private static OpenAiApi.ChatCompletionRequest patchReasoningContent(OpenAiApi.ChatCompletionRequest request) { + static OpenAiApi.ChatCompletionRequest patchReasoningContent( + OpenAiApi.ChatCompletionRequest request, ModelProviderEntity provider) { if (request.messages() == null || request.messages().isEmpty()) { return request; } - // 判断是否处于 thinking 模式 - boolean thinkingMode = request.reasoningEffort() != null; + // 1. Consume relay (if any) and compute the sanitized user field. + AssistantThinkingRelay.RelayEntry entry = AssistantThinkingRelay.take(request.user()); + String sanitizedUser = (entry != null) + ? entry.originalUser() + : (AssistantThinkingRelay.isToken(request.user()) ? null : request.user()); + + // 2. Detect thinking mode — unchanged from the prior design except that relay + // presence is also a trigger. + boolean thinkingMode = request.reasoningEffort() != null + || requiresReasoningContentPatch(request.model()) + || request.messages().stream().anyMatch(m -> + m.role() == OpenAiApi.ChatCompletionMessage.Role.ASSISTANT + && m.reasoningContent() != null) + || entry != null; if (!thinkingMode) { - thinkingMode = requiresReasoningContentPatch(request.model()); - } - if (!thinkingMode) { - thinkingMode = request.messages().stream().anyMatch(msg -> - msg.role() == OpenAiApi.ChatCompletionMessage.Role.ASSISTANT - && msg.reasoningContent() != null); - } - if (!thinkingMode) { - return request; + // Nothing to patch but we may still need to strip a leaked relay token from user. + return request.user() != null && !request.user().equals(sanitizedUser) + ? rebuildWithUser(request, sanitizedUser) + : request; } - // 检查是否有需要补丁的消息 - boolean needsPatch = request.messages().stream().anyMatch(msg -> - msg.role() == OpenAiApi.ChatCompletionMessage.Role.ASSISTANT - && msg.toolCalls() != null && !msg.toolCalls().isEmpty() - && msg.reasoningContent() == null); - if (!needsPatch) { - return request; - } - - // 重建消息列表，为缺失 reasoningContent 的 assistant tool call 消息注入 "" - List patched = request.messages().stream().map(msg -> { - if (msg.role() == OpenAiApi.ChatCompletionMessage.Role.ASSISTANT - && msg.toolCalls() != null && !msg.toolCalls().isEmpty() - && msg.reasoningContent() == null) { - return new OpenAiApi.ChatCompletionMessage( - msg.rawContent(), msg.role(), msg.name(), msg.toolCallId(), - msg.toolCalls(), msg.refusal(), msg.audioOutput(), - msg.annotations(), " "); + // 3. Find lastUserIdx so we can skip cross-turn assistants. + int lastUserIdx = -1; + for (int i = request.messages().size() - 1; i >= 0; i--) { + if (request.messages().get(i).role() == OpenAiApi.ChatCompletionMessage.Role.USER) { + lastUserIdx = i; + break; } - return msg; - }).toList(); + } - // 用 record canonical constructor 重建 request（不用反射） + FallbackPolicy policy = FallbackPolicy.forProvider(provider); + java.util.Iterator it = (entry != null) + ? entry.thinkings().iterator() + : java.util.Collections.emptyIterator(); + + // 4. Walk messages, patching only in-turn assistants; always advance iterator + // for all assistants so producer/consumer positions stay aligned. + boolean anyPatched = false; + List patched = new ArrayList<>(request.messages().size()); + for (int i = 0; i < request.messages().size(); i++) { + OpenAiApi.ChatCompletionMessage msg = request.messages().get(i); + if (msg.role() != OpenAiApi.ChatCompletionMessage.Role.ASSISTANT) { + patched.add(msg); + continue; + } + + String next = it.hasNext() ? it.next() : null; + + // Already has a real value: leave alone + if (msg.reasoningContent() != null && !msg.reasoningContent().isBlank()) { + patched.add(msg); + continue; + } + + // Cross-turn assistant: never patch (symmetric with stripThinkingFromPrompt) + if (i <= lastUserIdx) { + patched.add(msg); + continue; + } + + boolean hasToolCalls = msg.toolCalls() != null && !msg.toolCalls().isEmpty(); + if (!hasToolCalls && !policy.patchNonToolCall) { + patched.add(msg); + continue; + } + + String injected; + if (next != null && !next.isEmpty()) { + injected = next; + } else { + injected = policy.emptyFallback; + if (injected == null && policy.warnOnMissingReal) { + log.warn("[patchReasoningContent] provider={} requires real reasoning_content " + + "but relay has no value for assistant message at index {}; " + + "leaving null so provider returns explicit error.", + providerIdOrUnknown(provider), i); + } + } + if (injected == null && msg.reasoningContent() == null) { + // No change — keep original + patched.add(msg); + continue; + } + patched.add(new OpenAiApi.ChatCompletionMessage( + msg.rawContent(), msg.role(), msg.name(), msg.toolCallId(), + msg.toolCalls(), msg.refusal(), msg.audioOutput(), + msg.annotations(), injected)); + anyPatched = true; + } + + boolean userChanged = request.user() != null && !request.user().equals(sanitizedUser) + || (request.user() == null && sanitizedUser != null); + if (!anyPatched && !userChanged) { + return request; + } + + // 5. Rebuild with patched messages + sanitized user. return new OpenAiApi.ChatCompletionRequest( patched, request.model(), @@ -1458,7 +1555,7 @@ public class AgentGraphBuilder { request.tools(), request.toolChoice(), request.parallelToolCalls(), - request.user(), + sanitizedUser, request.reasoningEffort(), request.webSearchOptions(), request.verbosity(), @@ -1468,6 +1565,211 @@ public class AgentGraphBuilder { ); } + /** + * PR-2 (RFC-049 §2.3.2): Provider-keyed policy for how {@code patchReasoningContent} + * should behave when the relay has no real thinking for an in-turn assistant message. + * + *

    + *
  • {@code emptyFallback}: value to inject when relay has no real value — + * {@code null} means leave {@code reasoning_content} null (DeepSeek); + * {@code " "} preserves Spring AI 1.1.4 legacy tolerance (Kimi/OpenAI/unknown).
    • + *
  • {@code warnOnMissingReal}: emit WARN when {@code emptyFallback==null} fires — + * only DeepSeek wants this, because there a missing value means we have a bug.
    • + *
  • {@code patchNonToolCall}: whether to patch assistant messages without tool_calls — + * DeepSeek's contract applies to all in-turn assistant messages, others only + * to tool_call messages (historical behavior).
    • + *
+ * + * {@code DEFAULT} intentionally keeps the legacy {@code " "} tolerance rather than + * going no-op: an unrecognized provider (self-hosted DeepSeek-like backend, custom + * OpenAI-compatible gateway) might still require the patch — noop would regress + * those into new 400s. + */ + private enum FallbackPolicy { + DEEPSEEK(null, true, true), + KIMI (" ", false, false), + OPENAI (" ", false, false), + DEFAULT (" ", false, false); + + final String emptyFallback; + final boolean warnOnMissingReal; + final boolean patchNonToolCall; + + FallbackPolicy(String emptyFallback, boolean warnOnMissingReal, boolean patchNonToolCall) { + this.emptyFallback = emptyFallback; + this.warnOnMissingReal = warnOnMissingReal; + this.patchNonToolCall = patchNonToolCall; + } + + static FallbackPolicy forProvider(ModelProviderEntity provider) { + if (provider == null || provider.getProviderId() == null) { + return DEFAULT; + } + String id = provider.getProviderId().toLowerCase(); + return switch (id) { + case "deepseek" -> DEEPSEEK; + case "kimi-cn", "kimi-intl", "kimi-code" -> KIMI; + case "openai", "azure-openai" -> OPENAI; + default -> DEFAULT; + }; + } + } + + /** + * Rebuild a {@link OpenAiApi.ChatCompletionRequest} with only the {@code user} field + * replaced. Used when {@code patchReasoningContent} has no assistant-message changes + * but must strip a relay token from the outbound {@code user} field. + */ + private static OpenAiApi.ChatCompletionRequest rebuildWithUser( + OpenAiApi.ChatCompletionRequest request, String newUser) { + return new OpenAiApi.ChatCompletionRequest( + request.messages(), + request.model(), + request.store(), + request.metadata(), + request.frequencyPenalty(), + request.logitBias(), + request.logprobs(), + request.topLogprobs(), + request.maxTokens(), + request.maxCompletionTokens(), + request.n(), + request.outputModalities(), + request.audioParameters(), + request.presencePenalty(), + request.responseFormat(), + request.seed(), + request.serviceTier(), + request.stop(), + request.stream(), + request.streamOptions(), + request.temperature(), + request.topP(), + request.tools(), + request.toolChoice(), + request.parallelToolCalls(), + newUser, + request.reasoningEffort(), + request.webSearchOptions(), + request.verbosity(), + request.promptCacheKey(), + request.safetyIdentifier(), + request.extraBody() + ); + } + + /** + * PR-1.3 (RFC-049 L1-C): Provider-first sanitization of {@code reasoning_effort}. + * + *

Authoritative judgement uses the target {@code provider.getProviderId()} as a + * whitelist (default-deny). Only OpenAI official providers are allowed to carry + * {@code reasoning_effort}; everything else — known non-supporters (DeepSeek / Kimi / + * DashScope / Ollama / …) and any unrecognized providerId (self-hosted gateways, + * OpenRouter / Together / aggregators) — is stripped unconditionally. + * + *

The reason we intentionally distrust {@code request.model()} here: MateClaw's + * failover chain (RFC-009) can reuse the same {@code Prompt} and {@code OpenAiChatOptions} + * across providers, and {@code OpenAiChatOptions.model} was set to the primary's model + * name (e.g. {@code gpt-5}). If the sanitizer only checked {@code ModelFamily.detect( + * request.model())}, a failover hop from GPT-5 → DeepSeek would see model name + * "gpt-5" → OPENAI_REASONING → {@code supportsReasoningEffort == true} and quietly + * forward the primary's {@code reasoning_effort} to DeepSeek, re-triggering the + * incident this RFC exists to fix. + * + *

Only when the provider is on the whitelist do we fall through to the + * {@link ModelFamily} check (e.g. within OpenAI, {@code gpt-4} still wouldn't support + * reasoning_effort). Outside the whitelist, no runtime check on model is trusted. + * + *

Adding a new provider to the whitelist must be an explicit PR with a sanitizer + * test — do not add a catch-all default-allow branch. + */ + static OpenAiApi.ChatCompletionRequest sanitizeReasoningEffortForProvider( + OpenAiApi.ChatCompletionRequest request, ModelProviderEntity provider) { + if (request == null || request.reasoningEffort() == null) { + return request; + } + + if (!isReasoningEffortWhitelistedProvider(provider)) { + log.warn("[reasoning_effort sanitizer] provider={} is not on the reasoning_effort " + + "whitelist (only openai/azure-openai are); stripping value='{}' " + + "(request.model()='{}' may be leaked from failover primary).", + providerIdOrUnknown(provider), request.reasoningEffort(), request.model()); + return rebuildWithReasoningEffort(request, null); + } + + ModelFamily targetFamily = ModelFamily.detect(request.model()); + if (!targetFamily.supportsReasoningEffort()) { + log.warn("[reasoning_effort sanitizer] provider={} model={} family={} does not " + + "support reasoning_effort; stripping value='{}'.", + provider.getProviderId(), request.model(), targetFamily, request.reasoningEffort()); + return rebuildWithReasoningEffort(request, null); + } + return request; + } + + /** + * Whitelist of providers known to accept {@code reasoning_effort} on + * {@code /v1/chat/completions} (or {@code /v1/responses}). Anything else is denied. + * Adding a provider here must come with a corresponding sanitizer test case. + */ + static boolean isReasoningEffortWhitelistedProvider(ModelProviderEntity provider) { + if (provider == null || provider.getProviderId() == null) { + return false; + } + String id = provider.getProviderId().toLowerCase(); + return switch (id) { + case "openai", "azure-openai" -> true; + default -> false; + }; + } + + private static String providerIdOrUnknown(ModelProviderEntity p) { + return (p == null || p.getProviderId() == null) ? "" : p.getProviderId(); + } + + /** + * Rebuild a {@link OpenAiApi.ChatCompletionRequest} with a new {@code reasoningEffort} + * value (typically {@code null} to strip). Mirrors the record canonical-constructor + * pattern used by {@link #stripReasoningEffortIfIncompatible}. + */ + private static OpenAiApi.ChatCompletionRequest rebuildWithReasoningEffort( + OpenAiApi.ChatCompletionRequest request, String newReasoningEffort) { + return new OpenAiApi.ChatCompletionRequest( + request.messages(), + request.model(), + request.store(), + request.metadata(), + request.frequencyPenalty(), + request.logitBias(), + request.logprobs(), + request.topLogprobs(), + request.maxTokens(), + request.maxCompletionTokens(), + request.n(), + request.outputModalities(), + request.audioParameters(), + request.presencePenalty(), + request.responseFormat(), + request.seed(), + request.serviceTier(), + request.stop(), + request.stream(), + request.streamOptions(), + request.temperature(), + request.topP(), + request.tools(), + request.toolChoice(), + request.parallelToolCalls(), + request.user(), + newReasoningEffort, + request.webSearchOptions(), + request.verbosity(), + request.promptCacheKey(), + request.safetyIdentifier(), + request.extraBody() + ); + } + /** * GPT-5 兼容性：在 /v1/chat/completions 路径下，tools 与 reasoning_effort 不可同时存在。 *

diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AssistantThinkingRelay.java b/mateclaw-server/src/main/java/vip/mate/agent/AssistantThinkingRelay.java new file mode 100644 index 00000000..88736364 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/AssistantThinkingRelay.java @@ -0,0 +1,101 @@ +package vip.mate.agent; + +import java.util.List; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Relays per-request assistant {@code reasoning_content} from the producer + * ({@code NodeStreamingChatHelper}, which sees {@code AssistantMessage.metadata}) + * to the consumer ({@code AgentGraphBuilder.patchReasoningContent}, which rebuilds + * the outbound {@code ChatCompletionRequest}). + * + *

Why not {@link ThreadLocal}: {@code OpenAiChatModel.stream()} hops to + * {@code boundedElastic} via {@code subscribeOn}, so a {@code ThreadLocal} on the + * caller does not propagate across the producer/consumer boundary. The relay + * token travels inside the request object itself + * ({@code OpenAiApi.ChatCompletionRequest.user}), which survives scheduler hops + * without needing Reactor context propagation config. + * + *

The {@link RelayEntry} carries both the per-assistant thinking list and the + * caller's original {@code user} field — the producer overwrites + * {@code OpenAiChatOptions.user} with the relay token before handing the + * {@code Prompt} to Spring AI, so by the time the consumer runs, + * {@code request.user()} only contains the token. The consumer restores the + * caller's original value from the entry when rebuilding the outbound request. + * The internal token is never sent to the provider. + * + *

Ownership: the producer is responsible for calling {@link #discard(String)} + * in a {@code finally} block as a belt-and-suspenders cleanup. The consumer's + * {@link #take(String)} already removes the entry on the happy path, so + * {@code discard} is a no-op in that case; it becomes the only cleanup when the + * consumer never runs (e.g., a Reactor error before the request is dispatched). + * + * @author MateClaw Team + */ +public final class AssistantThinkingRelay { + + /** + * Per-request relay payload. + * + * @param thinkings per-assistant {@code reasoning_content} in message order; + * empty string means "this assistant had no thinking" + * @param originalUser the caller's original {@code OpenAiChatOptions.user} value + * before the producer overwrote it with the relay token; + * may be {@code null} + */ + public record RelayEntry(List thinkings, String originalUser) { + public RelayEntry { + thinkings = List.copyOf(thinkings); + } + } + + private static final ConcurrentHashMap MAP = new ConcurrentHashMap<>(); + + /** Prefix must be distinctive enough that a caller-provided {@code user} value + * can never collide with a relay token. */ + public static final String TOKEN_PREFIX = "__mc_thinking_"; + + private AssistantThinkingRelay() {} + + /** + * Stash per-assistant thinking (in message order) plus the caller's original + * {@code user} field. Returns the token to embed in + * {@code OpenAiChatOptions.user}. + */ + public static String stash(List thinkingsInOrder, String originalUser) { + String token = TOKEN_PREFIX + UUID.randomUUID(); + MAP.put(token, new RelayEntry(thinkingsInOrder, originalUser)); + return token; + } + + /** Consume and remove entry. Returns {@code null} if {@code user} is not a + * relay token or the entry was already taken. */ + public static RelayEntry take(String user) { + if (!isToken(user)) return null; + return MAP.remove(user); + } + + /** Whether the given {@code user} field value is a relay token produced by + * {@link #stash(List, String)}. */ + public static boolean isToken(String user) { + return user != null && user.startsWith(TOKEN_PREFIX); + } + + /** Defensive cleanup; idempotent — safe to call even after {@link #take}. */ + public static void discard(String token) { + if (token != null) MAP.remove(token); + } + + // ---------- test hooks ---------- + + /** Visible for tests: current map size. Production code must not use. */ + static int size() { + return MAP.size(); + } + + /** Visible for tests: clear all entries. Production code must not use. */ + static void clearAll() { + MAP.clear(); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java index 01bf666a..7cf4d910 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java @@ -3,10 +3,12 @@ package vip.mate.agent.graph; import lombok.extern.slf4j.Slf4j; import org.springframework.ai.chat.messages.AssistantMessage; import org.springframework.ai.chat.messages.Message; +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.web.reactive.function.client.WebClientResponseException; +import vip.mate.agent.AssistantThinkingRelay; import vip.mate.channel.web.ChatStreamTracker; import reactor.core.Disposable; @@ -567,6 +569,67 @@ public class NodeStreamingChatHelper { 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); + + // 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 thinkings = extractAssistantThinkings(outbound); + if (thinkings.stream().anyMatch(s -> !s.isEmpty())) { + originalUser = oaiOpts.getUser(); + relayToken = AssistantThinkingRelay.stash(thinkings, originalUser); + oaiOpts.setUser(relayToken); + oaiOptsForRelay = oaiOpts; + } + } + + try { + return doStreamCallInner(chatModel, outbound, conversationId, phase, broadcast, attempt); + } finally { + // Idempotent: if consumer already took the entry, discard is a no-op. + if (relayToken != null) { + AssistantThinkingRelay.discard(relayToken); + if (oaiOptsForRelay != null) { + oaiOptsForRelay.setUser(originalUser); + } + } + } + } + + /** + * PR-2: Extract per-assistant {@code reasoningContent} from a Prompt's messages in + * order. Non-assistant messages are skipped; assistants with no metadata or no + * reasoningContent yield {@code ""} so the returned list's positional index aligns + * with the assistant-message index as seen by the consumer. + */ + private static List extractAssistantThinkings(Prompt prompt) { + List out = new ArrayList<>(); + for (Message m : prompt.getInstructions()) { + if (m instanceof AssistantMessage am) { + Map meta = am.getMetadata(); + Object rc = meta != null ? meta.get("reasoningContent") : null; + out.add(rc instanceof String s ? s : ""); + } + } + return out; + } + + private StreamResult doStreamCallInner(ChatModel chatModel, Prompt prompt, + String conversationId, String phase, + boolean broadcast, int attempt) { if (attempt > 0) { long delay = Math.min(BACKOFF_BASE_MS * (1L << (attempt - 1)), BACKOFF_CAP_MS); // 加入 jitter 防止雷群效应（Hermes 风格） @@ -852,9 +915,7 @@ public class NodeStreamingChatHelper { } } - AssistantMessage assembledMessage = !finalToolCalls.isEmpty() - ? AssistantMessage.builder().content(fullContent).toolCalls(finalToolCalls).build() - : new AssistantMessage(fullContent); + AssistantMessage assembledMessage = buildAssistantMessageWithThinking(fullContent, fullThinking, finalToolCalls); recordCacheMetrics(phase, promptTok, completionTok, cacheReadTok, cacheWriteTok); return new StreamResult(fullContent, fullThinking, assembledMessage, @@ -883,15 +944,7 @@ public class NodeStreamingChatHelper { } } - AssistantMessage assembledMessage; - if (!finalToolCalls.isEmpty()) { - assembledMessage = AssistantMessage.builder() - .content(fullContent) - .toolCalls(finalToolCalls) - .build(); - } else { - assembledMessage = new AssistantMessage(fullContent); - } + AssistantMessage assembledMessage = buildAssistantMessageWithThinking(fullContent, fullThinking, finalToolCalls); recordCacheMetrics(phase, promptTok, completionTok, cacheReadTok, cacheWriteTok); return new StreamResult(fullContent, fullThinking, assembledMessage, @@ -899,6 +952,33 @@ public class NodeStreamingChatHelper { partial, errorMsg, ErrorType.NONE, false, cacheReadTok, cacheWriteTok); } + /** + * PR-2 L2 (RFC-049): Build an {@link AssistantMessage} that persists the per-turn + * {@code fullThinking} into the message's properties under key {@code "reasoningContent"}. + * + *

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 finalToolCalls) { + AssistantMessage.Builder builder = AssistantMessage.builder().content(fullContent); + if (finalToolCalls != null && !finalToolCalls.isEmpty()) { + builder.toolCalls(finalToolCalls); + } + if (fullThinking != null && !fullThinking.isEmpty()) { + builder.properties(Map.of("reasoningContent", fullThinking)); + } + return builder.build(); + } + /** * Record token / cache usage to the optional metrics aggregator. * Called only from successful assembly paths ({@link #assembleResult} @@ -917,26 +997,47 @@ public class NodeStreamingChatHelper { /** 构建纯错误 StreamResult（无任何内容） */ /** - * 从 Prompt 中剥离旧 AssistantMessage 的 thinking/reasoningContent metadata。 - * 保留最新一条 AssistantMessage 的 thinking（可能是模型需要的签名）。 + * Strip {@code reasoningContent} from AssistantMessages that belong to prior + * user turns, keeping thinking for messages within the current user + * turn intact. + * + *

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. */ - private Prompt stripThinkingFromPrompt(Prompt prompt) { + static Prompt stripThinkingFromPrompt(Prompt prompt) { List messages = prompt.getInstructions(); - // 找最后一个 AssistantMessage - int lastAssistantIdx = -1; + + // Find most recent UserMessage — boundary of the current user turn + int lastUserIdx = -1; for (int i = messages.size() - 1; i >= 0; i--) { - if (messages.get(i) instanceof AssistantMessage) { - lastAssistantIdx = i; + if (messages.get(i) instanceof UserMessage) { + lastUserIdx = i; break; } } - List cleaned = new ArrayList<>(); + + int strippedCount = 0; + List cleaned = new ArrayList<>(messages.size()); for (int i = 0; i < messages.size(); i++) { Message msg = messages.get(i); - if (msg instanceof AssistantMessage am && i != lastAssistantIdx) { + // Only strip prior-turn assistant thinking (i <= lastUserIdx); in-turn (i > lastUserIdx) stays + if (msg instanceof AssistantMessage am && i <= lastUserIdx) { Map meta = am.getMetadata(); if (meta != null && meta.containsKey("reasoningContent")) { - // 用 builder 重建 AssistantMessage，去掉 reasoningContent Map cleanMeta = new java.util.HashMap<>(meta); cleanMeta.remove("reasoningContent"); AssistantMessage.Builder builder = AssistantMessage.builder() @@ -949,13 +1050,17 @@ public class NodeStreamingChatHelper { builder.media(am.getMedia()); } cleaned.add(builder.build()); + strippedCount++; continue; } } cleaned.add(msg); } - log.info("[ThinkingRecovery] Stripped thinking blocks from {} messages, last assistant at index {}", - messages.size(), lastAssistantIdx); + if (strippedCount > 0) { + log.debug("[ThinkingRecovery] Stripped reasoningContent from {} prior-turn assistant messages " + + "(lastUserIdx={}, total={})", + strippedCount, lastUserIdx, messages.size()); + } return new Prompt(cleaned, prompt.getOptions()); } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java index 50cef765..ea324c46 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java @@ -58,6 +58,13 @@ public class ReasoningNode implements NodeAction { private final ChatModel chatModel; private final List toolCallbacks; private final String reasoningEffort; + /** + * PR-1.2 (RFC-049 L1-B): Whether the bound model's {@code ModelFamily} accepts + * {@code reasoning_effort}. Drives the capability gate in + * {@link #resolveEffectiveReasoningEffort()} so that a front-end {@code ThinkingLevelHolder} + * override is dropped on chat-type models that cannot honor it. + */ + private final boolean supportsReasoningEffort; private final NodeStreamingChatHelper streamingHelper; private final ConversationWindowManager conversationWindowManager; private final ChatStreamTracker streamTracker; @@ -86,9 +93,31 @@ public class ReasoningNode implements NodeAction { ConversationWindowManager conversationWindowManager, ChatStreamTracker streamTracker, int maxOutputTokens, vip.mate.wiki.service.WikiContextService wikiContextService) { + // Backward-compatible delegate. Callers that have not migrated to the explicit + // supportsReasoningEffort parameter inherit the pre-PR-1 behavior: treat the bound + // model as supporting reasoning_effort iff reasoningEffort was resolved to a non-null + // value at construction time. New callers (AgentGraphBuilder) should use the + // 9-arg constructor below. + this(chatModel, toolSet, reasoningEffort, reasoningEffort != null, + streamingHelper, conversationWindowManager, streamTracker, + maxOutputTokens, wikiContextService); + } + + /** + * PR-1.2 (RFC-049): Primary constructor with explicit {@code supportsReasoningEffort} + * capability flag — avoids inferring capability from {@code reasoningEffort == null}, + * which fails for a future "supports but not auto-enabled" scenario. + */ + public ReasoningNode(ChatModel chatModel, AgentToolSet toolSet, String reasoningEffort, + boolean supportsReasoningEffort, + NodeStreamingChatHelper streamingHelper, + ConversationWindowManager conversationWindowManager, + ChatStreamTracker streamTracker, int maxOutputTokens, + vip.mate.wiki.service.WikiContextService wikiContextService) { this.chatModel = chatModel; this.toolCallbacks = toolSet.callbacks(); this.reasoningEffort = reasoningEffort; + this.supportsReasoningEffort = supportsReasoningEffort; this.streamingHelper = streamingHelper; this.conversationWindowManager = conversationWindowManager; this.streamTracker = streamTracker; @@ -120,6 +149,7 @@ public class ReasoningNode implements NodeAction { this.chatModel = chatModel; this.toolCallbacks = toolCallbacks; this.reasoningEffort = null; + this.supportsReasoningEffort = false; this.streamingHelper = null; this.conversationWindowManager = null; this.streamTracker = null; @@ -520,6 +550,13 @@ public class ReasoningNode implements NodeAction { * 解析有效的 reasoningEffort。 * 优先级：ThinkingLevelHolder（请求级） > 构造时的 reasoningEffort（Agent/模型默认）。 * "off" 会清除 reasoningEffort（返回 null）。 + * + *

PR-1.2 (RFC-049 L1-B): If the bound model's family does not support + * {@code reasoning_effort} (as declared via {@link #supportsReasoningEffort} at + * construction time), the front-end thinking-level override is ignored. + * Chat-type models like {@code deepseek-chat} must not be forced into thinking mode + * just because the user ticked "deep thinking" in the UI — this is a product + * contract, not a runtime option. */ private String resolveEffectiveReasoningEffort() { String requestLevel = ThinkingLevelHolder.get(); @@ -527,6 +564,11 @@ public class ReasoningNode implements NodeAction { if ("off".equalsIgnoreCase(requestLevel)) { return null; } + if (!this.supportsReasoningEffort) { + log.debug("[ReasoningNode] Ignoring thinkingLevel='{}' — bound model family does not support reasoning_effort", + requestLevel); + return null; + } // thinkingLevel → reasoningEffort 映射 return switch (requestLevel.toLowerCase()) { case "low" -> "low"; diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java index cfa521ec..680c66e3 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java @@ -11,7 +11,6 @@ import org.springframework.ai.chat.messages.UserMessage; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.prompt.ChatOptions; import org.springframework.ai.chat.prompt.Prompt; -import org.springframework.ai.model.tool.ToolCallingChatOptions; import org.springframework.ai.openai.OpenAiChatOptions; import org.springframework.util.StringUtils; import com.fasterxml.jackson.databind.JsonNode; @@ -119,20 +118,20 @@ public class StepExecutionNode implements NodeAction { try { while (toolCallCount < MAX_TOOL_CALLS_PER_STEP) { - ChatOptions options; + // PR-2 (RFC-049 §2.3.4): always use OpenAiChatOptions so the relay + // producer in NodeStreamingChatHelper.doStreamCall can attach the + // user-token. Using ToolCallingChatOptions when reasoningEffort is + // null (e.g. DeepSeek-Reasoner whose thinking is model-inherent, + // or Kimi-K2.5) would bypass the relay and multi-round tool-calls + // would 400 again. + OpenAiChatOptions oaiOpts = OpenAiChatOptions.builder() + .toolCallbacks(toolSet.callbacks()) + .build(); if (StringUtils.hasText(reasoningEffort)) { - OpenAiChatOptions oaiOpts = OpenAiChatOptions.builder() - .toolCallbacks(toolSet.callbacks()) - .reasoningEffort(reasoningEffort) - .build(); - oaiOpts.setInternalToolExecutionEnabled(false); - options = oaiOpts; - } else { - options = ToolCallingChatOptions.builder() - .toolCallbacks(toolSet.callbacks()) - .internalToolExecutionEnabled(false) - .build(); + oaiOpts.setReasoningEffort(reasoningEffort); } + oaiOpts.setInternalToolExecutionEnabled(false); + ChatOptions options = oaiOpts; NodeStreamingChatHelper.StreamResult result = streamingHelper.streamCall( chatModel, new Prompt(messages, options), conversationId,