From cef18b4ad3882429923a3d976bbb1f95759f6ae9 Mon Sep 17 00:00:00 2001 From: xuhaihan Date: Sat, 22 Aug 2026 01:04:11 +0800 Subject: [PATCH] fix(deepseek): route client tools through native responses --- .../service/openai_gateway_forward.go | 22 +++- .../service/openai_gateway_passthrough.go | 44 +++++--- .../openai_gateway_response_handling.go | 8 ++ ...nai_gateway_responses_client_tools_test.go | 104 +++++++++++++----- 4 files changed, 129 insertions(+), 49 deletions(-) diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 99d89a7fac..957d38a8c7 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -115,12 +115,22 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco } } + nativeDeepSeekResponses := account.Platform == PlatformDeepseek && + (account.GetAPIProtocol() == APIProtocolResponses || account.IsAdaptiveAPIProtocol()) + if nativeDeepSeekResponses && account.Type == AccountTypeAPIKey && !compactPath && + needsOpenAIResponsesClientToolAdaptation(body) { + adaptedBody, mapping, adaptErr := adaptOpenAIResponsesClientTools(body) + if adaptErr != nil { + return nil, fmt.Errorf("adapt DeepSeek Responses client tools: %w", adaptErr) + } + body = adaptedBody + setOpenAIResponsesClientToolMapping(c, mapping) + } + originalBody := body requestView := newOpenAIRequestView(body) reqModel, reqStream, promptCacheKey := requestView.Model, requestView.Stream, requestView.PromptCacheKey originalModel := reqModel - nativeDeepSeekResponses := account.Platform == PlatformDeepseek && - (account.GetAPIProtocol() == APIProtocolResponses || account.IsAdaptiveAPIProtocol()) if account.Platform == PlatformGrok { return s.forwardGrokResponses(ctx, c, account, body, originalModel, reqStream, startTime) @@ -1054,6 +1064,14 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco } defer func() { _ = resp.Body.Close() }() + if mapping, ok := openAIResponsesClientToolMapping(c); ok && isEventStreamResponse(resp.Header) { + maxLineSize := defaultMaxLineSize + if s.cfg != nil && s.cfg.Gateway.MaxLineSize > 0 { + maxLineSize = s.cfg.Gateway.MaxLineSize + } + resp.Body = newResponsesClientToolStreamBody(resp.Body, mapping, maxLineSize) + } + serviceTier := extractOpenAIServiceTierFromBody(body) // 上游接受后只保留计费需要的标量，避免响应处理期间继续保活完整 input/tools map。 reqBody = nil diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 3e82586e43..23c283529d 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -83,17 +83,6 @@ func needsOpenAIResponsesClientToolAdaptation(body []byte) bool { return needsAdaptation } -func shouldAdaptOpenAIResponsesClientTools(account *Account, c *gin.Context, body []byte) bool { - if account == nil || account.Type != AccountTypeAPIKey || isOpenAIResponsesCompactPath(c) || - !needsOpenAIResponsesClientToolAdaptation(body) { - return false - } - if account.Platform == PlatformOpenAI { - return true - } - return account.Platform == PlatformDeepseek && account.GetAPIProtocol() == APIProtocolResponses -} - func openAIResponsesClientToolMapping(c *gin.Context) (apicompat.ResponsesClientToolMapping, bool) { if c == nil { return apicompat.ResponsesClientToolMapping{}, false @@ -103,6 +92,17 @@ func openAIResponsesClientToolMapping(c *gin.Context) (apicompat.ResponsesClient return mapping, ok && typed && hasOpenAIResponsesClientToolMapping(mapping) } +func setOpenAIResponsesClientToolMapping(c *gin.Context, mapping apicompat.ResponsesClientToolMapping) { + if c == nil { + return + } + if !hasOpenAIResponsesClientToolMapping(mapping) { + clearOpenAIResponsesClientToolMapping(c) + return + } + c.Set(openAIResponsesClientToolMappingContextKey, mapping) +} + // clearOpenAIResponsesClientToolMapping removes mapping state from the prior // forwarding attempt. Forward retries accounts on the same Gin context. func clearOpenAIResponsesClientToolMapping(c *gin.Context) { @@ -114,6 +114,15 @@ func clearOpenAIResponsesClientToolMapping(c *gin.Context) { } } +func restoreOpenAIResponsesClientToolPayload(c *gin.Context, payload []byte) ([]byte, error) { + mapping, ok := openAIResponsesClientToolMapping(c) + if !ok || !bytes.Contains(payload, []byte(`"function_call"`)) || !json.Valid(payload) { + return payload, nil + } + restored, _, err := apicompat.RestoreResponsesClientToolPayload(payload, mapping) + return restored, err +} + func (s *OpenAIGatewayService) forwardOpenAIPassthrough( ctx context.Context, c *gin.Context, @@ -214,13 +223,14 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( } } - if shouldAdaptOpenAIResponsesClientTools(account, c, body) { + if account != nil && account.Platform == PlatformOpenAI && account.Type == AccountTypeAPIKey && + !isOpenAIResponsesCompactPath(c) && needsOpenAIResponsesClientToolAdaptation(body) { adaptedBody, mapping, adaptErr := adaptOpenAIResponsesClientTools(body) if adaptErr != nil { return nil, adaptErr } body = adaptedBody - c.Set(openAIResponsesClientToolMappingContextKey, mapping) + setOpenAIResponsesClientToolMapping(c, mapping) } sanitizedBody, sanitized, err := sanitizeEmptyBase64InputImagesInOpenAIBody(body) @@ -2109,11 +2119,9 @@ func (s *OpenAIGatewayService) handleNonStreamingResponsePassthrough( return nil, fmt.Errorf("restore OpenAI passthrough namespace response: %w", err) } body = restoreCodexToolNamesFromContext(c, body) - if mapping, ok := openAIResponsesClientToolMapping(c); ok && json.Valid(body) { - body, _, err = apicompat.RestoreResponsesClientToolPayload(body, mapping) - if err != nil { - return nil, fmt.Errorf("restore OpenAI Responses client tools: %w", err) - } + body, err = restoreOpenAIResponsesClientToolPayload(c, body) + if err != nil { + return nil, fmt.Errorf("restore OpenAI Responses client tools: %w", err) } if !writeOpenAICompactSSEBridge(c, resp.StatusCode, body) { c.Data(resp.StatusCode, contentType, body) diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 2675ebb995..0b26a580cf 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -1614,6 +1614,10 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r if err != nil { return nil, fmt.Errorf("restore Grok Responses client tool response: %w", err) } + body, err = restoreOpenAIResponsesClientToolPayload(c, body) + if err != nil { + return nil, fmt.Errorf("restore OpenAI Responses client tool response: %w", err) + } body, err = restoreOpenAIResponsesNamespacePayload(c, body) if err != nil { return nil, fmt.Errorf("restore OpenAI namespace response: %w", err) @@ -1707,6 +1711,10 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte if restoreErr != nil { return nil, fmt.Errorf("restore Grok Responses client tool response: %w", restoreErr) } + restoredBody, restoreErr = restoreOpenAIResponsesClientToolPayload(c, restoredBody) + if restoreErr != nil { + return nil, fmt.Errorf("restore OpenAI Responses client tool response: %w", restoreErr) + } restoredBody, restoreErr = restoreOpenAIResponsesNamespacePayload(c, restoredBody) if restoreErr != nil { return nil, fmt.Errorf("restore OpenAI namespace response: %w", restoreErr) diff --git a/backend/internal/service/openai_gateway_responses_client_tools_test.go b/backend/internal/service/openai_gateway_responses_client_tools_test.go index 1cb2fe57c2..fa4866decf 100644 --- a/backend/internal/service/openai_gateway_responses_client_tools_test.go +++ b/backend/internal/service/openai_gateway_responses_client_tools_test.go @@ -108,32 +108,7 @@ func TestClearOpenAIResponsesClientToolMappingRemovesStaleContextState(t *testin require.False(t, ok) } -func TestShouldAdaptOpenAIResponsesClientToolsForDeepSeekResponses(t *testing.T) { - gin.SetMode(gin.TestMode) - c, _ := gin.CreateTestContext(httptest.NewRecorder()) - c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) - body := openAIClientToolsRequest(true) - - deepSeekResponses := &Account{ - Platform: PlatformDeepseek, - Type: AccountTypeAPIKey, - Credentials: map[string]any{ - "api_protocol": APIProtocolResponses, - }, - } - require.True(t, shouldAdaptOpenAIResponsesClientTools(deepSeekResponses, c, body)) - - deepSeekChat := &Account{ - Platform: PlatformDeepseek, - Type: AccountTypeAPIKey, - Credentials: map[string]any{ - "api_protocol": APIProtocolChatCompletions, - }, - } - require.False(t, shouldAdaptOpenAIResponsesClientTools(deepSeekChat, c, body)) -} - -func TestDeepSeekResponsesPassthroughRestoresClientToolsStreaming(t *testing.T) { +func TestDeepSeekResponsesForwardRestoresClientToolsStreaming(t *testing.T) { gin.SetMode(gin.TestMode) body := openAIClientToolsRequest(true) recorder := httptest.NewRecorder() @@ -163,19 +138,90 @@ func TestDeepSeekResponsesPassthroughRestoresClientToolsStreaming(t *testing.T) }, } - result, err := svc.forwardOpenAIPassthrough( - context.Background(), c, account, body, body, "gpt-5.4", false, nil, true, time.Now(), - ) + result, err := svc.Forward(context.Background(), c, account, body) require.NoError(t, err) require.NotNil(t, result) assertOpenAIClientToolsLowered(t, upstream.lastBody) + require.Equal(t, "/responses", upstream.lastReq.URL.Path) output := recorder.Body.String() require.Contains(t, output, `"type":"custom_tool_call"`) require.Contains(t, output, `"type":"response.custom_tool_call_input.done"`) require.Contains(t, output, `"input":"pwd"`) } +func TestDeepSeekAdaptiveResponsesForwardRestoresClientToolsNonStreaming(t *testing.T) { + gin.SetMode(gin.TestMode) + body := openAIClientToolsRequest(false) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"id":"resp_ds_adaptive_tools","status":"completed","output":[ + {"type":"function_call","id":"i1","call_id":"c1","name":"exec","arguments":"{\"input\":\"pwd\"}"}, + {"type":"function_call","id":"i2","call_id":"c2","name":"apply_patch","arguments":"{\"input\":\"*** Begin Patch\"}"}], + "usage":{"input_tokens":1,"output_tokens":1}}`)), + }} + svc := openAIClientToolsTestService(upstream) + account := &Account{ + ID: 5662, + Platform: PlatformDeepseek, + Type: AccountTypeAPIKey, + Credentials: map[string]any{ + "api_key": "test-key", + "api_protocol": APIProtocolAdaptive, + "api_base_urls": map[string]any{ + APIProtocolResponses: "https://relay.example", + }, + }, + } + + result, err := svc.Forward(context.Background(), c, account, body) + + require.NoError(t, err) + require.NotNil(t, result) + assertOpenAIClientToolsLowered(t, upstream.lastBody) + require.Equal(t, "/responses", upstream.lastReq.URL.Path) + require.Equal(t, "custom_tool_call", gjson.Get(recorder.Body.String(), "output.0.type").String()) + require.Equal(t, "pwd", gjson.Get(recorder.Body.String(), "output.0.input").String()) + require.Equal(t, "custom_tool_call", gjson.Get(recorder.Body.String(), "output.1.type").String()) + require.Equal(t, "*** Begin Patch", gjson.Get(recorder.Body.String(), "output.1.input").String()) +} + +func TestDeepSeekResponsesCompactSkipsClientToolAdaptation(t *testing.T) { + gin.SetMode(gin.TestMode) + body := openAIClientToolsRequest(false) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses/compact", bytes.NewReader(body)) + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"id":"resp_compact","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1}}`)), + }} + svc := openAIClientToolsTestService(upstream) + account := &Account{ + ID: 5663, + Platform: PlatformDeepseek, + Type: AccountTypeAPIKey, + Credentials: map[string]any{ + "api_key": "test-key", + "api_protocol": APIProtocolResponses, + "base_url": "https://relay.example", + }, + } + + _, err := svc.Forward(context.Background(), c, account, body) + + require.NoError(t, err) + require.Equal(t, "custom", gjson.GetBytes(upstream.lastBody, "tools.0.type").String()) + require.Equal(t, "/responses/compact", upstream.lastReq.URL.Path) +} + func TestOpenAIPassthroughAPIKeyRestoresClientToolsNonStreaming(t *testing.T) { gin.SetMode(gin.TestMode) body := openAIClientToolsRequest(false)