Copilot commented on code in PR #6341: URL: https://github.com/apache/shenyu/pull/6341#discussion_r3252993780
########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapter.java: ########## @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.shenyu.plugin.ai.common.protocol; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.shenyu.common.utils.JsonUtils; +import org.apache.shenyu.plugin.ai.common.config.AiCommonConfig; +import org.springframework.ai.openai.api.OpenAiApi.ChatCompletionRequest; + +import java.util.Objects; + +/** + * Adapts between OpenAI Chat Completions wire format and internal representations. + * + * <p>Note: deserialization into Spring AI's {@link ChatCompletionRequest} preserves fields + * modeled by that class (messages, model, temperature, maxTokens, stream, tools, etc.). + * Provider-specific extension fields not modeled by {@code ChatCompletionRequest} are dropped + * during deserialization. For the OpenAI-compatible providers this plugin currently supports, + * all required fields are covered. + */ +public final class OpenAiProtocolAdapter { + + private static final com.fasterxml.jackson.databind.ObjectMapper MAPPER = + new com.fasterxml.jackson.databind.ObjectMapper(); + + private OpenAiProtocolAdapter() { + } + + /** + * Resolve the stream flag: client request body takes priority, + * falls back to the provided default value. + * + * @param requestBody the raw JSON request body + * @param fallbackStream the default stream value from admin config + * @return true if streaming, false otherwise + */ + public static boolean resolveStream(final String requestBody, final Boolean fallbackStream) { + if (Objects.isNull(requestBody) || requestBody.isEmpty()) { + return Boolean.TRUE.equals(fallbackStream); + } + final JsonNode root = parseStrict(requestBody); + if (Objects.isNull(root)) { + return Boolean.TRUE.equals(fallbackStream); + } + if (root.hasNonNull("stream")) { + return root.get("stream").asBoolean(); + } + return Boolean.TRUE.equals(fallbackStream); + } + + /** + * Parse raw request body directly into ChatCompletionRequest, preserving fields + * modeled by Spring AI (including reasoning_content in assistant messages). + * + * <p>Spring AI's createRequest() loses reasoning_content, refusal, and annotations + * when reconstructing ChatCompletionMessage from AssistantMessage. + * This method avoids that loss by deserializing the raw JSON directly. + * + * <p>Also converts max_completion_tokens to max_tokens for broader API compatibility. + * + * @param requestBody the raw JSON request body in OpenAI Chat Completions format + * @param stream whether this is a streaming request (sets the stream field) + * @return a ChatCompletionRequest with all fields preserved from the original request + */ + public static ChatCompletionRequest toChatCompletionRequest(final String requestBody, final boolean stream) { + return toChatCompletionRequest(requestBody, stream, null); + } + + /** + * Parse raw request body directly into ChatCompletionRequest, preserving modeled fields. + * For model, temperature, max_tokens: client request takes priority; + * if missing, falls back to the corresponding field in fallbackConfig. + * + * @param requestBody the raw JSON request body in OpenAI Chat Completions format + * @param stream whether this is a streaming request + * @param fallbackConfig the admin config used as fallback when client omits fields + * @return a ChatCompletionRequest with all fields preserved + */ + public static ChatCompletionRequest toChatCompletionRequest(final String requestBody, + final boolean stream, final AiCommonConfig fallbackConfig) { + if (Objects.isNull(requestBody) || requestBody.isEmpty()) { + throw new IllegalArgumentException("Request body must not be empty"); + } + final JsonNode root = parseStrict(requestBody); + if (Objects.isNull(root) || !root.isObject()) { + throw new IllegalArgumentException("Invalid request body: expected a JSON object"); + } + final ObjectNode mutableRoot = (ObjectNode) root; + + if (root.hasNonNull("max_completion_tokens") && !root.hasNonNull("max_tokens")) { + final JsonNode tokenNode = root.get("max_completion_tokens"); + if (!tokenNode.isNumber()) { + throw new IllegalArgumentException( + "max_completion_tokens must be a number, got: " + tokenNode.getNodeType()); + } + mutableRoot.put("max_tokens", tokenNode.asInt()); + mutableRoot.remove("max_completion_tokens"); + } + + if (Objects.nonNull(fallbackConfig)) { + if (!root.hasNonNull("model") && Objects.nonNull(fallbackConfig.getModel()) && !fallbackConfig.getModel().isEmpty()) { + mutableRoot.put("model", fallbackConfig.getModel()); + } + if (!root.hasNonNull("temperature") && Objects.nonNull(fallbackConfig.getTemperature())) { + mutableRoot.put("temperature", fallbackConfig.getTemperature()); + } + if (!root.hasNonNull("max_tokens") && Objects.nonNull(fallbackConfig.getMaxTokens())) { + mutableRoot.put("max_tokens", fallbackConfig.getMaxTokens()); + } + } + + mutableRoot.put("stream", stream); + + final ChatCompletionRequest result = JsonUtils.jsonToObject(mutableRoot.toString(), ChatCompletionRequest.class); + if (Objects.isNull(result)) { + throw new IllegalArgumentException("Failed to parse request body into ChatCompletionRequest"); + } + return result; Review Comment: The adapter's core behavior is meant to preserve modeled message fields such as reasoning_content/refusal/annotations, but the new tests only cover stream, model, temperature, and token merging. Add a serialization/deserialization test that includes those assistant-message fields so regressions in the OpenAI compatibility fix are caught. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/handler/AiProxyPluginHandler.java: ########## @@ -53,10 +51,10 @@ public void handlerPlugin(final PluginData pluginData) { @Override public void handlerSelector(final SelectorData selectorData) { // Invalidate the cache first when the selector is updated. - chatClientCache.remove(selectorData.getId()); + OpenAiApiCache.getInstance().remove(selectorData.getId()); // Do NOT remove AiProxyApiKeyCache here. Admin will push updated AI_PROXY_API_KEY events // with refreshed realApiKey after selector changes. Removing here introduces a window of misses. - if (Objects.isNull(selectorData.getHandle())) { + if (Objects.isNull(selectorData.getHandle()) || selectorData.getHandle().isEmpty()) { return; Review Comment: Returning early for a null or empty selector handle leaves any previously cached selector handle in selectorCachedHandle. If a selector is updated to clear its AI proxy handle, requests can continue using the stale configuration instead of seeing the handle as removed. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/AiProxyPlugin.java: ########## @@ -214,86 +213,46 @@ private Optional<ChatClient> resolveFallbackClient( if (LOG.isDebugEnabled()) { LOG.debug("[AiProxy] dynamic fallback config: {}", cfg); } - return createDynamicFallbackClient(cfg); + return new FallbackContext(createOpenAiApi(cfg), cfg); }) - .or( - () -> aiProxyConfigService - .resolveAdminFallbackConfig(primaryConfig, selectorHandle) - .map(adminFallbackConfig -> { - LOG.info("[AiProxy] use admin fallback"); - if (LOG.isDebugEnabled()) { - LOG.debug("[AiProxy] admin fallback config: {}", adminFallbackConfig); - } - return createAdminFallbackClient(selectorId, adminFallbackConfig); - })); + .or(() -> aiProxyConfigService + .resolveAdminFallbackConfig(primaryConfig, selectorHandle) + .map(adminFallbackConfig -> { + LOG.info("[AiProxy] use admin fallback"); + if (LOG.isDebugEnabled()) { + LOG.debug("[AiProxy] admin fallback config: {}", adminFallbackConfig); + } + return new FallbackContext( + getCachedOpenAiApi(selectorId, "adminFallback", adminFallbackConfig), + adminFallbackConfig); + })); + } + + private OpenAiApi getCachedOpenAiApi(final String selectorId, final String type, final AiCommonConfig config) { + final String cacheKey = selectorId + "|" + type + "_" + generateConfigCacheKey(config); + return OpenAiApiCache.getInstance().computeIfAbsent(cacheKey, () -> createOpenAiApi(config)); } - /** - * Generate cache key based on config fields excluding apiKey. - * This ensures cache consistency even when apiKey is updated at runtime. - * - * @param config the config - * @return cache key hash - */ private int generateConfigCacheKey(final AiCommonConfig config) { return Objects.hash( - config.getProvider(), config.getBaseUrl(), Review Comment: The OpenAiApi cache key omits the API key even though createOpenAiApi binds that key into the cached client. In proxy-key mode, two different proxy keys for the same selector/baseUrl/model will reuse the first cached OpenAiApi and send subsequent requests with the wrong real API key, which can leak traffic across tenants or keep using a rotated key. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/AiProxyPluginTest.java: ########## @@ -262,61 +251,57 @@ public void testExecuteWithAdminFallback() { StepVerifier.create(plugin.doExecute(exchange, mock(ShenyuPluginChain.class), selector, rule)) .verifyComplete(); - verify(executorService).execute(any(ChatClient.class), any(Optional.class), any()); + verify(executorService).executeDirectCall(any(OpenAiApi.class), any(Optional.class), any(ChatCompletionRequest.class), any(String.class)); } @Test public void testCacheIsUsedForAdminFallbackClient() { final AiProxyHandle handle = new AiProxyHandle(); final AiCommonConfig primaryConfig = new AiCommonConfig(); primaryConfig.setProvider(AiModelProviderEnum.OPEN_AI.getName()); + primaryConfig.setBaseUrl("https://api.openai.com";); + primaryConfig.setApiKey("test-key"); final AiCommonConfig fallbackConfig = new AiCommonConfig(); fallbackConfig.setProvider(AiModelProviderEnum.DEEP_SEEK.getName()); + fallbackConfig.setBaseUrl("https://api.deepseek.com";); + fallbackConfig.setApiKey("fallback-key"); // Cache the handle for the test aiProxyPluginHandler.getSelectorCachedHandle().cachedHandle(CacheKeyUtils.INST.getKey(SELECTOR_ID, Constants.DEFAULT_RULE), handle); - final ChatResponse chatResponse = mock(ChatResponse.class); + final ChatCompletion chatCompletion = mock(ChatCompletion.class); + final ResponseEntity<ChatCompletion> responseEntity = ResponseEntity.ok(chatCompletion); // Setup all necessary mocks when(configService.resolvePrimaryConfig(handle)).thenReturn(primaryConfig); when(configService.resolveDynamicFallbackConfig(primaryConfig, REQUEST_BODY)).thenReturn(Optional.empty()); when(configService.resolveAdminFallbackConfig(primaryConfig, handle)).thenReturn(Optional.of(fallbackConfig)); - when(configService.extractPrompt(anyString())).thenAnswer(invocation -> invocation.getArgument(0)); - when(executorService.execute(any(), any(), any())).thenReturn(Mono.just(chatResponse)); + when(executorService.executeDirectCall(any(OpenAiApi.class), any(Optional.class), any(ChatCompletionRequest.class), any(String.class))).thenReturn(Mono.just(responseEntity)); // Execute the test - focus on successful execution rather than cache verification StepVerifier.create(plugin.doExecute(exchange, mock(ShenyuPluginChain.class), selector, rule)) .verifyComplete(); - + // Verify that the configuration methods were called correctly verify(configService).resolvePrimaryConfig(handle); verify(configService).resolveAdminFallbackConfig(primaryConfig, handle); - verify(executorService).execute(any(), any(), any()); + verify(executorService).executeDirectCall(any(OpenAiApi.class), any(Optional.class), any(ChatCompletionRequest.class), any(String.class)); Review Comment: The updated test explicitly avoids verifying cache behavior, but it remains the only admin-fallback cache test path. Either assert OpenAiApiCache reuse here or rename/split the test so cache behavior is covered elsewhere. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapter.java: ########## @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.shenyu.plugin.ai.common.protocol; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.shenyu.common.utils.JsonUtils; +import org.apache.shenyu.plugin.ai.common.config.AiCommonConfig; +import org.springframework.ai.openai.api.OpenAiApi.ChatCompletionRequest; + +import java.util.Objects; + +/** + * Adapts between OpenAI Chat Completions wire format and internal representations. + * + * <p>Note: deserialization into Spring AI's {@link ChatCompletionRequest} preserves fields + * modeled by that class (messages, model, temperature, maxTokens, stream, tools, etc.). + * Provider-specific extension fields not modeled by {@code ChatCompletionRequest} are dropped + * during deserialization. For the OpenAI-compatible providers this plugin currently supports, + * all required fields are covered. + */ +public final class OpenAiProtocolAdapter { + + private static final com.fasterxml.jackson.databind.ObjectMapper MAPPER = + new com.fasterxml.jackson.databind.ObjectMapper(); + + private OpenAiProtocolAdapter() { + } + + /** + * Resolve the stream flag: client request body takes priority, + * falls back to the provided default value. + * + * @param requestBody the raw JSON request body + * @param fallbackStream the default stream value from admin config + * @return true if streaming, false otherwise + */ + public static boolean resolveStream(final String requestBody, final Boolean fallbackStream) { + if (Objects.isNull(requestBody) || requestBody.isEmpty()) { + return Boolean.TRUE.equals(fallbackStream); + } + final JsonNode root = parseStrict(requestBody); + if (Objects.isNull(root)) { + return Boolean.TRUE.equals(fallbackStream); + } + if (root.hasNonNull("stream")) { + return root.get("stream").asBoolean(); + } + return Boolean.TRUE.equals(fallbackStream); + } + + /** + * Parse raw request body directly into ChatCompletionRequest, preserving fields + * modeled by Spring AI (including reasoning_content in assistant messages). + * + * <p>Spring AI's createRequest() loses reasoning_content, refusal, and annotations + * when reconstructing ChatCompletionMessage from AssistantMessage. + * This method avoids that loss by deserializing the raw JSON directly. + * + * <p>Also converts max_completion_tokens to max_tokens for broader API compatibility. + * + * @param requestBody the raw JSON request body in OpenAI Chat Completions format + * @param stream whether this is a streaming request (sets the stream field) + * @return a ChatCompletionRequest with all fields preserved from the original request + */ + public static ChatCompletionRequest toChatCompletionRequest(final String requestBody, final boolean stream) { + return toChatCompletionRequest(requestBody, stream, null); + } + + /** + * Parse raw request body directly into ChatCompletionRequest, preserving modeled fields. + * For model, temperature, max_tokens: client request takes priority; + * if missing, falls back to the corresponding field in fallbackConfig. + * + * @param requestBody the raw JSON request body in OpenAI Chat Completions format + * @param stream whether this is a streaming request + * @param fallbackConfig the admin config used as fallback when client omits fields + * @return a ChatCompletionRequest with all fields preserved + */ + public static ChatCompletionRequest toChatCompletionRequest(final String requestBody, + final boolean stream, final AiCommonConfig fallbackConfig) { + if (Objects.isNull(requestBody) || requestBody.isEmpty()) { + throw new IllegalArgumentException("Request body must not be empty"); + } + final JsonNode root = parseStrict(requestBody); + if (Objects.isNull(root) || !root.isObject()) { + throw new IllegalArgumentException("Invalid request body: expected a JSON object"); + } + final ObjectNode mutableRoot = (ObjectNode) root; + + if (root.hasNonNull("max_completion_tokens") && !root.hasNonNull("max_tokens")) { + final JsonNode tokenNode = root.get("max_completion_tokens"); + if (!tokenNode.isNumber()) { + throw new IllegalArgumentException( + "max_completion_tokens must be a number, got: " + tokenNode.getNodeType()); + } + mutableRoot.put("max_tokens", tokenNode.asInt()); + mutableRoot.remove("max_completion_tokens"); + } + + if (Objects.nonNull(fallbackConfig)) { + if (!root.hasNonNull("model") && Objects.nonNull(fallbackConfig.getModel()) && !fallbackConfig.getModel().isEmpty()) { + mutableRoot.put("model", fallbackConfig.getModel()); + } + if (!root.hasNonNull("temperature") && Objects.nonNull(fallbackConfig.getTemperature())) { + mutableRoot.put("temperature", fallbackConfig.getTemperature()); + } + if (!root.hasNonNull("max_tokens") && Objects.nonNull(fallbackConfig.getMaxTokens())) { + mutableRoot.put("max_tokens", fallbackConfig.getMaxTokens()); + } + } + + mutableRoot.put("stream", stream); + + final ChatCompletionRequest result = JsonUtils.jsonToObject(mutableRoot.toString(), ChatCompletionRequest.class); + if (Objects.isNull(result)) { + throw new IllegalArgumentException("Failed to parse request body into ChatCompletionRequest"); Review Comment: JsonUtils.jsonToObject logs the full JSON string when deserialization fails. Here that JSON is the client's chat-completion request and can contain sensitive prompts or tool data, so malformed/unsupported requests may leak user content into gateway logs; parse with a mapper that does not log the payload or log only sanitized metadata. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapter.java: ########## @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.shenyu.plugin.ai.common.protocol; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.shenyu.common.utils.JsonUtils; +import org.apache.shenyu.plugin.ai.common.config.AiCommonConfig; +import org.springframework.ai.openai.api.OpenAiApi.ChatCompletionRequest; + +import java.util.Objects; + +/** + * Adapts between OpenAI Chat Completions wire format and internal representations. + * + * <p>Note: deserialization into Spring AI's {@link ChatCompletionRequest} preserves fields + * modeled by that class (messages, model, temperature, maxTokens, stream, tools, etc.). + * Provider-specific extension fields not modeled by {@code ChatCompletionRequest} are dropped + * during deserialization. For the OpenAI-compatible providers this plugin currently supports, + * all required fields are covered. + */ +public final class OpenAiProtocolAdapter { + + private static final com.fasterxml.jackson.databind.ObjectMapper MAPPER = + new com.fasterxml.jackson.databind.ObjectMapper(); + + private OpenAiProtocolAdapter() { + } + + /** + * Resolve the stream flag: client request body takes priority, + * falls back to the provided default value. + * + * @param requestBody the raw JSON request body + * @param fallbackStream the default stream value from admin config + * @return true if streaming, false otherwise + */ + public static boolean resolveStream(final String requestBody, final Boolean fallbackStream) { + if (Objects.isNull(requestBody) || requestBody.isEmpty()) { + return Boolean.TRUE.equals(fallbackStream); + } + final JsonNode root = parseStrict(requestBody); + if (Objects.isNull(root)) { + return Boolean.TRUE.equals(fallbackStream); + } + if (root.hasNonNull("stream")) { + return root.get("stream").asBoolean(); + } + return Boolean.TRUE.equals(fallbackStream); + } + + /** + * Parse raw request body directly into ChatCompletionRequest, preserving fields + * modeled by Spring AI (including reasoning_content in assistant messages). + * + * <p>Spring AI's createRequest() loses reasoning_content, refusal, and annotations + * when reconstructing ChatCompletionMessage from AssistantMessage. + * This method avoids that loss by deserializing the raw JSON directly. + * + * <p>Also converts max_completion_tokens to max_tokens for broader API compatibility. + * + * @param requestBody the raw JSON request body in OpenAI Chat Completions format + * @param stream whether this is a streaming request (sets the stream field) + * @return a ChatCompletionRequest with all fields preserved from the original request + */ + public static ChatCompletionRequest toChatCompletionRequest(final String requestBody, final boolean stream) { + return toChatCompletionRequest(requestBody, stream, null); + } + + /** + * Parse raw request body directly into ChatCompletionRequest, preserving modeled fields. + * For model, temperature, max_tokens: client request takes priority; + * if missing, falls back to the corresponding field in fallbackConfig. + * + * @param requestBody the raw JSON request body in OpenAI Chat Completions format + * @param stream whether this is a streaming request + * @param fallbackConfig the admin config used as fallback when client omits fields + * @return a ChatCompletionRequest with all fields preserved + */ + public static ChatCompletionRequest toChatCompletionRequest(final String requestBody, + final boolean stream, final AiCommonConfig fallbackConfig) { + if (Objects.isNull(requestBody) || requestBody.isEmpty()) { + throw new IllegalArgumentException("Request body must not be empty"); + } + final JsonNode root = parseStrict(requestBody); + if (Objects.isNull(root) || !root.isObject()) { + throw new IllegalArgumentException("Invalid request body: expected a JSON object"); + } + final ObjectNode mutableRoot = (ObjectNode) root; + + if (root.hasNonNull("max_completion_tokens") && !root.hasNonNull("max_tokens")) { + final JsonNode tokenNode = root.get("max_completion_tokens"); + if (!tokenNode.isNumber()) { + throw new IllegalArgumentException( + "max_completion_tokens must be a number, got: " + tokenNode.getNodeType()); + } + mutableRoot.put("max_tokens", tokenNode.asInt()); + mutableRoot.remove("max_completion_tokens"); + } + + if (Objects.nonNull(fallbackConfig)) { + if (!root.hasNonNull("model") && Objects.nonNull(fallbackConfig.getModel()) && !fallbackConfig.getModel().isEmpty()) { + mutableRoot.put("model", fallbackConfig.getModel()); + } + if (!root.hasNonNull("temperature") && Objects.nonNull(fallbackConfig.getTemperature())) { + mutableRoot.put("temperature", fallbackConfig.getTemperature()); + } + if (!root.hasNonNull("max_tokens") && Objects.nonNull(fallbackConfig.getMaxTokens())) { + mutableRoot.put("max_tokens", fallbackConfig.getMaxTokens()); + } Review Comment: Because fallbackConfig values are only applied when the client omits the field, an admin/dynamic fallback model is ignored for normal OpenAI requests that include a model. When the fallback provider uses a different model name, the fallback request will still send the primary client model and can fail instead of using the configured fallback model. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyExecutorService.java: ########## @@ -39,78 +45,124 @@ public class AiProxyExecutorService { private static final Logger LOG = LoggerFactory.getLogger(AiProxyExecutorService.class); /** - * Execute the AI call with retry and fallback. + * Execute a streaming AI call directly via {@link OpenAiApi}, bypassing Spring AI's + * {@code createRequest()} which loses fields like {@code reasoning_content}. * - * @param mainClient the main chat client - * @param fallbackClientOpt the optional fallback chat client - * @param requestBody the request body - * @return a Mono containing the ChatResponse + * @param mainApi the main OpenAiApi + * @param fallbackCtxOpt the optional fallback context (api + config) + * @param request the ChatCompletionRequest with all fields preserved + * @param requestBody the original request body for rebuilding fallback request + * @param stream whether this is a streaming request + * @return a Flux of ChatCompletionChunk */ - public Mono<ChatResponse> execute(final ChatClient mainClient, final Optional<ChatClient> fallbackClientOpt, final String requestBody) { - final Mono<ChatResponse> mainCall = doChatCall(mainClient, requestBody); + public Flux<ChatCompletionChunk> executeDirectStream(final OpenAiApi mainApi, + final Optional<FallbackContext> fallbackCtxOpt, final ChatCompletionRequest request, + final String requestBody, final boolean stream) { + return mainApi.chatCompletionStream(request) + .doOnError(e -> UpstreamErrorLogger.logUpstreamError(LOG, e, "direct stream")) + .retryWhen(Retry.max(1) + .filter(AiProxyExecutorService::isRetryable) + .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> { + LOG.warn("Direct stream retry exhausted. Triggering fallback.", + retrySignal.failure()); + return new NonTransientAiException( + "Direct stream failed after 1 retry. Triggering fallback.", + retrySignal.failure()); + })) + .onErrorResume(NonTransientAiException.class, + throwable -> handleDirectFallbackStream(throwable, fallbackCtxOpt, requestBody, stream)); + } - return mainCall + /** + * Execute a non-streaming AI call directly via {@link OpenAiApi}. + * + * @param mainApi the main OpenAiApi + * @param fallbackCtxOpt the optional fallback context (api + config) + * @param request the ChatCompletionRequest with all fields preserved + * @param requestBody the original request body for rebuilding fallback request + * @return a Mono of ResponseEntity containing ChatCompletion + */ + public Mono<ResponseEntity<ChatCompletion>> executeDirectCall(final OpenAiApi mainApi, + final Optional<FallbackContext> fallbackCtxOpt, final ChatCompletionRequest request, + final String requestBody) { + return Mono.fromCallable(() -> mainApi.chatCompletionEntity(request)) + .subscribeOn(Schedulers.boundedElastic()) + .doOnError(e -> UpstreamErrorLogger.logUpstreamError(LOG, e, "direct call")) .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)) - .filter(throwable -> !(throwable instanceof NonTransientAiException)) + .filter(AiProxyExecutorService::isRetryable) .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> { - LOG.warn("Retries exhausted for AI call after {} attempts.", + LOG.warn("Direct call retries exhausted after {} attempts. Triggering fallback.", retrySignal.totalRetries(), retrySignal.failure()); - return new NonTransientAiException("Retries exhausted. Triggering fallback.", + return new NonTransientAiException("Direct call retries exhausted. Triggering fallback.", retrySignal.failure()); })) .onErrorResume(NonTransientAiException.class, - throwable -> handleFallback(throwable, fallbackClientOpt, requestBody)); + throwable -> handleDirectFallbackCall(throwable, fallbackCtxOpt, requestBody)); } - protected Mono<ChatResponse> doChatCall(final ChatClient client, final String requestBody) { - return Mono.fromCallable(() -> client.prompt().user(requestBody).call().chatResponse()) - .subscribeOn(Schedulers.boundedElastic()); + private Flux<ChatCompletionChunk> handleDirectFallbackStream(final Throwable throwable, + final Optional<FallbackContext> fallbackCtxOpt, final String requestBody, final boolean stream) { + LOG.warn("Main direct stream failed, attempting fallback...", throwable); + + if (fallbackCtxOpt.isEmpty()) { + return Flux.error(throwable); + } + + final FallbackContext ctx = fallbackCtxOpt.get(); + LOG.info("Using fallback OpenAiApi for direct stream"); + final ChatCompletionRequest fallbackRequest = OpenAiProtocolAdapter.toChatCompletionRequest( + requestBody, stream, ctx.config()); + return ctx.api().chatCompletionStream(fallbackRequest); } - private Mono<ChatResponse> handleFallback(final Throwable throwable, final Optional<ChatClient> fallbackClientOpt, final String requestBody) { - LOG.warn("AI main call failed or retries exhausted, attempting to fallback...", throwable); + private Mono<ResponseEntity<ChatCompletion>> handleDirectFallbackCall(final Throwable throwable, + final Optional<FallbackContext> fallbackCtxOpt, final String requestBody) { + LOG.warn("Main direct call failed, attempting fallback...", throwable); - if (fallbackClientOpt.isEmpty()) { + if (fallbackCtxOpt.isEmpty()) { return Mono.error(throwable); } - return SimpleModelFallbackStrategy.INSTANCE.fallback(fallbackClientOpt.get(), requestBody, throwable); + final FallbackContext ctx = fallbackCtxOpt.get(); + LOG.info("Using fallback OpenAiApi for direct call"); + final ChatCompletionRequest fallbackRequest = OpenAiProtocolAdapter.toChatCompletionRequest( + requestBody, false, ctx.config()); + return Mono.fromCallable(() -> ctx.api().chatCompletionEntity(fallbackRequest)) + .subscribeOn(Schedulers.boundedElastic()); } /** - * Execute the AI call with retry and fallback. - * - * @param mainClient the main chat client - * @param fallbackClientOpt the optional fallback chat client - * @param requestBody the request body - * @return a Flux containing the ChatResponse + * Determine if the error is retryable. + * Retries transient network errors and retryable HTTP statuses (429, 5xx). + * Non-retryable: NonTransientAiException, client errors (400/401/403/404). */ - public Flux<ChatResponse> executeStream(final ChatClient mainClient, final Optional<ChatClient> fallbackClientOpt, final String requestBody) { - final Flux<ChatResponse> mainStream = doChatStream(mainClient, requestBody); - - return mainStream - .retryWhen(Retry.max(1) - .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> { - LOG.warn("Retrying stream once failed. Attempts: {}. Triggering fallback.", - retrySignal.totalRetries(), retrySignal.failure()); - return new NonTransientAiException("Stream failed after 1 retry. Triggering fallback.", retrySignal.failure()); - })) - .onErrorResume(NonTransientAiException.class, - throwable -> handleFallbackStream(throwable, fallbackClientOpt, requestBody)); - } - - protected Flux<ChatResponse> doChatStream(final ChatClient client, final String requestBody) { - return Flux.defer(() -> client.prompt().user(requestBody).stream().chatResponse()) - .subscribeOn(Schedulers.boundedElastic()); + private static boolean isRetryable(final Throwable throwable) { + if (throwable instanceof NonTransientAiException) { + return false; + } + final WebClientResponseException webClientEx = findWebClientResponseException(throwable); + if (Objects.nonNull(webClientEx)) { + final int status = webClientEx.getStatusCode().value(); + return status == 429 || status >= 500; + } Review Comment: For WebClientResponseException statuses that isRetryable returns false for (for example 401/403/404), Reactor propagates the original exception from retryWhen, so the onErrorResume(NonTransientAiException.class) fallback path is never invoked. That means a configured fallback provider will be skipped for non-retryable primary failures such as expired primary credentials, whereas fallback should be independent from retry eligibility. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapter.java: ########## @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.shenyu.plugin.ai.common.protocol; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.shenyu.common.utils.JsonUtils; +import org.apache.shenyu.plugin.ai.common.config.AiCommonConfig; +import org.springframework.ai.openai.api.OpenAiApi.ChatCompletionRequest; + +import java.util.Objects; + +/** + * Adapts between OpenAI Chat Completions wire format and internal representations. + * + * <p>Note: deserialization into Spring AI's {@link ChatCompletionRequest} preserves fields + * modeled by that class (messages, model, temperature, maxTokens, stream, tools, etc.). + * Provider-specific extension fields not modeled by {@code ChatCompletionRequest} are dropped + * during deserialization. For the OpenAI-compatible providers this plugin currently supports, + * all required fields are covered. + */ +public final class OpenAiProtocolAdapter { + + private static final com.fasterxml.jackson.databind.ObjectMapper MAPPER = + new com.fasterxml.jackson.databind.ObjectMapper(); + + private OpenAiProtocolAdapter() { + } + + /** + * Resolve the stream flag: client request body takes priority, + * falls back to the provided default value. + * + * @param requestBody the raw JSON request body + * @param fallbackStream the default stream value from admin config + * @return true if streaming, false otherwise + */ + public static boolean resolveStream(final String requestBody, final Boolean fallbackStream) { + if (Objects.isNull(requestBody) || requestBody.isEmpty()) { + return Boolean.TRUE.equals(fallbackStream); + } + final JsonNode root = parseStrict(requestBody); + if (Objects.isNull(root)) { + return Boolean.TRUE.equals(fallbackStream); + } + if (root.hasNonNull("stream")) { + return root.get("stream").asBoolean(); + } + return Boolean.TRUE.equals(fallbackStream); + } + + /** + * Parse raw request body directly into ChatCompletionRequest, preserving fields + * modeled by Spring AI (including reasoning_content in assistant messages). + * + * <p>Spring AI's createRequest() loses reasoning_content, refusal, and annotations + * when reconstructing ChatCompletionMessage from AssistantMessage. + * This method avoids that loss by deserializing the raw JSON directly. + * + * <p>Also converts max_completion_tokens to max_tokens for broader API compatibility. + * + * @param requestBody the raw JSON request body in OpenAI Chat Completions format + * @param stream whether this is a streaming request (sets the stream field) + * @return a ChatCompletionRequest with all fields preserved from the original request Review Comment: This return description still says the request has all fields preserved, but the class-level documentation states provider-specific fields not modeled by ChatCompletionRequest are dropped during deserialization. Narrow this wording to modeled fields to avoid documenting a stronger guarantee than the adapter provides. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/config/AiCommonConfig.java: ########## @@ -50,7 +50,7 @@ public class AiCommonConfig { /** * temperature. */ - private Double temperature = 0.8; + private Double temperature; Review Comment: Removing the default temperature from AiCommonConfig changes behavior for every AI plugin/factory that creates a new AiCommonConfig, not just the ai-proxy adapter: OpenAiModelFactory and DeepSeekModelFactory will no longer set the previous 0.8 default when callers omit temperature. If only the proxy request adapter needs to avoid injecting a default, keep the shared config default or isolate that behavior to the adapter. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapter.java: ########## @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.shenyu.plugin.ai.common.protocol; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.shenyu.common.utils.JsonUtils; +import org.apache.shenyu.plugin.ai.common.config.AiCommonConfig; +import org.springframework.ai.openai.api.OpenAiApi.ChatCompletionRequest; + +import java.util.Objects; + +/** + * Adapts between OpenAI Chat Completions wire format and internal representations. + * + * <p>Note: deserialization into Spring AI's {@link ChatCompletionRequest} preserves fields + * modeled by that class (messages, model, temperature, maxTokens, stream, tools, etc.). + * Provider-specific extension fields not modeled by {@code ChatCompletionRequest} are dropped + * during deserialization. For the OpenAI-compatible providers this plugin currently supports, + * all required fields are covered. + */ +public final class OpenAiProtocolAdapter { + + private static final com.fasterxml.jackson.databind.ObjectMapper MAPPER = + new com.fasterxml.jackson.databind.ObjectMapper(); + + private OpenAiProtocolAdapter() { + } + + /** + * Resolve the stream flag: client request body takes priority, + * falls back to the provided default value. + * + * @param requestBody the raw JSON request body + * @param fallbackStream the default stream value from admin config + * @return true if streaming, false otherwise + */ + public static boolean resolveStream(final String requestBody, final Boolean fallbackStream) { + if (Objects.isNull(requestBody) || requestBody.isEmpty()) { + return Boolean.TRUE.equals(fallbackStream); + } + final JsonNode root = parseStrict(requestBody); + if (Objects.isNull(root)) { + return Boolean.TRUE.equals(fallbackStream); + } + if (root.hasNonNull("stream")) { + return root.get("stream").asBoolean(); + } + return Boolean.TRUE.equals(fallbackStream); + } + + /** + * Parse raw request body directly into ChatCompletionRequest, preserving fields + * modeled by Spring AI (including reasoning_content in assistant messages). + * + * <p>Spring AI's createRequest() loses reasoning_content, refusal, and annotations + * when reconstructing ChatCompletionMessage from AssistantMessage. + * This method avoids that loss by deserializing the raw JSON directly. + * + * <p>Also converts max_completion_tokens to max_tokens for broader API compatibility. + * + * @param requestBody the raw JSON request body in OpenAI Chat Completions format + * @param stream whether this is a streaming request (sets the stream field) + * @return a ChatCompletionRequest with all fields preserved from the original request + */ + public static ChatCompletionRequest toChatCompletionRequest(final String requestBody, final boolean stream) { + return toChatCompletionRequest(requestBody, stream, null); + } + + /** + * Parse raw request body directly into ChatCompletionRequest, preserving modeled fields. + * For model, temperature, max_tokens: client request takes priority; + * if missing, falls back to the corresponding field in fallbackConfig. + * + * @param requestBody the raw JSON request body in OpenAI Chat Completions format + * @param stream whether this is a streaming request + * @param fallbackConfig the admin config used as fallback when client omits fields + * @return a ChatCompletionRequest with all fields preserved + */ + public static ChatCompletionRequest toChatCompletionRequest(final String requestBody, + final boolean stream, final AiCommonConfig fallbackConfig) { + if (Objects.isNull(requestBody) || requestBody.isEmpty()) { + throw new IllegalArgumentException("Request body must not be empty"); + } + final JsonNode root = parseStrict(requestBody); + if (Objects.isNull(root) || !root.isObject()) { + throw new IllegalArgumentException("Invalid request body: expected a JSON object"); + } + final ObjectNode mutableRoot = (ObjectNode) root; + + if (root.hasNonNull("max_completion_tokens") && !root.hasNonNull("max_tokens")) { + final JsonNode tokenNode = root.get("max_completion_tokens"); + if (!tokenNode.isNumber()) { + throw new IllegalArgumentException( + "max_completion_tokens must be a number, got: " + tokenNode.getNodeType()); + } + mutableRoot.put("max_tokens", tokenNode.asInt()); Review Comment: This accepts any JSON number for max_completion_tokens and converts it with asInt(), so a decimal value such as 10.5 is silently truncated to 10 instead of being rejected as an invalid token count. Check for an integral integer value before converting to avoid changing the client's requested limit. ########## shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyExecutorService.java: ########## @@ -39,78 +45,124 @@ public class AiProxyExecutorService { private static final Logger LOG = LoggerFactory.getLogger(AiProxyExecutorService.class); /** - * Execute the AI call with retry and fallback. + * Execute a streaming AI call directly via {@link OpenAiApi}, bypassing Spring AI's + * {@code createRequest()} which loses fields like {@code reasoning_content}. * - * @param mainClient the main chat client - * @param fallbackClientOpt the optional fallback chat client - * @param requestBody the request body - * @return a Mono containing the ChatResponse + * @param mainApi the main OpenAiApi + * @param fallbackCtxOpt the optional fallback context (api + config) + * @param request the ChatCompletionRequest with all fields preserved + * @param requestBody the original request body for rebuilding fallback request + * @param stream whether this is a streaming request + * @return a Flux of ChatCompletionChunk */ - public Mono<ChatResponse> execute(final ChatClient mainClient, final Optional<ChatClient> fallbackClientOpt, final String requestBody) { - final Mono<ChatResponse> mainCall = doChatCall(mainClient, requestBody); + public Flux<ChatCompletionChunk> executeDirectStream(final OpenAiApi mainApi, + final Optional<FallbackContext> fallbackCtxOpt, final ChatCompletionRequest request, + final String requestBody, final boolean stream) { + return mainApi.chatCompletionStream(request) + .doOnError(e -> UpstreamErrorLogger.logUpstreamError(LOG, e, "direct stream")) + .retryWhen(Retry.max(1) + .filter(AiProxyExecutorService::isRetryable) + .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> { + LOG.warn("Direct stream retry exhausted. Triggering fallback.", + retrySignal.failure()); + return new NonTransientAiException( + "Direct stream failed after 1 retry. Triggering fallback.", + retrySignal.failure()); + })) + .onErrorResume(NonTransientAiException.class, + throwable -> handleDirectFallbackStream(throwable, fallbackCtxOpt, requestBody, stream)); + } - return mainCall + /** + * Execute a non-streaming AI call directly via {@link OpenAiApi}. + * + * @param mainApi the main OpenAiApi + * @param fallbackCtxOpt the optional fallback context (api + config) + * @param request the ChatCompletionRequest with all fields preserved + * @param requestBody the original request body for rebuilding fallback request + * @return a Mono of ResponseEntity containing ChatCompletion + */ + public Mono<ResponseEntity<ChatCompletion>> executeDirectCall(final OpenAiApi mainApi, + final Optional<FallbackContext> fallbackCtxOpt, final ChatCompletionRequest request, + final String requestBody) { + return Mono.fromCallable(() -> mainApi.chatCompletionEntity(request)) + .subscribeOn(Schedulers.boundedElastic()) + .doOnError(e -> UpstreamErrorLogger.logUpstreamError(LOG, e, "direct call")) .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)) - .filter(throwable -> !(throwable instanceof NonTransientAiException)) + .filter(AiProxyExecutorService::isRetryable) .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> { - LOG.warn("Retries exhausted for AI call after {} attempts.", + LOG.warn("Direct call retries exhausted after {} attempts. Triggering fallback.", retrySignal.totalRetries(), retrySignal.failure()); - return new NonTransientAiException("Retries exhausted. Triggering fallback.", + return new NonTransientAiException("Direct call retries exhausted. Triggering fallback.", retrySignal.failure()); })) .onErrorResume(NonTransientAiException.class, - throwable -> handleFallback(throwable, fallbackClientOpt, requestBody)); + throwable -> handleDirectFallbackCall(throwable, fallbackCtxOpt, requestBody)); } - protected Mono<ChatResponse> doChatCall(final ChatClient client, final String requestBody) { - return Mono.fromCallable(() -> client.prompt().user(requestBody).call().chatResponse()) - .subscribeOn(Schedulers.boundedElastic()); + private Flux<ChatCompletionChunk> handleDirectFallbackStream(final Throwable throwable, + final Optional<FallbackContext> fallbackCtxOpt, final String requestBody, final boolean stream) { + LOG.warn("Main direct stream failed, attempting fallback...", throwable); + + if (fallbackCtxOpt.isEmpty()) { + return Flux.error(throwable); + } + + final FallbackContext ctx = fallbackCtxOpt.get(); + LOG.info("Using fallback OpenAiApi for direct stream"); + final ChatCompletionRequest fallbackRequest = OpenAiProtocolAdapter.toChatCompletionRequest( + requestBody, stream, ctx.config()); + return ctx.api().chatCompletionStream(fallbackRequest); } - private Mono<ChatResponse> handleFallback(final Throwable throwable, final Optional<ChatClient> fallbackClientOpt, final String requestBody) { - LOG.warn("AI main call failed or retries exhausted, attempting to fallback...", throwable); + private Mono<ResponseEntity<ChatCompletion>> handleDirectFallbackCall(final Throwable throwable, + final Optional<FallbackContext> fallbackCtxOpt, final String requestBody) { + LOG.warn("Main direct call failed, attempting fallback...", throwable); - if (fallbackClientOpt.isEmpty()) { + if (fallbackCtxOpt.isEmpty()) { return Mono.error(throwable); } - return SimpleModelFallbackStrategy.INSTANCE.fallback(fallbackClientOpt.get(), requestBody, throwable); + final FallbackContext ctx = fallbackCtxOpt.get(); + LOG.info("Using fallback OpenAiApi for direct call"); + final ChatCompletionRequest fallbackRequest = OpenAiProtocolAdapter.toChatCompletionRequest( + requestBody, false, ctx.config()); + return Mono.fromCallable(() -> ctx.api().chatCompletionEntity(fallbackRequest)) + .subscribeOn(Schedulers.boundedElastic()); } /** - * Execute the AI call with retry and fallback. - * - * @param mainClient the main chat client - * @param fallbackClientOpt the optional fallback chat client - * @param requestBody the request body - * @return a Flux containing the ChatResponse + * Determine if the error is retryable. + * Retries transient network errors and retryable HTTP statuses (429, 5xx). + * Non-retryable: NonTransientAiException, client errors (400/401/403/404). */ - public Flux<ChatResponse> executeStream(final ChatClient mainClient, final Optional<ChatClient> fallbackClientOpt, final String requestBody) { - final Flux<ChatResponse> mainStream = doChatStream(mainClient, requestBody); - - return mainStream - .retryWhen(Retry.max(1) - .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> { - LOG.warn("Retrying stream once failed. Attempts: {}. Triggering fallback.", - retrySignal.totalRetries(), retrySignal.failure()); - return new NonTransientAiException("Stream failed after 1 retry. Triggering fallback.", retrySignal.failure()); - })) - .onErrorResume(NonTransientAiException.class, - throwable -> handleFallbackStream(throwable, fallbackClientOpt, requestBody)); - } - - protected Flux<ChatResponse> doChatStream(final ChatClient client, final String requestBody) { - return Flux.defer(() -> client.prompt().user(requestBody).stream().chatResponse()) - .subscribeOn(Schedulers.boundedElastic()); + private static boolean isRetryable(final Throwable throwable) { + if (throwable instanceof NonTransientAiException) { + return false; + } + final WebClientResponseException webClientEx = findWebClientResponseException(throwable); + if (Objects.nonNull(webClientEx)) { + final int status = webClientEx.getStatusCode().value(); + return status == 429 || status >= 500; + } + return true; } - private Flux<ChatResponse> handleFallbackStream(final Throwable throwable, final Optional<ChatClient> fallbackClientOpt, final String requestBody) { - LOG.warn("AI main stream failed or retries exhausted, attempting to fallback...", throwable); - - if (fallbackClientOpt.isEmpty()) { - return Flux.error(throwable); + private static WebClientResponseException findWebClientResponseException(final Throwable e) { + Throwable current = e; + while (Objects.nonNull(current)) { + if (current instanceof WebClientResponseException ex) { + return ex; + } + current = current.getCause(); } + return null; Review Comment: This duplicates the same cause-chain scan that UpstreamErrorLogger already implements. Keeping two copies makes it easy for retry classification and logging to diverge; expose a shared helper or centralize WebClientResponseException extraction in one place. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
