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 1f0a43d6..7b31185b 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 @@ -1177,6 +1177,7 @@ public class NodeStreamingChatHelper { List toolCallAccumulators = new ArrayList<>(); AtomicReference lastAssistantMessage = new AtomicReference<>(); AtomicReference errorRef = new AtomicReference<>(); + AtomicReference finishReason = new AtomicReference<>(); AtomicInteger promptTokens = new AtomicInteger(0); AtomicInteger completionTokens = new AtomicInteger(0); // Prompt cache / reasoning counters; providers that don't report them stay 0. @@ -1292,6 +1293,11 @@ public class NodeStreamingChatHelper { return; } var generation = chatResponse.getResult(); + if (generation.getMetadata() != null + && generation.getMetadata().getFinishReason() != null + && !generation.getMetadata().getFinishReason().isBlank()) { + finishReason.set(generation.getMetadata().getFinishReason()); + } AssistantMessage msg = generation.getOutput(); lastAssistantMessage.set(msg); @@ -1580,7 +1586,10 @@ public class NodeStreamingChatHelper { // ===== 成功（检查是否因 thinking-only 软上限或内容重复被截断） ===== boolean truncatedByThinkingCap = thinkingOnlyCapTriggered.get(); boolean truncatedByContentRepeat = contentRepeatCapTriggered.get(); - boolean truncated = truncatedByThinkingCap || truncatedByContentRepeat; + boolean thinkingTokenLimit = "length".equalsIgnoreCase(finishReason.get()) + && thinkingAccum.length() > 0 && contentAccum.toString().isBlank() + && toolCallAccumulators.isEmpty(); + boolean truncated = truncatedByThinkingCap || truncatedByContentRepeat || thinkingTokenLimit; if (truncatedByThinkingCap) { log.warn("[{}] LLM stream disposed: thinking-only soft cap reached for conversation {}", phase, conversationId); @@ -1609,6 +1618,7 @@ public class NodeStreamingChatHelper { String truncationReason = truncatedByThinkingCap ? "thinking_only_no_content" : truncatedByContentRepeat ? "content_repetition" + : thinkingTokenLimit ? "thinking_token_limit" : null; return assembleResult(contentAccum, thinkingAccum, toolCallAccumulators, promptTokens.get(), completionTokens.get(), 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 d546aef0..c1cd8635 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 @@ -1160,21 +1160,20 @@ public class ReasoningNode implements NodeAction { // soft cap would be re-promoted to ERROR_FALLBACK and we'd lose the // INCOMPLETE semantics. - if (result.partial() && "thinking_only_no_content".equals(result.errorMessage())) { - // Soft thinking-only loop: the helper disposed the upstream stream - // because the model accumulated >= THINKING_ONLY_HARD_CAP_CHARS of - // reasoning_content without emitting any visible content or tool - // calls. Treat as INCOMPLETE rather than fatal — the thinking text - // has already been streamed and is preserved for the UI's collapse - // panel; the user gets a short fallback line they can retry from. + if (result.partial() && ("thinking_only_no_content".equals(result.errorMessage()) + || "thinking_token_limit".equals(result.errorMessage()))) { + // Either our thinking-only cap or the provider's output-token budget + // ended reasoning before any answer/tool call. Preserve the transcript + // and surface INCOMPLETE rather than retrying a supposed empty response. String partialThinking = result.thinking() != null ? result.thinking() : ""; - log.warn("[ReasoningNode] Thinking-only soft cap hit ({} thinking chars, no content/tools); " + - "INCOMPLETE", - partialThinking.length()); + log.warn("[ReasoningNode] Thinking-only turn ended: {} ({} thinking chars); INCOMPLETE", + result.errorMessage(), partialThinking.length()); var builder = reasonOutput() .needsToolCall(false) .shouldSummarize(false) - .finalAnswer("（模型在思考阶段停留过久且未给出最终答案，请重试或拆分问题。）") + .finalAnswer("thinking_token_limit".equals(result.errorMessage()) + ? "（模型在输出最终答案前已耗尽输出 token 预算。请关闭思考、适当增加模型最大输出 token 数，或拆分问题后重试。）" + : "（模型在思考阶段停留过久且未给出最终答案，请重试或拆分问题。）") .llmCallCount(nextLlmCallCount) .finishReason(FinishReason.INCOMPLETE) .contentStreamed(false) diff --git a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiCompatibleChatModelBuilder.java b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiCompatibleChatModelBuilder.java index cebe0f92..2ab97a10 100644 --- a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiCompatibleChatModelBuilder.java +++ b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiCompatibleChatModelBuilder.java @@ -93,7 +93,7 @@ public class OpenAiCompatibleChatModelBuilder implements ChatModelBuilder { if (ModelFamily.detect(model.getModelName()) == ModelFamily.DEEPSEEK_V4_REASONING) { return new DeepSeekV4ThinkingDecorator(raw); } - return raw; + return VllmThinkingDecorator.supports(provider, options) ? new VllmThinkingDecorator(raw) : raw; } // ==================== chat options ==================== @@ -252,6 +252,9 @@ public class OpenAiCompatibleChatModelBuilder implements ChatModelBuilder { WebClient.Builder webClientBuilder = applyHttpTimeoutsToWebClient( webClientBuilderProvider.getIfAvailable(WebClient::builder), readTimeoutOverride); + restClientBuilder.requestInterceptor(OpenAiReasoningResponseNormalizer.blockingInterceptor()); + webClientBuilder.filter(OpenAiReasoningResponseNormalizer.streamingFilter()); + // Spring AI's OpenAiApi constructor sets User-Agent to "spring-ai" first, then addAll's // our headers, so a custom User-Agent is appended rather than replaced. For providers // that must masquerade as a specific client (e.g. kimi-code), force-override headers diff --git a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiReasoningResponseNormalizer.java b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiReasoningResponseNormalizer.java new file mode 100644 index 00000000..e4990010 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiReasoningResponseNormalizer.java @@ -0,0 +1,97 @@ +package vip.mate.llm.chatmodel; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.springframework.core.io.buffer.DefaultDataBufferFactory; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatusCode; +import org.springframework.http.MediaType; +import org.springframework.http.client.ClientHttpRequestInterceptor; +import org.springframework.http.client.ClientHttpResponse; +import org.springframework.web.reactive.function.client.ExchangeFilterFunction; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; + +/** Bridges vLLM's newer reasoning field to Spring AI 1.1.x's reasoning_content. */ +final class OpenAiReasoningResponseNormalizer { + private static final ObjectMapper JSON = new ObjectMapper(); + + private OpenAiReasoningResponseNormalizer() {} + + static String normalize(String body) { + if (!body.contains("\"reasoning\"")) return body; + try { + JsonNode root = JSON.readTree(body); + boolean changed = false; + for (JsonNode choice : root.path("choices")) { + changed |= normalizeMessage(choice.path("delta")); + changed |= normalizeMessage(choice.path("message")); + } + return changed ? JSON.writeValueAsString(root) : body; + } catch (JsonProcessingException ignored) { + // Keep malformed input intact: the SDK owns protocol error handling. + return body; + } + } + + private static boolean normalizeMessage(JsonNode node) { + if (node instanceof ObjectNode message && !message.hasNonNull("reasoning_content") + && message.path("reasoning").isTextual()) { + message.set("reasoning_content", message.get("reasoning")); + return true; + } + return false; + } + + static ExchangeFilterFunction streamingFilter() { + return (request, next) -> next.exchange(request).map(response -> { + if (!response.statusCode().is2xxSuccessful() + || !response.headers().contentType().map(MediaType.TEXT_EVENT_STREAM::isCompatibleWith).orElse(false)) { + return response; + } + // Decode complete SSE data events, not TCP/DataBuffer fragments. This + // preserves split UTF-8, multiline data, [DONE], backpressure and cancellation. + var events = response.bodyToFlux(String.class).map(OpenAiReasoningResponseNormalizer::normalize) + .map(data -> "data: " + data.replace("\n", "\ndata: ") + "\n\n") + .map(data -> DefaultDataBufferFactory.sharedInstance.wrap(data.getBytes(StandardCharsets.UTF_8))); + // The function overload retains the source body. body(Flux) would + // release it immediately and subscribe to the HTTP body twice. + return response.mutate().headers(headers -> headers.remove(HttpHeaders.CONTENT_LENGTH)).body(original -> events).build(); + }); + } + + static ClientHttpRequestInterceptor blockingInterceptor() { + return (request, body, execution) -> { + ClientHttpResponse response = execution.execute(request, body); + if (!response.getStatusCode().is2xxSuccessful() + || response.getHeaders().getContentType() == null + || !MediaType.APPLICATION_JSON.isCompatibleWith(response.getHeaders().getContentType())) { + return response; + } + try { + byte[] bytes = normalize(new String(response.getBody().readAllBytes(), StandardCharsets.UTF_8)) + .getBytes(StandardCharsets.UTF_8); + HttpHeaders headers = new HttpHeaders(); + headers.putAll(response.getHeaders()); + headers.setContentLength(bytes.length); + return new ClientHttpResponse() { + private final InputStream input = new ByteArrayInputStream(bytes); + @Override public HttpStatusCode getStatusCode() throws IOException { return response.getStatusCode(); } + @Override public String getStatusText() throws IOException { return response.getStatusText(); } + @Override public HttpHeaders getHeaders() { return headers; } + @Override public InputStream getBody() { return input; } + @Override public void close() { response.close(); } + }; + } catch (IOException | RuntimeException error) { + response.close(); + throw error; + } + }; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/VllmThinkingDecorator.java b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/VllmThinkingDecorator.java new file mode 100644 index 00000000..e4165a15 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/VllmThinkingDecorator.java @@ -0,0 +1,70 @@ +package vip.mate.llm.chatmodel; + +import org.springframework.ai.chat.model.ChatModel; +import org.springframework.ai.chat.model.ChatResponse; +import org.springframework.ai.chat.prompt.ChatOptions; +import org.springframework.ai.chat.prompt.Prompt; +import org.springframework.ai.model.ModelOptionsUtils; +import org.springframework.ai.model.tool.ToolCallingChatOptions; +import org.springframework.ai.openai.OpenAiChatOptions; +import reactor.core.publisher.Flux; +import vip.mate.llm.model.ModelProviderEntity; + +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; + +/** Per-request template switch for vLLM/Qwen, including arbitrary served model aliases. */ +final class VllmThinkingDecorator implements ChatModel { + private final ChatModel delegate; + + VllmThinkingDecorator(ChatModel delegate) { + this.delegate = delegate; + } + + static boolean supports(ModelProviderEntity provider, OpenAiChatOptions defaults) { + String id = provider.getProviderId(); + if (id != null && id.toLowerCase(Locale.ROOT).contains("vllm")) return true; + // Custom providers can opt in by explicitly configuring the template switch. + return defaults.getExtraBody() != null + && defaults.getExtraBody().get("chat_template_kwargs") instanceof Map template + && template.containsKey("enable_thinking"); + } + + @Override public ChatResponse call(Prompt prompt) { return delegate.call(transform(prompt)); } + @Override public Flux stream(Prompt prompt) { return delegate.stream(transform(prompt)); } + @Override public ChatOptions getDefaultOptions() { return delegate.getDefaultOptions(); } + + private Prompt transform(Prompt prompt) { + // Capture on the caller's thread, before Reactor subscription or worker handoff. + String level = ThinkingLevelHolder.get(); + if (level == null || level.isBlank()) return prompt; + ChatOptions runtime = prompt.getOptions(); + OpenAiChatOptions patched = runtime instanceof OpenAiChatOptions options + ? OpenAiChatOptions.fromOptions(options) + : runtime == null ? new OpenAiChatOptions() + : runtime instanceof ToolCallingChatOptions toolOptions + ? ModelOptionsUtils.copyToTarget(toolOptions, ToolCallingChatOptions.class, OpenAiChatOptions.class) + : ModelOptionsUtils.copyToTarget(runtime, ChatOptions.class, OpenAiChatOptions.class); + Map extra = new LinkedHashMap<>(); + Map template = new LinkedHashMap<>(); + if (delegate.getDefaultOptions() instanceof OpenAiChatOptions defaults) { + mergeExtra(extra, template, defaults.getExtraBody()); + } + mergeExtra(extra, template, patched.getExtraBody()); + template.put("enable_thinking", !"off".equalsIgnoreCase(level)); + extra.put("chat_template_kwargs", template); + patched.setExtraBody(extra); + // vLLM template switching does not use OpenAI's effort levels. + patched.setReasoningEffort(null); + return new Prompt(prompt.getInstructions(), patched); + } + + private static void mergeExtra(Map extra, Map template, Map source) { + if (source == null) return; + extra.putAll(source); + if (source.get("chat_template_kwargs") instanceof Map values) { + values.forEach((key, value) -> { if (key instanceof String name) template.put(name, value); }); + } + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperThinkingCapTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperThinkingCapTest.java index 7c88026e..60e9d13a 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperThinkingCapTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperThinkingCapTest.java @@ -112,4 +112,36 @@ class NodeStreamingChatHelperThinkingCapTest { assertEquals(hugeThinking, result.thinking(), "Thinking transcript is preserved so the UI can show it in a collapse panel"); } + @Test + void reasoningOnlyTokenLimitIsIncompleteRatherThanEmptyOrSuccessful() { + var message = AssistantMessage.builder().content("") + .properties(Map.of("reasoningContent", "still reasoning")).build(); + var generation = new Generation(message, ChatGenerationMetadata.builder().finishReason("length").build()); + var model = mock(ChatModel.class); + when(model.stream(any(Prompt.class))).thenReturn(Flux.just(new ChatResponse(List.of(generation)))); + var result = new NodeStreamingChatHelper(streamTracker) + .streamCall(model, smallPrompt(), "conv-length", "reasoning"); + assertTrue(result.partial()); + assertEquals("thinking_token_limit", result.errorMessage()); + assertEquals("still reasoning", result.thinking()); + assertEquals("", result.text()); + assertEquals(NodeStreamingChatHelper.ErrorType.NONE, result.errorType()); + } + + @Test + void tokenLimitWithContentOrToolsIsNotMisclassifiedAsThinkingOnly() { + for (var message : List.of( + AssistantMessage.builder().content("partial answer") + .properties(Map.of("reasoningContent", "reasoning")).build(), + AssistantMessage.builder().content("").properties(Map.of("reasoningContent", "reasoning")) + .toolCalls(List.of(new AssistantMessage.ToolCall("id", "function", "search", "{}"))).build())) { + var generation = new Generation(message, ChatGenerationMetadata.builder().finishReason("length").build()); + var model = mock(ChatModel.class); + when(model.stream(any(Prompt.class))).thenReturn(Flux.just(new ChatResponse(List.of(generation)))); + var result = new NodeStreamingChatHelper(streamTracker) + .streamCall(model, smallPrompt(), "conv-length-progress", "reasoning"); + assertNotEquals("thinking_token_limit", result.errorMessage()); + } + } + } diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ReasoningNodeOutputTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ReasoningNodeOutputTest.java index 6a1e96ce..6c2217ef 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ReasoningNodeOutputTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/ReasoningNodeOutputTest.java @@ -428,6 +428,23 @@ class ReasoningNodeOutputTest { "Thinking transcript must be preserved for the UI's collapse panel"); } + @Test + void thinkingTokenLimit_explainsBudgetAndDoesNotRetryOrExposeReasoningAsAnswer() throws Exception { + var result = new NodeStreamingChatHelper.StreamResult( + "", "unfinished reasoning", new AssistantMessage(""), List.of(), false, + 100, 256, true, "thinking_token_limit", NodeStreamingChatHelper.ErrorType.NONE); + when(streamingHelper.streamCall(any(), any(), anyString(), anyString())).thenReturn(result); + Map output = createNode().apply(buildStaleState()); + assertControlFlagsCleared(output, "thinkingTokenLimit"); + assertEquals("incomplete", output.get(FINISH_REASON)); + assertEquals("unfinished reasoning", output.get(FINAL_THINKING)); + String answer = (String) output.get(FINAL_ANSWER); + assertTrue(answer.contains("token 预算")); + assertTrue(answer.contains("关闭思考")); + assertFalse(answer.contains("unfinished reasoning")); + verify(streamingHelper, times(1)).streamCall(any(), any(), anyString(), anyString()); + } + // ===== CancellationException (no content stop) ===== @Test diff --git a/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/OpenAiReasoningResponseNormalizerTest.java b/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/OpenAiReasoningResponseNormalizerTest.java new file mode 100644 index 00000000..d87ef93f --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/OpenAiReasoningResponseNormalizerTest.java @@ -0,0 +1,51 @@ +package vip.mate.llm.chatmodel; + +import org.junit.jupiter.api.Test; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DefaultDataBufferFactory; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.web.reactive.function.client.ClientRequest; +import org.springframework.web.reactive.function.client.ClientResponse; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.*; + +class OpenAiReasoningResponseNormalizerTest { + @Test + void eventMappingSubscribesOnceAndPropagatesCancellation() { + AtomicInteger subscriptions = new AtomicInteger(); + AtomicInteger cancellations = new AtomicInteger(); + Flux body = Flux.defer(() -> { + subscriptions.incrementAndGet(); + String event = "data: {\"choices\":[{\"delta\":{\"reasoning\":\"thinking\"}}]}\n\n"; + return Flux.concat(Flux.just(DefaultDataBufferFactory.sharedInstance.wrap(event.getBytes(StandardCharsets.UTF_8))), + Flux.never()); + }).doOnCancel(cancellations::incrementAndGet); + ClientResponse source = ClientResponse.create(HttpStatus.OK) + .header("Content-Type", MediaType.TEXT_EVENT_STREAM_VALUE).body(body).build(); + var request = ClientRequest.create(HttpMethod.POST, URI.create("http://localhost/v1/chat/completions")).build(); + var filtered = OpenAiReasoningResponseNormalizer.streamingFilter() + .filter(request, ignored -> Mono.just(source)).block(); + assertEquals(0, subscriptions.get(), "filter must not eagerly drain the HTTP body"); + var events = filtered.bodyToFlux(String.class).take(1).collectList().block(Duration.ofSeconds(2)); + assertEquals(1, events.size()); + assertTrue(events.getFirst().contains("reasoning_content")); + assertEquals(1, subscriptions.get()); + assertEquals(1, cancellations.get()); + } + + @Test + void malformedFramesAndDoneRemainUnchanged() { + assertEquals("[DONE]", OpenAiReasoningResponseNormalizer.normalize("[DONE]")); + String malformed = "{\"reasoning\": broken"; + assertEquals(malformed, OpenAiReasoningResponseNormalizer.normalize(malformed)); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/VllmThinkingCompatibilityTest.java b/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/VllmThinkingCompatibilityTest.java new file mode 100644 index 00000000..6ecd9221 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/VllmThinkingCompatibilityTest.java @@ -0,0 +1,233 @@ +package vip.mate.llm.chatmodel; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpServer; +import io.micrometer.observation.ObservationRegistry; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.ai.chat.model.ChatModel; +import org.springframework.ai.chat.prompt.Prompt; +import org.springframework.ai.openai.OpenAiChatOptions; +import org.springframework.beans.factory.support.StaticListableBeanFactory; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.web.client.RestClient; +import org.springframework.web.reactive.function.client.WebClient; +import vip.mate.llm.model.ModelConfigEntity; +import vip.mate.llm.model.ModelProviderEntity; +import vip.mate.llm.service.ModelProviderService; + +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** Exercises the real Spring AI HTTP/SSE boundary, including fragmented UTF-8 events. */ +class VllmThinkingCompatibilityTest { + private final ObjectMapper json = new ObjectMapper(); + private final LinkedBlockingQueue requests = new LinkedBlockingQueue<>(); + private HttpServer server; + private org.springframework.ai.openai.api.OpenAiApi api; + private volatile String response; + private volatile boolean streaming = true; + + @BeforeEach + void start() throws Exception { + response = "data: " + chunk("\"content\":\"answer\"", "stop") + "\n\ndata: [DONE]\n\n"; + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/v1/chat/completions", exchange -> { + requests.add(json.readTree(exchange.getRequestBody())); + exchange.getResponseHeaders().set("Content-Type", streaming ? "text/event-stream" : "application/json"); + byte[] bytes = response.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, 0); + try (var out = exchange.getResponseBody()) { + for (int offset = 0; offset < bytes.length; offset += 7) { + out.write(bytes, offset, Math.min(7, bytes.length - offset)); + out.flush(); + } + } + }); + server.start(); + } + + @AfterEach + void stop() { + ThinkingLevelHolder.clear(); + server.stop(0); + } + + private ChatModel model(String providerId, Map kwargs) { + return model(providerId, kwargs, false); + } + + private ChatModel model(String providerId, Map kwargs, boolean apiOnly) { + var service = mock(ModelProviderService.class); + when(service.isProviderConfigured(providerId)).thenReturn(true); + var provider = new ModelProviderEntity(); + provider.setProviderId(providerId); + provider.setRequireApiKey(false); + provider.setBaseUrl("http://127.0.0.1:" + server.getAddress().getPort()); + when(service.readProviderGenerateKwargs(provider)).thenReturn(kwargs); + var beans = new StaticListableBeanFactory(); + beans.addBean("rest", RestClient.builder()); + // The application enables this restricted header at process startup; this + // fixture only tests payload compatibility and does not run that bootstrap. + beans.addBean("web", WebClient.builder().filter((request, next) -> next.exchange( + org.springframework.web.reactive.function.client.ClientRequest.from(request) + .headers(headers -> headers.remove("Connection")).build()))); + beans.addBean("observations", ObservationRegistry.NOOP); + var builder = new OpenAiCompatibleChatModelBuilder(service, + beans.getBeanProvider(RestClient.Builder.class), beans.getBeanProvider(WebClient.Builder.class), + beans.getBeanProvider(ObservationRegistry.class)); + var config = new ModelConfigEntity(); + config.setModelName("sdhs-model"); // A custom alias must not hide vLLM capabilities. + config.setMaxTokens(256); + if (apiOnly) { + api = builder.buildOpenAiApi(provider, null); + return null; + } + return builder.build(config, provider, RetryTemplate.builder().maxAttempts(1).build()); + } + + private static String chunk(String delta, String finish) { + return "{\"id\":\"test\",\"object\":\"chat.completion.chunk\",\"created\":1,\"model\":\"sdhs-model\"," + + "\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"," + delta + + "},\"finish_reason\":" + (finish == null ? "null" : "\"" + finish + "\"") + "}]}"; + } + + @Test + void streamAcceptsBothReasoningFieldsWithoutChangingContent() { + response = ": heartbeat\r\n\r\ndata: " + chunk("\"reasoning\":\"推理一\"", null) + + "\r\n\r\ndata: " + chunk("\"reasoning_content\":\"推理二\"", null) + + "\n\ndata: " + chunk("\"content\":\"literal reasoning and reasoning_content\"", "stop") + + "\n\ndata: [DONE]\n\n"; + var responses = model("vllm", Map.of()).stream(new Prompt("hi")).collectList().block(Duration.ofSeconds(10)); + String reasoning = responses.stream().map(r -> r.getResult().getOutput().getMetadata().get("reasoningContent")) + .filter(java.util.Objects::nonNull).map(Object::toString).reduce("", String::concat); + assertEquals("推理一推理二", reasoning); + assertEquals("literal reasoning and reasoning_content", responses.stream() + .map(r -> r.getResult().getOutput().getText()).filter(java.util.Objects::nonNull).reduce("", String::concat)); + } + + @Test + void blockingCallAcceptsReasoningAlias() { + streaming = false; + response = "{\"id\":\"test\",\"created\":1,\"model\":\"sdhs-model\",\"choices\":[{\"index\":0," + + "\"message\":{\"role\":\"assistant\",\"content\":\"answer\",\"reasoning\":\"分析\"},\"finish_reason\":\"stop\"}]}"; + model("vllm", Map.of(), true); + var result = api.chatCompletionEntity(new org.springframework.ai.openai.api.OpenAiApi.ChatCompletionRequest(List.of(), false)); + assertEquals("分析", result.getBody().choices().getFirst().message().reasoningContent()); + assertEquals("answer", result.getBody().choices().getFirst().message().content()); + } + + @Test + void switchIsPerRequestAndPreservesTemplateDefaultsAndRuntimeOptions() throws Exception { + var model = model("vllm", Map.of("chat_template_kwargs", Map.of("enable_thinking", true, "custom", "keep"))); + var options = OpenAiChatOptions.builder().temperature(0.3).extraBody(Map.of("top_k", 20)).build(); + ThinkingLevelHolder.set("off"); + var stream = model.stream(new Prompt("hi", options)); + ThinkingLevelHolder.clear(); // Capture before a subscription moves to a Reactor worker. + stream.collectList().block(Duration.ofSeconds(10)); + var off = requests.poll(1, TimeUnit.SECONDS); + assertEquals(false, off.at("/chat_template_kwargs/enable_thinking").booleanValue()); + assertEquals("keep", off.at("/chat_template_kwargs/custom").asText()); + assertEquals(20, off.path("top_k").asInt()); + assertEquals(0.3, off.path("temperature").asDouble()); + assertFalse(off.has("reasoning_effort")); + ThinkingLevelHolder.set("high"); + model.stream(new Prompt("hi")).collectList().block(Duration.ofSeconds(10)); + var on = requests.poll(1, TimeUnit.SECONDS); + assertTrue(on.at("/chat_template_kwargs/enable_thinking").booleanValue()); + assertEquals(Map.of("top_k", 20), options.getExtraBody()); + assertEquals(true, ((Map) ((OpenAiChatOptions) model.getDefaultOptions()).getExtraBody() + .get("chat_template_kwargs")).get("enable_thinking")); + } + + @Test + void unspecifiedLevelPreservesExplicitProviderDefault() throws Exception { + model("vllm", Map.of("chat_template_kwargs", Map.of("enable_thinking", false))) + .stream(new Prompt("hi")).collectList().block(Duration.ofSeconds(10)); + assertFalse(requests.poll(1, TimeUnit.SECONDS).at("/chat_template_kwargs/enable_thinking").booleanValue()); + } + + @Test + void otherProvidersDoNotReceiveVllmOptions() throws Exception { + ThinkingLevelHolder.set("off"); + model("azure", Map.of()).stream(new Prompt("hi")).collectList().block(Duration.ofSeconds(10)); + assertFalse(requests.poll(1, TimeUnit.SECONDS).has("chat_template_kwargs")); + } + @Test + void explicitTemplateSwitchOptsCustomProviderIn() throws Exception { + ThinkingLevelHolder.set("off"); + model("local-inference", Map.of("chat_template_kwargs", Map.of("enable_thinking", true))) + .stream(new Prompt("hi")).collectList().block(Duration.ofSeconds(10)); + var sent = requests.poll(1, TimeUnit.SECONDS); + assertTrue(sent.at("/chat_template_kwargs/enable_thinking").isBoolean()); + assertFalse(sent.at("/chat_template_kwargs/enable_thinking").booleanValue()); + } + + @Test + void vllmWithoutTemplateDefaultsReceivesExplicitOff() throws Exception { + ThinkingLevelHolder.set("off"); + model("vllm", Map.of()).stream(new Prompt("hi")).collectList().block(Duration.ofSeconds(10)); + var sent = requests.poll(1, TimeUnit.SECONDS); + assertTrue(sent.at("/chat_template_kwargs/enable_thinking").isBoolean()); + assertFalse(sent.at("/chat_template_kwargs/enable_thinking").booleanValue()); + } + + @Test + void canonicalReasoningWinsAndStructuredToolArgumentsAreUntouched() throws Exception { + // Build arguments as JSON rather than relying on nested Java/JSON escaping. + var root = json.createObjectNode(); + var delta = root.putArray("choices").addObject().putObject("delta"); + delta.put("reasoning", "alias"); + delta.put("reasoning_content", "canonical"); + delta.putArray("tool_calls").addObject().putObject("function") + .put("arguments", "{reasoning: keep}"); + assertEquals(root.toString(), OpenAiReasoningResponseNormalizer.normalize(root.toString())); + delta.remove("reasoning_content"); + JsonNode normalized = json.readTree(OpenAiReasoningResponseNormalizer.normalize(root.toString())); + assertEquals("alias", normalized.at("/choices/0/delta/reasoning_content").asText()); + assertEquals("{reasoning: keep}", normalized.at("/choices/0/delta/tool_calls/0/function/arguments").asText()); + } + + @Test + void genericToolCallingOptionsSurviveThinkingSwitch() throws Exception { + var callback = mock(org.springframework.ai.tool.ToolCallback.class); + when(callback.getToolDefinition()).thenReturn(org.springframework.ai.tool.definition.ToolDefinition.builder() + .name("lookup").description("lookup a value").inputSchema("{} ").build()); + var options = org.springframework.ai.model.tool.ToolCallingChatOptions.builder() + .toolCallbacks(List.of(callback)).internalToolExecutionEnabled(false).build(); + ThinkingLevelHolder.set("off"); + model("vllm", Map.of()).stream(new Prompt("hi", options)).collectList().block(Duration.ofSeconds(10)); + var sent = requests.poll(1, TimeUnit.SECONDS); + assertEquals("lookup", sent.at("/tools/0/function/name").asText()); + } + + @Test + void streamedToolCallFragmentsSurviveReasoningNormalization() throws Exception { + var first = json.readTree(chunk("\"reasoning\":\"planning\"", null)); + var firstDelta = (com.fasterxml.jackson.databind.node.ObjectNode) first.at("/choices/0/delta"); + var firstTool = firstDelta.putArray("tool_calls").addObject(); + firstTool.put("index", 0).put("id", "call-1").put("type", "function"); + firstTool.putObject("function").put("name", "lookup").put("arguments", "{\"reasoning\":"); + var second = json.readTree(chunk("\"content\":null", "tool_calls")); + var secondDelta = (com.fasterxml.jackson.databind.node.ObjectNode) second.at("/choices/0/delta"); + secondDelta.putArray("tool_calls").addObject().put("index", 0) + .putObject("function").put("arguments", "\"keep\"}"); + response = "data: " + first + "\n\ndata: " + second + "\n\ndata: [DONE]\n\n"; + var result = model("vllm", Map.of()).stream(new Prompt("hi")).collectList().block(Duration.ofSeconds(10)); + var calls = result.stream().flatMap(r -> r.getResult().getOutput().getToolCalls().stream()).toList(); + assertEquals(1, calls.size()); + assertEquals("lookup", calls.getFirst().name()); + assertEquals("keep", json.readTree(calls.getFirst().arguments()).path("reasoning").asText()); + } + +}