|
- package openai
-
- import (
- "fmt"
- "io"
- "net/http"
- "strings"
- "time"
-
- "github.com/QuantumNous/new-api/common"
- "github.com/QuantumNous/new-api/dto"
- "github.com/QuantumNous/new-api/logger"
- relaycommon "github.com/QuantumNous/new-api/relay/common"
- "github.com/QuantumNous/new-api/relay/helper"
- "github.com/QuantumNous/new-api/service"
- "github.com/QuantumNous/new-api/types"
-
- "github.com/gin-gonic/gin"
- )
-
- func responsesStreamIndexKey(itemID string, idx *int) string {
- if itemID == "" {
- return ""
- }
- if idx == nil {
- return itemID
- }
- return fmt.Sprintf("%s:%d", itemID, *idx)
- }
-
- func stringDeltaFromPrefix(prev string, next string) string {
- if next == "" {
- return ""
- }
- if prev != "" && strings.HasPrefix(next, prev) {
- return next[len(prev):]
- }
- return next
- }
-
- 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)
- }
-
- defer service.CloseResponseBodyGracefully(resp)
-
- var responsesResp dto.OpenAIResponsesResponse
- body, err := io.ReadAll(resp.Body)
- if err != nil {
- return nil, types.NewOpenAIError(err, types.ErrorCodeReadResponseBodyFailed, http.StatusInternalServerError)
- }
-
- if err := common.Unmarshal(body, &responsesResp); err != nil {
- return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError)
- }
-
- if oaiError := responsesResp.GetOpenAIError(); oaiError != nil && oaiError.Type != "" {
- return nil, types.WithOpenAIError(*oaiError, resp.StatusCode)
- }
-
- chatId := helper.GetResponseID(c)
- chatResp, usage, err := service.ResponsesResponseToChatCompletionsResponse(&responsesResp, chatId)
- if err != nil {
- return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError)
- }
-
- if usage == nil || usage.TotalTokens == 0 {
- text := service.ExtractOutputTextFromResponses(&responsesResp)
- usage = service.ResponseText2Usage(c, text, info.UpstreamModelName, info.GetEstimatePromptTokens())
- chatResp.Usage = *usage
- }
-
- var responseBody []byte
- 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, resp, 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)
- }
-
- defer service.CloseResponseBodyGracefully(resp)
-
- responseId := helper.GetResponseID(c)
- createAt := time.Now().Unix()
- model := info.UpstreamModelName
-
- var (
- usage = &dto.Usage{}
- outputText strings.Builder
- usageText strings.Builder
- sentStart bool
- sentStop bool
- sawToolCall 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)
-
- if info.RelayFormat == types.RelayFormatClaude && info.ClaudeConvertInfo == nil {
- info.ClaudeConvertInfo = &relaycommon.ClaudeConvertInfo{LastMessagesType: relaycommon.LastMessageTypeNone}
- }
-
- sendChatChunk := func(chunk *dto.ChatCompletionsStreamResponse) bool {
- if chunk == nil {
- return true
- }
- if info.RelayFormat == types.RelayFormatOpenAI {
- if err := helper.ObjectData(c, chunk); err != nil {
- streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponse, http.StatusInternalServerError)
- return false
- }
- return true
- }
-
- chunkData, err := common.Marshal(chunk)
- if err != nil {
- streamErr = types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError)
- return false
- }
- if err := HandleStreamFormat(c, info, string(chunkData), false, false); err != nil {
- streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponse, http.StatusInternalServerError)
- return false
- }
- return true
- }
-
- sendStartIfNeeded := func() bool {
- if sentStart {
- return true
- }
- if !sendChatChunk(helper.GenerateStartEmptyResponse(responseId, createAt, model, nil)) {
- return false
- }
- sentStart = true
- return true
- }
-
- //sendReasoningDelta := func(delta string) bool {
- // if delta == "" {
- // return true
- // }
- // if !sendStartIfNeeded() {
- // return false
- // }
- //
- // usageText.WriteString(delta)
- // chunk := &dto.ChatCompletionsStreamResponse{
- // Id: responseId,
- // Object: "chat.completion.chunk",
- // Created: createAt,
- // Model: model,
- // Choices: []dto.ChatCompletionsStreamResponseChoice{
- // {
- // Index: 0,
- // Delta: dto.ChatCompletionsStreamResponseChoiceDelta{
- // ReasoningContent: &delta,
- // },
- // },
- // },
- // }
- // if err := helper.ObjectData(c, chunk); err != nil {
- // streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponse, http.StatusInternalServerError)
- // return false
- // }
- // return true
- //}
-
- sendReasoningSummaryDelta := func(delta string) bool {
- if delta == "" {
- return true
- }
- if needsReasoningSummarySeparator {
- if strings.HasPrefix(delta, "\n\n") {
- needsReasoningSummarySeparator = false
- } else if strings.HasPrefix(delta, "\n") {
- delta = "\n" + delta
- needsReasoningSummarySeparator = false
- } else {
- delta = "\n\n" + delta
- needsReasoningSummarySeparator = false
- }
- }
- if !sendStartIfNeeded() {
- return false
- }
-
- usageText.WriteString(delta)
- chunk := &dto.ChatCompletionsStreamResponse{
- Id: responseId,
- Object: "chat.completion.chunk",
- Created: createAt,
- Model: model,
- Choices: []dto.ChatCompletionsStreamResponseChoice{
- {
- Index: 0,
- Delta: dto.ChatCompletionsStreamResponseChoiceDelta{
- ReasoningContent: &delta,
- },
- },
- },
- }
- if !sendChatChunk(chunk) {
- return false
- }
- hasSentReasoningSummary = true
- return true
- }
-
- sendToolCallDelta := func(callID string, name string, argsDelta string) bool {
- if callID == "" {
- return true
- }
- if outputText.Len() > 0 {
- // Prefer streaming assistant text over tool calls to match non-stream behavior.
- return true
- }
- if !sendStartIfNeeded() {
- return false
- }
-
- idx, ok := toolCallIndexByID[callID]
- if !ok {
- idx = len(toolCallIndexByID)
- toolCallIndexByID[callID] = idx
- }
- if name != "" {
- toolCallNameByID[callID] = name
- }
- if toolCallNameByID[callID] != "" {
- name = toolCallNameByID[callID]
- }
-
- tool := dto.ToolCallResponse{
- ID: callID,
- Type: "function",
- Function: dto.FunctionResponse{
- Arguments: argsDelta,
- },
- }
- tool.SetIndex(idx)
- if name != "" && !toolCallNameSent[callID] {
- tool.Function.Name = name
- toolCallNameSent[callID] = true
- }
-
- chunk := &dto.ChatCompletionsStreamResponse{
- Id: responseId,
- Object: "chat.completion.chunk",
- Created: createAt,
- Model: model,
- Choices: []dto.ChatCompletionsStreamResponseChoice{
- {
- Index: 0,
- Delta: dto.ChatCompletionsStreamResponseChoiceDelta{
- ToolCalls: []dto.ToolCallResponse{tool},
- },
- },
- },
- }
- 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
- }
-
- helper.StreamScannerHandler(c, resp, info, func(data string) bool {
- if streamErr != nil {
- return false
- }
-
- var streamResp dto.ResponsesStreamResponse
- if err := common.UnmarshalJsonStr(data, &streamResp); err != nil {
- logger.LogError(c, "failed to unmarshal responses stream event: "+err.Error())
- return true
- }
-
- 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)
- }
- }
-
- //case "response.reasoning_text.delta":
- //if !sendReasoningDelta(streamResp.Delta) {
- // return false
- //}
-
- //case "response.reasoning_text.done":
-
- case "response.reasoning_summary_text.delta":
- if !sendReasoningSummaryDelta(streamResp.Delta) {
- return false
- }
-
- case "response.reasoning_summary_text.done":
- if hasSentReasoningSummary {
- needsReasoningSummarySeparator = true
- }
-
- //case "response.reasoning_summary_part.added", "response.reasoning_summary_part.done":
- // key := responsesStreamIndexKey(strings.TrimSpace(streamResp.ItemID), streamResp.SummaryIndex)
- // if key == "" || streamResp.Part == nil {
- // break
- // }
- // // Only handle summary text parts, ignore other part types.
- // if streamResp.Part.Type != "" && streamResp.Part.Type != "summary_text" {
- // break
- // }
- // prev := reasoningSummaryTextByKey[key]
- // next := streamResp.Part.Text
- // delta := stringDeltaFromPrefix(prev, next)
- // reasoningSummaryTextByKey[key] = next
- // if !sendReasoningSummaryDelta(delta) {
- // return false
- // }
-
- case "response.output_text.delta":
- if !sendStartIfNeeded() {
- return false
- }
-
- if streamResp.Delta != "" {
- outputText.WriteString(streamResp.Delta)
- usageText.WriteString(streamResp.Delta)
- delta := streamResp.Delta
- chunk := &dto.ChatCompletionsStreamResponse{
- Id: responseId,
- Object: "chat.completion.chunk",
- Created: createAt,
- Model: model,
- Choices: []dto.ChatCompletionsStreamResponseChoice{
- {
- Index: 0,
- Delta: dto.ChatCompletionsStreamResponseChoiceDelta{
- Content: &delta,
- },
- },
- },
- }
- if !sendChatChunk(chunk) {
- return false
- }
- }
-
- 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) {
- return false
- }
-
- 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) {
- return false
- }
-
- 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
- }
- }
- }
-
- if !sendStartIfNeeded() {
- return false
- }
- if !sentStop {
- if info.RelayFormat == types.RelayFormatClaude && info.ClaudeConvertInfo != nil {
- info.ClaudeConvertInfo.Usage = usage
- }
- finishReason := "stop"
- if sawToolCall && outputText.Len() == 0 {
- finishReason = "tool_calls"
- }
- stop := helper.GenerateStopResponse(responseId, createAt, model, finishReason)
- if !sendChatChunk(stop) {
- return false
- }
- 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)
- return false
-
- default:
- }
-
- return true
- })
-
- if streamErr != nil {
- return nil, streamErr
- }
-
- if usage.TotalTokens == 0 {
- usage = service.ResponseText2Usage(c, usageText.String(), info.UpstreamModelName, info.GetEstimatePromptTokens())
- }
-
- if !sentStart {
- if !sendChatChunk(helper.GenerateStartEmptyResponse(responseId, createAt, model, nil)) {
- return nil, streamErr
- }
- }
- if !sentStop {
- if info.RelayFormat == types.RelayFormatClaude && info.ClaudeConvertInfo != nil {
- info.ClaudeConvertInfo.Usage = usage
- }
- finishReason := "stop"
- if sawToolCall && outputText.Len() == 0 {
- finishReason = "tool_calls"
- }
- stop := helper.GenerateStopResponse(responseId, createAt, 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 {
- return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponse, http.StatusInternalServerError)
- }
- }
-
- if info.RelayFormat == types.RelayFormatOpenAI {
- helper.Done(c)
- }
- return usage, nil
- }
|