fix(gateway): restore public response model aliases

This commit is contained in:
wucm667
2026-09-18 13:31:58 +08:00
parent efe9aab1e4
commit 0920e06a42
7 changed files with 165 additions and 35 deletions
@@ -353,6 +353,7 @@ func (s *OpenAIGatewayService) streamRawChatCompletions(
line = applyOllamaCloudRawChatCompletionsSSELine(account, line)
line = stripEmptyChatToolCallIdentityFromSSELine(line)
line = s.replaceModelInSSELine(line, upstreamModel, originalModel)
writeLine(line)
if line == "" {
if !clientDisconnected && clientOutputStarted {
@@ -523,6 +524,7 @@ func (s *OpenAIGatewayService) bufferRawChatCompletions(
return nil, newGrokMissingUsageFailoverError(c, account, upstreamRequestID)
}
respBody = applyOllamaCloudRawChatCompletionsResponse(account, respBody)
respBody = s.replaceModelInResponseBody(respBody, upstreamModel, originalModel)
if s.responseHeaderFilter != nil {
responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
@@ -6,6 +6,7 @@ import (
"bytes"
"context"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
@@ -1240,3 +1241,39 @@ func largeRawChatCompletionsBody() []byte {
strings.Repeat("x", openAISilentRefusalMinRequestBodyBytes) +
`"}],"stream":true}`)
}
func TestForwardAsRawChatCompletions_RestoresMappedResponseModel(t *testing.T) {
gin.SetMode(gin.TestMode)
for _, stream := range []bool{false, true} {
for _, mapped := range []bool{false, true} {
for _, returned := range []string{"zhipu/glm-5.3", "glm-5.3-alias"} {
t.Run(fmt.Sprintf("stream=%v/mapped=%v/%s", stream, mapped, returned), func(t *testing.T) {
body := []byte(fmt.Sprintf(`{"model":"public","messages":[{"role":"user","content":"hello"}],"stream":%v}`, stream))
payload := `{"id":"chatcmpl_1","model":"` + returned + `","choices":[{"index":0,"delta":{"content":"keep alias"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`
upstreamBody, contentType := payload, "application/json"
if stream {
upstreamBody = "data: " + payload + "\n\ndata: [DONE]\n\n"
contentType = "text/event-stream"
}
upstream := &httpUpstreamRecorder{resp: &http.Response{StatusCode: 200, Header: http.Header{"Content-Type": {contentType}}, Body: io.NopCloser(strings.NewReader(upstreamBody))}}
svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream}
account := rawChatCompletionsTestAccount()
expectedModel, expectedUpstream := returned, "public"
if mapped {
account.Credentials["model_mapping"] = map[string]any{"public": "ZHIPU/GLM-5.3"}
expectedModel = "public"
expectedUpstream = "ZHIPU/GLM-5.3"
}
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body))
result, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "")
require.NoError(t, err)
require.Equal(t, expectedUpstream, gjson.GetBytes(upstream.lastBody, "model").String())
require.Contains(t, rec.Body.String(), strings.Replace(payload, `"model":"`+returned+`"`, `"model":"`+expectedModel+`"`, 1))
require.Equal(t, returned, result.UpstreamResponseModel)
})
}
}
}
}
@@ -1995,7 +1995,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough(
trimmedData := strings.TrimSpace(data)
rawEventType := effectiveOpenAISSEEventType(dataBytes, pendingSSEEventType)
observer.ObserveOpenAI(dataBytes, rawEventType)
if needModelReplace && strings.Contains(data, mappedModel) {
if needModelReplace {
line = s.replaceModelInSSELine(line, mappedModel, originalModel)
if replacedData, replaced := extractOpenAISSEDataLine(line); replaced {
dataBytes = []byte(replacedData)
@@ -794,8 +794,11 @@ func appendOpenAIResponsesRequestPathSuffix(baseURL, suffix string) string {
}
func (s *OpenAIGatewayService) replaceModelInResponseBody(body []byte, fromModel, toModel string) []byte {
// 使用 gjson/sjson 精确替换 model 字段,避免全量 JSON 反序列化
if m := gjson.GetBytes(body, "model"); m.Exists() && m.Str == fromModel {
// A mapped request must retain its public name even when upstream uses an alias.
if fromModel == "" || toModel == "" || fromModel == toModel || !gjson.ValidBytes(body) {
return body
}
if m := gjson.GetBytes(body, "model"); m.Type == gjson.String {
newBody, err := sjson.SetBytes(body, "model", toModel)
if err != nil {
return body
@@ -663,8 +663,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context.
line = "data: " + data
}
// Replace model in response if needed.
// Fast path: most events do not contain model field values.
if needModelReplace && mappedModel != "" && strings.Contains(line, mappedModel) {
if needModelReplace {
line = s.replaceModelInSSELine(line, mappedModel, originalModel)
}
startsClientOutput := forceFlushFailedEvent || openAIStreamDataStartsClientOutput(data, eventType)
@@ -1070,33 +1069,28 @@ func effectiveOpenAISSEEventType(payload []byte, eventType string) string {
}
func (s *OpenAIGatewayService) replaceModelInSSELine(line, fromModel, toModel string) string {
if fromModel == "" || toModel == "" || fromModel == toModel {
return line
}
data, ok := extractOpenAISSEDataLine(line)
if !ok {
if !ok || !gjson.Valid(data) {
return line
}
if data == "" || data == "[DONE]" {
updated := data
// Only protocol model fields are rewritten; text and tool payloads are untouched.
for _, path := range []string{"model", "response.model"} {
if m := gjson.Get(updated, path); m.Type == gjson.String {
var err error
updated, err = sjson.Set(updated, path, toModel)
if err != nil {
return line
}
}
}
if updated == data {
return line
}
// 使用 gjson 精确检查 model 字段,避免全量 JSON 反序列化
if m := gjson.Get(data, "model"); m.Exists() && m.Str == fromModel {
newData, err := sjson.Set(data, "model", toModel)
if err != nil {
return line
}
return "data: " + newData
}
// 检查嵌套的 response.model 字段
if m := gjson.Get(data, "response.model"); m.Exists() && m.Str == fromModel {
newData, err := sjson.Set(data, "response.model", toModel)
if err != nil {
return line
}
return "data: " + newData
}
return line
return "data: " + updated
}
// correctToolCallsInResponseBody 修正响应体中的工具调用
@@ -3324,11 +3324,11 @@ func TestReplaceModelInSSELine(t *testing.T) {
expected: `data: {"type":"response","response":{"id":"resp-1","model":"my-model","output":[]}}`,
},
{
name: "model 不匹配时不替换",
name: "上游别名仍替换",
line: `data: {"id":"chatcmpl-123","model":"gpt-3.5-turbo","choices":[]}`,
from: "gpt-4o",
to: "my-model",
expected: `data: {"id":"chatcmpl-123","model":"gpt-3.5-turbo","choices":[]}`,
expected: `data: {"id":"chatcmpl-123","model":"my-model","choices":[]}`,
},
{
name: "无 model 字段时不替换",
@@ -3394,11 +3394,11 @@ func TestReplaceModelInSSELine(t *testing.T) {
expected: `data: {"id":"abc","object":"chat.completion.chunk","model":"alias","created":1234567890,"choices":[{"index":0,"delta":{"content":"hi"}}]}`,
},
{
name: "顶层优先于嵌套:同时存在两个 model",
name: "同时替换两个 model",
line: `data: {"model":"gpt-4o","response":{"model":"gpt-4o"}}`,
from: "gpt-4o",
to: "replaced",
expected: `data: {"model":"replaced","response":{"model":"gpt-4o"}}`,
expected: `data: {"model":"replaced","response":{"model":"replaced"}}`,
},
}
@@ -3428,11 +3428,11 @@ func TestReplaceModelInSSEBody(t *testing.T) {
expected: "data: {\"model\":\"alias\",\"choices\":[]}\n\ndata: {\"model\":\"alias\",\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\ndata: [DONE]\n",
},
{
name: "无需替换的 body",
name: "上游别名 body",
body: "data: {\"model\":\"gpt-3.5-turbo\"}\n\ndata: [DONE]\n",
from: "gpt-4o",
to: "alias",
expected: "data: {\"model\":\"gpt-3.5-turbo\"}\n\ndata: [DONE]\n",
expected: "data: {\"model\":\"alias\"}\n\ndata: [DONE]\n",
},
{
name: "混合 event 和 data 行",
@@ -3476,11 +3476,11 @@ func TestReplaceModelInResponseBody(t *testing.T) {
expected: `{"id":"chatcmpl-123","model":"alias","choices":[]}`,
},
{
name: "model 不匹配不替换",
name: "上游别名仍替换",
body: `{"id":"chatcmpl-123","model":"gpt-3.5-turbo","choices":[]}`,
from: "gpt-4o",
to: "alias",
expected: `{"id":"chatcmpl-123","model":"gpt-3.5-turbo","choices":[]}`,
expected: `{"id":"chatcmpl-123","model":"alias","choices":[]}`,
},
{
name: "无 model 字段不替换",
@@ -0,0 +1,94 @@
package service
import (
"context"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
func TestMappedResponseModelPreservesOtherData(t *testing.T) {
svc := &OpenAIGatewayService{}
body := `{"model":"alias","text":"mapped alias","tool":{"model":"mapped","arguments":"{\"model\":\"alias\"}"}}`
want := `{"model":"public","text":"mapped alias","tool":{"model":"mapped","arguments":"{\"model\":\"alias\"}"}}`
require.Equal(t, want, string(svc.replaceModelInResponseBody([]byte(body), "mapped", "public")))
require.Equal(t, "data: "+want, svc.replaceModelInSSELine("data: "+body, "mapped", "public"))
for _, body := range []string{
`{"model":"alias",`, `{"model":"alias"} trailing`, `{"model":null}`, `{"model":42}`, `{"model":{}}`, `{"model":[]}`, `{"model":true}`,
`{"text":"alias","tool":{"model":"alias"}}`,
} {
require.Equal(t, body, string(svc.replaceModelInResponseBody([]byte(body), "mapped", "public")))
require.Equal(t, "data: "+body, svc.replaceModelInSSELine("data: "+body, "mapped", "public"))
}
for _, models := range [][2]string{{"same", "same"}, {"", "public"}, {"mapped", ""}} {
body := `{"model":"alias","response":{"model":"alias"}}`
require.Equal(t, body, string(svc.replaceModelInResponseBody([]byte(body), models[0], models[1])))
require.Equal(t, "data: "+body, svc.replaceModelInSSELine("data: "+body, models[0], models[1]))
}
require.Equal(t, `data: {"response":{"model":42}}`, svc.replaceModelInSSELine(`data: {"response":{"model":42}}`, "mapped", "public"))
require.Equal(t, `{"model":"public"}`, string(svc.replaceModelInResponseBody([]byte(`{"model":""}`), "mapped", "public")))
}
// Exercise both streaming processors so substring fast paths cannot bypass the rewrite.
func TestMappedResponseModelForwarding(t *testing.T) {
gin.SetMode(gin.TestMode)
for _, passthrough := range []bool{false, true} {
for _, returned := range []string{"zhipu/glm-5.3", "glm-5.3-alias"} {
for _, mapped := range []string{"ZHIPU/GLM-5.3", "public"} {
for _, kind := range []string{"json", "chat", "responses"} {
name := kind + "/" + returned + "/" + mapped
if passthrough {
name += "/passthrough"
}
t.Run(name, func(t *testing.T) {
model := returned
if mapped != "public" {
model = "public"
}
payload := `{"model":"` + returned + `","choices":[{"delta":{"content":"keep alias","tool_calls":[{"function":{"arguments":"{\"model\":\"alias\"}"}}]}}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`
want := strings.Replace(payload, `"model":"`+returned+`"`, `"model":"`+model+`"`, 1)
contentType := "application/json"
body := payload
if kind == "responses" {
payload = `{"type":"response.completed","response":{"id":"resp_1","model":"` + returned + `","output":[],"usage":{"input_tokens":1,"output_tokens":1}}}`
want = strings.Replace(payload, `"model":"`+returned+`"`, `"model":"`+model+`"`, 1)
}
if kind != "json" {
contentType = "text/event-stream"
body = "data: " + payload + "\n\ndata: [DONE]\n\n"
}
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
resp := &http.Response{StatusCode: 200, Header: http.Header{"Content-Type": {contentType}}, Body: io.NopCloser(strings.NewReader(body))}
svc := &OpenAIGatewayService{}
account := &Account{ID: 1}
var err error
if kind == "json" {
if passthrough {
_, err = svc.handleNonStreamingResponsePassthrough(context.Background(), resp, c, account, "public", mapped)
} else {
_, err = svc.handleNonStreamingResponse(context.Background(), resp, c, account, "public", mapped)
}
} else {
if passthrough {
_, err = svc.handleStreamingResponsePassthrough(context.Background(), resp, c, account, time.Now(), "public", mapped)
} else {
_, err = svc.handleStreamingResponse(context.Background(), resp, c, account, time.Now(), "public", mapped)
}
}
require.NoError(t, err)
require.Contains(t, rec.Body.String(), want)
require.Equal(t, returned, observedUpstreamResponseModel(c))
})
}
}
}
}
}