- 提取 TruncateBody 公共函数,截断到 2KB 避免日志过大 - 新增 handleResponsesStreamError 统一 SSE 错误提取逻辑 - 流式/非流式 Responses API 错误路径均捕获 UpstreamBody - 添加 upstream_body 和 truncate_body 单元测试 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>master
| @@ -56,7 +56,9 @@ func OaiResponsesToChatHandler(c *gin.Context, info *relaycommon.RelayInfo, resp | |||||
| } | } | ||||
| if oaiError := responsesResp.GetOpenAIError(); oaiError != nil && oaiError.Type != "" { | if oaiError := responsesResp.GetOpenAIError(); oaiError != nil && oaiError.Type != "" { | ||||
| return nil, types.WithOpenAIError(*oaiError, resp.StatusCode) | |||||
| apiErr := types.WithOpenAIError(*oaiError, resp.StatusCode) | |||||
| apiErr.UpstreamBody = service.TruncateBody(string(body)) | |||||
| return nil, apiErr | |||||
| } | } | ||||
| chatId := helper.GetResponseID(c) | chatId := helper.GetResponseID(c) | ||||
| @@ -484,14 +486,8 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo | |||||
| sentStop = true | sentStop = true | ||||
| } | } | ||||
| case "response.error", "response.failed": | |||||
| if streamResp.Response != nil { | |||||
| if oaiErr := streamResp.Response.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" { | |||||
| streamErr = types.WithOpenAIError(*oaiErr, http.StatusInternalServerError) | |||||
| return false | |||||
| } | |||||
| } | |||||
| streamErr = types.NewOpenAIError(fmt.Errorf("responses stream error: %s", streamResp.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError) | |||||
| case "response.error", "response.failed", "error": | |||||
| streamErr = handleResponsesStreamError(streamResp, data) | |||||
| return false | return false | ||||
| default: | default: | ||||
| @@ -31,7 +31,9 @@ func OaiResponsesHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http | |||||
| return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) | return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) | ||||
| } | } | ||||
| if oaiError := responsesResponse.GetOpenAIError(); oaiError != nil && oaiError.Type != "" { | if oaiError := responsesResponse.GetOpenAIError(); oaiError != nil && oaiError.Type != "" { | ||||
| return nil, types.WithOpenAIError(*oaiError, resp.StatusCode) | |||||
| apiErr := types.WithOpenAIError(*oaiError, resp.StatusCode) | |||||
| apiErr.UpstreamBody = service.TruncateBody(string(responseBody)) | |||||
| return nil, apiErr | |||||
| } | } | ||||
| if responsesResponse.HasImageGenerationCall() { | if responsesResponse.HasImageGenerationCall() { | ||||
| @@ -78,10 +80,10 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp | |||||
| var usage = &dto.Usage{} | var usage = &dto.Usage{} | ||||
| var responseTextBuilder strings.Builder | var responseTextBuilder strings.Builder | ||||
| var streamErr *types.NewAPIError | |||||
| helper.StreamScannerHandler(c, resp, info, func(data string) bool { | helper.StreamScannerHandler(c, resp, info, func(data string) bool { | ||||
| // 检查当前数据是否包含 completed 状态和 usage 信息 | |||||
| var streamResponse dto.ResponsesStreamResponse | var streamResponse dto.ResponsesStreamResponse | ||||
| if err := common.UnmarshalJsonStr(data, &streamResponse); err == nil { | if err := common.UnmarshalJsonStr(data, &streamResponse); err == nil { | ||||
| sendResponsesStreamData(c, streamResponse, data) | sendResponsesStreamData(c, streamResponse, data) | ||||
| @@ -109,10 +111,8 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp | |||||
| } | } | ||||
| } | } | ||||
| case "response.output_text.delta": | case "response.output_text.delta": | ||||
| // 处理输出文本 | |||||
| responseTextBuilder.WriteString(streamResponse.Delta) | responseTextBuilder.WriteString(streamResponse.Delta) | ||||
| case dto.ResponsesOutputTypeItemDone: | case dto.ResponsesOutputTypeItemDone: | ||||
| // 函数调用处理 | |||||
| if streamResponse.Item != nil { | if streamResponse.Item != nil { | ||||
| switch streamResponse.Item.Type { | switch streamResponse.Item.Type { | ||||
| case dto.BuildInCallWebSearchCall: | case dto.BuildInCallWebSearchCall: | ||||
| @@ -123,6 +123,9 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp | |||||
| } | } | ||||
| } | } | ||||
| } | } | ||||
| case "response.error", "response.failed", "error": | |||||
| streamErr = handleResponsesStreamError(streamResponse, data) | |||||
| return false | |||||
| } | } | ||||
| } else { | } else { | ||||
| logger.LogError(c, "failed to unmarshal stream response: "+err.Error()) | logger.LogError(c, "failed to unmarshal stream response: "+err.Error()) | ||||
| @@ -130,6 +133,10 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp | |||||
| return true | return true | ||||
| }) | }) | ||||
| if streamErr != nil { | |||||
| return nil, streamErr | |||||
| } | |||||
| if usage.CompletionTokens == 0 { | if usage.CompletionTokens == 0 { | ||||
| // 计算输出文本的 token 数量 | // 计算输出文本的 token 数量 | ||||
| tempStr := responseTextBuilder.String() | tempStr := responseTextBuilder.String() | ||||
| @@ -0,0 +1,29 @@ | |||||
| package openai | |||||
| import ( | |||||
| "fmt" | |||||
| "net/http" | |||||
| "github.com/QuantumNous/new-api/dto" | |||||
| "github.com/QuantumNous/new-api/service" | |||||
| "github.com/QuantumNous/new-api/types" | |||||
| ) | |||||
| // handleResponsesStreamError extracts error from a Responses API SSE event and attaches the raw data as UpstreamBody. | |||||
| func handleResponsesStreamError(streamResp dto.ResponsesStreamResponse, data string) *types.NewAPIError { | |||||
| if streamResp.Response != nil { | |||||
| if oaiErr := streamResp.Response.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" { | |||||
| err := types.WithOpenAIError(*oaiErr, http.StatusInternalServerError) | |||||
| err.UpstreamBody = service.TruncateBody(data) | |||||
| return err | |||||
| } | |||||
| } | |||||
| if oaiErr := dto.GetOpenAIError(streamResp.Error); oaiErr != nil && oaiErr.Type != "" { | |||||
| err := types.WithOpenAIError(*oaiErr, http.StatusInternalServerError) | |||||
| err.UpstreamBody = service.TruncateBody(data) | |||||
| return err | |||||
| } | |||||
| err := types.NewOpenAIError(fmt.Errorf("responses stream error: %s", streamResp.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError) | |||||
| err.UpstreamBody = service.TruncateBody(data) | |||||
| return err | |||||
| } | |||||
| @@ -0,0 +1,64 @@ | |||||
| package openai | |||||
| import ( | |||||
| "testing" | |||||
| "github.com/QuantumNous/new-api/common" | |||||
| "github.com/QuantumNous/new-api/dto" | |||||
| "github.com/stretchr/testify/assert" | |||||
| "github.com/stretchr/testify/require" | |||||
| ) | |||||
| func TestResponsesStreamEventErrorParsing(t *testing.T) { | |||||
| t.Run("standalone error event", func(t *testing.T) { | |||||
| data := `{"type":"error","error":{"type":"too_many_requests","code":"too_many_requests","message":"Too Many Requests","param":null}}` | |||||
| var streamResp dto.ResponsesStreamResponse | |||||
| err := common.UnmarshalJsonStr(data, &streamResp) | |||||
| require.NoError(t, err) | |||||
| assert.Equal(t, "error", streamResp.Type) | |||||
| assert.NotNil(t, streamResp.Error) | |||||
| oaiErr := dto.GetOpenAIError(streamResp.Error) | |||||
| require.NotNil(t, oaiErr) | |||||
| assert.Equal(t, "too_many_requests", oaiErr.Type) | |||||
| assert.Equal(t, "Too Many Requests", oaiErr.Message) | |||||
| }) | |||||
| t.Run("response.failed event", func(t *testing.T) { | |||||
| data := `{"type":"response.failed","response":{"id":"test-id","status":"failed","error":{"type":"server_error","message":"Internal server error"}}}` | |||||
| var streamResp dto.ResponsesStreamResponse | |||||
| err := common.UnmarshalJsonStr(data, &streamResp) | |||||
| require.NoError(t, err) | |||||
| assert.Equal(t, "response.failed", streamResp.Type) | |||||
| require.NotNil(t, streamResp.Response) | |||||
| oaiErr := streamResp.Response.GetOpenAIError() | |||||
| require.NotNil(t, oaiErr) | |||||
| assert.Equal(t, "server_error", oaiErr.Type) | |||||
| }) | |||||
| t.Run("response.error event with error in response object", func(t *testing.T) { | |||||
| data := `{"type":"response.error","response":{"id":"test-id","status":"failed","error":{"type":"invalid_request_error","message":"Invalid model"}}}` | |||||
| var streamResp dto.ResponsesStreamResponse | |||||
| err := common.UnmarshalJsonStr(data, &streamResp) | |||||
| require.NoError(t, err) | |||||
| assert.Equal(t, "response.error", streamResp.Type) | |||||
| oaiErr := streamResp.Response.GetOpenAIError() | |||||
| require.NotNil(t, oaiErr) | |||||
| assert.Equal(t, "invalid_request_error", oaiErr.Type) | |||||
| }) | |||||
| t.Run("normal event has no error", func(t *testing.T) { | |||||
| data := `{"type":"response.output_text.delta","delta":"hello"}` | |||||
| var streamResp dto.ResponsesStreamResponse | |||||
| err := common.UnmarshalJsonStr(data, &streamResp) | |||||
| require.NoError(t, err) | |||||
| assert.Equal(t, "response.output_text.delta", streamResp.Type) | |||||
| assert.Nil(t, streamResp.Error) | |||||
| }) | |||||
| } | |||||
| @@ -58,6 +58,14 @@ func MidjourneyErrorWithStatusCodeWrapper(code int, desc string, statusCode int) | |||||
| // return openaiErr | // return openaiErr | ||||
| //} | //} | ||||
| func TruncateBody(body string) string { | |||||
| const maxBodyLen = 2048 | |||||
| if len(body) > maxBodyLen { | |||||
| return body[:maxBodyLen] + "...(truncated)" | |||||
| } | |||||
| return body | |||||
| } | |||||
| func ClaudeErrorWrapper(err error, code string, statusCode int) *dto.ClaudeErrorWithStatusCode { | func ClaudeErrorWrapper(err error, code string, statusCode int) *dto.ClaudeErrorWithStatusCode { | ||||
| text := err.Error() | text := err.Error() | ||||
| lowerText := strings.ToLower(text) | lowerText := strings.ToLower(text) | ||||
| @@ -98,12 +106,7 @@ func RelayErrorHandler(ctx context.Context, resp *http.Response, showBodyWhenFai | |||||
| return | return | ||||
| } | } | ||||
| CloseResponseBodyGracefully(resp) | CloseResponseBodyGracefully(resp) | ||||
| // Capture upstream response body (truncate to 2KB to avoid oversized logs) | |||||
| const maxBodyLen = 2048 | |||||
| bodyStr := string(responseBody) | |||||
| if len(bodyStr) > maxBodyLen { | |||||
| bodyStr = bodyStr[:maxBodyLen] + "...(truncated)" | |||||
| } | |||||
| bodyStr := TruncateBody(string(responseBody)) | |||||
| newApiErr.UpstreamBody = bodyStr | newApiErr.UpstreamBody = bodyStr | ||||
| var errResponse dto.GeneralErrorResponse | var errResponse dto.GeneralErrorResponse | ||||
| buildErrWithBody := func(message string) error { | buildErrWithBody := func(message string) error { | ||||
| @@ -0,0 +1,32 @@ | |||||
| package service | |||||
| import ( | |||||
| "strings" | |||||
| "testing" | |||||
| "github.com/stretchr/testify/assert" | |||||
| ) | |||||
| func TestTruncateBody(t *testing.T) { | |||||
| t.Parallel() | |||||
| t.Run("short body unchanged", func(t *testing.T) { | |||||
| body := `{"error":{"type":"too_many_requests","message":"Too Many Requests"}}` | |||||
| assert.Equal(t, body, TruncateBody(body)) | |||||
| }) | |||||
| t.Run("empty body unchanged", func(t *testing.T) { | |||||
| assert.Equal(t, "", TruncateBody("")) | |||||
| }) | |||||
| t.Run("exact max length unchanged", func(t *testing.T) { | |||||
| body := strings.Repeat("a", 2048) | |||||
| assert.Equal(t, body, TruncateBody(body)) | |||||
| }) | |||||
| t.Run("over max length truncated", func(t *testing.T) { | |||||
| body := strings.Repeat("a", 3000) | |||||
| result := TruncateBody(body) | |||||
| assert.Equal(t, strings.Repeat("a", 2048)+"...(truncated)", result) | |||||
| }) | |||||
| } | |||||