mirror of
https://github.com/QuantumNous/new-api.git
synced 2026-09-12 23:30:35 +00:00
feat: support Responses to Chat (#5787)
* fix(openai): harden Chat-to-Responses compatibility Add a shared Responses-to-Chat stream state machine and use it from the OpenAI relay path. Preserve assistant text alongside tool calls, bind tool argument deltas by output_index, map incomplete finish reasons, support reasoning/custom tool events, and buffer upstream SSE for non-stream Chat clients. Add deterministic service tests and relay SSE tests for the conversion path. Related to #5745. * refactor: rename openaicompat to relayconvert for improved clarity * feat(gemini): support responses request conversion * feat: add responses to chat conversion support * fix: harden responses chat conversion edge cases
This commit is contained in:
@@ -14,7 +14,7 @@ import (
|
||||
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/service/openaicompat"
|
||||
"github.com/QuantumNous/new-api/service/relayconvert"
|
||||
"github.com/QuantumNous/new-api/types"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
@@ -78,7 +78,7 @@ func OaiResponsesToChatBufferedStreamHandler(c *gin.Context, info *relaycommon.R
|
||||
}
|
||||
defer service.CloseResponseBodyGracefully(resp)
|
||||
|
||||
accumulator := openaicompat.NewResponsesBufferedAccumulator()
|
||||
accumulator := relayconvert.NewResponsesBufferedAccumulator()
|
||||
var finalResponse *dto.OpenAIResponsesResponse
|
||||
var streamErr *types.NewAPIError
|
||||
|
||||
@@ -184,7 +184,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
|
||||
|
||||
responseId := helper.GetResponseID(c)
|
||||
createAt := time.Now().Unix()
|
||||
state := openaicompat.NewResponsesToChatStreamState(info.UpstreamModelName, false)
|
||||
state := relayconvert.NewResponsesToChatStreamState(info.UpstreamModelName, false)
|
||||
state.ID = responseId
|
||||
state.Created = createAt
|
||||
streamErr := (*types.NewAPIError)(nil)
|
||||
@@ -243,7 +243,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
|
||||
return
|
||||
}
|
||||
|
||||
chunks, err := openaicompat.ResponsesStreamEventToChatChunks(&streamResp, state)
|
||||
chunks, err := relayconvert.ResponsesStreamEventToChatChunks(&streamResp, state)
|
||||
if err != nil {
|
||||
streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponse, http.StatusInternalServerError)
|
||||
sr.Stop(streamErr)
|
||||
@@ -270,7 +270,7 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
|
||||
if info.RelayFormat == types.RelayFormatClaude && info.ClaudeConvertInfo != nil {
|
||||
info.ClaudeConvertInfo.Usage = usage
|
||||
}
|
||||
for _, chunk := range openaicompat.FinalizeResponsesToChatStream(state) {
|
||||
for _, chunk := range relayconvert.FinalizeResponsesToChatStream(state) {
|
||||
if !sendChatChunk(chunk) {
|
||||
return nil, streamErr
|
||||
}
|
||||
|
||||
@@ -116,6 +116,58 @@ func TestOaiResponsesToChatBufferedStreamHandlerReturnsJSONFromSSE(t *testing.T)
|
||||
require.Contains(t, got, `"finish_reason":"tool_calls"`)
|
||||
}
|
||||
|
||||
func TestOaiChatToResponsesStreamHandlerConvertsSSEOrderAndUsage(t *testing.T) {
|
||||
oldMode := gin.Mode()
|
||||
gin.SetMode(gin.TestMode)
|
||||
t.Cleanup(func() { gin.SetMode(oldMode) })
|
||||
|
||||
oldTimeout := constant.StreamingTimeout
|
||||
constant.StreamingTimeout = 30
|
||||
t.Cleanup(func() { constant.StreamingTimeout = oldTimeout })
|
||||
|
||||
body := strings.Join([]string{
|
||||
`data: {"id":"chatcmpl_1","object":"chat.completion.chunk","created":1710000000,"model":"gpt-test","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}`,
|
||||
`data: {"id":"chatcmpl_1","object":"chat.completion.chunk","created":1710000000,"model":"gpt-test","choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":null}]}`,
|
||||
`data: {"id":"chatcmpl_1","object":"chat.completion.chunk","created":1710000000,"model":"gpt-test","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"lookup"}}]},"finish_reason":null}]}`,
|
||||
`data: {"id":"chatcmpl_1","object":"chat.completion.chunk","created":1710000000,"model":"gpt-test","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"q\":\"x\"}"}}]},"finish_reason":null}]}`,
|
||||
`data: {"id":"chatcmpl_1","object":"chat.completion.chunk","created":1710000000,"model":"gpt-test","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}`,
|
||||
`data: {"id":"chatcmpl_1","object":"chat.completion.chunk","created":1710000000,"model":"gpt-test","choices":[],"usage":{"prompt_tokens":2,"completion_tokens":3,"total_tokens":5}}`,
|
||||
`data: [DONE]`,
|
||||
``,
|
||||
}, "\n")
|
||||
|
||||
c, recorder, resp, info := newResponsesChatTestContext(t, body, true)
|
||||
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
|
||||
|
||||
usage, err := OaiChatToResponsesStreamHandler(c, info, resp)
|
||||
require.Nil(t, err)
|
||||
require.NotNil(t, usage)
|
||||
require.Equal(t, 2, usage.PromptTokens)
|
||||
require.Equal(t, 3, usage.CompletionTokens)
|
||||
require.Equal(t, 5, usage.TotalTokens)
|
||||
|
||||
got := recorder.Body.String()
|
||||
require.Equal(t, "text/event-stream", recorder.Header().Get("Content-Type"))
|
||||
require.Contains(t, got, `event: response.created`)
|
||||
require.Contains(t, got, `event: response.output_text.delta`)
|
||||
require.Contains(t, got, `"delta":"hello"`)
|
||||
require.Contains(t, got, `event: response.function_call_arguments.delta`)
|
||||
require.Contains(t, got, `"delta":"{\"q\":\"x\"}"`)
|
||||
require.Contains(t, got, `event: response.completed`)
|
||||
require.Contains(t, got, `"input_tokens":2`)
|
||||
require.Contains(t, got, `"output_tokens":3`)
|
||||
requireOrderedSubstrings(t, got,
|
||||
`event: response.created`,
|
||||
`event: response.output_item.added`,
|
||||
`event: response.output_text.delta`,
|
||||
`event: response.output_item.added`,
|
||||
`event: response.function_call_arguments.delta`,
|
||||
`event: response.output_text.done`,
|
||||
`event: response.function_call_arguments.done`,
|
||||
`event: response.completed`,
|
||||
)
|
||||
}
|
||||
|
||||
func requireOrderedSubstrings(t *testing.T, s string, parts ...string) {
|
||||
t.Helper()
|
||||
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
package openai
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
|
||||
"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/service/relayconvert"
|
||||
"github.com/QuantumNous/new-api/types"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func OaiChatToResponsesHandler(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)
|
||||
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, types.NewOpenAIError(err, types.ErrorCodeReadResponseBodyFailed, http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
var chatResp dto.OpenAITextResponse
|
||||
if err := common.Unmarshal(body, &chatResp); err != nil {
|
||||
return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError)
|
||||
}
|
||||
if oaiError := chatResp.GetOpenAIError(); oaiError != nil && oaiError.Type != "" {
|
||||
return nil, types.WithOpenAIError(*oaiError, resp.StatusCode)
|
||||
}
|
||||
|
||||
responseID := helper.GetResponseID(c)
|
||||
responsesResp, usage, err := service.ChatCompletionsResponseToResponsesResponse(&chatResp, responseID)
|
||||
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())
|
||||
responsesResp.Usage = relayconvert.UsageFromChatUsage(usage)
|
||||
}
|
||||
|
||||
responseBody, err := common.Marshal(responsesResp)
|
||||
if err != nil {
|
||||
return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
service.IOCopyBytesGracefully(c, resp, responseBody)
|
||||
return usage, nil
|
||||
}
|
||||
|
||||
func OaiChatToResponsesStreamHandler(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)
|
||||
state := relayconvert.NewChatToResponsesStreamState(responseID, info.UpstreamModelName)
|
||||
streamErr := (*types.NewAPIError)(nil)
|
||||
|
||||
sendEvent := func(event relayconvert.ChatToResponsesStreamEvent) bool {
|
||||
data, err := common.Marshal(event.Payload)
|
||||
if err != nil {
|
||||
streamErr = types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError)
|
||||
return false
|
||||
}
|
||||
helper.ResponseChunkData(c, dto.ResponsesStreamResponse{Type: event.Type}, string(data))
|
||||
return true
|
||||
}
|
||||
|
||||
helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
|
||||
if streamErr != nil {
|
||||
sr.Stop(streamErr)
|
||||
return
|
||||
}
|
||||
|
||||
var errorResp dto.OpenAITextResponse
|
||||
if err := common.UnmarshalJsonStr(data, &errorResp); err == nil {
|
||||
if oaiError := errorResp.GetOpenAIError(); oaiError != nil && oaiError.Type != "" {
|
||||
streamErr = types.WithOpenAIError(*oaiError, resp.StatusCode)
|
||||
sr.Stop(streamErr)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
var chunk dto.ChatCompletionsStreamResponse
|
||||
if err := common.UnmarshalJsonStr(data, &chunk); err != nil {
|
||||
logger.LogError(c, "failed to unmarshal chat stream response: "+err.Error())
|
||||
sr.Error(err)
|
||||
return
|
||||
}
|
||||
|
||||
events, err := relayconvert.ChatCompletionsStreamChunkToResponsesEvents(&chunk, state)
|
||||
if err != nil {
|
||||
streamErr = types.NewOpenAIError(err, types.ErrorCodeBadResponse, http.StatusInternalServerError)
|
||||
sr.Stop(streamErr)
|
||||
return
|
||||
}
|
||||
for _, event := range events {
|
||||
if !sendEvent(event) {
|
||||
sr.Stop(streamErr)
|
||||
return
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
if streamErr != nil {
|
||||
return nil, streamErr
|
||||
}
|
||||
|
||||
usage := state.Usage
|
||||
if usage == nil || usage.TotalTokens == 0 {
|
||||
usage = service.ResponseText2Usage(c, state.UsageText(), info.UpstreamModelName, info.GetEstimatePromptTokens())
|
||||
state.Usage = relayconvert.UsageFromChatUsage(usage)
|
||||
}
|
||||
|
||||
for _, event := range relayconvert.FinalizeChatCompletionsStreamToResponses(state) {
|
||||
if !sendEvent(event) {
|
||||
return nil, streamErr
|
||||
}
|
||||
}
|
||||
|
||||
return usage, nil
|
||||
}
|
||||
Reference in New Issue
Block a user