Copilot commented on code in PR #6341:
URL: https://github.com/apache/shenyu/pull/6341#discussion_r3245197977
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/AiProxyPlugin.java:
##########
@@ -188,24 +183,26 @@ private Mono<Void> handleNonStreamRequest(
final String requestBody,
final AiCommonConfig primaryConfig,
final AiProxyHandle selectorHandle) {
- final ChatClient mainClient = createMainChatClient(selector.getId(),
primaryConfig);
- final String prompt = aiProxyConfigService.extractPrompt(requestBody);
- final Optional<ChatClient> fallbackClient =
resolveFallbackClient(primaryConfig, selectorHandle,
- selector.getId(), requestBody);
+ final OpenAiApi mainApi = getCachedOpenAiApi(selector.getId(), "main",
primaryConfig);
+ final ChatCompletionRequest request =
OpenAiProtocolAdapter.toChatCompletionRequest(requestBody, false,
primaryConfig);
+ final Optional<OpenAiApi> fallbackApi =
resolveFallbackOpenAiApi(selector.getId(), primaryConfig, selectorHandle,
+ requestBody);
Review Comment:
The non-streaming fallback path also reuses a ChatCompletionRequest built
with the primary config, so fallback-specific model/temperature/max_tokens are
never applied. With OpenAiApi the fallback OpenAiApi only changes endpoint
credentials; the request still targets the primary/client model, which breaks
configured fallback models.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/AiProxyPlugin.java:
##########
@@ -161,23 +153,26 @@ private Mono<Void> handleStreamRequest(
final String requestBody,
final AiCommonConfig primaryConfig,
final AiProxyHandle selectorHandle) {
- final ChatClient mainClient = createMainChatClient(selector.getId(),
primaryConfig);
- final String prompt = aiProxyConfigService.extractPrompt(requestBody);
- final Optional<ChatClient> fallbackClient =
resolveFallbackClient(primaryConfig, selectorHandle,
- selector.getId(), requestBody);
+ final OpenAiApi mainApi = getCachedOpenAiApi(selector.getId(), "main",
primaryConfig);
+ final ChatCompletionRequest request =
OpenAiProtocolAdapter.toChatCompletionRequest(requestBody, true, primaryConfig);
+ final Optional<OpenAiApi> fallbackApi =
resolveFallbackOpenAiApi(selector.getId(), primaryConfig, selectorHandle,
+ requestBody);
Review Comment:
The streaming fallback path builds a single ChatCompletionRequest with the
primary config and then passes that same request to any fallback OpenAiApi.
Because model/temperature/max_tokens live in the request body for
OpenAI-compatible calls, an admin or dynamic fallback with a different model or
generation settings will still receive the primary/client values and may fail
against the fallback provider.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.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.
+ */
+public final class OpenAiProtocolAdapter {
+
+ 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 = JsonUtils.toJsonNode(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
ALL fields
+ * 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
ALL 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 = JsonUtils.toJsonNode(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")) {
+ mutableRoot.put("max_tokens",
root.get("max_completion_tokens").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);
+
+ return JsonUtils.jsonToObject(mutableRoot.toString(),
ChatCompletionRequest.class);
Review Comment:
Deserializing into Spring AI's ChatCompletionRequest does not preserve
arbitrary OpenAI/provider-specific fields from the raw JSON; fields not modeled
by that class are dropped by JsonUtils before OpenAiApi serializes the request
again. That undermines the stated goal of preserving all fields such as
provider extensions from the client request.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/test/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapterTest.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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 org.apache.shenyu.plugin.ai.common.config.AiCommonConfig;
+import org.junit.jupiter.api.Test;
+import org.springframework.ai.openai.api.OpenAiApi.ChatCompletionRequest;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public final class OpenAiProtocolAdapterTest {
+
+ private static final String BASE_BODY =
"{\"messages\":[{\"role\":\"user\",\"content\":\"hello\"}]}";
+
+ @Test
+ void testNullRequestBody() {
+ assertThrows(IllegalArgumentException.class,
+ () -> OpenAiProtocolAdapter.toChatCompletionRequest(null,
false));
+ }
+
+ @Test
+ void testEmptyRequestBody() {
+ assertThrows(IllegalArgumentException.class,
+ () -> OpenAiProtocolAdapter.toChatCompletionRequest("",
false));
+ }
+
+ @Test
+ void testStreamFlagSet() {
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(BASE_BODY, true);
+ assertNotNull(req);
+ assertTrue(req.stream());
+ }
+
+ @Test
+ void testResolveStreamClientTrueOverridesFallbackFalse() {
+ final String body =
"{\"model\":\"m\",\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}],\"stream\":true}";
+ assertTrue(OpenAiProtocolAdapter.resolveStream(body, false));
+ }
+
+ @Test
+ void testResolveStreamClientFalseOverridesFallbackTrue() {
+ final String body =
"{\"model\":\"m\",\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}],\"stream\":false}";
+ assertFalse(OpenAiProtocolAdapter.resolveStream(body, true));
+ }
+
+ @Test
+ void testResolveStreamFallbackWhenClientMissing() {
+ assertTrue(OpenAiProtocolAdapter.resolveStream(BASE_BODY, true));
+ assertFalse(OpenAiProtocolAdapter.resolveStream(BASE_BODY, false));
+ }
+
+ @Test
+ void testResolveStreamFallbackWhenClientMissingAndConfigNull() {
+ assertFalse(OpenAiProtocolAdapter.resolveStream(BASE_BODY, null));
+ }
+
+ @Test
+ void testResolveStreamFallbackWhenBodyEmpty() {
+ assertTrue(OpenAiProtocolAdapter.resolveStream("", true));
+ assertFalse(OpenAiProtocolAdapter.resolveStream("", false));
+ }
+
+ @Test
+ void testStreamFlagUnset() {
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(BASE_BODY, false);
+ assertNotNull(req);
+ assertFalse(req.stream());
+ }
+
+ @Test
+ void testMaxCompletionTokensConvertedToMaxTokens() {
+ final String body =
"{\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}],\"max_completion_tokens\":100}";
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(body, false);
+ assertNotNull(req);
+ assertEquals(100, req.maxTokens());
+ }
+
+ @Test
+ void testClientModelTakesPriority() {
+ final String body =
"{\"model\":\"client-model\",\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}]}";
+ final AiCommonConfig config = new AiCommonConfig();
+ config.setModel("admin-model");
+
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(body, false, config);
+ assertEquals("client-model", req.model());
+ }
+
+ @Test
+ void testFallbackModelWhenClientMissing() {
+ final AiCommonConfig config = new AiCommonConfig();
+ config.setModel("admin-model");
+
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(BASE_BODY, false, config);
+ assertEquals("admin-model", req.model());
+ }
+
+ @Test
+ void testNoFallbackModelWhenConfigIsNull() {
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(BASE_BODY, false, null);
+ assertNotNull(req);
+ }
+
+ @Test
+ void testNoFallbackWhenConfigModelIsNull() {
+ final AiCommonConfig config = new AiCommonConfig();
+ config.setModel(null);
+
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(BASE_BODY, false, config);
+ assertNotNull(req);
+ }
+
+ @Test
+ void testNoFallbackWhenConfigModelIsEmpty() {
+ final AiCommonConfig config = new AiCommonConfig();
+ config.setModel("");
+
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(BASE_BODY, false, config);
+ assertNotNull(req);
+ }
+
+ @Test
+ void testClientTemperatureTakesPriority() {
+ final String body =
"{\"model\":\"m\",\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}],\"temperature\":0.5}";
+ final AiCommonConfig config = new AiCommonConfig();
+ config.setTemperature(0.9);
+
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(body, false, config);
+ assertEquals(0.5, req.temperature(), 0.001);
+ }
+
+ @Test
+ void testFallbackTemperatureWhenClientMissing() {
+ final AiCommonConfig config = new AiCommonConfig();
+ config.setTemperature(0.7);
+
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(BASE_BODY, false, config);
+ assertEquals(0.7, req.temperature(), 0.001);
+ }
+
+ @Test
+ void testNoFallbackTemperatureWhenConfigNull() {
+ final AiCommonConfig config = new AiCommonConfig();
+ config.setTemperature(null);
+
+ final ChatCompletionRequest req =
OpenAiProtocolAdapter.toChatCompletionRequest(BASE_BODY, false, config);
Review Comment:
The test name says the config is null, but the test creates a non-null
AiCommonConfig with a null temperature. Rename it to describe the actual case
so failures are easier to diagnose.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.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.
+ */
+public final class OpenAiProtocolAdapter {
+
+ 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 = JsonUtils.toJsonNode(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
ALL fields
+ * 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
ALL 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 = JsonUtils.toJsonNode(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")) {
+ mutableRoot.put("max_tokens",
root.get("max_completion_tokens").asInt());
Review Comment:
Using asInt() silently coerces non-numeric max_completion_tokens values to 0
and can truncate out-of-range numbers, so malformed client input is changed
before validation instead of being rejected or forwarded accurately. Preserve
the JsonNode value or validate that it is an integer before converting.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyExecutorService.java:
##########
@@ -39,78 +41,77 @@ 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 fallbackApiOpt the optional fallback OpenAiApi
+ * @param request the ChatCompletionRequest with all fields preserved
+ * @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);
-
- return mainCall
- .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
- .filter(throwable -> !(throwable instanceof
NonTransientAiException))
+ public Flux<ChatCompletionChunk> executeDirectStream(final OpenAiApi
mainApi,
+ final Optional<OpenAiApi> fallbackApiOpt, final
ChatCompletionRequest request) {
+ return mainApi.chatCompletionStream(request)
+ .doOnError(e -> UpstreamErrorLogger.logUpstreamError(LOG, e,
"direct stream"))
+ .retryWhen(Retry.max(1)
.onRetryExhaustedThrow((retryBackoffSpec, retrySignal)
-> {
- LOG.warn("Retries exhausted for AI call after {}
attempts.",
- retrySignal.totalRetries(),
retrySignal.failure());
- return new NonTransientAiException("Retries
exhausted. Triggering fallback.",
+ 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 -> handleFallback(throwable,
fallbackClientOpt, requestBody));
- }
-
- protected Mono<ChatResponse> doChatCall(final ChatClient client, final
String requestBody) {
- return Mono.fromCallable(() ->
client.prompt().user(requestBody).call().chatResponse())
- .subscribeOn(Schedulers.boundedElastic());
- }
-
- 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);
-
- if (fallbackClientOpt.isEmpty()) {
- return Mono.error(throwable);
- }
-
- return
SimpleModelFallbackStrategy.INSTANCE.fallback(fallbackClientOpt.get(),
requestBody, throwable);
+ throwable -> handleDirectFallbackStream(throwable,
fallbackApiOpt, request));
}
/**
- * Execute the AI call with retry and fallback.
+ * Execute a non-streaming AI call directly via {@link OpenAiApi}.
*
- * @param mainClient the main chat client
- * @param fallbackClientOpt the optional fallback chat client
- * @param requestBody the request body
- * @return a Flux containing the ChatResponse
+ * @param mainApi the main OpenAiApi
+ * @param fallbackApiOpt the optional fallback OpenAiApi
+ * @param request the ChatCompletionRequest with all fields preserved
+ * @return a Mono of ResponseEntity containing ChatCompletion
*/
- 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)
+ public Mono<ResponseEntity<ChatCompletion>> executeDirectCall(final
OpenAiApi mainApi,
+ final Optional<OpenAiApi> fallbackApiOpt, final
ChatCompletionRequest request) {
+ 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))
Review Comment:
The retry filter retries every exception except NonTransientAiException,
including upstream 400/401/403 validation or authentication errors. Those
failures are not transient, so each bad request waits through the full backoff
sequence and may trigger fallback unnecessarily; filter retries to transient
network errors and retryable statuses such as 429/5xx.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/AiProxyPlugin.java:
##########
@@ -214,86 +211,35 @@ private Optional<ChatClient> resolveFallbackClient(
if (LOG.isDebugEnabled()) {
LOG.debug("[AiProxy] dynamic fallback config: {}",
cfg);
}
- return createDynamicFallbackClient(cfg);
+ return getCachedOpenAiApi(selectorId, "dynamicFallback",
cfg);
Review Comment:
Dynamic fallback config is parsed from the client request body, but this now
caches the resulting OpenAiApi globally. That lets clients retain
request-supplied API keys/base URLs in process memory beyond the request and
churn the shared cache with arbitrary fallback configs; dynamic fallbacks
should remain per-request or otherwise be bounded/redacted separately from
admin-configured clients.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.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.
+ */
+public final class OpenAiProtocolAdapter {
+
+ 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 = JsonUtils.toJsonNode(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
ALL fields
+ * 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
ALL 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 = JsonUtils.toJsonNode(requestBody);
+ if (Objects.isNull(root) || !root.isObject()) {
+ throw new IllegalArgumentException("Invalid request body: expected
a JSON object");
Review Comment:
JsonUtils.toJsonNode swallows parse errors and returns an empty ObjectNode,
so malformed JSON passes this validation as an empty request instead of being
rejected. This can forward invalid client input to the upstream provider with
missing messages/model and produce misleading upstream failures.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/UpstreamErrorLogger.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.proxy.enhanced.service;
+
+import org.slf4j.Logger;
+import
org.springframework.web.reactive.function.client.WebClientResponseException;
+
+import java.util.Objects;
+
+/**
+ * Shared utility for logging upstream AI service errors with
WebClientResponseException details.
+ */
+public final class UpstreamErrorLogger {
+
+ private UpstreamErrorLogger() {
+ }
+
+ public static void logUpstreamError(final Logger log, final Throwable e,
final String mode) {
+ final WebClientResponseException webClientEx =
findWebClientResponseException(e);
+ if (Objects.nonNull(webClientEx)) {
+ log.error("[AiProxy] {} failed, status={}, upstreamBody={}",
+ mode, webClientEx.getStatusCode(),
webClientEx.getResponseBodyAsString(), e);
Review Comment:
This logs the full upstream response body at error level. AI provider error
bodies can include echoed prompt content, account details, or other sensitive
data, and they can be large; redact or truncate the body, or gate full bodies
behind debug logging.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyExecutorService.java:
##########
@@ -39,78 +41,77 @@ 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 fallbackApiOpt the optional fallback OpenAiApi
+ * @param request the ChatCompletionRequest with all fields preserved
+ * @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);
-
- return mainCall
- .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
- .filter(throwable -> !(throwable instanceof
NonTransientAiException))
+ public Flux<ChatCompletionChunk> executeDirectStream(final OpenAiApi
mainApi,
+ final Optional<OpenAiApi> fallbackApiOpt, final
ChatCompletionRequest request) {
+ return mainApi.chatCompletionStream(request)
+ .doOnError(e -> UpstreamErrorLogger.logUpstreamError(LOG, e,
"direct stream"))
+ .retryWhen(Retry.max(1)
Review Comment:
Streaming retries are unconditional, so non-retryable upstream responses
such as invalid request, unauthorized, or forbidden are sent twice before
falling back. This adds latency and load for client/configuration errors;
restrict retries to transient failures or retryable HTTP statuses.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/handler/AiProxyPluginHandlerTest.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.proxy.enhanced.handler;
+
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.dto.convert.rule.AiProxyHandle;
+import org.apache.shenyu.common.enums.PluginEnum;
+import org.apache.shenyu.plugin.base.utils.CacheKeyUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class AiProxyPluginHandlerTest {
+
+ private AiProxyPluginHandler handler;
+
+ @BeforeEach
+ void setUp() {
+ handler = new AiProxyPluginHandler();
+ }
+
+ @Test
+ void testPluginNamed() {
+ assertEquals(PluginEnum.AI_PROXY.getName(), handler.pluginNamed());
+ }
+
+ @Test
+ void testHandlerSelectorCachesHandle() {
+ SelectorData selector = new SelectorData();
+ selector.setId("sel-1");
+
selector.setHandle("{\"provider\":\"open_ai\",\"baseUrl\":\"https://api.openai.com\",\"apiKey\":\"sk-test\",\"model\":\"gpt-4\"}";);
+
+ handler.handlerSelector(selector);
+
+ String key = CacheKeyUtils.INST.getKey("sel-1",
Constants.DEFAULT_RULE);
+ AiProxyHandle cached =
handler.getSelectorCachedHandle().obtainHandle(key);
+ assertNotNull(cached);
+ assertEquals("open_ai", cached.getProvider());
+ assertEquals("https://api.openai.com";, cached.getBaseUrl());
+ assertEquals("sk-test", cached.getApiKey());
+ assertEquals("gpt-4", cached.getModel());
+ }
+
+ @Test
+ void testHandlerSelectorWithNullHandle() {
+ SelectorData selector = new SelectorData();
+ selector.setId("sel-2");
+ selector.setHandle(null);
+
+ handler.handlerSelector(selector);
+
+ String key = CacheKeyUtils.INST.getKey("sel-2",
Constants.DEFAULT_RULE);
+ assertNull(handler.getSelectorCachedHandle().obtainHandle(key));
+ }
+
+ @Test
+ void testHandlerSelectorWithEmptyHandle() {
+ SelectorData selector = new SelectorData();
+ selector.setId("sel-3");
+ selector.setHandle("");
+
Review Comment:
This test never calls handler.handlerSelector(selector), so it passes
without exercising the empty-handle behavior it is named to cover. Add the
invocation before asserting the cache state so regressions in empty-handle
handling are caught.
##########
shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-common/src/main/java/org/apache/shenyu/plugin/ai/common/protocol/OpenAiProtocolAdapter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.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.
+ */
+public final class OpenAiProtocolAdapter {
+
+ 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 = JsonUtils.toJsonNode(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
ALL fields
+ * 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
ALL 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 = JsonUtils.toJsonNode(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")) {
+ mutableRoot.put("max_tokens",
root.get("max_completion_tokens").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);
+
+ return JsonUtils.jsonToObject(mutableRoot.toString(),
ChatCompletionRequest.class);
Review Comment:
JsonUtils.jsonToObject returns null when Jackson cannot map the body to
ChatCompletionRequest, but this method returns that null without checking.
Callers then pass a null request into OpenAiApi, turning malformed input into a
downstream NullPointerException instead of a clear validation error.
--
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]
