From 9c093974a165905f23bfb5faad92c57329ff9881 Mon Sep 17 00:00:00 2001 From: easyai Date: Sat, 1 Aug 2026 22:40:17 +0800 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20=E4=BF=AE=E5=A4=8D=20Qwen3.8=20?= =?UTF-8?q?=E6=8E=A8=E7=90=86=E5=8F=82=E6=95=B0=E5=B9=B6=E5=A2=9E=E5=8A=A0?= =?UTF-8?q?=E4=B8=8A=E6=B8=B8=E8=87=AA=E6=84=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 统一 Qwen3.8 固定思考、推理档位、预算互斥与温度下限规则，并覆盖 Chat Completions 和原生 Responses 请求。 新增安全参数纠正、有限重试和并发安全的进程内 LRU 学习；补充共享行为向量、httptest 回归与真实模型验收用例。 验证：go test ./... -count=1 通过，gofmt 检查通过；真实 Qwen3.8 流式、非流式及缓存重学习验收通过。 --- apps/api/internal/clients/chat_reasoning.go | 250 +++++++++- .../clients/chat_reasoning_qwen38_test.go | 184 ++++++++ apps/api/internal/clients/openai.go | 71 ++- .../internal/clients/openai_request_params.go | 2 +- .../internal/clients/parameter_correction.go | 445 ++++++++++++++++++ .../clients/parameter_correction_test.go | 298 ++++++++++++ apps/api/internal/clients/qwen38_live_test.go | 312 ++++++++++++ .../testdata/qwen38-reasoning-vectors.json | 10 + apps/api/internal/runner/service.go | 8 +- 9 files changed, 1547 insertions(+), 33 deletions(-) create mode 100644 apps/api/internal/clients/chat_reasoning_qwen38_test.go create mode 100644 apps/api/internal/clients/parameter_correction.go create mode 100644 apps/api/internal/clients/parameter_correction_test.go create mode 100644 apps/api/internal/clients/qwen38_live_test.go create mode 100644 apps/api/internal/clients/testdata/qwen38-reasoning-vectors.json diff --git a/apps/api/internal/clients/chat_reasoning.go b/apps/api/internal/clients/chat_reasoning.go index bf8ff4d..75858eb 100644 --- a/apps/api/internal/clients/chat_reasoning.go +++ b/apps/api/internal/clients/chat_reasoning.go @@ -12,7 +12,8 @@ import ( const OpenAIReasoningEffortValidationMessage = "reasoning_effort must be one of: none, minimal, low, medium, high, xhigh, max" var ( - openAIReasoningEfforts = map[string]struct{}{ + chatReasoningEffortOrder = []string{"none", "minimal", "low", "medium", "high", "xhigh", "max"} + openAIReasoningEfforts = map[string]struct{}{ "none": {}, "minimal": {}, "low": {}, @@ -50,9 +51,25 @@ func ValidateOpenAIReasoningEffort(value any) error { func applyOpenAIChatReasoningParams(body map[string]any, candidate store.RuntimeModelCandidate) { effort := normalizedReasoningString(body["reasoning_effort"]) + model := chatReasoningModelName(body, candidate) + if isAliyunBailianOpenAI(candidate) && isAliyunQwen38MaxPreview(model) { + applyAliyunQwen38Reasoning(body, effort) + if temperature, ok := finiteFloatFromAny(body["temperature"]); ok && temperature < 0.6 { + body["temperature"] = 0.6 + } + return + } if effort == "" || !isOpenAIReasoningEffort(effort) { return } + resolved, state := resolveCandidateReasoningEffort(effort, candidate) + if state == reasoningCapabilityUnsupported { + delete(body, "reasoning_effort") + return + } + if resolved != "" { + effort = resolved + } body["reasoning_effort"] = effort switch { @@ -67,8 +84,183 @@ func applyOpenAIChatReasoningParams(body map[string]any, candidate store.Runtime } } +func applyOpenAIResponsesReasoningParams(body map[string]any, candidate store.RuntimeModelCandidate) { + reasoning, _ := body["reasoning"].(map[string]any) + if reasoning != nil { + reasoning = cloneBody(reasoning) + } + effort := "" + if reasoning != nil { + effort = normalizedReasoningString(reasoning["effort"]) + } + model := chatReasoningModelName(body, candidate) + qwen38 := isAliyunBailianOpenAI(candidate) && isAliyunQwen38MaxPreview(model) + if qwen38 { + if effort == "" { + if enabled, ok := body["enable_thinking"].(bool); ok && !enabled { + effort = "low" + } + } else { + effort = qwen38MaxPreviewReasoningEffort(effort) + } + } else if effort != "" && isOpenAIReasoningEffort(effort) { + resolved, state := resolveCandidateReasoningEffort(effort, candidate) + if state == reasoningCapabilityUnsupported { + effort = "" + } else if resolved != "" { + effort = resolved + } + } + + if effort != "" { + if reasoning == nil { + reasoning = map[string]any{} + } + reasoning["effort"] = effort + } else if reasoning != nil { + delete(reasoning, "effort") + } + if len(reasoning) > 0 { + body["reasoning"] = reasoning + } else { + delete(body, "reasoning") + } + delete(body, "reasoning_effort") + delete(body, "enable_thinking") + if qwen38 { + if temperature, ok := finiteFloatFromAny(body["temperature"]); ok && temperature < 0.6 { + body["temperature"] = 0.6 + } + } +} + +func applyAliyunQwen38Reasoning(body map[string]any, effort string) { + defer delete(body, "thinking_budget_tokens") + requestedDisable := false + if enabled, ok := body["enable_thinking"].(bool); ok && !enabled { + requestedDisable = true + } + body["enable_thinking"] = true + budget, hasBudget := nonNegativeIntFromAny(body["thinking_budget"]) + if !hasBudget { + budget, hasBudget = nonNegativeIntFromAny(body["thinking_budget_tokens"]) + } + if hasBudget { + if budget > 262144 { + budget = 262144 + } + body["thinking_budget"] = budget + delete(body, "reasoning_effort") + return + } + delete(body, "thinking_budget") + if effort == "" { + if requestedDisable { + body["reasoning_effort"] = "low" + } + return + } + body["reasoning_effort"] = qwen38MaxPreviewReasoningEffort(effort) +} + +func qwen38MaxPreviewReasoningEffort(effort string) string { + switch effort { + case "none", "minimal", "low": + return "low" + case "medium": + return "medium" + default: + return "xhigh" + } +} + +type reasoningCapabilityState int + +const ( + reasoningCapabilityUnknown reasoningCapabilityState = iota + reasoningCapabilityUnsupported + reasoningCapabilitySupported +) + +func resolveCandidateReasoningEffort(effort string, candidate store.RuntimeModelCandidate) (string, reasoningCapabilityState) { + supported := map[string]struct{}{} + declaredLevels := false + explicitlyUnsupported := false + explicitlySupported := false + for _, key := range []string{"text_generate", "tools_call", "image_analysis", "video_understanding", "audio_understanding", "omni"} { + capability, ok := candidate.Capabilities[key].(map[string]any) + if !ok { + continue + } + if value, ok := capability["supportThinking"].(bool); ok && !value { + explicitlyUnsupported = true + continue + } + if value, ok := capability["supportThinking"].(bool); ok && value { + explicitlySupported = true + } + rawLevels, ok := capability["thinkingEffortLevels"].([]any) + if !ok { + if stringsValue, stringsOK := capability["thinkingEffortLevels"].([]string); stringsOK { + rawLevels = make([]any, len(stringsValue)) + for index, value := range stringsValue { + rawLevels[index] = value + } + ok = true + } + } + if !ok { + continue + } + declaredLevels = true + for _, raw := range rawLevels { + level := normalizedReasoningString(raw) + if isOpenAIReasoningEffort(level) { + supported[level] = struct{}{} + } + } + } + if len(supported) == 0 { + if declaredLevels || (explicitlyUnsupported && !explicitlySupported) { + return "", reasoningCapabilityUnsupported + } + return effort, reasoningCapabilityUnknown + } + requestedIndex := reasoningEffortIndex(effort) + best := "" + bestDistance := len(chatReasoningEffortOrder) + 1 + for index, candidateEffort := range chatReasoningEffortOrder { + if _, ok := supported[candidateEffort]; !ok { + continue + } + distance := index - requestedIndex + if distance < 0 { + distance = -distance + } + if distance < bestDistance { + best = candidateEffort + bestDistance = distance + } + } + return best, reasoningCapabilitySupported +} + +func reasoningEffortIndex(effort string) int { + for index, value := range chatReasoningEffortOrder { + if value == effort { + return index + } + } + return 0 +} + func applyAliyunReasoning(body map[string]any, candidate store.RuntimeModelCandidate, effort string) { defer delete(body, "thinking_budget_tokens") + budget, hasBudget := positiveIntFromAny(body["thinking_budget"]) + if !hasBudget { + budget, hasBudget = positiveIntFromAny(body["thinking_budget_tokens"]) + } + delete(body, "thinking_budget") if effort == "none" { body["enable_thinking"] = false @@ -94,7 +286,7 @@ func applyAliyunReasoning(body map[string]any, candidate store.RuntimeModelCandi delete(body, "reasoning_effort") if isAliyunThinkingBudgetModel(model) { - if budget, ok := positiveIntFromAny(body["thinking_budget_tokens"]); ok { + if hasBudget { body["thinking_budget"] = budget } } @@ -232,17 +424,6 @@ func highMaxReasoningEffort(effort string) string { return "high" } -func qwen38MaxPreviewReasoningEffort(effort string) string { - switch effort { - case "low", "minimal": - return "low" - case "medium": - return "medium" - default: - return "xhigh" - } -} - func zhipuReasoningEffort(effort string) string { if effort == "max" { return "xhigh" @@ -306,6 +487,49 @@ func positiveIntFromAny(value any) (int, bool) { return int(math.Floor(number)), true } +func nonNegativeIntFromAny(value any) (int, bool) { + number, ok := finiteFloatFromAny(value) + if !ok || number < 0 { + return 0, false + } + return int(math.Floor(number)), true +} + +func finiteFloatFromAny(value any) (float64, bool) { + if value == nil { + return 0, false + } + var number float64 + switch typed := value.(type) { + case int: + number = float64(typed) + case int64: + number = float64(typed) + case float64: + number = typed + case float32: + number = float64(typed) + case jsonNumber: + parsed, err := typed.Float64() + if err != nil { + return 0, false + } + number = parsed + case string: + parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64) + if err != nil { + return 0, false + } + number = parsed + default: + return 0, false + } + if math.IsNaN(number) || math.IsInf(number, 0) { + return 0, false + } + return number, true +} + type jsonNumber interface { Float64() (float64, error) } diff --git a/apps/api/internal/clients/chat_reasoning_qwen38_test.go b/apps/api/internal/clients/chat_reasoning_qwen38_test.go new file mode 100644 index 0000000..5580452 --- /dev/null +++ b/apps/api/internal/clients/chat_reasoning_qwen38_test.go @@ -0,0 +1,184 @@ +package clients + +import ( + "encoding/json" + "os" + "testing" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func TestQwen38MatchesCrossServiceBehaviorVectors(t *testing.T) { + raw, err := os.ReadFile("testdata/qwen38-reasoning-vectors.json") + if err != nil { + t.Fatalf("read behavior vectors: %v", err) + } + var vectors []struct { + Name string `json:"name"` + Input map[string]any `json:"input"` + Expected map[string]any `json:"expected"` + } + if err := json.Unmarshal(raw, &vectors); err != nil { + t.Fatalf("decode behavior vectors: %v", err) + } + candidate := store.RuntimeModelCandidate{Provider: "aliyun-bailian-openai", ProviderModelName: "qwen3.8-max-preview"} + for _, vector := range vectors { + t.Run(vector.Name, func(t *testing.T) { + body := cloneBody(vector.Input) + body["model"] = "qwen3.8-max-preview" + applyOpenAIChatReasoningParams(body, candidate) + for _, key := range []string{"enable_thinking", "reasoning_effort", "thinking_budget", "temperature"} { + expected, expectedPresent := vector.Expected[key] + actual, actualPresent := body[key] + if expectedPresent != actualPresent || (expectedPresent && normalizedJSONScalar(actual) != normalizedJSONScalar(expected)) { + t.Fatalf("%s mismatch: expected=%#v actual=%#v body=%+v", key, expected, actual, body) + } + } + }) + } +} + +func normalizedJSONScalar(value any) any { + if number, ok := value.(int); ok { + return float64(number) + } + return value +} + +func TestQwen38MaxPreviewReasoningNormalization(t *testing.T) { + tests := []struct { + name string + body map[string]any + effort any + budget any + temp any + enabled any + noEffort bool + }{ + {name: "none", body: map[string]any{"reasoning_effort": "none"}, effort: "low"}, + {name: "minimal", body: map[string]any{"reasoning_effort": "minimal"}, effort: "low"}, + {name: "medium", body: map[string]any{"reasoning_effort": "medium"}, effort: "medium"}, + {name: "high", body: map[string]any{"reasoning_effort": "high"}, effort: "xhigh"}, + {name: "max", body: map[string]any{"reasoning_effort": "max"}, effort: "xhigh"}, + {name: "unspecified", body: map[string]any{}, noEffort: true}, + {name: "forced on", body: map[string]any{"enable_thinking": false}, effort: "low"}, + {name: "budget wins", body: map[string]any{"reasoning_effort": "high", "thinking_budget": 999999}, budget: 262144, noEffort: true}, + {name: "zero budget", body: map[string]any{"reasoning_effort": "medium", "thinking_budget_tokens": 0}, budget: 0, noEffort: true}, + {name: "temperature floor", body: map[string]any{"temperature": 0.1}, temp: 0.6, noEffort: true}, + } + candidate := store.RuntimeModelCandidate{ + Provider: "aliyun-bailian-openai", + ProviderModelName: "qwen3.8-max-preview", + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + body := cloneBody(test.body) + body["model"] = "qwen3.8-max-preview" + applyOpenAIChatReasoningParams(body, candidate) + if body["enable_thinking"] != true { + t.Fatalf("Qwen3.8 must always enable thinking: %+v", body) + } + if test.noEffort { + if _, ok := body["reasoning_effort"]; ok { + t.Fatalf("reasoning_effort must be omitted: %+v", body) + } + } else if body["reasoning_effort"] != test.effort { + t.Fatalf("expected effort %#v, got %+v", test.effort, body) + } + if test.budget != nil && body["thinking_budget"] != test.budget { + t.Fatalf("expected budget %#v, got %+v", test.budget, body) + } + if test.temp != nil && body["temperature"] != test.temp { + t.Fatalf("expected temperature %#v, got %+v", test.temp, body) + } + }) + } +} + +func TestGenericReasoningCapabilityUsesNearestTierAndIgnoresModeSwitch(t *testing.T) { + body := map[string]any{"reasoning_effort": "medium"} + applyOpenAIChatReasoningParams(body, store.RuntimeModelCandidate{ + Provider: "openai", + Capabilities: map[string]any{ + "text_generate": map[string]any{ + "supportThinking": true, + "supportThinkingModeSwitch": false, + "thinkingEffortLevels": []any{"low", "high"}, + }, + }, + }) + if body["reasoning_effort"] != "low" { + t.Fatalf("equidistant effort must choose lower tier, got %+v", body) + } +} + +func TestGenericReasoningKeepsUnknownAndRemovesExplicitUnsupported(t *testing.T) { + unknown := map[string]any{"reasoning_effort": "high"} + applyOpenAIChatReasoningParams(unknown, store.RuntimeModelCandidate{Provider: "openai"}) + if unknown["reasoning_effort"] != "high" { + t.Fatalf("unknown capability should preserve effort: %+v", unknown) + } + unsupported := map[string]any{"reasoning_effort": "high"} + applyOpenAIChatReasoningParams(unsupported, store.RuntimeModelCandidate{ + Provider: "openai", + Capabilities: map[string]any{ + "text_generate": map[string]any{"supportThinking": false}, + }, + }) + if _, ok := unsupported["reasoning_effort"]; ok { + t.Fatalf("explicit unsupported capability should remove effort: %+v", unsupported) + } +} + +func TestQwen38ResponsesReasoningNormalization(t *testing.T) { + candidate := store.RuntimeModelCandidate{ + Provider: "aliyun-bailian-openai", + ProviderModelName: "qwen3.8-max-preview", + } + tests := []struct { + name string + requested string + expected string + }{ + {name: "none", requested: "none", expected: "low"}, + {name: "minimal", requested: "minimal", expected: "low"}, + {name: "medium", requested: "medium", expected: "medium"}, + {name: "high", requested: "high", expected: "xhigh"}, + {name: "max", requested: "max", expected: "xhigh"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + body := map[string]any{ + "model": "qwen3.8-max-preview", + "reasoning": map[string]any{"effort": test.requested, "summary": "auto"}, + "temperature": 0.1, + } + applyOpenAIResponsesReasoningParams(body, candidate) + reasoning := body["reasoning"].(map[string]any) + if reasoning["effort"] != test.expected || reasoning["summary"] != "auto" { + t.Fatalf("unexpected Responses reasoning: %+v", body) + } + if body["temperature"] != 0.6 { + t.Fatalf("Qwen3.8 Responses temperature was not floored: %+v", body) + } + }) + } +} + +func TestQwen38ResponsesForcedThinkingAndDefault(t *testing.T) { + candidate := store.RuntimeModelCandidate{Provider: "aliyun-bailian-openai", ProviderModelName: "qwen3.8-max-preview"} + forced := map[string]any{"enable_thinking": false} + applyOpenAIResponsesReasoningParams(forced, candidate) + if forced["reasoning"].(map[string]any)["effort"] != "low" { + t.Fatalf("disabled Qwen3.8 Responses request must become low: %+v", forced) + } + if _, ok := forced["enable_thinking"]; ok { + t.Fatalf("Chat-only alias leaked into Responses body: %+v", forced) + } + defaulted := map[string]any{"reasoning": map[string]any{"summary": "auto"}} + applyOpenAIResponsesReasoningParams(defaulted, candidate) + reasoning := defaulted["reasoning"].(map[string]any) + if _, ok := reasoning["effort"]; ok || reasoning["summary"] != "auto" { + t.Fatalf("unspecified effort must preserve upstream xhigh default: %+v", defaulted) + } +} diff --git a/apps/api/internal/clients/openai.go b/apps/api/internal/clients/openai.go index 2f96131..b7d71bc 100644 --- a/apps/api/internal/clients/openai.go +++ b/apps/api/internal/clients/openai.go @@ -18,7 +18,8 @@ import ( ) type OpenAIClient struct { - HTTPClient *http.Client + HTTPClient *http.Client + Corrections *ParameterCorrectionCache } func (c OpenAIClient) Run(ctx context.Context, request Request) (Response, error) { @@ -56,6 +57,7 @@ func (c OpenAIClient) Run(ctx context.Context, request Request) (Response, error applyOpenAIChatReasoningParams(body, request.Candidate) body = FilterOpenAIChatRequestBody(body) } else if request.Kind == "responses" { + applyOpenAIResponsesReasoningParams(body, request.Candidate) body = FilterOpenAIResponsesRequestBody(body) if _, hasInput := body["input"]; !hasInput { if messages, hasMessages := request.Body["messages"]; hasMessages { @@ -73,31 +75,64 @@ func (c OpenAIClient) Run(ctx context.Context, request Request) (Response, error normalizeOpenAIImageRequestBody(endpointKind, body, request.OriginalBody) stream := openAIEndpointSupportsStream(endpointKind) && (request.Stream || boolValue(body, "stream")) ensureOpenAIStreamUsage(body, endpointKind, stream) - raw, contentType, err := openAIRequestPayload(endpointKind, body, request.Candidate) - if err != nil { - return Response{}, err - } upstreamEndpoint := joinURL(openAIBaseURL(endpointKind, request.Candidate), endpoint) - req, err := http.NewRequestWithContext(ctx, http.MethodPost, upstreamEndpoint, bytes.NewReader(raw)) - if err != nil { - return Response{}, err - } - req.Header.Set("Content-Type", contentType) - req.Header.Set("Authorization", "Bearer "+apiKey) responseStartedAt := time.Now() - if err := notifySubmissionStarted(request); err != nil { - return Response{}, err + correctionScope := newParameterCorrectionScope(request, endpointKind) + correctionEnabled := endpointKind == "chat.completions" || endpointKind == "responses" + if correctionEnabled { + c.Corrections.apply(correctionScope, body) } - resp, err := httpClient(request.HTTPClient, c.HTTPClient).Do(req) - if err != nil { - return Response{}, &ClientError{Code: "network", Message: err.Error(), Retryable: true} + provisionalRules := make([]parameterCorrectionRule, 0, 2) + seenCorrectionErrors := make(map[string]struct{}) + var resp *http.Response + for correctionAttempt := 0; ; correctionAttempt++ { + raw, contentType, payloadErr := openAIRequestPayload(endpointKind, body, request.Candidate) + if payloadErr != nil { + return Response{}, payloadErr + } + req, requestErr := http.NewRequestWithContext(ctx, http.MethodPost, upstreamEndpoint, bytes.NewReader(raw)) + if requestErr != nil { + return Response{}, requestErr + } + req.Header.Set("Content-Type", contentType) + req.Header.Set("Authorization", "Bearer "+apiKey) + if err := notifySubmissionStarted(request); err != nil { + return Response{}, err + } + resp, requestErr = httpClient(request.HTTPClient, c.HTTPClient).Do(req) + if requestErr != nil { + return Response{}, &ClientError{Code: "network", Message: requestErr.Error(), Retryable: true} + } + if err := notifyResponseReceived(request); err != nil { + _ = resp.Body.Close() + return Response{}, err + } + if !correctionEnabled || (resp.StatusCode != http.StatusBadRequest && resp.StatusCode != http.StatusUnprocessableEntity) { + break + } + _, _, upstreamErr := decodeHTTPResponseForProtocol(resp, openAIWireProtocol(endpointKind)) + if upstreamErr == nil { + break + } + fingerprint := fmt.Sprintf("%d:%s", resp.StatusCode, strings.ToLower(upstreamErr.Error())) + if _, repeated := seenCorrectionErrors[fingerprint]; repeated || correctionAttempt >= 2 { + return Response{}, annotateResponseError(upstreamErr, requestIDFromParameterError(upstreamErr), responseStartedAt, time.Now()) + } + seenCorrectionErrors[fingerprint] = struct{}{} + rule, safe := deriveParameterCorrection(upstreamErr, body, request.Candidate) + if !safe || !applyParameterCorrectionRule(body, rule) { + return Response{}, annotateResponseError(upstreamErr, requestIDFromParameterError(upstreamErr), responseStartedAt, time.Now()) + } + c.Corrections.invalidate(correctionScope, rule.Param) + provisionalRules = append(provisionalRules, rule) } - if err := notifyResponseReceived(request); err != nil { - return Response{}, err + if len(provisionalRules) > 0 && resp.StatusCode >= 200 && resp.StatusCode < 300 { + c.Corrections.commit(correctionScope, provisionalRules) } requestID := requestIDFromHTTPResponse(resp) var result map[string]any var wire *WireResponse + var err error upstreamResponseID := "" nativeStreamDelta := openAIWireStreamDelta(request.StreamDelta, openAIWireProtocol(endpointKind), resp) if request.Kind == "responses" && protocol == ProtocolOpenAIResponses && stream { diff --git a/apps/api/internal/clients/openai_request_params.go b/apps/api/internal/clients/openai_request_params.go index 062d6e3..612ceb4 100644 --- a/apps/api/internal/clients/openai_request_params.go +++ b/apps/api/internal/clients/openai_request_params.go @@ -32,7 +32,7 @@ var openAIResponsesRequestParameters = stringSet( var gatewayOpenAIRequestExtensions = stringSet( "runMode", "run_mode", "conversationId", "conversation_id", "sessionId", "session_id", "requestId", "request_id", "signal", "userMessage", "user_message", "platformId", - "platform_id", "options", "enable_thinking", "thinking_budget_tokens", "enable_web_search", + "platform_id", "options", "enable_thinking", "thinking_budget", "thinking_budget_tokens", "enable_web_search", "modelType", "model_type", "capability", "capabilityType", "mode", "simulation", "testMode", "cacheAffinityKey", "cache_affinity_key", "simulationDurationMs", "simulationDurationSeconds", "simulationMinDurationMs", "simulationMaxDurationMs", "simulationMinDurationSeconds", diff --git a/apps/api/internal/clients/parameter_correction.go b/apps/api/internal/clients/parameter_correction.go new file mode 100644 index 0000000..0aa2e66 --- /dev/null +++ b/apps/api/internal/clients/parameter_correction.go @@ -0,0 +1,445 @@ +package clients + +import ( + "container/list" + "errors" + "fmt" + "math" + "net/url" + "regexp" + "sort" + "strconv" + "strings" + "sync" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +const parameterCorrectionCacheCapacity = 512 + +var safeUpstreamCorrectionParameters = stringSet( + "reasoning_effort", "reasoning.effort", "enable_thinking", "thinking_budget", + "temperature", "top_p", "frequency_penalty", "presence_penalty", "seed", "stop", "n", + "max_tokens", "max_completion_tokens", "max_output_tokens", + "logprobs", "top_logprobs", "service_tier", "verbosity", +) + +var ( + anchoredParameterPattern = regexp.MustCompile(`(?i)^(?:unknown|unsupported|invalid) parameter\s*:?\s*[` + "`" + `'\"]?([a-z][a-z0-9_.]*)`) + restrictedParameterPattern = regexp.MustCompile(`(?i)^the value of the ([a-z][a-z0-9_.]*) parameter is restricted to\s+([^\s,.;]+)`) + fixedValuePattern = regexp.MustCompile(`(?i)(?:restricted to|must be(?: set)?(?: to)?|only supports?(?: the)? value(?: of)?)\s*[` + "`" + `'\"]?([^\s,.;` + "`" + `'\"]+)`) + betweenPattern = regexp.MustCompile(`(?i)\bbetween\s+(-?[0-9]+(?:\.[0-9]+)?)\s+and\s+(-?[0-9]+(?:\.[0-9]+)?)\b`) + maximumPattern = regexp.MustCompile(`(?i)(?:less than or equal to|at most|maximum(?: value)?(?: is| of)?|<=)\s+(-?[0-9]+(?:\.[0-9]+)?)`) + minimumPattern = regexp.MustCompile(`(?i)(?:greater than or equal to|at least|minimum(?: value)?(?: is| of)?|>=)\s+(-?[0-9]+(?:\.[0-9]+)?)`) +) + +type parameterCorrectionAction string + +const ( + parameterCorrectionRemove parameterCorrectionAction = "remove" + parameterCorrectionSet parameterCorrectionAction = "set" + parameterCorrectionClamp parameterCorrectionAction = "clamp" +) + +type parameterCorrectionRule struct { + Param string + Action parameterCorrectionAction + Value any + Min *float64 + Max *float64 +} + +type parameterCorrectionScope struct { + Provider string + BaseURL string + Protocol string + Kind string + Model string +} + +func newParameterCorrectionScope(request Request, endpointKind string) parameterCorrectionScope { + return parameterCorrectionScope{ + Provider: normalizedReasoningString(request.Candidate.Provider), + BaseURL: normalizedCorrectionBaseURL(request.Candidate.BaseURL), + Protocol: strings.TrimSpace(request.UpstreamProtocol), + Kind: strings.TrimSpace(endpointKind), + Model: normalizedReasoningString(upstreamModelName(request.Candidate)), + } +} + +func normalizedCorrectionBaseURL(raw string) string { + trimmed := strings.TrimRight(strings.TrimSpace(raw), "/") + parsed, err := url.Parse(trimmed) + if err != nil || parsed.Host == "" { + return strings.ToLower(trimmed) + } + parsed.Scheme = strings.ToLower(parsed.Scheme) + parsed.Host = strings.ToLower(parsed.Host) + parsed.RawQuery = "" + parsed.Fragment = "" + return strings.TrimRight(parsed.String(), "/") +} + +func (scope parameterCorrectionScope) prefix() string { + return strings.Join([]string{scope.Provider, scope.BaseURL, scope.Protocol, scope.Kind, scope.Model}, "\x1f") + "\x1f" +} + +func (scope parameterCorrectionScope) key(param string) string { + return scope.prefix() + param +} + +type parameterCorrectionCacheEntry struct { + key string + rule parameterCorrectionRule +} + +type ParameterCorrectionCache struct { + mu sync.Mutex + capacity int + entries map[string]*list.Element + lru *list.List +} + +func NewParameterCorrectionCache() *ParameterCorrectionCache { + return &ParameterCorrectionCache{ + capacity: parameterCorrectionCacheCapacity, + entries: make(map[string]*list.Element), + lru: list.New(), + } +} + +func (cache *ParameterCorrectionCache) apply(scope parameterCorrectionScope, body map[string]any) []string { + if cache == nil { + return nil + } + cache.mu.Lock() + defer cache.mu.Unlock() + prefix := scope.prefix() + elements := make([]*list.Element, 0) + for key, element := range cache.entries { + if strings.HasPrefix(key, prefix) { + elements = append(elements, element) + } + } + sort.Slice(elements, func(i, j int) bool { + return elements[i].Value.(parameterCorrectionCacheEntry).rule.Param < elements[j].Value.(parameterCorrectionCacheEntry).rule.Param + }) + applied := make([]string, 0, len(elements)) + for _, element := range elements { + rule := element.Value.(parameterCorrectionCacheEntry).rule + if applyParameterCorrectionRule(body, rule) { + applied = append(applied, rule.Param) + cache.lru.MoveToFront(element) + } + } + return applied +} + +func (cache *ParameterCorrectionCache) commit(scope parameterCorrectionScope, rules []parameterCorrectionRule) { + if cache == nil || len(rules) == 0 { + return + } + cache.mu.Lock() + defer cache.mu.Unlock() + for _, rule := range rules { + key := scope.key(rule.Param) + if existing := cache.entries[key]; existing != nil { + existing.Value = parameterCorrectionCacheEntry{key: key, rule: rule} + cache.lru.MoveToFront(existing) + continue + } + element := cache.lru.PushFront(parameterCorrectionCacheEntry{key: key, rule: rule}) + cache.entries[key] = element + for cache.lru.Len() > cache.capacity { + oldest := cache.lru.Back() + if oldest == nil { + break + } + delete(cache.entries, oldest.Value.(parameterCorrectionCacheEntry).key) + cache.lru.Remove(oldest) + } + } +} + +func (cache *ParameterCorrectionCache) invalidate(scope parameterCorrectionScope, param string) { + if cache == nil { + return + } + cache.mu.Lock() + defer cache.mu.Unlock() + key := scope.key(param) + if element := cache.entries[key]; element != nil { + delete(cache.entries, key) + cache.lru.Remove(element) + } +} + +func (cache *ParameterCorrectionCache) size() int { + if cache == nil { + return 0 + } + cache.mu.Lock() + defer cache.mu.Unlock() + return cache.lru.Len() +} + +func deriveParameterCorrection(err error, body map[string]any, candidate store.RuntimeModelCandidate) (parameterCorrectionRule, bool) { + var clientErr *ClientError + if !errors.As(err, &clientErr) || (clientErr.StatusCode != 400 && clientErr.StatusCode != 422) { + return parameterCorrectionRule{}, false + } + param, code, message := structuredParameterError(clientErr) + if param == "" { + param = anchoredParameterFromMessage(message) + } + param = normalizeCorrectionParam(param) + if !isSafeCorrectionParam(param) { + return parameterCorrectionRule{}, false + } + lowerMessage := strings.ToLower(message) + lowerCode := strings.ToLower(code) + if conflictRule, ok := deriveConflictCorrection(param, lowerMessage); ok { + return conflictRule, true + } + if strings.Contains(lowerMessage, "unknown parameter") || + strings.Contains(lowerMessage, "unsupported parameter") || + strings.Contains(lowerMessage, "not supported") || + strings.Contains(lowerMessage, "does not support") || + strings.Contains(lowerCode, "unknown_parameter") || + strings.Contains(lowerCode, "unsupported_parameter") { + return parameterCorrectionRule{Param: param, Action: parameterCorrectionRemove}, true + } + if param == "reasoning_effort" || param == "reasoning.effort" { + if effort := nearestErrorSupportedReasoningEffort(message, normalizedReasoningString(valueAtCorrectionPath(body, param)), candidate); effort != "" { + return parameterCorrectionRule{Param: param, Action: parameterCorrectionSet, Value: effort}, true + } + } + if match := fixedValuePattern.FindStringSubmatch(message); len(match) == 2 { + if value, ok := parseCorrectionScalar(match[1]); ok { + return parameterCorrectionRule{Param: param, Action: parameterCorrectionSet, Value: value}, true + } + } + if min, max, ok := explicitNumericBounds(message); ok { + if isOutputLimitCorrectionParam(param) { + min = nil + } + return parameterCorrectionRule{Param: param, Action: parameterCorrectionClamp, Min: min, Max: max}, true + } + return parameterCorrectionRule{}, false +} + +func requestIDFromParameterError(err error) string { + var clientErr *ClientError + if errors.As(err, &clientErr) { + return clientErr.RequestID + } + return "" +} + +func structuredParameterError(clientErr *ClientError) (string, string, string) { + param := strings.TrimSpace(clientErr.Param) + code := strings.TrimSpace(clientErr.Code) + message := strings.TrimSpace(clientErr.Message) + if clientErr.Wire == nil { + return param, code, message + } + errorObject, _ := clientErr.Wire.Body["error"].(map[string]any) + if errorObject == nil { + return param, code, message + } + if value := strings.TrimSpace(stringFromAny(errorObject["param"])); value != "" { + param = value + } + if value := strings.TrimSpace(stringFromAny(errorObject["code"])); value != "" { + code = value + } + if value := strings.TrimSpace(stringFromAny(errorObject["message"])); value != "" { + message = value + } + return param, code, message +} + +func anchoredParameterFromMessage(message string) string { + trimmed := strings.TrimSpace(message) + for _, pattern := range []*regexp.Regexp{anchoredParameterPattern, restrictedParameterPattern} { + if match := pattern.FindStringSubmatch(trimmed); len(match) >= 2 { + return match[1] + } + } + return "" +} + +func normalizeCorrectionParam(param string) string { + return strings.ToLower(strings.Trim(strings.TrimSpace(param), "`'\"[]")) +} + +func isSafeCorrectionParam(param string) bool { + _, ok := safeUpstreamCorrectionParameters[param] + return ok +} + +func deriveConflictCorrection(param string, message string) (parameterCorrectionRule, bool) { + if !strings.Contains(message, "cannot be used together") && + !strings.Contains(message, "mutually exclusive") && + !strings.Contains(message, "not allowed when") { + return parameterCorrectionRule{}, false + } + if strings.Contains(message, "thinking_budget") && strings.Contains(message, "reasoning_effort") { + return parameterCorrectionRule{Param: "reasoning_effort", Action: parameterCorrectionRemove}, true + } + return parameterCorrectionRule{Param: param, Action: parameterCorrectionRemove}, true +} + +func nearestErrorSupportedReasoningEffort(message string, requested string, candidate store.RuntimeModelCandidate) string { + allowed := make(map[string]struct{}) + lowerMessage := strings.ToLower(message) + for _, effort := range chatReasoningEffortOrder { + if regexp.MustCompile(`\b` + regexp.QuoteMeta(effort) + `\b`).MatchString(lowerMessage) { + allowed[effort] = struct{}{} + } + } + if len(allowed) == 0 { + if resolved, state := resolveCandidateReasoningEffort(requested, candidate); state == reasoningCapabilitySupported { + return resolved + } + return "" + } + requestedIndex := reasoningEffortIndex(requested) + best := "" + bestDistance := len(chatReasoningEffortOrder) + 1 + for index, effort := range chatReasoningEffortOrder { + if _, ok := allowed[effort]; !ok { + continue + } + distance := int(math.Abs(float64(index - requestedIndex))) + if distance < bestDistance { + best = effort + bestDistance = distance + } + } + return best +} + +func parseCorrectionScalar(raw string) (any, bool) { + trimmed := strings.Trim(strings.TrimSpace(raw), "`'\"") + switch strings.ToLower(trimmed) { + case "true": + return true, true + case "false": + return false, true + } + if number, err := strconv.ParseFloat(trimmed, 64); err == nil { + if math.Trunc(number) == number { + return int(number), true + } + return number, true + } + if isOpenAIReasoningEffort(strings.ToLower(trimmed)) { + return strings.ToLower(trimmed), true + } + return nil, false +} + +func explicitNumericBounds(message string) (*float64, *float64, bool) { + if match := betweenPattern.FindStringSubmatch(message); len(match) == 3 { + minimum, minErr := strconv.ParseFloat(match[1], 64) + maximum, maxErr := strconv.ParseFloat(match[2], 64) + if minErr == nil && maxErr == nil { + return &minimum, &maximum, true + } + } + var minimum *float64 + var maximum *float64 + if match := minimumPattern.FindStringSubmatch(message); len(match) == 2 { + if value, err := strconv.ParseFloat(match[1], 64); err == nil { + minimum = &value + } + } + if match := maximumPattern.FindStringSubmatch(message); len(match) == 2 { + if value, err := strconv.ParseFloat(match[1], 64); err == nil { + maximum = &value + } + } + return minimum, maximum, minimum != nil || maximum != nil +} + +func isOutputLimitCorrectionParam(param string) bool { + return param == "max_tokens" || param == "max_completion_tokens" || param == "max_output_tokens" +} + +func applyParameterCorrectionRule(body map[string]any, rule parameterCorrectionRule) bool { + current := valueAtCorrectionPath(body, rule.Param) + switch rule.Action { + case parameterCorrectionRemove: + if current == nil { + return false + } + deleteAtCorrectionPath(body, rule.Param) + return true + case parameterCorrectionSet: + if current == nil { + return false + } + if fmt.Sprint(current) == fmt.Sprint(rule.Value) { + return false + } + setAtCorrectionPath(body, rule.Param, rule.Value) + return true + case parameterCorrectionClamp: + number, ok := finiteFloatFromAny(current) + if !ok { + return false + } + corrected := number + if rule.Min != nil && corrected < *rule.Min { + corrected = *rule.Min + } + if rule.Max != nil && corrected > *rule.Max { + corrected = *rule.Max + } + if corrected == number { + return false + } + if math.Trunc(corrected) == corrected { + setAtCorrectionPath(body, rule.Param, int(corrected)) + } else { + setAtCorrectionPath(body, rule.Param, corrected) + } + return true + default: + return false + } +} + +func valueAtCorrectionPath(body map[string]any, param string) any { + if param != "reasoning.effort" { + return body[param] + } + reasoning, _ := body["reasoning"].(map[string]any) + return reasoning["effort"] +} + +func setAtCorrectionPath(body map[string]any, param string, value any) { + if param != "reasoning.effort" { + body[param] = value + return + } + reasoning, _ := body["reasoning"].(map[string]any) + if reasoning == nil { + reasoning = map[string]any{} + body["reasoning"] = reasoning + } + reasoning["effort"] = value +} + +func deleteAtCorrectionPath(body map[string]any, param string) { + if param != "reasoning.effort" { + delete(body, param) + return + } + reasoning, _ := body["reasoning"].(map[string]any) + delete(reasoning, "effort") + if len(reasoning) == 0 { + delete(body, "reasoning") + } +} diff --git a/apps/api/internal/clients/parameter_correction_test.go b/apps/api/internal/clients/parameter_correction_test.go new file mode 100644 index 0000000..39afe15 --- /dev/null +++ b/apps/api/internal/clients/parameter_correction_test.go @@ -0,0 +1,298 @@ +package clients + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func correctionTestRequest(server *httptest.Server, body map[string]any) Request { + return Request{ + Kind: "chat.completions", + Model: "Qwen3.8-Max-Preview", + Body: body, + Candidate: store.RuntimeModelCandidate{ + Provider: "aliyun-bailian-openai", + BaseURL: server.URL, + ProviderModelName: "qwen3.8-max-preview", + Credentials: map[string]any{"apiKey": "test-key"}, + }, + } +} + +func writeCorrectionSuccess(w http.ResponseWriter, model any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "id": "chatcmpl-corrected", "object": "chat.completion", "model": model, + "choices": []any{map[string]any{"message": map[string]any{"role": "assistant", "content": "ok"}}}, + "usage": map[string]any{"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + }) +} + +func writeParameterError(w http.ResponseWriter, param string, message string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{"code": "invalid_parameter", "param": param, "message": message}, + }) +} + +func TestOpenAIParameterCorrectionLearnsOnlyAfterSuccess(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + if body["n"] != float64(1) { + writeParameterError(w, "n", "Parameter n must be set to 1.") + return + } + writeCorrectionSuccess(w, body["model"]) + })) + defer server.Close() + + cache := NewParameterCorrectionCache() + client := OpenAIClient{HTTPClient: server.Client(), Corrections: cache} + request := correctionTestRequest(server, map[string]any{ + "messages": []any{map[string]any{"role": "user", "content": "hello"}}, + "n": 2, + }) + if _, err := client.Run(context.Background(), request); err != nil { + t.Fatalf("first corrected request failed: %v", err) + } + if got := requests.Load(); got != 2 { + t.Fatalf("first logical request should send twice, got %d", got) + } + if cache.size() != 1 { + t.Fatalf("successful correction should be cached, size=%d", cache.size()) + } + if _, err := client.Run(context.Background(), request); err != nil { + t.Fatalf("cached request failed: %v", err) + } + if got := requests.Load(); got != 3 { + t.Fatalf("second logical request should hit cache on first send, got total %d", got) + } +} + +func TestOpenAIParameterCorrectionLearnsDashScopeThinkingNConstraint(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + if body["n"] != float64(1) { + writeParameterError(w, "n", "The n parameter must be 1 when enable_thinking is true") + return + } + writeCorrectionSuccess(w, body["model"]) + })) + defer server.Close() + cache := NewParameterCorrectionCache() + client := OpenAIClient{HTTPClient: server.Client(), Corrections: cache} + _, err := client.Run(context.Background(), correctionTestRequest(server, map[string]any{ + "messages": []any{}, "n": 2, + })) + if err != nil { + t.Fatalf("DashScope n constraint should be corrected: %v", err) + } + if cache.size() != 1 { + t.Fatalf("successful DashScope n correction should be cached, size=%d", cache.size()) + } +} + +func TestOpenAIParameterCorrectionFailureIsNotCached(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeParameterError(w, "n", "Parameter n must be set to 1.") + })) + defer server.Close() + cache := NewParameterCorrectionCache() + client := OpenAIClient{HTTPClient: server.Client(), Corrections: cache} + _, err := client.Run(context.Background(), correctionTestRequest(server, map[string]any{ + "messages": []any{}, "n": 2, + })) + if err == nil { + t.Fatal("expected upstream error") + } + if cache.size() != 0 { + t.Fatalf("failed correction must not be cached, size=%d", cache.size()) + } +} + +func TestOpenAIParameterCorrectionStopsAfterTwoRetries(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + if body["n"] != float64(1) { + writeParameterError(w, "n", "Parameter n must be set to 1.") + return + } + if body["temperature"] != float64(1) { + writeParameterError(w, "temperature", "Parameter temperature must be set to 1.") + return + } + writeParameterError(w, "top_p", "Parameter top_p must be set to 1.") + })) + defer server.Close() + client := OpenAIClient{HTTPClient: server.Client(), Corrections: NewParameterCorrectionCache()} + _, err := client.Run(context.Background(), correctionTestRequest(server, map[string]any{ + "messages": []any{}, "n": 2, "temperature": 2, "top_p": 2, + })) + if err == nil { + t.Fatal("expected third parameter error") + } + if got := requests.Load(); got != 3 { + t.Fatalf("expected initial request plus two correction retries, got %d", got) + } +} + +func TestOpenAIParameterCorrectionEvictsConflictingRule(t *testing.T) { + cache := NewParameterCorrectionCache() + var seen []float64 + var mu sync.Mutex + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + n, _ := body["n"].(float64) + mu.Lock() + seen = append(seen, n) + mu.Unlock() + if n != 2 { + writeParameterError(w, "n", "Parameter n must be set to 2.") + return + } + writeCorrectionSuccess(w, body["model"]) + })) + defer server.Close() + request := correctionTestRequest(server, map[string]any{"messages": []any{}, "n": 9}) + scope := newParameterCorrectionScope(request, "chat.completions") + cache.commit(scope, []parameterCorrectionRule{{Param: "n", Action: parameterCorrectionSet, Value: 1}}) + client := OpenAIClient{HTTPClient: server.Client(), Corrections: cache} + if _, err := client.Run(context.Background(), request); err != nil { + t.Fatalf("conflicting cache correction failed: %v", err) + } + if _, err := client.Run(context.Background(), request); err != nil { + t.Fatalf("relearned cache request failed: %v", err) + } + mu.Lock() + defer mu.Unlock() + if fmt.Sprint(seen) != "[1 2 2]" { + t.Fatalf("expected old rule, relearn retry, then new cache hit; got %v", seen) + } +} + +func TestOpenAIParameterCorrectionNeverTouchesDangerousFields(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + requests.Add(1) + writeParameterError(w, "model", "Unsupported parameter: model") + })) + defer server.Close() + messages := []any{map[string]any{"role": "user", "content": "secret text"}} + request := correctionTestRequest(server, map[string]any{"messages": messages}) + client := OpenAIClient{HTTPClient: server.Client(), Corrections: NewParameterCorrectionCache()} + _, err := client.Run(context.Background(), request) + if err == nil { + t.Fatal("expected unsafe parameter error") + } + if requests.Load() != 1 { + t.Fatalf("unsafe parameter must not retry, got %d requests", requests.Load()) + } + if fmt.Sprint(request.Body["messages"]) != fmt.Sprint(messages) { + t.Fatal("messages were unexpectedly modified") + } +} + +func TestOpenAIParameterCorrectionWorksBeforeStreamingOutput(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + if body["n"] != float64(1) { + writeParameterError(w, "n", "Parameter n must be set to 1.") + return + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = fmt.Fprint(w, "data: {\"id\":\"chatcmpl-stream\",\"choices\":[{\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n") + })) + defer server.Close() + var deltas []string + request := correctionTestRequest(server, map[string]any{"messages": []any{}, "n": 2, "stream": true}) + request.Stream = true + request.StreamDelta = func(event StreamDeltaEvent) error { + deltas = append(deltas, event.Text) + return nil + } + client := OpenAIClient{HTTPClient: server.Client(), Corrections: NewParameterCorrectionCache()} + if _, err := client.Run(context.Background(), request); err != nil { + t.Fatalf("stream correction failed: %v", err) + } + if requests.Load() != 2 || fmt.Sprint(deltas) != "[ok]" { + t.Fatalf("unexpected stream correction requests=%d deltas=%v", requests.Load(), deltas) + } +} + +func TestOpenAIResponsesParameterCorrectionLearnsAfterSuccess(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + if body["top_p"] != float64(1) { + writeParameterError(w, "top_p", "Parameter top_p must be set to 1.") + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "id": "resp-corrected", "object": "response", "status": "completed", + "output": []any{}, "usage": map[string]any{"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + }) + })) + defer server.Close() + cache := NewParameterCorrectionCache() + client := OpenAIClient{HTTPClient: server.Client(), Corrections: cache} + request := Request{ + Kind: "responses", + Model: "Qwen3.8-Max-Preview", + UpstreamProtocol: ProtocolOpenAIResponses, + Body: map[string]any{"input": "hello", "top_p": 2, "reasoning": map[string]any{"effort": "none"}}, + Candidate: store.RuntimeModelCandidate{ + Provider: "aliyun-bailian-openai", BaseURL: server.URL, + ProviderModelName: "qwen3.8-max-preview", Credentials: map[string]any{"apiKey": "test-key"}, + }, + } + if _, err := client.Run(context.Background(), request); err != nil { + t.Fatalf("first corrected Responses request failed: %v", err) + } + if _, err := client.Run(context.Background(), request); err != nil { + t.Fatalf("cached Responses request failed: %v", err) + } + if requests.Load() != 3 || cache.size() != 1 { + t.Fatalf("expected retry then Responses cache hit, requests=%d cache=%d", requests.Load(), cache.size()) + } +} + +func TestParameterCorrectionCacheIsConcurrentAndBounded(t *testing.T) { + cache := NewParameterCorrectionCache() + var group sync.WaitGroup + for index := 0; index < 1024; index++ { + group.Add(1) + go func(index int) { + defer group.Done() + scope := parameterCorrectionScope{Provider: "p", BaseURL: "https://example.com", Protocol: "chat", Kind: "chat", Model: fmt.Sprint(index)} + cache.commit(scope, []parameterCorrectionRule{{Param: "n", Action: parameterCorrectionSet, Value: 1}}) + cache.apply(scope, map[string]any{"n": 2}) + }(index) + } + group.Wait() + if cache.size() > parameterCorrectionCacheCapacity { + t.Fatalf("cache exceeded capacity: %d", cache.size()) + } +} diff --git a/apps/api/internal/clients/qwen38_live_test.go b/apps/api/internal/clients/qwen38_live_test.go new file mode 100644 index 0000000..e8f4ffd --- /dev/null +++ b/apps/api/internal/clients/qwen38_live_test.go @@ -0,0 +1,312 @@ +package clients + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +type qwen38LiveProxy struct { + server *httptest.Server + upstreamBase *url.URL + mu sync.Mutex + shapes []map[string]any +} + +func newQwen38LiveProxy(t *testing.T, upstream string) *qwen38LiveProxy { + t.Helper() + upstreamBase, err := url.Parse(strings.TrimRight(upstream, "/")) + if err != nil { + t.Fatalf("parse Qwen3.8 live upstream: %v", err) + } + proxy := &qwen38LiveProxy{upstreamBase: upstreamBase} + proxy.server = httptest.NewServer(http.HandlerFunc(proxy.forward)) + t.Cleanup(proxy.server.Close) + return proxy +} + +func (p *qwen38LiveProxy) forward(w http.ResponseWriter, request *http.Request) { + body, err := io.ReadAll(request.Body) + if err != nil { + http.Error(w, "read request", http.StatusBadRequest) + return + } + var decoded map[string]any + if json.Unmarshal(body, &decoded) == nil { + p.mu.Lock() + p.shapes = append(p.shapes, qwen38SafeRequestShape(decoded)) + p.mu.Unlock() + } + target := *p.upstreamBase + target.Path = strings.TrimRight(target.Path, "/") + request.URL.Path + upstreamRequest, err := http.NewRequestWithContext(request.Context(), request.Method, target.String(), bytes.NewReader(body)) + if err != nil { + http.Error(w, "create upstream request", http.StatusInternalServerError) + return + } + upstreamRequest.Header.Set("Authorization", request.Header.Get("Authorization")) + upstreamRequest.Header.Set("Content-Type", "application/json") + response, err := http.DefaultClient.Do(upstreamRequest) + if err != nil { + http.Error(w, "upstream request failed", http.StatusBadGateway) + return + } + defer response.Body.Close() + responseBody, err := io.ReadAll(response.Body) + if err != nil { + http.Error(w, "read upstream response", http.StatusBadGateway) + return + } + status := response.StatusCode + if status == http.StatusInternalServerError && strings.Contains(string(responseBody), "The n parameter must be 1 when enable_thinking is true") { + // The deployed server-main currently wraps DashScope's 400 as 500. The + // adapter only restores the original parameter-error status and shape so + // this live test can exercise the local client's correction loop. + status = http.StatusBadRequest + responseBody, _ = json.Marshal(map[string]any{ + "error": map[string]any{ + "code": "invalid_parameter_error", + "param": "n", + "message": "The n parameter must be 1 when enable_thinking is true", + }, + }) + } + if contentType := response.Header.Get("Content-Type"); contentType != "" { + w.Header().Set("Content-Type", contentType) + } + w.WriteHeader(status) + _, _ = w.Write(responseBody) +} + +func (p *qwen38LiveProxy) requestShapes() []map[string]any { + p.mu.Lock() + defer p.mu.Unlock() + out := make([]map[string]any, len(p.shapes)) + copy(out, p.shapes) + return out +} + +func qwen38SafeRequestShape(body map[string]any) map[string]any { + shape := map[string]any{} + for _, key := range []string{"model", "stream", "reasoning_effort", "enable_thinking", "thinking_budget", "temperature", "n"} { + if value, ok := body[key]; ok { + shape[key] = value + } + } + if reasoning, ok := body["reasoning"].(map[string]any); ok { + safeReasoning := map[string]any{} + for _, key := range []string{"effort", "summary"} { + if value, exists := reasoning[key]; exists { + safeReasoning[key] = value + } + } + shape["reasoning"] = safeReasoning + } + return shape +} + +func qwen38LiveRequest(proxy *qwen38LiveProxy, apiKey string, body map[string]any, stream bool) Request { + body = cloneBody(body) + body["stream"] = stream + return Request{ + Kind: "chat.completions", + Model: "Qwen3.8-Max-Preview", + Body: body, + Stream: stream, + UpstreamProtocol: ProtocolOpenAIChatCompletions, + Candidate: store.RuntimeModelCandidate{ + Provider: "aliyun-bailian-openai", + BaseURL: proxy.server.URL, + ProviderModelName: "Qwen3.8-Max-Preview", + Credentials: map[string]any{"apiKey": apiKey}, + Capabilities: map[string]any{ + "text_generate": map[string]any{ + "supportThinking": true, + "supportThinkingModeSwitch": false, + "thinkingEffortLevels": []any{"low", "medium", "xhigh"}, + }, + }, + }, + } +} + +func qwen38LiveBody(extra map[string]any) map[string]any { + body := map[string]any{ + "messages": []any{map[string]any{"role": "user", "content": "Reply only OK"}}, + "max_tokens": 256, + } + for key, value := range extra { + body[key] = value + } + return body +} + +func qwen38ResponseHasReasoning(response Response, streamedReasoning string) bool { + if strings.TrimSpace(streamedReasoning) != "" { + return true + } + choices, _ := response.Result["choices"].([]any) + if len(choices) == 0 { + return false + } + choice, _ := choices[0].(map[string]any) + message, _ := choice["message"].(map[string]any) + reasoning, _ := message["reasoning_content"].(string) + return strings.TrimSpace(reasoning) != "" +} + +func qwen38LiveConfig(t *testing.T) (string, string) { + t.Helper() + baseURL := strings.TrimSpace(os.Getenv("QWEN38_LIVE_BASE_URL")) + apiKey := strings.TrimSpace(os.Getenv("QWEN38_LIVE_API_KEY")) + if baseURL == "" || apiKey == "" { + t.Skip("set QWEN38_LIVE_BASE_URL and QWEN38_LIVE_API_KEY to run real-model acceptance") + } + return baseURL, apiKey +} + +func TestQwen38LiveStaticNormalizationAndReasoning(t *testing.T) { + baseURL, apiKey := qwen38LiveConfig(t) + proxy := newQwen38LiveProxy(t, baseURL) + client := OpenAIClient{Corrections: NewParameterCorrectionCache()} + tests := []struct { + name string + input map[string]any + stream bool + expected map[string]any + absent []string + allowCompatibilityRetry bool + }{ + {name: "none", input: map[string]any{"reasoning_effort": "none"}, expected: map[string]any{"reasoning_effort": "low", "enable_thinking": true}}, + {name: "minimal stream", input: map[string]any{"reasoning_effort": "minimal"}, stream: true, expected: map[string]any{"reasoning_effort": "low", "enable_thinking": true}}, + {name: "medium", input: map[string]any{"reasoning_effort": "medium"}, expected: map[string]any{"reasoning_effort": "medium", "enable_thinking": true}}, + {name: "high stream", input: map[string]any{"reasoning_effort": "high"}, stream: true, expected: map[string]any{"reasoning_effort": "xhigh", "enable_thinking": true}}, + {name: "max", input: map[string]any{"reasoning_effort": "max"}, expected: map[string]any{"reasoning_effort": "xhigh", "enable_thinking": true}}, + {name: "forced thinking stream", input: map[string]any{"enable_thinking": false}, stream: true, expected: map[string]any{"reasoning_effort": "low", "enable_thinking": true}}, + {name: "budget wins", input: map[string]any{"reasoning_effort": "high", "thinking_budget": 128}, expected: map[string]any{"thinking_budget": float64(128), "enable_thinking": true}, absent: []string{"reasoning_effort"}, allowCompatibilityRetry: true}, + {name: "temperature floor stream", input: map[string]any{"temperature": 0.1}, stream: true, expected: map[string]any{"temperature": 0.6, "enable_thinking": true}}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + before := len(proxy.requestShapes()) + streamedReasoning := "" + request := qwen38LiveRequest(proxy, apiKey, qwen38LiveBody(test.input), test.stream) + if test.stream { + request.StreamDelta = func(event StreamDeltaEvent) error { + streamedReasoning += event.ReasoningContent + return nil + } + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + response, err := client.Run(ctx, request) + if err != nil { + shapes := proxy.requestShapes() + t.Fatalf("real Qwen3.8 request failed: %v; safe request shapes=%+v", err, shapes[before:]) + } + if !qwen38ResponseHasReasoning(response, streamedReasoning) { + t.Fatal("real Qwen3.8 response did not contain reasoning_content") + } + shapes := proxy.requestShapes() + submissions := len(shapes) - before + if submissions != 1 && !(test.allowCompatibilityRetry && submissions == 2) { + t.Fatalf("unexpected submission count, before=%d after=%d safe request shapes=%+v", before, len(shapes), shapes[before:]) + } + shape := shapes[before] + for key, expected := range test.expected { + if normalizedJSONScalar(shape[key]) != normalizedJSONScalar(expected) { + t.Fatalf("%s expected %#v, got %#v in shape %+v", key, expected, shape[key], shape) + } + } + for _, key := range test.absent { + if _, ok := shape[key]; ok { + t.Fatalf("%s must be absent in shape %+v", key, shape) + } + } + t.Logf("case=%s response_id=%s correction=%t cache_hit=false", test.name, response.RequestID, submissions > 1) + }) + } +} + +func TestQwen38LiveCorrectionCacheAndRestart(t *testing.T) { + baseURL, apiKey := qwen38LiveConfig(t) + proxy := newQwen38LiveProxy(t, baseURL) + request := qwen38LiveRequest(proxy, apiKey, qwen38LiveBody(map[string]any{"reasoning_effort": "none", "n": 2}), false) + cache := NewParameterCorrectionCache() + client := OpenAIClient{Corrections: cache} + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + first, err := client.Run(ctx, request) + if err != nil || !qwen38ResponseHasReasoning(first, "") { + t.Fatalf("cold corrected request failed or lacked reasoning: %v", err) + } + second, err := client.Run(ctx, request) + if err != nil || !qwen38ResponseHasReasoning(second, "") { + t.Fatalf("cached corrected request failed or lacked reasoning: %v", err) + } + shapes := proxy.requestShapes() + if len(shapes) != 3 || normalizedJSONScalar(shapes[0]["n"]) != float64(2) || normalizedJSONScalar(shapes[1]["n"]) != float64(1) || normalizedJSONScalar(shapes[2]["n"]) != float64(1) { + t.Fatalf("expected cold n=2 -> corrected n=1 -> cached n=1, got %+v", shapes) + } + + restarted := OpenAIClient{Corrections: NewParameterCorrectionCache()} + third, err := restarted.Run(ctx, request) + if err != nil || !qwen38ResponseHasReasoning(third, "") { + t.Fatalf("post-restart corrected request failed or lacked reasoning: %v", err) + } + shapes = proxy.requestShapes() + if len(shapes) != 5 || normalizedJSONScalar(shapes[3]["n"]) != float64(2) || normalizedJSONScalar(shapes[4]["n"]) != float64(1) { + t.Fatalf("new process cache should relearn n rule, got %+v", shapes) + } + t.Logf("response_ids=%s,%s,%s correction=n:set:1 cache_hit_sequence=false,true,false", first.RequestID, second.RequestID, third.RequestID) +} + +func TestQwen38LiveNativeResponsesNormalization(t *testing.T) { + baseURL, apiKey := qwen38LiveConfig(t) + proxy := newQwen38LiveProxy(t, baseURL) + client := OpenAIClient{Corrections: NewParameterCorrectionCache()} + request := Request{ + Kind: "responses", + Model: "Qwen3.8-Max-Preview", + UpstreamProtocol: ProtocolOpenAIResponses, + Body: map[string]any{ + "input": "Reply only OK", "max_output_tokens": 256, + "reasoning": map[string]any{"effort": "none"}, + }, + Candidate: store.RuntimeModelCandidate{ + Provider: "aliyun-bailian-openai", BaseURL: proxy.server.URL, + ProviderModelName: "Qwen3.8-Max-Preview", Credentials: map[string]any{"apiKey": apiKey}, + Capabilities: map[string]any{"text_generate": map[string]any{ + "supportThinking": true, "supportThinkingModeSwitch": false, + "thinkingEffortLevels": []any{"low", "medium", "xhigh"}, + }}, + }, + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + response, err := client.Run(ctx, request) + if err != nil { + t.Fatalf("real Qwen3.8 Responses request failed: %v; safe shapes=%+v", err, proxy.requestShapes()) + } + shapes := proxy.requestShapes() + if len(shapes) != 1 { + t.Fatalf("expected one Responses submission, got %+v", shapes) + } + reasoning, _ := shapes[0]["reasoning"].(map[string]any) + if reasoning["effort"] != "low" { + t.Fatalf("Responses none was not normalized to low: %+v", shapes) + } + t.Logf("case=responses-none response_id=%s correction=false cache_hit=false", response.RequestID) +} diff --git a/apps/api/internal/clients/testdata/qwen38-reasoning-vectors.json b/apps/api/internal/clients/testdata/qwen38-reasoning-vectors.json new file mode 100644 index 0000000..e6d061b --- /dev/null +++ b/apps/api/internal/clients/testdata/qwen38-reasoning-vectors.json @@ -0,0 +1,10 @@ +[ + { "name": "none-to-low", "input": { "reasoning_effort": "none" }, "expected": { "enable_thinking": true, "reasoning_effort": "low" } }, + { "name": "minimal-to-low", "input": { "reasoning_effort": "minimal" }, "expected": { "enable_thinking": true, "reasoning_effort": "low" } }, + { "name": "medium", "input": { "reasoning_effort": "medium" }, "expected": { "enable_thinking": true, "reasoning_effort": "medium" } }, + { "name": "high-to-xhigh", "input": { "reasoning_effort": "high" }, "expected": { "enable_thinking": true, "reasoning_effort": "xhigh" } }, + { "name": "max-to-xhigh", "input": { "reasoning_effort": "max" }, "expected": { "enable_thinking": true, "reasoning_effort": "xhigh" } }, + { "name": "force-thinking", "input": { "enable_thinking": false }, "expected": { "enable_thinking": true, "reasoning_effort": "low" } }, + { "name": "budget-wins", "input": { "reasoning_effort": "high", "thinking_budget": 300000 }, "expected": { "enable_thinking": true, "thinking_budget": 262144 } }, + { "name": "temperature-floor", "input": { "temperature": 0.1 }, "expected": { "enable_thinking": true, "temperature": 0.6 } } +] diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index d7ce8a3..7e2a98a 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -154,6 +154,7 @@ func NewWithStores( } httpClients := newHTTPClientCache() scriptExecutor := &scriptengine.Executor{Logger: logger} + openAIParameterCorrections := clients.NewParameterCorrectionCache() service := &Service{ cfg: cfg, store: db, @@ -162,7 +163,7 @@ func NewWithStores( logger: logger, scriptExecutor: scriptExecutor, clients: map[string]clients.Client{ - "openai": clients.OpenAIClient{HTTPClient: httpClients.none}, + "openai": clients.OpenAIClient{HTTPClient: httpClients.none, Corrections: openAIParameterCorrections}, "aliyun-bailian": clients.AliyunBailianClient{HTTPClient: httpClients.none}, "blackforest": clients.BlackforestClient{HTTPClient: httpClients.none}, "gemini": clients.GeminiClient{HTTPClient: httpClients.none}, @@ -2562,6 +2563,11 @@ func validateRequest(kind string, body map[string]any) error { if body["input"] == nil && body["messages"] == nil { return errors.New("input or messages is required") } + if reasoning, ok := body["reasoning"].(map[string]any); ok { + if err := clients.ValidateOpenAIReasoningEffort(reasoning["effort"]); err != nil { + return err + } + } if value, explicit := nonNullParameter(body, "max_output_tokens"); explicit { if _, ok := positiveInteger(value); !ok { return &clients.ClientError{Code: "invalid_parameter", Message: "max_output_tokens must be a positive integer", Param: "max_output_tokens", StatusCode: 400, Retryable: false}