mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-11-02 14:11:49 +08:00
fix(llm): preserve vLLM reasoning and honor thinking switch (#633)
This commit is contained in:
matevip
parent
8337d13403
commit
3643052e9f
@ -1177,6 +1177,7 @@ public class NodeStreamingChatHelper {
|
||||
List<ToolCallAccumulator> toolCallAccumulators = new ArrayList<>();
|
||||
AtomicReference<AssistantMessage> lastAssistantMessage = new AtomicReference<>();
|
||||
AtomicReference<Throwable> errorRef = new AtomicReference<>();
|
||||
AtomicReference<String> 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(),
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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")
|
||||
.<DataBuffer>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;
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
@ -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<ChatResponse> 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<String, Object> extra = new LinkedHashMap<>();
|
||||
Map<String, Object> 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<String, Object> extra, Map<String, Object> template, Map<String, Object> 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); });
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -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<String, Object> 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
|
||||
|
||||
@ -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<DataBuffer> body = Flux.<DataBuffer>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));
|
||||
}
|
||||
}
|
||||
@ -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<JsonNode> 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<String, Object> kwargs) {
|
||||
return model(providerId, kwargs, false);
|
||||
}
|
||||
|
||||
private ChatModel model(String providerId, Map<String, Object> 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());
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user