From 4ee6bee45fe3be4a98c0338d9e074517a10b618b Mon Sep 17 00:00:00 2001 From: Sheng Fan Date: Sat, 25 Apr 2026 02:12:46 +0800 Subject: [PATCH 1/8] feat(openai): add stream-to-nonstream chat handling --- dto/channel_settings.go | 13 +- relay/channel/openai/adaptor.go | 6 +- relay/channel/openai/chat_via_responses.go | 265 ++++++++++++++++++ relay/channel/openai/relay-openai.go | 214 ++++++++++++++ relay/chat_completions_via_responses.go | 9 + relay/common/relay_info.go | 51 ++-- relay/compatible_handler.go | 16 +- .../channels/modals/EditChannelModal.jsx | 13 + 8 files changed, 553 insertions(+), 34 deletions(-) diff --git a/dto/channel_settings.go b/dto/channel_settings.go index b6a1ab9f7138..86687c3f5a16 100644 --- a/dto/channel_settings.go +++ b/dto/channel_settings.go @@ -1,12 +1,13 @@ package dto type ChannelSettings struct { - ForceFormat bool `json:"force_format,omitempty"` - ThinkingToContent bool `json:"thinking_to_content,omitempty"` - Proxy string `json:"proxy"` - PassThroughBodyEnabled bool `json:"pass_through_body_enabled,omitempty"` - SystemPrompt string `json:"system_prompt,omitempty"` - SystemPromptOverride bool `json:"system_prompt_override,omitempty"` + ForceFormat bool `json:"force_format,omitempty"` + ThinkingToContent bool `json:"thinking_to_content,omitempty"` + Proxy string `json:"proxy"` + PassThroughBodyEnabled bool `json:"pass_through_body_enabled,omitempty"` + NonStreamUpstreamStream bool `json:"non_stream_upstream_stream,omitempty"` + SystemPrompt string `json:"system_prompt,omitempty"` + SystemPromptOverride bool `json:"system_prompt_override,omitempty"` } type VertexKeyType string diff --git a/relay/channel/openai/adaptor.go b/relay/channel/openai/adaptor.go index 3fd8d76a49c6..f8e777befd57 100644 --- a/relay/channel/openai/adaptor.go +++ b/relay/channel/openai/adaptor.go @@ -626,7 +626,11 @@ func (a *Adaptor) DoResponse(c *gin.Context, resp *http.Response, info *relaycom usage, err = OaiResponsesCompactionHandler(c, resp) default: if info.IsStream { - usage, err = OaiStreamHandler(c, info, resp) + if info.UpstreamStreamForNonStream && info.RelayMode == relayconstant.RelayModeChatCompletions { + usage, err = OaiStreamToNonStreamHandler(c, info, resp) + } else { + usage, err = OaiStreamHandler(c, info, resp) + } } else { usage, err = OpenaiHandler(c, info, resp) } diff --git a/relay/channel/openai/chat_via_responses.go b/relay/channel/openai/chat_via_responses.go index 5e8ec173de1a..24417d7078c4 100644 --- a/relay/channel/openai/chat_via_responses.go +++ b/relay/channel/openai/chat_via_responses.go @@ -1,6 +1,7 @@ package openai import ( + "bufio" "fmt" "io" "net/http" @@ -90,6 +91,270 @@ func OaiResponsesToChatHandler(c *gin.Context, info *relaycommon.RelayInfo, resp return usage, nil } +func OaiResponsesToChatStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { + if resp == nil || resp.Body == nil { + return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError) + } + + defer service.CloseResponseBodyGracefully(resp) + + responseId := helper.GetResponseID(c) + createdAt := time.Now().Unix() + model := info.UpstreamModelName + usage := &dto.Usage{} + var outputText strings.Builder + var reasoningText strings.Builder + var usageText strings.Builder + var sawToolCall bool + var streamErr *types.NewAPIError + + toolCallIndexByID := make(map[string]int) + toolCallNameByID := make(map[string]string) + toolCallArgsByID := make(map[string]string) + toolCallCanonicalIDByItemID := make(map[string]string) + + mergeUsage := func(response *dto.OpenAIResponsesResponse) { + if response == nil || response.Usage == nil { + return + } + if response.Usage.InputTokens != 0 { + usage.PromptTokens = response.Usage.InputTokens + usage.InputTokens = response.Usage.InputTokens + } + if response.Usage.OutputTokens != 0 { + usage.CompletionTokens = response.Usage.OutputTokens + usage.OutputTokens = response.Usage.OutputTokens + } + if response.Usage.TotalTokens != 0 { + usage.TotalTokens = response.Usage.TotalTokens + } else { + usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens + } + if response.Usage.InputTokensDetails != nil { + usage.PromptTokensDetails.CachedTokens = response.Usage.InputTokensDetails.CachedTokens + usage.PromptTokensDetails.ImageTokens = response.Usage.InputTokensDetails.ImageTokens + usage.PromptTokensDetails.AudioTokens = response.Usage.InputTokensDetails.AudioTokens + } + if response.Usage.CompletionTokenDetails.ReasoningTokens != 0 { + usage.CompletionTokenDetails.ReasoningTokens = response.Usage.CompletionTokenDetails.ReasoningTokens + } + } + + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.DefaultMaxScannerBufferSize) + for scanner.Scan() { + data := strings.TrimSpace(scanner.Text()) + if data == "" { + continue + } + if !strings.HasPrefix(data, "data:") && !strings.HasPrefix(data, "[DONE]") && !strings.HasPrefix(data, "event:") { + continue + } + if strings.HasPrefix(data, "event:") { + continue + } + if strings.HasPrefix(data, "data:") { + data = strings.TrimSpace(data[5:]) + } + if data == "" || strings.HasPrefix(data, "[DONE]") { + break + } + + var streamResp dto.ResponsesStreamResponse + if err := common.UnmarshalJsonStr(data, &streamResp); err != nil { + logger.LogError(c, "failed to unmarshal responses stream event: "+err.Error()) + streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) + break + } + + info.SetFirstResponseTime() + info.ReceivedResponseCount++ + + switch streamResp.Type { + case "response.created": + if streamResp.Response != nil { + if streamResp.Response.ID != "" { + responseId = streamResp.Response.ID + } + if streamResp.Response.Model != "" { + model = streamResp.Response.Model + } + if streamResp.Response.CreatedAt != 0 { + createdAt = int64(streamResp.Response.CreatedAt) + } + } + + case "response.output_text.delta": + if streamResp.Delta != "" { + outputText.WriteString(streamResp.Delta) + usageText.WriteString(streamResp.Delta) + } + + case "response.reasoning_summary_text.delta": + if streamResp.Delta != "" { + reasoningText.WriteString(streamResp.Delta) + usageText.WriteString(streamResp.Delta) + } + + case "response.output_item.added", "response.output_item.done": + if streamResp.Item == nil || streamResp.Item.Type != "function_call" { + break + } + itemID := strings.TrimSpace(streamResp.Item.ID) + callID := strings.TrimSpace(streamResp.Item.CallId) + if callID == "" { + callID = itemID + } + if itemID != "" && callID != "" { + toolCallCanonicalIDByItemID[itemID] = callID + } + if callID == "" { + break + } + if _, ok := toolCallIndexByID[callID]; !ok { + toolCallIndexByID[callID] = len(toolCallIndexByID) + } + if name := strings.TrimSpace(streamResp.Item.Name); name != "" { + toolCallNameByID[callID] = name + } + if streamResp.Item.Arguments != "" { + toolCallArgsByID[callID] = streamResp.Item.Arguments + } + sawToolCall = true + + case "response.function_call_arguments.delta": + itemID := strings.TrimSpace(streamResp.ItemID) + callID := toolCallCanonicalIDByItemID[itemID] + if callID == "" { + callID = itemID + } + if callID == "" { + break + } + if _, ok := toolCallIndexByID[callID]; !ok { + toolCallIndexByID[callID] = len(toolCallIndexByID) + } + toolCallArgsByID[callID] += streamResp.Delta + usageText.WriteString(streamResp.Delta) + sawToolCall = true + + case "response.completed": + if streamResp.Response != nil { + if streamResp.Response.ID != "" { + responseId = streamResp.Response.ID + } + if streamResp.Response.Model != "" { + model = streamResp.Response.Model + } + if streamResp.Response.CreatedAt != 0 { + createdAt = int64(streamResp.Response.CreatedAt) + } + mergeUsage(streamResp.Response) + } + + case "response.error", "response.failed": + if streamResp.Response != nil { + if oaiErr := streamResp.Response.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" { + streamErr = types.WithOpenAIError(*oaiErr, http.StatusInternalServerError) + break + } + } + streamErr = types.NewOpenAIError(fmt.Errorf("responses stream error: %s", streamResp.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError) + + default: + } + if streamErr != nil { + break + } + } + if err := scanner.Err(); err != nil && streamErr == nil { + streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) + } + + if streamErr != nil { + return nil, streamErr + } + + if usage.TotalTokens == 0 { + usage = service.ResponseText2Usage(c, usageText.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) + } + + msg := dto.Message{ + Role: "assistant", + Content: outputText.String(), + } + if reasoningText.Len() > 0 { + msg.ReasoningContent = reasoningText.String() + } + + var toolCalls []dto.ToolCallResponse + if sawToolCall && outputText.Len() == 0 { + toolCalls = make([]dto.ToolCallResponse, 0, len(toolCallIndexByID)) + callIDs := make([]string, len(toolCallIndexByID)) + for callID, idx := range toolCallIndexByID { + if idx >= 0 && idx < len(callIDs) { + callIDs[idx] = callID + } + } + for _, callID := range callIDs { + if callID == "" { + continue + } + toolCalls = append(toolCalls, dto.ToolCallResponse{ + ID: callID, + Type: "function", + Function: dto.FunctionResponse{ + Name: toolCallNameByID[callID], + Arguments: toolCallArgsByID[callID], + }, + }) + } + if len(toolCalls) > 0 { + msg.SetToolCalls(toolCalls) + msg.Content = "" + } + } + + finishReason := "stop" + if len(toolCalls) > 0 { + finishReason = "tool_calls" + } + + chatResp := &dto.OpenAITextResponse{ + Id: responseId, + Object: "chat.completion", + Created: createdAt, + Model: model, + Choices: []dto.OpenAITextResponseChoice{ + { + Index: 0, + Message: msg, + FinishReason: finishReason, + }, + }, + Usage: *usage, + } + + var responseBody []byte + var err error + switch info.RelayFormat { + case types.RelayFormatClaude: + claudeResp := service.ResponseOpenAI2Claude(chatResp, info) + responseBody, err = common.Marshal(claudeResp) + case types.RelayFormatGemini: + geminiResp := service.ResponseOpenAI2Gemini(chatResp, info) + responseBody, err = common.Marshal(geminiResp) + default: + responseBody, err = common.Marshal(chatResp) + } + if err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) + } + + service.IOCopyBytesGracefully(c, nil, responseBody) + return usage, nil +} + func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { if resp == nil || resp.Body == nil { return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError) diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index d33c5555f267..4ced867b7c96 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -1,10 +1,12 @@ package openai import ( + "bufio" "fmt" "io" "net/http" "strings" + "time" "github.com/QuantumNous/new-api/common" "github.com/QuantumNous/new-api/constant" @@ -192,6 +194,218 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re return usage, nil } +func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { + if resp == nil || resp.Body == nil { + logger.LogError(c, "invalid response or response body") + return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError) + } + + defer service.CloseResponseBodyGracefully(resp) + + responseId := helper.GetResponseID(c) + createdAt := time.Now().Unix() + model := info.UpstreamModelName + object := "chat.completion" + var usage = &dto.Usage{} + containStreamUsage := false + var responseTextBuilder strings.Builder + var toolCount int + var streamItems []string + var lastStreamData string + choices := make(map[int]*dto.OpenAITextResponseChoice) + toolCallsByChoice := make(map[int]map[int]*dto.ToolCallResponse) + var streamErr *types.NewAPIError + + getChoice := func(index int) *dto.OpenAITextResponseChoice { + choice, ok := choices[index] + if !ok { + choice = &dto.OpenAITextResponseChoice{Index: index} + choices[index] = choice + } + return choice + } + + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.DefaultMaxScannerBufferSize) + for scanner.Scan() { + data := strings.TrimSpace(scanner.Text()) + if data == "" { + continue + } + if !strings.HasPrefix(data, "data:") && !strings.HasPrefix(data, "[DONE]") { + continue + } + if strings.HasPrefix(data, "data:") { + data = strings.TrimSpace(data[5:]) + } + if data == "" || strings.HasPrefix(data, "[DONE]") { + break + } + + info.SetFirstResponseTime() + info.ReceivedResponseCount++ + lastStreamData = data + streamItems = append(streamItems, data) + var streamResponse dto.ChatCompletionsStreamResponse + if err := common.UnmarshalJsonStr(data, &streamResponse); err != nil { + streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) + break + } + + if streamResponse.Id != "" { + responseId = streamResponse.Id + } + if streamResponse.Created != 0 { + createdAt = streamResponse.Created + } + if streamResponse.Model != "" { + model = streamResponse.Model + } + if service.ValidUsage(streamResponse.Usage) { + usage = streamResponse.Usage + if usage.TotalTokens == 0 { + usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens + } + containStreamUsage = true + } + + for _, streamChoice := range streamResponse.Choices { + choice := getChoice(streamChoice.Index) + if streamChoice.FinishReason != nil { + choice.FinishReason = *streamChoice.FinishReason + } + if content := streamChoice.Delta.GetContentString(); content != "" { + choice.Message.SetStringContent(choice.Message.StringContent() + content) + } + if reasoning := streamChoice.Delta.GetReasoningContent(); reasoning != "" { + choice.Message.ReasoningContent += reasoning + } + if len(streamChoice.Delta.ToolCalls) > 0 { + choice.Message.Role = "assistant" + choiceToolCalls := toolCallsByChoice[streamChoice.Index] + if choiceToolCalls == nil { + choiceToolCalls = make(map[int]*dto.ToolCallResponse) + toolCallsByChoice[streamChoice.Index] = choiceToolCalls + } + for i, deltaToolCall := range streamChoice.Delta.ToolCalls { + idx := i + if deltaToolCall.Index != nil { + idx = *deltaToolCall.Index + } + toolCall := choiceToolCalls[idx] + if toolCall == nil { + toolCall = &dto.ToolCallResponse{} + choiceToolCalls[idx] = toolCall + } + if deltaToolCall.ID != "" { + toolCall.ID = deltaToolCall.ID + } + if deltaToolCall.Type != nil { + toolCall.Type = deltaToolCall.Type + } + if deltaToolCall.Function.Name != "" { + toolCall.Function.Name = deltaToolCall.Function.Name + } + if deltaToolCall.Function.Arguments != "" { + toolCall.Function.Arguments += deltaToolCall.Function.Arguments + } + } + } + } + } + if err := scanner.Err(); err != nil && streamErr == nil { + streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) + } + + if streamErr != nil { + return nil, streamErr + } + + if err := processTokens(info.RelayMode, streamItems, &responseTextBuilder, &toolCount); err != nil { + logger.LogError(c, "error processing tokens: "+err.Error()) + } + if !containStreamUsage { + usage = service.ResponseText2Usage(c, responseTextBuilder.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) + usage.CompletionTokens += toolCount * 7 + } + + responseChoices := make([]dto.OpenAITextResponseChoice, 0, len(choices)) + for index := 0; index < len(choices); index++ { + choice, ok := choices[index] + if !ok || choice == nil { + continue + } + choice.Message.Role = "assistant" + if choice.Message.Content == nil { + choice.Message.SetStringContent("") + } + if choice.FinishReason == "" { + choice.FinishReason = constant.FinishReasonStop + } + if choice.FinishReason == constant.FinishReasonContentFilter { + common.SetContextKey(c, constant.ContextKeyAdminRejectReason, "openai_finish_reason=content_filter") + } + if choiceToolCalls := toolCallsByChoice[index]; len(choiceToolCalls) > 0 { + toolCalls := make([]dto.ToolCallResponse, 0, len(choiceToolCalls)) + for toolIndex := 0; toolIndex < len(choiceToolCalls); toolIndex++ { + toolCall := choiceToolCalls[toolIndex] + if toolCall == nil { + continue + } + toolCalls = append(toolCalls, *toolCall) + } + choice.Message.SetToolCalls(toolCalls) + } + responseChoices = append(responseChoices, *choice) + } + + if len(responseChoices) == 0 { + responseChoices = append(responseChoices, dto.OpenAITextResponseChoice{ + Index: 0, + FinishReason: constant.FinishReasonStop, + Message: dto.Message{ + Role: "assistant", + Content: "", + }, + }) + } + + createdAny := any(createdAt) + simpleResponse := dto.OpenAITextResponse{ + Id: responseId, + Model: model, + Object: object, + Created: createdAny, + Choices: responseChoices, + Usage: *usage, + } + + responseBody, err := common.Marshal(simpleResponse) + if err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) + } + applyUsagePostProcessing(info, &simpleResponse.Usage, common.StringToByteSlice(lastStreamData)) + responseBody, err = common.Marshal(simpleResponse) + if err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) + } + + switch info.RelayFormat { + case types.RelayFormatClaude: + claudeResp := service.ResponseOpenAI2Claude(&simpleResponse, info) + responseBody, err = common.Marshal(claudeResp) + case types.RelayFormatGemini: + geminiResp := service.ResponseOpenAI2Gemini(&simpleResponse, info) + responseBody, err = common.Marshal(geminiResp) + } + if err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) + } + + service.IOCopyBytesGracefully(c, nil, responseBody) + return &simpleResponse.Usage, nil +} + func OpenaiHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { defer service.CloseResponseBodyGracefully(resp) diff --git a/relay/chat_completions_via_responses.go b/relay/chat_completions_via_responses.go index 7a2eb9aa6dac..13c907f441c5 100644 --- a/relay/chat_completions_via_responses.go +++ b/relay/chat_completions_via_responses.go @@ -146,6 +146,15 @@ func chatCompletionsViaResponses(c *gin.Context, info *relaycommon.RelayInfo, ad return nil, newApiErr } + if info.UpstreamStreamForNonStream { + usage, newApiErr := openaichannel.OaiResponsesToChatStreamToNonStreamHandler(c, info, httpResp) + if newApiErr != nil { + service.ResetStatusCode(newApiErr, statusCodeMappingStr) + return nil, newApiErr + } + return usage, nil + } + if info.IsStream { usage, newApiErr := openaichannel.OaiResponsesToChatStreamHandler(c, info, httpResp) if newApiErr != nil { diff --git a/relay/common/relay_info.go b/relay/common/relay_info.go index 64d4d4eedfaa..4f8920fd05c4 100644 --- a/relay/common/relay_info.go +++ b/relay/common/relay_info.go @@ -96,31 +96,32 @@ type RelayInfo struct { FirstResponseTime time.Time isFirstResponse bool //SendLastReasoningResponse bool - IsStream bool - IsGeminiBatchEmbedding bool - IsPlayground bool - UsePrice bool - RelayMode int - OriginModelName string - RequestURLPath string - RequestHeaders map[string]string - ShouldIncludeUsage bool - DisablePing bool // 是否禁止向下游发送自定义 Ping - ClientWs *websocket.Conn - TargetWs *websocket.Conn - InputAudioFormat string - OutputAudioFormat string - RealtimeTools []dto.RealTimeTool - IsFirstRequest bool - AudioUsage bool - ReasoningEffort string - UserSetting dto.UserSetting - UserEmail string - UserQuota int - RelayFormat types.RelayFormat - SendResponseCount int - ReceivedResponseCount int - FinalPreConsumedQuota int // 最终预消耗的配额 + IsStream bool + IsGeminiBatchEmbedding bool + IsPlayground bool + UsePrice bool + RelayMode int + OriginModelName string + RequestURLPath string + RequestHeaders map[string]string + ShouldIncludeUsage bool + UpstreamStreamForNonStream bool // 客户端非流式请求是否被转换为上游流式请求 + DisablePing bool // 是否禁止向下游发送自定义 Ping + ClientWs *websocket.Conn + TargetWs *websocket.Conn + InputAudioFormat string + OutputAudioFormat string + RealtimeTools []dto.RealTimeTool + IsFirstRequest bool + AudioUsage bool + ReasoningEffort string + UserSetting dto.UserSetting + UserEmail string + UserQuota int + RelayFormat types.RelayFormat + SendResponseCount int + ReceivedResponseCount int + FinalPreConsumedQuota int // 最终预消耗的配额 // ForcePreConsume 为 true 时禁用 BillingSession 的信任额度旁路， // 强制预扣全额。用于异步任务（视频/音乐生成等），因为请求返回后任务仍在运行， // 必须在提交前锁定全额。 diff --git a/relay/compatible_handler.go b/relay/compatible_handler.go index 7a5624eb34d6..13f68df5c414 100644 --- a/relay/compatible_handler.go +++ b/relay/compatible_handler.go @@ -40,6 +40,8 @@ func TextHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError *types c.Set("chat_completion_web_search_context_size", request.WebSearchOptions.SearchContextSize) } + passThroughGlobal := model_setting.GetGlobalSettings().PassThroughRequestEnabled + err = helper.ModelMappedHelper(c, info, request) if err != nil { return types.NewError(err, types.ErrorCodeChannelModelMappedError, types.ErrOptionWithSkipRetry()) @@ -51,12 +53,23 @@ func TextHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError *types includeUsage = request.StreamOptions.IncludeUsage } + convertNonStreamUpstreamStream := info.RelayMode == relayconstant.RelayModeChatCompletions && + !lo.FromPtrOr(request.Stream, false) && + info.ChannelSetting.NonStreamUpstreamStream && + !passThroughGlobal && + !info.ChannelSetting.PassThroughBodyEnabled + if convertNonStreamUpstreamStream { + stream := true + request.Stream = &stream + info.UpstreamStreamForNonStream = true + } + // 如果不支持StreamOptions，将StreamOptions设置为nil if !info.SupportStreamOptions || !lo.FromPtrOr(request.Stream, false) { request.StreamOptions = nil } else { // 如果支持StreamOptions，且请求中没有设置StreamOptions，根据配置文件设置StreamOptions - if constant.ForceStreamOption { + if constant.ForceStreamOption || convertNonStreamUpstreamStream { request.StreamOptions = &dto.StreamOptions{ IncludeUsage: true, } @@ -71,7 +84,6 @@ func TextHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError *types } adaptor.Init(info) - passThroughGlobal := model_setting.GetGlobalSettings().PassThroughRequestEnabled if info.RelayMode == relayconstant.RelayModeChatCompletions && !passThroughGlobal && !info.ChannelSetting.PassThroughBodyEnabled && diff --git a/web/src/components/table/channels/modals/EditChannelModal.jsx b/web/src/components/table/channels/modals/EditChannelModal.jsx index f0395d2f810c..d8288d339f9a 100644 --- a/web/src/components/table/channels/modals/EditChannelModal.jsx +++ b/web/src/components/table/channels/modals/EditChannelModal.jsx @@ -193,6 +193,7 @@ const EditChannelModal = (props) => { thinking_to_content: false, proxy: '', pass_through_body_enabled: false, + non_stream_upstream_stream: false, system_prompt: '', system_prompt_override: false, settings: '', @@ -496,6 +497,7 @@ const EditChannelModal = (props) => { const [channelSettings, setChannelSettings] = useState({ force_format: false, thinking_to_content: false, + non_stream_upstream_stream: false, proxy: '', pass_through_body_enabled: false, system_prompt: '', @@ -849,6 +851,8 @@ const EditChannelModal = (props) => { data.proxy = parsedSettings.proxy || ''; data.pass_through_body_enabled = parsedSettings.pass_through_body_enabled || false; + data.non_stream_upstream_stream = + parsedSettings.non_stream_upstream_stream || false; data.system_prompt = parsedSettings.system_prompt || ''; data.system_prompt_override = parsedSettings.system_prompt_override || false; @@ -858,6 +862,7 @@ const EditChannelModal = (props) => { data.thinking_to_content = false; data.proxy = ''; data.pass_through_body_enabled = false; + data.non_stream_upstream_stream = false; data.system_prompt = ''; data.system_prompt_override = false; } @@ -866,6 +871,7 @@ const EditChannelModal = (props) => { data.thinking_to_content = false; data.proxy = ''; data.pass_through_body_enabled = false; + data.non_stream_upstream_stream = false; data.system_prompt = ''; data.system_prompt_override = false; } @@ -975,6 +981,7 @@ const EditChannelModal = (props) => { thinking_to_content: data.thinking_to_content, proxy: data.proxy, pass_through_body_enabled: data.pass_through_body_enabled, + non_stream_upstream_stream: data.non_stream_upstream_stream, system_prompt: data.system_prompt, system_prompt_override: data.system_prompt_override || false, }); @@ -1017,6 +1024,7 @@ const EditChannelModal = (props) => { (data.system_prompt && data.system_prompt.trim()) || data.thinking_to_content || data.pass_through_body_enabled || + data.non_stream_upstream_stream || data.force_format || data.claude_beta_query || data.system_prompt_override; @@ -1364,6 +1372,7 @@ const EditChannelModal = (props) => { thinking_to_content: false, proxy: '', pass_through_body_enabled: false, + non_stream_upstream_stream: false, system_prompt: '', system_prompt_override: false, }); @@ -1734,6 +1743,7 @@ const EditChannelModal = (props) => { thinking_to_content: localInputs.thinking_to_content || false, proxy: localInputs.proxy || '', pass_through_body_enabled: localInputs.pass_through_body_enabled || false, + non_stream_upstream_stream: localInputs.non_stream_upstream_stream || false, system_prompt: localInputs.system_prompt || '', system_prompt_override: localInputs.system_prompt_override || false, }; @@ -1815,6 +1825,7 @@ const EditChannelModal = (props) => { delete localInputs.thinking_to_content; delete localInputs.proxy; delete localInputs.pass_through_body_enabled; + delete localInputs.non_stream_upstream_stream; delete localInputs.system_prompt; delete localInputs.system_prompt_override; delete localInputs.is_enterprise_account; @@ -2508,6 +2519,8 @@ const EditChannelModal = (props) => { handleChannelSettingsChange('thinking_to_content', value)} extraText={t('将 reasoning_content 转换为 标签拼接到内容中')} /> handleChannelSettingsChange('pass_through_body_enabled', value)} extraText={t('启用请求体透传功能')} /> + handleChannelSettingsChange('non_stream_upstream_stream', value)} extraText={t('客户端非流式 Chat Completions 请求将以流式方式请求上游，并在服务端聚合后返回非流式响应；透传请求体时不生效')} /> + handleChannelSettingsChange('proxy', value)} showClear extraText={t('用于配置网络代理，支持 socks5 协议')} /> handleChannelSettingsChange('system_prompt', value)} autosize showClear extraText={t('用户优先：如果用户在请求中指定了系统提示词，将优先使用用户的设置')} /> From d704da8ecabcb252ce13a9c755089b7f83044755 Mon Sep 17 00:00:00 2001 From: Sheng Fan Date: Sat, 25 Apr 2026 12:13:25 +0800 Subject: [PATCH 2/8] Fix stream fallback usage accounting --- relay/channel/openai/chat_via_responses.go | 16 ++++++++++++++-- relay/channel/openai/relay-openai.go | 2 ++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/relay/channel/openai/chat_via_responses.go b/relay/channel/openai/chat_via_responses.go index 24417d7078c4..5b3995510a09 100644 --- a/relay/channel/openai/chat_via_responses.go +++ b/relay/channel/openai/chat_via_responses.go @@ -215,10 +215,22 @@ func OaiResponsesToChatStreamToNonStreamHandler(c *gin.Context, info *relaycommo toolCallIndexByID[callID] = len(toolCallIndexByID) } if name := strings.TrimSpace(streamResp.Item.Name); name != "" { + if toolCallNameByID[callID] != name { + usageText.WriteString(name) + } toolCallNameByID[callID] = name } - if streamResp.Item.Arguments != "" { - toolCallArgsByID[callID] = streamResp.Item.Arguments + if args := streamResp.Item.Arguments; args != "" { + prevArgs := toolCallArgsByID[callID] + switch { + case prevArgs == "": + usageText.WriteString(args) + case strings.HasPrefix(args, prevArgs): + usageText.WriteString(args[len(prevArgs):]) + case prevArgs != args: + usageText.WriteString(args) + } + toolCallArgsByID[callID] = args } sawToolCall = true diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 4ced867b7c96..4f1e6bbaa251 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -185,6 +185,7 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re if !containStreamUsage { usage = service.ResponseText2Usage(c, responseTextBuilder.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) usage.CompletionTokens += toolCount * 7 + usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens } applyUsagePostProcessing(info, usage, common.StringToByteSlice(lastStreamData)) @@ -327,6 +328,7 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re if !containStreamUsage { usage = service.ResponseText2Usage(c, responseTextBuilder.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) usage.CompletionTokens += toolCount * 7 + usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens } responseChoices := make([]dto.OpenAITextResponseChoice, 0, len(choices)) From 4debfcb1acfbaa29b141cb70f50b0f40a25454b4 Mon Sep 17 00:00:00 2001 From: Sheng Fan Date: Sat, 25 Apr 2026 15:12:46 +0800 Subject: [PATCH 3/8] fix(openai): unify responses stream accumulation logic --- relay/channel/openai/chat_via_responses.go | 655 +++++++++++---------- relay/channel/openai/relay-openai.go | 7 +- 2 files changed, 336 insertions(+), 326 deletions(-) diff --git a/relay/channel/openai/chat_via_responses.go b/relay/channel/openai/chat_via_responses.go index 5b3995510a09..4ed44ac545b0 100644 --- a/relay/channel/openai/chat_via_responses.go +++ b/relay/channel/openai/chat_via_responses.go @@ -39,6 +39,267 @@ func stringDeltaFromPrefix(prev string, next string) string { return next } +type responsesStreamAccumulator struct { + c *gin.Context + info *relaycommon.RelayInfo + + responseID string + createdAt int64 + model string + + usage *dto.Usage + outputText strings.Builder + reasoningText strings.Builder + usageText strings.Builder + + sawToolCall bool + toolCallIndexByID map[string]int + toolCallNameByID map[string]string + toolCallArgsByID map[string]string + toolCallCanonicalIDByItemID map[string]string + lastToolCallID string + lastToolCallName string + lastToolCallArgsDelta string + err *types.NewAPIError +} + +func newResponsesStreamAccumulator(c *gin.Context, info *relaycommon.RelayInfo, responseID string, createdAt int64, model string) *responsesStreamAccumulator { + return &responsesStreamAccumulator{ + c: c, + info: info, + responseID: responseID, + createdAt: createdAt, + model: model, + usage: &dto.Usage{}, + toolCallIndexByID: make(map[string]int), + toolCallNameByID: make(map[string]string), + toolCallArgsByID: make(map[string]string), + toolCallCanonicalIDByItemID: make(map[string]string), + } +} + +func (a *responsesStreamAccumulator) Apply(ev *dto.ResponsesStreamResponse) error { + if a == nil || ev == nil { + return nil + } + if a.err != nil { + return a.applyError() + } + + a.lastToolCallID = "" + a.lastToolCallName = "" + a.lastToolCallArgsDelta = "" + + switch ev.Type { + case "response.created": + a.applyResponseMeta(ev.Response) + + case "response.output_text.delta": + if ev.Delta != "" { + a.outputText.WriteString(ev.Delta) + a.usageText.WriteString(ev.Delta) + } + + case "response.reasoning_summary_text.delta": + if ev.Delta != "" { + a.reasoningText.WriteString(ev.Delta) + a.usageText.WriteString(ev.Delta) + } + + case "response.output_item.added", "response.output_item.done": + a.applyToolCallItem(ev.Item) + + case "response.function_call_arguments.delta": + a.applyToolCallArgumentsDelta(ev.ItemID, ev.Delta) + + case "response.function_call_arguments.done": + + case "response.completed": + a.applyResponseMeta(ev.Response) + a.mergeUsage(ev.Response) + + case "response.error", "response.failed": + if ev.Response != nil { + if oaiErr := ev.Response.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" { + a.err = types.WithOpenAIError(*oaiErr, http.StatusInternalServerError) + } else { + a.err = types.NewOpenAIError(fmt.Errorf("responses stream error: %s", ev.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError) + } + } else { + a.err = types.NewOpenAIError(fmt.Errorf("responses stream error: %s", ev.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError) + } + return a.applyError() + + default: + } + + return nil +} + +func (a *responsesStreamAccumulator) Result() (messageText string, reasoningText string, usage *dto.Usage, toolCalls []dto.ToolCallResponse, err *types.NewAPIError) { + if a == nil { + return "", "", nil, nil, nil + } + if a.err != nil { + return "", "", nil, nil, a.err + } + usage = a.usage + if usage == nil { + usage = &dto.Usage{} + } + if usage.TotalTokens == 0 { + usage = service.ResponseText2Usage(a.c, a.usageText.String(), a.info.UpstreamModelName, a.info.GetEstimatePromptTokens()) + a.usage = usage + } + if a.sawToolCall && a.outputText.Len() == 0 { + toolCalls = a.orderedToolCalls() + } + return a.outputText.String(), a.reasoningText.String(), usage, toolCalls, nil +} + +func (a *responsesStreamAccumulator) applyResponseMeta(response *dto.OpenAIResponsesResponse) { + if response == nil { + return + } + if response.ID != "" { + a.responseID = response.ID + } + if response.Model != "" { + a.model = response.Model + } + if response.CreatedAt != 0 { + a.createdAt = int64(response.CreatedAt) + } +} + +func (a *responsesStreamAccumulator) mergeUsage(response *dto.OpenAIResponsesResponse) { + if response == nil || response.Usage == nil { + return + } + if a.usage == nil { + a.usage = &dto.Usage{} + } + if response.Usage.InputTokens != 0 { + a.usage.PromptTokens = response.Usage.InputTokens + a.usage.InputTokens = response.Usage.InputTokens + } + if response.Usage.OutputTokens != 0 { + a.usage.CompletionTokens = response.Usage.OutputTokens + a.usage.OutputTokens = response.Usage.OutputTokens + } + if response.Usage.TotalTokens != 0 { + a.usage.TotalTokens = response.Usage.TotalTokens + } else { + a.usage.TotalTokens = a.usage.PromptTokens + a.usage.CompletionTokens + } + if response.Usage.InputTokensDetails != nil { + a.usage.PromptTokensDetails.CachedTokens = response.Usage.InputTokensDetails.CachedTokens + a.usage.PromptTokensDetails.ImageTokens = response.Usage.InputTokensDetails.ImageTokens + a.usage.PromptTokensDetails.AudioTokens = response.Usage.InputTokensDetails.AudioTokens + } + if response.Usage.CompletionTokenDetails.ReasoningTokens != 0 { + a.usage.CompletionTokenDetails.ReasoningTokens = response.Usage.CompletionTokenDetails.ReasoningTokens + } +} + +func (a *responsesStreamAccumulator) applyToolCallItem(item *dto.ResponsesOutput) { + if item == nil || item.Type != "function_call" { + return + } + itemID := strings.TrimSpace(item.ID) + callID := strings.TrimSpace(item.CallId) + if callID == "" { + callID = itemID + } + if itemID != "" && callID != "" { + a.toolCallCanonicalIDByItemID[itemID] = callID + } + if callID == "" { + return + } + a.ensureToolCallIndex(callID) + if name := strings.TrimSpace(item.Name); name != "" { + if a.toolCallNameByID[callID] != name { + a.usageText.WriteString(name) + } + a.toolCallNameByID[callID] = name + a.lastToolCallName = name + } + if args := item.Arguments; args != "" { + prevArgs := a.toolCallArgsByID[callID] + switch { + case prevArgs == "": + a.lastToolCallArgsDelta = args + case strings.HasPrefix(args, prevArgs): + a.lastToolCallArgsDelta = args[len(prevArgs):] + case prevArgs != args: + a.lastToolCallArgsDelta = args + } + if a.lastToolCallArgsDelta != "" { + a.usageText.WriteString(a.lastToolCallArgsDelta) + } + a.toolCallArgsByID[callID] = args + } + a.lastToolCallID = callID + a.sawToolCall = true +} + +func (a *responsesStreamAccumulator) applyToolCallArgumentsDelta(itemID string, delta string) { + callID := a.toolCallCanonicalIDByItemID[strings.TrimSpace(itemID)] + if callID == "" { + callID = strings.TrimSpace(itemID) + } + if callID == "" { + return + } + a.ensureToolCallIndex(callID) + a.toolCallArgsByID[callID] += delta + a.usageText.WriteString(delta) + a.lastToolCallID = callID + a.lastToolCallArgsDelta = delta + a.sawToolCall = true +} + +func (a *responsesStreamAccumulator) ensureToolCallIndex(callID string) { + if _, ok := a.toolCallIndexByID[callID]; !ok { + a.toolCallIndexByID[callID] = len(a.toolCallIndexByID) + } +} + +func (a *responsesStreamAccumulator) orderedToolCalls() []dto.ToolCallResponse { + toolCalls := make([]dto.ToolCallResponse, 0, len(a.toolCallIndexByID)) + callIDs := make([]string, len(a.toolCallIndexByID)) + for callID, idx := range a.toolCallIndexByID { + if idx >= 0 && idx < len(callIDs) { + callIDs[idx] = callID + } + } + for _, callID := range callIDs { + if callID == "" { + continue + } + toolCalls = append(toolCalls, dto.ToolCallResponse{ + ID: callID, + Type: "function", + Function: dto.FunctionResponse{ + Name: a.toolCallNameByID[callID], + Arguments: a.toolCallArgsByID[callID], + }, + }) + } + return toolCalls +} + +func (a *responsesStreamAccumulator) applyError() error { + if a == nil || a.err == nil { + return nil + } + if a.err.Err != nil { + return a.err.Err + } + return fmt.Errorf("responses stream error") +} + func OaiResponsesToChatHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { if resp == nil || resp.Body == nil { return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError) @@ -101,44 +362,8 @@ func OaiResponsesToChatStreamToNonStreamHandler(c *gin.Context, info *relaycommo responseId := helper.GetResponseID(c) createdAt := time.Now().Unix() model := info.UpstreamModelName - usage := &dto.Usage{} - var outputText strings.Builder - var reasoningText strings.Builder - var usageText strings.Builder - var sawToolCall bool var streamErr *types.NewAPIError - - toolCallIndexByID := make(map[string]int) - toolCallNameByID := make(map[string]string) - toolCallArgsByID := make(map[string]string) - toolCallCanonicalIDByItemID := make(map[string]string) - - mergeUsage := func(response *dto.OpenAIResponsesResponse) { - if response == nil || response.Usage == nil { - return - } - if response.Usage.InputTokens != 0 { - usage.PromptTokens = response.Usage.InputTokens - usage.InputTokens = response.Usage.InputTokens - } - if response.Usage.OutputTokens != 0 { - usage.CompletionTokens = response.Usage.OutputTokens - usage.OutputTokens = response.Usage.OutputTokens - } - if response.Usage.TotalTokens != 0 { - usage.TotalTokens = response.Usage.TotalTokens - } else { - usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens - } - if response.Usage.InputTokensDetails != nil { - usage.PromptTokensDetails.CachedTokens = response.Usage.InputTokensDetails.CachedTokens - usage.PromptTokensDetails.ImageTokens = response.Usage.InputTokensDetails.ImageTokens - usage.PromptTokensDetails.AudioTokens = response.Usage.InputTokensDetails.AudioTokens - } - if response.Usage.CompletionTokenDetails.ReasoningTokens != 0 { - usage.CompletionTokenDetails.ReasoningTokens = response.Usage.CompletionTokenDetails.ReasoningTokens - } - } + acc := newResponsesStreamAccumulator(c, info, responseId, createdAt, model) scanner := bufio.NewScanner(resp.Body) scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.DefaultMaxScannerBufferSize) @@ -170,110 +395,8 @@ func OaiResponsesToChatStreamToNonStreamHandler(c *gin.Context, info *relaycommo info.SetFirstResponseTime() info.ReceivedResponseCount++ - switch streamResp.Type { - case "response.created": - if streamResp.Response != nil { - if streamResp.Response.ID != "" { - responseId = streamResp.Response.ID - } - if streamResp.Response.Model != "" { - model = streamResp.Response.Model - } - if streamResp.Response.CreatedAt != 0 { - createdAt = int64(streamResp.Response.CreatedAt) - } - } - - case "response.output_text.delta": - if streamResp.Delta != "" { - outputText.WriteString(streamResp.Delta) - usageText.WriteString(streamResp.Delta) - } - - case "response.reasoning_summary_text.delta": - if streamResp.Delta != "" { - reasoningText.WriteString(streamResp.Delta) - usageText.WriteString(streamResp.Delta) - } - - case "response.output_item.added", "response.output_item.done": - if streamResp.Item == nil || streamResp.Item.Type != "function_call" { - break - } - itemID := strings.TrimSpace(streamResp.Item.ID) - callID := strings.TrimSpace(streamResp.Item.CallId) - if callID == "" { - callID = itemID - } - if itemID != "" && callID != "" { - toolCallCanonicalIDByItemID[itemID] = callID - } - if callID == "" { - break - } - if _, ok := toolCallIndexByID[callID]; !ok { - toolCallIndexByID[callID] = len(toolCallIndexByID) - } - if name := strings.TrimSpace(streamResp.Item.Name); name != "" { - if toolCallNameByID[callID] != name { - usageText.WriteString(name) - } - toolCallNameByID[callID] = name - } - if args := streamResp.Item.Arguments; args != "" { - prevArgs := toolCallArgsByID[callID] - switch { - case prevArgs == "": - usageText.WriteString(args) - case strings.HasPrefix(args, prevArgs): - usageText.WriteString(args[len(prevArgs):]) - case prevArgs != args: - usageText.WriteString(args) - } - toolCallArgsByID[callID] = args - } - sawToolCall = true - - case "response.function_call_arguments.delta": - itemID := strings.TrimSpace(streamResp.ItemID) - callID := toolCallCanonicalIDByItemID[itemID] - if callID == "" { - callID = itemID - } - if callID == "" { - break - } - if _, ok := toolCallIndexByID[callID]; !ok { - toolCallIndexByID[callID] = len(toolCallIndexByID) - } - toolCallArgsByID[callID] += streamResp.Delta - usageText.WriteString(streamResp.Delta) - sawToolCall = true - - case "response.completed": - if streamResp.Response != nil { - if streamResp.Response.ID != "" { - responseId = streamResp.Response.ID - } - if streamResp.Response.Model != "" { - model = streamResp.Response.Model - } - if streamResp.Response.CreatedAt != 0 { - createdAt = int64(streamResp.Response.CreatedAt) - } - mergeUsage(streamResp.Response) - } - - case "response.error", "response.failed": - if streamResp.Response != nil { - if oaiErr := streamResp.Response.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" { - streamErr = types.WithOpenAIError(*oaiErr, http.StatusInternalServerError) - break - } - } - streamErr = types.NewOpenAIError(fmt.Errorf("responses stream error: %s", streamResp.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError) - - default: + if err := acc.Apply(&streamResp); err != nil { + _, _, _, _, streamErr = acc.Result() } if streamErr != nil { break @@ -287,44 +410,22 @@ func OaiResponsesToChatStreamToNonStreamHandler(c *gin.Context, info *relaycommo return nil, streamErr } - if usage.TotalTokens == 0 { - usage = service.ResponseText2Usage(c, usageText.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) + messageText, reasoning, usage, toolCalls, streamErr := acc.Result() + if streamErr != nil { + return nil, streamErr } msg := dto.Message{ Role: "assistant", - Content: outputText.String(), + Content: messageText, } - if reasoningText.Len() > 0 { - msg.ReasoningContent = reasoningText.String() + if reasoning != "" { + msg.ReasoningContent = reasoning } - var toolCalls []dto.ToolCallResponse - if sawToolCall && outputText.Len() == 0 { - toolCalls = make([]dto.ToolCallResponse, 0, len(toolCallIndexByID)) - callIDs := make([]string, len(toolCallIndexByID)) - for callID, idx := range toolCallIndexByID { - if idx >= 0 && idx < len(callIDs) { - callIDs[idx] = callID - } - } - for _, callID := range callIDs { - if callID == "" { - continue - } - toolCalls = append(toolCalls, dto.ToolCallResponse{ - ID: callID, - Type: "function", - Function: dto.FunctionResponse{ - Name: toolCallNameByID[callID], - Arguments: toolCallArgsByID[callID], - }, - }) - } - if len(toolCalls) > 0 { - msg.SetToolCalls(toolCalls) - msg.Content = "" - } + if len(toolCalls) > 0 { + msg.SetToolCalls(toolCalls) + msg.Content = "" } finishReason := "stop" @@ -333,10 +434,10 @@ func OaiResponsesToChatStreamToNonStreamHandler(c *gin.Context, info *relaycommo } chatResp := &dto.OpenAITextResponse{ - Id: responseId, + Id: acc.responseID, Object: "chat.completion", - Created: createdAt, - Model: model, + Created: acc.createdAt, + Model: acc.model, Choices: []dto.OpenAITextResponseChoice{ { Index: 0, @@ -377,22 +478,15 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo responseId := helper.GetResponseID(c) createAt := time.Now().Unix() model := info.UpstreamModelName + acc := newResponsesStreamAccumulator(c, info, responseId, createAt, model) var ( - usage = &dto.Usage{} - outputText strings.Builder - usageText strings.Builder - sentStart bool - sentStop bool - sawToolCall bool - streamErr *types.NewAPIError + sentStart bool + sentStop bool + streamErr *types.NewAPIError ) - toolCallIndexByID := make(map[string]int) - toolCallNameByID := make(map[string]string) - toolCallArgsByID := make(map[string]string) toolCallNameSent := make(map[string]bool) - toolCallCanonicalIDByItemID := make(map[string]string) hasSentReasoningSummary := false needsReasoningSummarySeparator := false //reasoningSummaryTextByKey := make(map[string]string) @@ -429,7 +523,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo if sentStart { return true } - if !sendChatChunk(helper.GenerateStartEmptyResponse(responseId, createAt, model, nil)) { + if !sendChatChunk(helper.GenerateStartEmptyResponse(acc.responseID, acc.createdAt, acc.model, nil)) { return false } sentStart = true @@ -485,12 +579,11 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo return false } - usageText.WriteString(delta) chunk := &dto.ChatCompletionsStreamResponse{ - Id: responseId, + Id: acc.responseID, Object: "chat.completion.chunk", - Created: createAt, - Model: model, + Created: acc.createdAt, + Model: acc.model, Choices: []dto.ChatCompletionsStreamResponseChoice{ { Index: 0, @@ -511,7 +604,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo if callID == "" { return true } - if outputText.Len() > 0 { + if acc.outputText.Len() > 0 { // Prefer streaming assistant text over tool calls to match non-stream behavior. return true } @@ -519,16 +612,13 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo return false } - idx, ok := toolCallIndexByID[callID] + idx, ok := acc.toolCallIndexByID[callID] if !ok { - idx = len(toolCallIndexByID) - toolCallIndexByID[callID] = idx - } - if name != "" { - toolCallNameByID[callID] = name + idx = len(acc.toolCallIndexByID) + acc.toolCallIndexByID[callID] = idx } - if toolCallNameByID[callID] != "" { - name = toolCallNameByID[callID] + if acc.toolCallNameByID[callID] != "" { + name = acc.toolCallNameByID[callID] } tool := dto.ToolCallResponse{ @@ -545,10 +635,10 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo } chunk := &dto.ChatCompletionsStreamResponse{ - Id: responseId, + Id: acc.responseID, Object: "chat.completion.chunk", - Created: createAt, - Model: model, + Created: acc.createdAt, + Model: acc.model, Choices: []dto.ChatCompletionsStreamResponseChoice{ { Index: 0, @@ -561,15 +651,6 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo if !sendChatChunk(chunk) { return false } - sawToolCall = true - - // Include tool call data in the local builder for fallback token estimation. - if tool.Function.Name != "" { - usageText.WriteString(tool.Function.Name) - } - if argsDelta != "" { - usageText.WriteString(argsDelta) - } return true } @@ -586,16 +667,27 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo return } - switch streamResp.Type { - case "response.created": - if streamResp.Response != nil { - if streamResp.Response.Model != "" { - model = streamResp.Response.Model - } - if streamResp.Response.CreatedAt != 0 { - createAt = int64(streamResp.Response.CreatedAt) - } + streamEvent := streamResp + if streamEvent.Type == "response.reasoning_summary_text.delta" && streamEvent.Delta != "" && needsReasoningSummarySeparator { + if strings.HasPrefix(streamEvent.Delta, "\n\n") { + needsReasoningSummarySeparator = false + } else if strings.HasPrefix(streamEvent.Delta, "\n") { + streamEvent.Delta = "\n" + streamEvent.Delta + needsReasoningSummarySeparator = false + } else { + streamEvent.Delta = "\n\n" + streamEvent.Delta + needsReasoningSummarySeparator = false } + } + + if err := acc.Apply(&streamEvent); err != nil { + _, _, _, _, streamErr = acc.Result() + sr.Stop(streamErr) + return + } + + switch streamEvent.Type { + case "response.created": //case "response.reasoning_text.delta": //if !sendReasoningDelta(streamResp.Delta) { @@ -606,7 +698,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo //case "response.reasoning_text.done": case "response.reasoning_summary_text.delta": - if !sendReasoningSummaryDelta(streamResp.Delta) { + if !sendReasoningSummaryDelta(streamEvent.Delta) { sr.Stop(streamErr) return } @@ -640,15 +732,13 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo return } - if streamResp.Delta != "" { - outputText.WriteString(streamResp.Delta) - usageText.WriteString(streamResp.Delta) - delta := streamResp.Delta + if streamEvent.Delta != "" { + delta := streamEvent.Delta chunk := &dto.ChatCompletionsStreamResponse{ - Id: responseId, + Id: acc.responseID, Object: "chat.completion.chunk", - Created: createAt, - Model: model, + Created: acc.createdAt, + Model: acc.model, Choices: []dto.ChatCompletionsStreamResponseChoice{ { Index: 0, @@ -665,54 +755,13 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo } case "response.output_item.added", "response.output_item.done": - if streamResp.Item == nil { - break - } - if streamResp.Item.Type != "function_call" { - break - } - - itemID := strings.TrimSpace(streamResp.Item.ID) - callID := strings.TrimSpace(streamResp.Item.CallId) - if callID == "" { - callID = itemID - } - if itemID != "" && callID != "" { - toolCallCanonicalIDByItemID[itemID] = callID - } - name := strings.TrimSpace(streamResp.Item.Name) - if name != "" { - toolCallNameByID[callID] = name - } - - newArgs := streamResp.Item.Arguments - prevArgs := toolCallArgsByID[callID] - argsDelta := "" - if newArgs != "" { - if strings.HasPrefix(newArgs, prevArgs) { - argsDelta = newArgs[len(prevArgs):] - } else { - argsDelta = newArgs - } - toolCallArgsByID[callID] = newArgs - } - - if !sendToolCallDelta(callID, name, argsDelta) { + if !sendToolCallDelta(acc.lastToolCallID, acc.lastToolCallName, acc.lastToolCallArgsDelta) { sr.Stop(streamErr) return } case "response.function_call_arguments.delta": - itemID := strings.TrimSpace(streamResp.ItemID) - callID := toolCallCanonicalIDByItemID[itemID] - if callID == "" { - callID = itemID - } - if callID == "" { - break - } - toolCallArgsByID[callID] += streamResp.Delta - if !sendToolCallDelta(callID, "", streamResp.Delta) { + if !sendToolCallDelta(acc.lastToolCallID, "", acc.lastToolCallArgsDelta) { sr.Stop(streamErr) return } @@ -720,36 +769,11 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo case "response.function_call_arguments.done": case "response.completed": - if streamResp.Response != nil { - if streamResp.Response.Model != "" { - model = streamResp.Response.Model - } - if streamResp.Response.CreatedAt != 0 { - createAt = int64(streamResp.Response.CreatedAt) - } - if streamResp.Response.Usage != nil { - if streamResp.Response.Usage.InputTokens != 0 { - usage.PromptTokens = streamResp.Response.Usage.InputTokens - usage.InputTokens = streamResp.Response.Usage.InputTokens - } - if streamResp.Response.Usage.OutputTokens != 0 { - usage.CompletionTokens = streamResp.Response.Usage.OutputTokens - usage.OutputTokens = streamResp.Response.Usage.OutputTokens - } - if streamResp.Response.Usage.TotalTokens != 0 { - usage.TotalTokens = streamResp.Response.Usage.TotalTokens - } else { - usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens - } - if streamResp.Response.Usage.InputTokensDetails != nil { - usage.PromptTokensDetails.CachedTokens = streamResp.Response.Usage.InputTokensDetails.CachedTokens - usage.PromptTokensDetails.ImageTokens = streamResp.Response.Usage.InputTokensDetails.ImageTokens - usage.PromptTokensDetails.AudioTokens = streamResp.Response.Usage.InputTokensDetails.AudioTokens - } - if streamResp.Response.Usage.CompletionTokenDetails.ReasoningTokens != 0 { - usage.CompletionTokenDetails.ReasoningTokens = streamResp.Response.Usage.CompletionTokenDetails.ReasoningTokens - } - } + _, _, usage, toolCalls, resultErr := acc.Result() + if resultErr != nil { + streamErr = resultErr + sr.Stop(streamErr) + return } if !sendStartIfNeeded() { @@ -761,10 +785,10 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo info.ClaudeConvertInfo.Usage = usage } finishReason := "stop" - if sawToolCall && outputText.Len() == 0 { + if len(toolCalls) > 0 { finishReason = "tool_calls" } - stop := helper.GenerateStopResponse(responseId, createAt, model, finishReason) + stop := helper.GenerateStopResponse(acc.responseID, acc.createdAt, acc.model, finishReason) if !sendChatChunk(stop) { sr.Stop(streamErr) return @@ -773,16 +797,6 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo } case "response.error", "response.failed": - if streamResp.Response != nil { - if oaiErr := streamResp.Response.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" { - streamErr = types.WithOpenAIError(*oaiErr, http.StatusInternalServerError) - sr.Stop(streamErr) - return - } - } - streamErr = types.NewOpenAIError(fmt.Errorf("responses stream error: %s", streamResp.Type), types.ErrorCodeBadResponse, http.StatusInternalServerError) - sr.Stop(streamErr) - return default: } @@ -792,12 +806,13 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo return nil, streamErr } - if usage.TotalTokens == 0 { - usage = service.ResponseText2Usage(c, usageText.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) + _, _, usage, toolCalls, streamErr := acc.Result() + if streamErr != nil { + return nil, streamErr } if !sentStart { - if !sendChatChunk(helper.GenerateStartEmptyResponse(responseId, createAt, model, nil)) { + if !sendChatChunk(helper.GenerateStartEmptyResponse(acc.responseID, acc.createdAt, acc.model, nil)) { return nil, streamErr } } @@ -806,16 +821,16 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo info.ClaudeConvertInfo.Usage = usage } finishReason := "stop" - if sawToolCall && outputText.Len() == 0 { + if len(toolCalls) > 0 { finishReason = "tool_calls" } - stop := helper.GenerateStopResponse(responseId, createAt, model, finishReason) + stop := helper.GenerateStopResponse(acc.responseID, acc.createdAt, acc.model, finishReason) if !sendChatChunk(stop) { return nil, streamErr } } if info.RelayFormat == types.RelayFormatOpenAI && info.ShouldIncludeUsage && usage != nil { - if err := helper.ObjectData(c, helper.GenerateFinalUsageResponse(responseId, createAt, model, *usage)); err != nil { + if err := helper.ObjectData(c, helper.GenerateFinalUsageResponse(acc.responseID, acc.createdAt, acc.model, *usage)); err != nil { return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponse, http.StatusInternalServerError) } } diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 4f1e6bbaa251..77cad1c148c3 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -372,12 +372,11 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re }) } - createdAny := any(createdAt) simpleResponse := dto.OpenAITextResponse{ Id: responseId, Model: model, Object: object, - Created: createdAny, + Created: createdAt, Choices: responseChoices, Usage: *usage, } @@ -387,10 +386,6 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) } applyUsagePostProcessing(info, &simpleResponse.Usage, common.StringToByteSlice(lastStreamData)) - responseBody, err = common.Marshal(simpleResponse) - if err != nil { - return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) - } switch info.RelayFormat { case types.RelayFormatClaude: From c5288f23c7cf895a0a003c30a31ed42f90725a7f Mon Sep 17 00:00:00 2001 From: Sheng Fan Date: Tue, 28 Apr 2026 14:35:36 +0800 Subject: [PATCH 4/8] fix openai stream aggregation postprocessing --- relay/channel/openai/relay-openai.go | 100 ++++++++++++++++++++++++--- 1 file changed, 90 insertions(+), 10 deletions(-) diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 77cad1c148c3..2dbeae06a0ba 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "net/http" + "sort" "strings" "time" @@ -188,7 +189,8 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens } - applyUsagePostProcessing(info, usage, common.StringToByteSlice(lastStreamData)) + postProcessingBody := buildStreamUsagePostProcessingBody(info, streamItems) + applyUsagePostProcessing(info, usage, postProcessingBody) HandleFinalResponse(c, info, lastStreamData, responseId, createAt, model, systemFingerprint, usage, containStreamUsage) @@ -212,7 +214,6 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re var responseTextBuilder strings.Builder var toolCount int var streamItems []string - var lastStreamData string choices := make(map[int]*dto.OpenAITextResponseChoice) toolCallsByChoice := make(map[int]map[int]*dto.ToolCallResponse) var streamErr *types.NewAPIError @@ -245,7 +246,6 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re info.SetFirstResponseTime() info.ReceivedResponseCount++ - lastStreamData = data streamItems = append(streamItems, data) var streamResponse dto.ChatCompletionsStreamResponse if err := common.UnmarshalJsonStr(data, &streamResponse); err != nil { @@ -331,8 +331,14 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens } + choiceIndexes := make([]int, 0, len(choices)) + for index := range choices { + choiceIndexes = append(choiceIndexes, index) + } + sort.Ints(choiceIndexes) + responseChoices := make([]dto.OpenAITextResponseChoice, 0, len(choices)) - for index := 0; index < len(choices); index++ { + for _, index := range choiceIndexes { choice, ok := choices[index] if !ok || choice == nil { continue @@ -348,8 +354,14 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re common.SetContextKey(c, constant.ContextKeyAdminRejectReason, "openai_finish_reason=content_filter") } if choiceToolCalls := toolCallsByChoice[index]; len(choiceToolCalls) > 0 { + toolCallIndexes := make([]int, 0, len(choiceToolCalls)) + for toolIndex := range choiceToolCalls { + toolCallIndexes = append(toolCallIndexes, toolIndex) + } + sort.Ints(toolCallIndexes) + toolCalls := make([]dto.ToolCallResponse, 0, len(choiceToolCalls)) - for toolIndex := 0; toolIndex < len(choiceToolCalls); toolIndex++ { + for _, toolIndex := range toolCallIndexes { toolCall := choiceToolCalls[toolIndex] if toolCall == nil { continue @@ -381,12 +393,11 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re Usage: *usage, } - responseBody, err := common.Marshal(simpleResponse) - if err != nil { - return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) - } - applyUsagePostProcessing(info, &simpleResponse.Usage, common.StringToByteSlice(lastStreamData)) + postProcessingBody := buildStreamUsagePostProcessingBody(info, streamItems) + applyUsagePostProcessing(info, &simpleResponse.Usage, postProcessingBody) + var responseBody []byte + var err error switch info.RelayFormat { case types.RelayFormatClaude: claudeResp := service.ResponseOpenAI2Claude(&simpleResponse, info) @@ -394,6 +405,8 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re case types.RelayFormatGemini: geminiResp := service.ResponseOpenAI2Gemini(&simpleResponse, info) responseBody, err = common.Marshal(geminiResp) + default: + responseBody, err = common.Marshal(simpleResponse) } if err != nil { return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) @@ -803,6 +816,73 @@ func OpenaiHandlerWithUsage(c *gin.Context, info *relaycommon.RelayInfo, resp *h return &usageResp.Usage, nil } +func buildStreamUsagePostProcessingBody(info *relaycommon.RelayInfo, streamItems []string) []byte { + if info == nil || len(streamItems) == 0 { + return nil + } + + switch info.ChannelType { + case constant.ChannelTypeZhipu_v4: + if cachedTokens, ok := findCachedTokensInStream(streamItems, extractCachedTokensFromBody); ok { + return marshalPostProcessingPayload(map[string]any{ + "usage": map[string]any{ + "prompt_tokens_details": map[string]any{ + "cached_tokens": cachedTokens, + }, + }, + }) + } + case constant.ChannelTypeMoonshot: + if cachedTokens, ok := findCachedTokensInStream(streamItems, extractMoonshotCachedTokensFromBody); ok { + return marshalPostProcessingPayload(map[string]any{ + "choices": []map[string]any{ + { + "usage": map[string]any{ + "cached_tokens": cachedTokens, + }, + }, + }, + }) + } + if cachedTokens, ok := findCachedTokensInStream(streamItems, extractCachedTokensFromBody); ok { + return marshalPostProcessingPayload(map[string]any{ + "usage": map[string]any{ + "prompt_tokens_details": map[string]any{ + "cached_tokens": cachedTokens, + }, + }, + }) + } + case constant.ChannelTypeOpenAI: + if cachedTokens, ok := findCachedTokensInStream(streamItems, extractLlamaCachedTokensFromBody); ok { + return marshalPostProcessingPayload(map[string]any{ + "timings": map[string]any{ + "cache_n": cachedTokens, + }, + }) + } + } + + return nil +} + +func findCachedTokensInStream(streamItems []string, extractor func([]byte) (int, bool)) (int, bool) { + for i := len(streamItems) - 1; i >= 0; i-- { + if cachedTokens, ok := extractor(common.StringToByteSlice(streamItems[i])); ok { + return cachedTokens, true + } + } + return 0, false +} + +func marshalPostProcessingPayload(payload any) []byte { + body, err := common.Marshal(payload) + if err != nil { + return nil + } + return body +} + func applyUsagePostProcessing(info *relaycommon.RelayInfo, usage *dto.Usage, responseBody []byte) { if info == nil || usage == nil { return From 97371f794e22dd230e99619dfc8422c683bbc27b Mon Sep 17 00:00:00 2001 From: Sheng Fan Date: Tue, 28 Apr 2026 14:38:43 +0800 Subject: [PATCH 5/8] fix responses stream usage fallback --- relay/channel/openai/chat_via_responses.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/relay/channel/openai/chat_via_responses.go b/relay/channel/openai/chat_via_responses.go index 4ed44ac545b0..ec5d806ea2d0 100644 --- a/relay/channel/openai/chat_via_responses.go +++ b/relay/channel/openai/chat_via_responses.go @@ -149,6 +149,8 @@ func (a *responsesStreamAccumulator) Result() (messageText string, reasoningText } if usage.TotalTokens == 0 { usage = service.ResponseText2Usage(a.c, a.usageText.String(), a.info.UpstreamModelName, a.info.GetEstimatePromptTokens()) + usage.CompletionTokens += len(a.toolCallIndexByID) * 7 + usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens a.usage = usage } if a.sawToolCall && a.outputText.Len() == 0 { @@ -797,6 +799,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo } case "response.error", "response.failed": + // Error events are converted to streamErr by acc.Apply before dispatch. default: } From 2d8ab7561f1e190cea104fceac47b062c2bf85aa Mon Sep 17 00:00:00 2001 From: Sheng Fan Date: Tue, 28 Apr 2026 17:34:21 +0800 Subject: [PATCH 6/8] fix(openai): return error for empty upstream choices --- relay/channel/openai/relay-openai.go | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 2dbeae06a0ba..3e72d29a2451 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -374,14 +374,7 @@ func OaiStreamToNonStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, re } if len(responseChoices) == 0 { - responseChoices = append(responseChoices, dto.OpenAITextResponseChoice{ - Index: 0, - FinishReason: constant.FinishReasonStop, - Message: dto.Message{ - Role: "assistant", - Content: "", - }, - }) + return nil, types.NewOpenAIError(fmt.Errorf("no choices parsed from upstream"), types.ErrorCodeBadResponseBody, http.StatusInternalServerError) } simpleResponse := dto.OpenAITextResponse{ From 597b45334804e35e860bd42f7a8d68ab5b7c3858 Mon Sep 17 00:00:00 2001 From: Sheng Fan Date: Tue, 28 Apr 2026 21:39:53 +0800 Subject: [PATCH 7/8] fix: reject empty responses chat streams --- relay/channel/openai/chat_via_responses.go | 32 ++++-- .../channel/openai/chat_via_responses_test.go | 103 ++++++++++++++++++ 2 files changed, 128 insertions(+), 7 deletions(-) create mode 100644 relay/channel/openai/chat_via_responses_test.go diff --git a/relay/channel/openai/chat_via_responses.go b/relay/channel/openai/chat_via_responses.go index 63c1fbe0d360..5e62d18d0e72 100644 --- a/relay/channel/openai/chat_via_responses.go +++ b/relay/channel/openai/chat_via_responses.go @@ -302,6 +302,13 @@ func (a *responsesStreamAccumulator) applyError() error { return fmt.Errorf("responses stream error") } +func emptyResponsesChatResultError(messageText string, reasoning string, toolCalls []dto.ToolCallResponse) *types.NewAPIError { + if messageText != "" || reasoning != "" || len(toolCalls) > 0 { + return nil + } + return types.NewOpenAIError(fmt.Errorf("responses stream returned empty assistant response"), types.ErrorCodeEmptyResponse, http.StatusInternalServerError) +} + func OaiResponsesToChatHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { if resp == nil || resp.Body == nil { return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError) @@ -416,6 +423,9 @@ func OaiResponsesToChatStreamToNonStreamHandler(c *gin.Context, info *relaycommo if streamErr != nil { return nil, streamErr } + if emptyErr := emptyResponsesChatResultError(messageText, reasoning, toolCalls); emptyErr != nil { + return nil, emptyErr + } msg := dto.Message{ Role: "assistant", @@ -729,12 +739,12 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo // } case "response.output_text.delta": - if !sendStartIfNeeded() { - sr.Stop(streamErr) - return - } - if streamEvent.Delta != "" { + if !sendStartIfNeeded() { + sr.Stop(streamErr) + return + } + delta := streamEvent.Delta chunk := &dto.ChatCompletionsStreamResponse{ Id: acc.responseID, @@ -772,12 +782,17 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo case "response.function_call_arguments.done": case "response.completed": - _, _, usage, toolCalls, resultErr := acc.Result() + messageText, reasoning, usage, toolCalls, resultErr := acc.Result() if resultErr != nil { streamErr = resultErr sr.Stop(streamErr) return } + if emptyErr := emptyResponsesChatResultError(messageText, reasoning, toolCalls); emptyErr != nil { + streamErr = emptyErr + sr.Stop(streamErr) + return + } if !sendStartIfNeeded() { sr.Stop(streamErr) @@ -810,10 +825,13 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo return nil, streamErr } - _, _, usage, toolCalls, streamErr := acc.Result() + messageText, reasoning, usage, toolCalls, streamErr := acc.Result() if streamErr != nil { return nil, streamErr } + if emptyErr := emptyResponsesChatResultError(messageText, reasoning, toolCalls); emptyErr != nil { + return nil, emptyErr + } if !sentStart { if !sendChatChunk(helper.GenerateStartEmptyResponse(acc.responseID, acc.createdAt, acc.model, nil)) { diff --git a/relay/channel/openai/chat_via_responses_test.go b/relay/channel/openai/chat_via_responses_test.go new file mode 100644 index 000000000000..fb7fe82d2a61 --- /dev/null +++ b/relay/channel/openai/chat_via_responses_test.go @@ -0,0 +1,103 @@ +package openai + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/QuantumNous/new-api/constant" + relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/QuantumNous/new-api/types" + + "github.com/gin-gonic/gin" +) + +func init() { + gin.SetMode(gin.TestMode) +} + +func setupResponsesChatTest(t *testing.T, body string) (*gin.Context, *httptest.ResponseRecorder, *http.Response, *relaycommon.RelayInfo) { + t.Helper() + + oldTimeout := constant.StreamingTimeout + constant.StreamingTimeout = 30 + t.Cleanup(func() { + constant.StreamingTimeout = oldTimeout + }) + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(body)), + } + + info := &relaycommon.RelayInfo{ + ChannelMeta: &relaycommon.ChannelMeta{ + UpstreamModelName: "gpt-test", + }, + RelayFormat: types.RelayFormatOpenAI, + DisablePing: true, + } + + return c, recorder, resp, info +} + +func emptyResponsesCompletedEvent() string { + return `{"type":"response.completed","response":{"id":"resp_empty","created_at":123,"model":"gpt-test","usage":{"input_tokens":1,"output_tokens":0,"total_tokens":1}}}` +} + +func TestOaiResponsesToChatStreamToNonStreamHandlerRejectsEmptyAssistantResult(t *testing.T) { + body := "data: " + emptyResponsesCompletedEvent() + "\n\ndata: [DONE]\n\n" + c, recorder, resp, info := setupResponsesChatTest(t, body) + + usage, err := OaiResponsesToChatStreamToNonStreamHandler(c, info, resp) + + if err == nil { + t.Fatal("expected empty assistant result to return an error") + } + if usage != nil { + t.Fatalf("expected nil usage on error, got %#v", usage) + } + if !strings.Contains(err.Error(), "empty assistant response") { + t.Fatalf("expected empty assistant response error, got %q", err.Error()) + } + if err.GetErrorCode() != types.ErrorCodeEmptyResponse { + t.Fatalf("expected empty_response error code, got %q", err.GetErrorCode()) + } + if recorder.Body.Len() != 0 { + t.Fatalf("expected no synthetic chat response, got %q", recorder.Body.String()) + } +} + +func TestOaiResponsesToChatStreamHandlerRejectsEmptyAssistantResult(t *testing.T) { + body := strings.Join([]string{ + `data: {"type":"response.output_text.delta","delta":""}`, + "data: " + emptyResponsesCompletedEvent(), + "data: [DONE]", + "", + }, "\n\n") + c, recorder, resp, info := setupResponsesChatTest(t, body) + + usage, err := OaiResponsesToChatStreamHandler(c, info, resp) + + if err == nil { + t.Fatal("expected empty assistant result to return an error") + } + if usage != nil { + t.Fatalf("expected nil usage on error, got %#v", usage) + } + if !strings.Contains(err.Error(), "empty assistant response") { + t.Fatalf("expected empty assistant response error, got %q", err.Error()) + } + if err.GetErrorCode() != types.ErrorCodeEmptyResponse { + t.Fatalf("expected empty_response error code, got %q", err.GetErrorCode()) + } + if strings.Contains(recorder.Body.String(), "chat.completion.chunk") { + t.Fatalf("expected no synthetic stream chunks, got %q", recorder.Body.String()) + } +} From 56361e3ad7e4b5fa16e6e432c477ad2d07b8522e Mon Sep 17 00:00:00 2001 From: Sheng Fan Date: Tue, 28 Apr 2026 22:06:43 +0800 Subject: [PATCH 8/8] fix: map responses terminal finish reasons --- dto/openai_response.go | 3 +- relay/channel/openai/chat_via_responses.go | 59 +++++++++++++++---- .../channel/openai/chat_via_responses_test.go | 48 +++++++++++++++ service/openai_chat_responses_compat.go | 4 ++ service/openaicompat/responses_to_chat.go | 48 +++++++++++++-- .../openaicompat/responses_to_chat_test.go | 57 ++++++++++++++++++ 6 files changed, 201 insertions(+), 18 deletions(-) create mode 100644 service/openaicompat/responses_to_chat_test.go diff --git a/dto/openai_response.go b/dto/openai_response.go index 0e6b818dbd8b..81cbe7c10aa9 100644 --- a/dto/openai_response.go +++ b/dto/openai_response.go @@ -334,7 +334,8 @@ func (o *OpenAIResponsesResponse) GetSize() string { } type IncompleteDetails struct { - Reasoning string `json:"reasoning"` + Reason string `json:"reason,omitempty"` + Reasoning string `json:"reasoning,omitempty"` } type ResponsesOutput struct { diff --git a/relay/channel/openai/chat_via_responses.go b/relay/channel/openai/chat_via_responses.go index 5e62d18d0e72..9b87a8f3ea77 100644 --- a/relay/channel/openai/chat_via_responses.go +++ b/relay/channel/openai/chat_via_responses.go @@ -52,6 +52,9 @@ type responsesStreamAccumulator struct { reasoningText strings.Builder usageText strings.Builder + terminalStatus string + incompleteReason string + sawToolCall bool toolCallIndexByID map[string]int toolCallNameByID map[string]string @@ -115,6 +118,12 @@ func (a *responsesStreamAccumulator) Apply(ev *dto.ResponsesStreamResponse) erro case "response.function_call_arguments.done": case "response.completed": + a.applyEventTerminalStatus(ev.Type) + a.applyResponseMeta(ev.Response) + a.mergeUsage(ev.Response) + + case "response.incomplete": + a.applyEventTerminalStatus(ev.Type) a.applyResponseMeta(ev.Response) a.mergeUsage(ev.Response) @@ -172,6 +181,39 @@ func (a *responsesStreamAccumulator) applyResponseMeta(response *dto.OpenAIRespo if response.CreatedAt != 0 { a.createdAt = int64(response.CreatedAt) } + if status := strings.TrimSpace(common.JsonRawMessageToString(response.Status)); status != "" { + a.terminalStatus = status + } + if response.IncompleteDetails != nil { + reason := strings.TrimSpace(response.IncompleteDetails.Reason) + if reason == "" { + reason = strings.TrimSpace(response.IncompleteDetails.Reasoning) + } + if reason != "" { + a.incompleteReason = reason + } + } +} + +func (a *responsesStreamAccumulator) applyEventTerminalStatus(eventType string) { + if a == nil { + return + } + switch eventType { + case "response.completed": + a.terminalStatus = "completed" + case "response.incomplete": + a.terminalStatus = "incomplete" + case "response.failed": + a.terminalStatus = "failed" + } +} + +func (a *responsesStreamAccumulator) finishReason(hasToolCalls bool) string { + if a == nil { + return service.MapResponsesTerminalStatusToFinishReason("", "", hasToolCalls) + } + return service.MapResponsesTerminalStatusToFinishReason(a.terminalStatus, a.incompleteReason, hasToolCalls) } func (a *responsesStreamAccumulator) mergeUsage(response *dto.OpenAIResponsesResponse) { @@ -440,10 +482,7 @@ func OaiResponsesToChatStreamToNonStreamHandler(c *gin.Context, info *relaycommo msg.Content = "" } - finishReason := "stop" - if len(toolCalls) > 0 { - finishReason = "tool_calls" - } + finishReason := acc.finishReason(len(toolCalls) > 0) chatResp := &dto.OpenAITextResponse{ Id: acc.responseID, @@ -781,7 +820,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo case "response.function_call_arguments.done": - case "response.completed": + case "response.completed", "response.incomplete": messageText, reasoning, usage, toolCalls, resultErr := acc.Result() if resultErr != nil { streamErr = resultErr @@ -802,10 +841,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo if info.RelayFormat == types.RelayFormatClaude && info.ClaudeConvertInfo != nil { info.ClaudeConvertInfo.Usage = usage } - finishReason := "stop" - if len(toolCalls) > 0 { - finishReason = "tool_calls" - } + finishReason := acc.finishReason(len(toolCalls) > 0) stop := helper.GenerateStopResponse(acc.responseID, acc.createdAt, acc.model, finishReason) if !sendChatChunk(stop) { sr.Stop(streamErr) @@ -842,10 +878,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo if info.RelayFormat == types.RelayFormatClaude && info.ClaudeConvertInfo != nil { info.ClaudeConvertInfo.Usage = usage } - finishReason := "stop" - if len(toolCalls) > 0 { - finishReason = "tool_calls" - } + finishReason := acc.finishReason(len(toolCalls) > 0) stop := helper.GenerateStopResponse(acc.responseID, acc.createdAt, acc.model, finishReason) if !sendChatChunk(stop) { return nil, streamErr diff --git a/relay/channel/openai/chat_via_responses_test.go b/relay/channel/openai/chat_via_responses_test.go index fb7fe82d2a61..0e87abd0f24a 100644 --- a/relay/channel/openai/chat_via_responses_test.go +++ b/relay/channel/openai/chat_via_responses_test.go @@ -51,6 +51,10 @@ func emptyResponsesCompletedEvent() string { return `{"type":"response.completed","response":{"id":"resp_empty","created_at":123,"model":"gpt-test","usage":{"input_tokens":1,"output_tokens":0,"total_tokens":1}}}` } +func incompleteResponsesEvent(reason string) string { + return `{"type":"response.incomplete","response":{"id":"resp_incomplete","created_at":123,"model":"gpt-test","status":"incomplete","incomplete_details":{"reason":"` + reason + `"},"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}` +} + func TestOaiResponsesToChatStreamToNonStreamHandlerRejectsEmptyAssistantResult(t *testing.T) { body := "data: " + emptyResponsesCompletedEvent() + "\n\ndata: [DONE]\n\n" c, recorder, resp, info := setupResponsesChatTest(t, body) @@ -101,3 +105,47 @@ func TestOaiResponsesToChatStreamHandlerRejectsEmptyAssistantResult(t *testing.T t.Fatalf("expected no synthetic stream chunks, got %q", recorder.Body.String()) } } + +func TestOaiResponsesToChatStreamToNonStreamHandlerMapsIncompleteFinishReason(t *testing.T) { + body := strings.Join([]string{ + `data: {"type":"response.output_text.delta","delta":"hello"}`, + "data: " + incompleteResponsesEvent("max_output_tokens"), + "data: [DONE]", + "", + }, "\n\n") + c, recorder, resp, info := setupResponsesChatTest(t, body) + + usage, err := OaiResponsesToChatStreamToNonStreamHandler(c, info, resp) + + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if usage == nil { + t.Fatal("expected usage") + } + if !strings.Contains(recorder.Body.String(), `"finish_reason":"length"`) { + t.Fatalf("expected length finish_reason, got %q", recorder.Body.String()) + } +} + +func TestOaiResponsesToChatStreamHandlerMapsIncompleteFinishReason(t *testing.T) { + body := strings.Join([]string{ + `data: {"type":"response.output_text.delta","delta":"hello"}`, + "data: " + incompleteResponsesEvent("content_filter"), + "data: [DONE]", + "", + }, "\n\n") + c, recorder, resp, info := setupResponsesChatTest(t, body) + + usage, err := OaiResponsesToChatStreamHandler(c, info, resp) + + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if usage == nil { + t.Fatal("expected usage") + } + if !strings.Contains(recorder.Body.String(), `"finish_reason":"content_filter"`) { + t.Fatalf("expected content_filter finish_reason, got %q", recorder.Body.String()) + } +} diff --git a/service/openai_chat_responses_compat.go b/service/openai_chat_responses_compat.go index 2e887386339d..1c8eb324e6c4 100644 --- a/service/openai_chat_responses_compat.go +++ b/service/openai_chat_responses_compat.go @@ -13,6 +13,10 @@ func ResponsesResponseToChatCompletionsResponse(resp *dto.OpenAIResponsesRespons return openaicompat.ResponsesResponseToChatCompletionsResponse(resp, id) } +func MapResponsesTerminalStatusToFinishReason(status string, incompleteReason string, hasToolCalls bool) string { + return openaicompat.MapResponsesTerminalStatusToFinishReason(status, incompleteReason, hasToolCalls) +} + func ExtractOutputTextFromResponses(resp *dto.OpenAIResponsesResponse) string { return openaicompat.ExtractOutputTextFromResponses(resp) } diff --git a/service/openaicompat/responses_to_chat.go b/service/openaicompat/responses_to_chat.go index d1c7473fe8ac..65cfb7e0983b 100644 --- a/service/openaicompat/responses_to_chat.go +++ b/service/openaicompat/responses_to_chat.go @@ -4,6 +4,8 @@ import ( "errors" "strings" + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/constant" "github.com/QuantumNous/new-api/dto" ) @@ -66,10 +68,7 @@ func ResponsesResponseToChatCompletionsResponse(resp *dto.OpenAIResponsesRespons } } - finishReason := "stop" - if len(toolCalls) > 0 { - finishReason = "tool_calls" - } + finishReason := ResponsesResponseFinishReason(resp, len(toolCalls) > 0) msg := dto.Message{ Role: "assistant", @@ -98,6 +97,47 @@ func ResponsesResponseToChatCompletionsResponse(resp *dto.OpenAIResponsesRespons return out, usage, nil } +func ResponsesResponseFinishReason(resp *dto.OpenAIResponsesResponse, hasToolCalls bool) string { + status := "" + incompleteReason := "" + if resp != nil { + status = common.JsonRawMessageToString(resp.Status) + if resp.IncompleteDetails != nil { + incompleteReason = resp.IncompleteDetails.Reason + if incompleteReason == "" { + incompleteReason = resp.IncompleteDetails.Reasoning + } + } + } + return MapResponsesTerminalStatusToFinishReason(status, incompleteReason, hasToolCalls) +} + +func MapResponsesTerminalStatusToFinishReason(status string, incompleteReason string, hasToolCalls bool) string { + normalizedStatus := strings.ToLower(strings.TrimSpace(status)) + normalizedReason := strings.ToLower(strings.TrimSpace(incompleteReason)) + + switch normalizedStatus { + case "incomplete": + switch normalizedReason { + case "content_filter", "safety", "safety_filter": + return constant.FinishReasonContentFilter + case "max_output_tokens", "max_tokens", "length": + return constant.FinishReasonLength + default: + if strings.Contains(normalizedReason, "filter") || strings.Contains(normalizedReason, "safety") { + return constant.FinishReasonContentFilter + } + return constant.FinishReasonLength + } + case "", "completed": + if hasToolCalls { + return constant.FinishReasonToolCalls + } + } + + return constant.FinishReasonStop +} + func ExtractOutputTextFromResponses(resp *dto.OpenAIResponsesResponse) string { if resp == nil || len(resp.Output) == 0 { return "" diff --git a/service/openaicompat/responses_to_chat_test.go b/service/openaicompat/responses_to_chat_test.go new file mode 100644 index 000000000000..11308df4629d --- /dev/null +++ b/service/openaicompat/responses_to_chat_test.go @@ -0,0 +1,57 @@ +package openaicompat + +import ( + "testing" + + "github.com/QuantumNous/new-api/constant" +) + +func TestMapResponsesTerminalStatusToFinishReason(t *testing.T) { + tests := []struct { + name string + status string + incompleteReason string + hasToolCalls bool + want string + }{ + { + name: "completed tool call", + status: "completed", + hasToolCalls: true, + want: constant.FinishReasonToolCalls, + }, + { + name: "incomplete max output tokens", + status: "incomplete", + incompleteReason: "max_output_tokens", + hasToolCalls: true, + want: constant.FinishReasonLength, + }, + { + name: "incomplete content filter", + status: "incomplete", + incompleteReason: "content_filter", + want: constant.FinishReasonContentFilter, + }, + { + name: "missing status falls back to tool calls", + hasToolCalls: true, + want: constant.FinishReasonToolCalls, + }, + { + name: "failed status does not fall back to tool calls", + status: "failed", + hasToolCalls: true, + want: constant.FinishReasonStop, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := MapResponsesTerminalStatusToFinishReason(tt.status, tt.incompleteReason, tt.hasToolCalls) + if got != tt.want { + t.Fatalf("expected %q, got %q", tt.want, got) + } + }) + } +}