diff --git a/api/admin/v1/channel.go b/api/admin/v1/channel.go index 8205d27..6e9cdbd 100644 --- a/api/admin/v1/channel.go +++ b/api/admin/v1/channel.go @@ -40,6 +40,7 @@ type ChannelItem struct { Remark string `json:"remark"` IsVIP bool `json:"is_vip"` UseProxy bool `json:"use_proxy"` + UpstreamResponses bool `json:"upstream_responses" dc:"上游使用 OpenAI Responses 协议（/v1/responses）"` SharingThreshold *float64 `json:"sharing_threshold"` PreemptionThreshold *float64 `json:"preemption_threshold"` BorrowingCooldownSeconds *int `json:"borrowing_cooldown_seconds"` @@ -65,6 +66,7 @@ type ChannelCreateReq struct { Remark string `json:"remark" dc:"备注"` IsVIP bool `json:"is_vip" d:"false" dc:"是否VIP专属渠道"` UseProxy bool `json:"use_proxy" d:"false" dc:"启用代理"` + UpstreamResponses bool `json:"upstream_responses" d:"false" dc:"上游使用 OpenAI Responses 协议（/v1/responses），responses 入站直连转发"` SharingThreshold float64 `json:"sharing_threshold" d:"0.6" dc:"普通租户借用阈值"` PreemptionThreshold float64 `json:"preemption_threshold" d:"0.8" dc:"VIP抢占阈值"` BorrowingCooldownSeconds int `json:"borrowing_cooldown_seconds" d:"30" dc:"被抢占后冷却时间(秒)"` @@ -93,6 +95,7 @@ type ChannelUpdateReq struct { StrictCapacity *bool `json:"strict_capacity" dc:"严格容量（fail-closed）"` IsVIP *bool `json:"is_vip" dc:"是否VIP专属渠道"` UseProxy *bool `json:"use_proxy" dc:"启用代理"` + UpstreamResponses *bool `json:"upstream_responses" dc:"上游使用 OpenAI Responses 协议（/v1/responses）"` SharingThreshold *float64 `json:"sharing_threshold" dc:"普通租户借用阈值"` PreemptionThreshold *float64 `json:"preemption_threshold" dc:"VIP抢占阈值"` BorrowingCooldownSeconds *int `json:"borrowing_cooldown_seconds" dc:"被抢占后冷却时间(秒)"` @@ -127,6 +130,7 @@ type ChannelDetailRes struct { Remark string `json:"remark"` IsVIP bool `json:"is_vip"` UseProxy bool `json:"use_proxy"` + UpstreamResponses bool `json:"upstream_responses" dc:"上游使用 OpenAI Responses 协议（/v1/responses）"` SharingThreshold *float64 `json:"sharing_threshold"` PreemptionThreshold *float64 `json:"preemption_threshold"` BorrowingCooldownSeconds *int `json:"borrowing_cooldown_seconds"` diff --git a/internal/logic/admin/channel.go b/internal/logic/admin/channel.go index b36ca6f..2794c4a 100644 --- a/internal/logic/admin/channel.go +++ b/internal/logic/admin/channel.go @@ -139,6 +139,7 @@ func (s *sAdmin) ListChannels(ctx context.Context, req *v1.ChannelListReq) (*v1. Remark: ch.Remark, IsVIP: ch.IsVIP, UseProxy: settings.UseProxy, + UpstreamResponses: settings.UpstreamResponses, SharingThreshold: ch.SharingThreshold, PreemptionThreshold: ch.PreemptionThreshold, BorrowingCooldownSeconds: ch.BorrowingCooldownSeconds, @@ -279,10 +280,15 @@ func (s *sAdmin) CreateChannel(ctx context.Context, req *v1.ChannelCreateReq) (* baseURL = defaultProviderURL(req.Type) } - // Build settings JSON + // Build settings JSON（use_proxy / upstream_responses 协议类开关） settingsJSON := "{}" - if req.UseProxy { - settingsJSON = `{"use_proxy":true}` + if req.UseProxy || req.UpstreamResponses { + settings := relay.DefaultChannelSettings() + settings.UseProxy = req.UseProxy + settings.UpstreamResponses = req.UpstreamResponses + if b, err := json.Marshal(settings); err == nil { + settingsJSON = string(b) + } } tier := req.Tier @@ -380,12 +386,17 @@ func (s *sAdmin) UpdateChannel(ctx context.Context, req *v1.ChannelUpdateReq) (* data.BorrowingCooldownSeconds = *req.BorrowingCooldownSeconds } - // Update use_proxy in settings JSONB - if req.UseProxy != nil { + // Update protocol-related settings in settings JSONB + if req.UseProxy != nil || req.UpstreamResponses != nil { var currentSettings string _ = dao.ChnChannels.Ctx(ctx).Where("id", req.ID).Fields("settings").Scan(¤tSettings) settings := relay.ParseChannelSettings(currentSettings) - settings.UseProxy = *req.UseProxy + if req.UseProxy != nil { + settings.UseProxy = *req.UseProxy + } + if req.UpstreamResponses != nil { + settings.UpstreamResponses = *req.UpstreamResponses + } if settingsJSON, err := json.Marshal(settings); err == nil { data.Settings = string(settingsJSON) } @@ -529,6 +540,7 @@ func (s *sAdmin) GetChannelDetail(ctx context.Context, req *v1.ChannelDetailReq) Remark: ch.Remark, IsVIP: ch.IsVIP, UseProxy: settings.UseProxy, + UpstreamResponses: settings.UpstreamResponses, SharingThreshold: ch.SharingThreshold, PreemptionThreshold: ch.PreemptionThreshold, BorrowingCooldownSeconds: ch.BorrowingCooldownSeconds, diff --git a/internal/logic/relay/apikey_test.go b/internal/logic/relay/apikey_test.go index 4502095..bfb2b48 100644 --- a/internal/logic/relay/apikey_test.go +++ b/internal/logic/relay/apikey_test.go @@ -76,3 +76,14 @@ func TestParseChannelSettings_FullOverride(t *testing.T) { t.Errorf("RetryCount = %d, want 3", s.RetryCount) } } + +func TestParseChannelSettings_UpstreamResponses(t *testing.T) { + // 未配置时默认 false + if s := ParseChannelSettings(`{}`); s.UpstreamResponses { + t.Error("UpstreamResponses should default to false") + } + // 开启后正确解析 + if s := ParseChannelSettings(`{"upstream_responses":true}`); !s.UpstreamResponses { + t.Error("UpstreamResponses should parse true") + } +} diff --git a/relay/channel/openai/adaptor.go b/relay/channel/openai/adaptor.go index 26c2cff..d277f25 100644 --- a/relay/channel/openai/adaptor.go +++ b/relay/channel/openai/adaptor.go @@ -58,7 +58,14 @@ func (a *Adaptor) GetRequestURL(info *common.RelayInfo) (string, error) { case constant.RelayModeRerank: return baseURL + "/v1/rerank", nil case constant.RelayModeResponses, constant.RelayModeResponsesCompact: - // Responses API 请求会被转换为 Chat Completions 格式发送 + // 渠道上游为 Responses 协议：直接打 /v1/responses 端点（响应按 Responses 格式原样透传） + if info.ChannelMeta.Settings.UpstreamResponses { + if constant.RelayMode(info.RelayMode) == constant.RelayModeResponsesCompact { + return baseURL + "/v1/responses/compact", nil + } + return baseURL + "/v1/responses", nil + } + // 兜底（chat-only 上游）：Responses API 请求转换为 Chat Completions 格式发送 return baseURL + "/v1/chat/completions", nil default: return "", fmt.Errorf("unsupported relay mode: %d", info.RelayMode) @@ -127,6 +134,8 @@ func (a *Adaptor) SetupRequestHeader(header http.Header, info *common.RelayInfo) // ConvertRequest 根据入站格式转换请求体为 OpenAI 格式，然后做 OpenAI 特有后处理。 func (a *Adaptor) ConvertRequest(ctx context.Context, info *common.RelayInfo, requestBody []byte) (io.Reader, error) { mode := constant.RelayMode(info.RelayMode) + // responses 入站 + 渠道声明上游为 Responses 协议：保持 Responses 格式直连，不做 chat 转换 + responsesUpstream := info.ChannelMeta.Settings.UpstreamResponses && info.InboundFormat == constant.RelayFormatResponses // Audio/Rerank: OpenAI 原生格式直接透传（不做 replaceModelIfNeeded 和 injectStreamOptions） switch mode { @@ -150,23 +159,33 @@ func (a *Adaptor) ConvertRequest(ctx context.Context, info *common.RelayInfo, re } converted = r case constant.RelayFormatResponses: - r, err := ConvertResponsesToOpenAI(requestBody, info) - if err != nil { - return nil, err + if responsesUpstream { + // 上游为 Responses 协议：保持 Responses 格式直连，仅做模型映射等后处理 + converted = bytes.NewReader(requestBody) + } else { + r, err := ConvertResponsesToOpenAI(requestBody, info) + if err != nil { + return nil, err + } + converted = r } - converted = r default: converted = bytes.NewReader(requestBody) } result := replaceModelIfNeeded(converted, info) - // stream_options 是 Chat Completions 专属字段，GPT Image 使用 stream/partial_images 原生参数 - if info.IsStream && mode != constant.RelayModeImagesGenerations && mode != constant.RelayModeImagesEdits { + // stream_options 是 Chat Completions 专属字段，GPT Image 使用 stream/partial_images 原生参数； + // Responses 上游走原生 stream 字段，不注入 chat 专属 stream_options + if info.IsStream && !responsesUpstream && mode != constant.RelayModeImagesGenerations && mode != constant.RelayModeImagesEdits { result = InjectStreamOptions(result, info) } - // Thinking 后缀路由：注入 reasoning_effort + // Thinking 后缀路由：Responses 上游映射为 reasoning.effort，其余注入 chat 的 reasoning_effort if info.ReasoningEffort != "" { - result = injectReasoningEffort(result, info.ReasoningEffort) + if responsesUpstream { + result = injectResponsesReasoning(result, info.ReasoningEffort) + } else { + result = injectReasoningEffort(result, info.ReasoningEffort) + } } return result, nil @@ -243,6 +262,13 @@ func (a *Adaptor) DoResponse(ctx context.Context, resp *http.Response, info *com } return handleGeminiInboundNonStream(ctx, resp, info, writer) case constant.RelayFormatResponses: + if info.ChannelMeta.Settings.UpstreamResponses { + // 上游为 Responses 协议：原样透传上游 Responses 响应 + if info.IsStream { + return a.handleResponsesUpstreamStream(ctx, resp, info, writer) + } + return a.handleResponsesUpstreamNonStream(ctx, resp, info, writer) + } if info.IsStream { return a.handleResponsesInboundStream(ctx, resp, info, writer) } @@ -445,6 +471,29 @@ func injectReasoningEffort(r io.Reader, effort string) io.Reader { return bytes.NewReader(body) } +// injectResponsesReasoning 为 Responses 请求注入 reasoning.effort（对应 chat 的 reasoning_effort）。 +// Responses 协议无 reasoning_effort 字段，thinking 后缀映射为 reasoning: {"effort": ...}。 +func injectResponsesReasoning(r io.Reader, effort string) io.Reader { + body, err := io.ReadAll(r) + if err != nil { + return r + } + var rawMap map[string]json.RawMessage + if err := json.Unmarshal(body, &rawMap); err != nil { + return bytes.NewReader(body) + } + // 仅在客户端未显式设置 reasoning 时注入 + if _, exists := rawMap["reasoning"]; !exists { + rawMap["reasoning"], _ = json.Marshal(map[string]any{"effort": effort}) + result, err := json.Marshal(rawMap) + if err != nil { + return bytes.NewReader(body) + } + return bytes.NewReader(result) + } + return bytes.NewReader(body) +} + // isUpstreamOpenAIError 检查上游响应是否为标准 OpenAI 错误格式 // 标准格式: {"error": {"type": "...", "message": "...", ...}} func isUpstreamOpenAIError(body []byte) bool { diff --git a/relay/channel/openai/adaptor_responses_test.go b/relay/channel/openai/adaptor_responses_test.go new file mode 100644 index 0000000..24f8ef7 --- /dev/null +++ b/relay/channel/openai/adaptor_responses_test.go @@ -0,0 +1,311 @@ +package openai + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/qianfree/team-api/relay/common" + "github.com/qianfree/team-api/relay/constant" + "github.com/qianfree/team-api/relay/dto" +) + +func responsesUpstreamInfo(relayMode constant.RelayMode, isStream bool) *common.RelayInfo { + return &common.RelayInfo{ + RelayMode: int(relayMode), + IsStream: isStream, + InboundFormat: constant.RelayFormatResponses, + ClientFormat: constant.RelayFormatResponses, + OriginModelName: "gpt-4o", + StreamStatus: common.NewStreamStatus(), + ChannelMeta: &common.ChannelMeta{ + ChannelType: int(constant.ProviderOpenAI), + BaseURL: "https://upstream.example.com", + UpstreamModelName: "gpt-4o-upstream", + IsModelMapped: false, + Settings: common.ChannelSettings{ + UpstreamResponses: true, + }, + }, + } +} + +// TestAdaptor_GetRequestURL_ResponsesUpstream 上游声明 Responses 协议时， +// responses / responses/compact 直达 /v1/responses，chat 仍走 /v1/chat/completions。 +func TestAdaptor_GetRequestURL_ResponsesUpstream(t *testing.T) { + cases := []struct { + mode constant.RelayMode + want string + }{ + {constant.RelayModeResponses, "https://upstream.example.com/v1/responses"}, + {constant.RelayModeResponsesCompact, "https://upstream.example.com/v1/responses/compact"}, + {constant.RelayModeChatCompletions, "https://upstream.example.com/v1/chat/completions"}, + } + for _, c := range cases { + a := &Adaptor{} + info := responsesUpstreamInfo(c.mode, false) + got, err := a.GetRequestURL(info) + if err != nil { + t.Fatalf("GetRequestURL(%v) error: %v", c.mode, err) + } + if got != c.want { + t.Errorf("GetRequestURL(%v) = %q, want %q", c.mode, got, c.want) + } + } +} + +// TestAdaptor_GetRequestURL_ResponsesChatFallback 未声明 Responses 协议时， +// responses 入站仍转换到 /v1/chat/completions（chat-only 上游兜底）。 +func TestAdaptor_GetRequestURL_ResponsesChatFallback(t *testing.T) { + info := responsesUpstreamInfo(constant.RelayModeResponses, false) + info.ChannelMeta.Settings.UpstreamResponses = false + a := &Adaptor{} + got, err := a.GetRequestURL(info) + if err != nil { + t.Fatalf("GetRequestURL error: %v", err) + } + if want := "https://upstream.example.com/v1/chat/completions"; got != want { + t.Errorf("GetRequestURL = %q, want %q", got, want) + } +} + +// TestAdaptor_ConvertRequest_ResponsesUpstream 上游声明 Responses 协议时， +// responses 请求保持 Responses 格式：模型名替换、不注入 chat 专属 stream_options/messages。 +func TestAdaptor_ConvertRequest_ResponsesUpstream(t *testing.T) { + info := responsesUpstreamInfo(constant.RelayModeResponses, true) + info.ChannelMeta.IsModelMapped = true + + body := []byte(`{"model":"gpt-4o","input":"say hi","stream":true}`) + a := &Adaptor{} + out, err := a.ConvertRequest(context.Background(), info, body) + if err != nil { + t.Fatalf("ConvertRequest error: %v", err) + } + raw, _ := io.ReadAll(out) + var m map[string]json.RawMessage + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("bad converted json: %v\n%s", err, raw) + } + if got := string(m["model"]); got != `"gpt-4o-upstream"` { + t.Errorf("model = %s, want gpt-4o-upstream", got) + } + if _, ok := m["input"]; !ok { + t.Error("responses body should keep input field") + } + if _, ok := m["messages"]; ok { + t.Error("should NOT inject chat messages field") + } + if _, ok := m["stream_options"]; ok { + t.Error("should NOT inject chat stream_options for responses upstream") + } +} + +// TestAdaptor_ConvertRequest_ResponsesUpstream_Thinking 上游声明 Responses 协议时， +// thinking 后缀映射为 reasoning.effort，不注入 chat 的 reasoning_effort。 +func TestAdaptor_ConvertRequest_ResponsesUpstream_Thinking(t *testing.T) { + info := responsesUpstreamInfo(constant.RelayModeResponses, false) + info.ReasoningEffort = "high" + + body := []byte(`{"model":"gpt-4o","input":"say hi"}`) + a := &Adaptor{} + out, err := a.ConvertRequest(context.Background(), info, body) + if err != nil { + t.Fatalf("ConvertRequest error: %v", err) + } + raw, _ := io.ReadAll(out) + var m map[string]json.RawMessage + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("bad converted json: %v\n%s", err, raw) + } + if _, ok := m["reasoning_effort"]; ok { + t.Error("should NOT inject chat reasoning_effort for responses upstream") + } + var reasoning struct { + Effort string `json:"effort"` + } + if err := json.Unmarshal(m["reasoning"], &reasoning); err != nil || reasoning.Effort != "high" { + t.Errorf("reasoning.effort = %+v (err=%v), want high", reasoning, err) + } +} + +// TestResponsesUsageToCommon 验证 Responses usage → common.Usage 映射。 +func TestResponsesUsageToCommon(t *testing.T) { + u := responsesUsageToCommon(&dto.ResponsesUsage{ + InputTokens: 100, + OutputTokens: 50, + TotalTokens: 150, + InputTokensDetails: &dto.InputTokenDetails{ + CachedTokens: 40, + AudioTokens: 10, + ImageTokens: 5, + }, + OutputTokenDetails: &dto.OutputTokenDetails{ + ReasoningTokens: 20, + AudioTokens: 3, + }, + }) + if u.PromptTokens != 100 || u.CompletionTokens != 50 || u.TotalTokens != 150 { + t.Errorf("bad token counts: %+v", u) + } + if !u.CacheIncludedInPrompt { + t.Error("OpenAI responses usage should set CacheIncludedInPrompt=true") + } + if u.PromptTokensDetails == nil || u.PromptTokensDetails.CachedTokens != 40 { + t.Errorf("cached tokens not mapped: %+v", u.PromptTokensDetails) + } + if u.CompletionTokenDetails == nil || u.CompletionTokenDetails.ReasoningTokens != 20 { + t.Errorf("reasoning tokens not mapped: %+v", u.CompletionTokenDetails) + } +} + +// TestAdaptor_DoResponse_ResponsesUpstreamNonStream 上游 Responses 响应原样透传，usage 正确解析。 +func TestAdaptor_DoResponse_ResponsesUpstreamNonStream(t *testing.T) { + respBody := `{"id":"resp_1","object":"response","status":"completed","model":"gpt-4o-upstream","output":[],"usage":{"input_tokens":100,"output_tokens":50,"total_tokens":150,"input_tokens_details":{"cached_tokens":40}}}` + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"application/json"}}, + Body: io.NopCloser(strings.NewReader(respBody)), + } + + info := responsesUpstreamInfo(constant.RelayModeResponses, false) + rec := httptest.NewRecorder() + a := &Adaptor{} + usage, err := a.DoResponse(context.Background(), resp, info, rec) + if err != nil { + t.Fatalf("DoResponse error: %v", err) + } + if got := rec.Body.String(); got != respBody { + t.Errorf("response body not passthrough:\n got %s\nwant %s", got, respBody) + } + if usage.PromptTokens != 100 || usage.CompletionTokens != 50 { + t.Errorf("usage = %+v, want prompt=100 completion=50", usage) + } +} + +// TestAdaptor_DoResponse_ResponsesUpstreamStream 上游 Responses SSE 逐行透传，usage 从 completed 事件解析。 +func TestAdaptor_DoResponse_ResponsesUpstreamStream(t *testing.T) { + ss := strings.Join([]string{ + "event: response.output_text.delta", + `data: {"type":"response.output_text.delta","delta":"Hello"}`, + "", + "event: response.completed", + `data: {"type":"response.completed","response":{"id":"resp_1","object":"response","status":"completed","usage":{"input_tokens":10,"output_tokens":20,"total_tokens":30,"input_tokens_details":{"cached_tokens":5}}}}`, + "", + "data: [DONE]", + "", + }, "\n") + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ss)), + } + + info := responsesUpstreamInfo(constant.RelayModeResponses, true) + rec := httptest.NewRecorder() + a := &Adaptor{} + usage, err := a.DoResponse(context.Background(), resp, info, rec) + if err != nil { + t.Fatalf("DoResponse error: %v", err) + } + out := rec.Body.String() + for _, want := range []string{"event: response.output_text.delta", `"delta":"Hello"`, "event: response.completed", "data: [DONE]"} { + if !strings.Contains(out, want) { + t.Errorf("stream output missing %q\n---\n%s", want, out) + } + } + if usage.PromptTokens != 10 || usage.CompletionTokens != 20 || usage.TotalTokens != 30 { + t.Errorf("usage = %+v, want prompt=10 completion=20 total=30", usage) + } +} + +// TestAdaptor_DoResponse_ResponsesUpstreamNonStream_Error 非 200 上游响应： +// 错误体透传给客户端的同时必须返回 upstream error（驱动重试/渠道健康上报）。 +func TestAdaptor_DoResponse_ResponsesUpstreamNonStream_Error(t *testing.T) { + respBody := `{"error":{"type":"rate_limit_error","message":"Rate limit reached"}}` + resp := &http.Response{ + StatusCode: http.StatusTooManyRequests, + Header: http.Header{"Content-Type": {"application/json"}}, + Body: io.NopCloser(strings.NewReader(respBody)), + } + + info := responsesUpstreamInfo(constant.RelayModeResponses, false) + rec := httptest.NewRecorder() + a := &Adaptor{} + usage, err := a.DoResponse(context.Background(), resp, info, rec) + if err == nil { + t.Fatal("non-200 upstream response should return an upstream error, got nil") + } + if usage == nil { + t.Fatal("usage should be non-nil for error path") + } + if rec.Code != http.StatusTooManyRequests { + t.Errorf("status = %d, want 429", rec.Code) + } + if !strings.Contains(rec.Body.String(), "rate_limit_error") { + t.Errorf("error body not passed through: %s", rec.Body.String()) + } +} + +// TestAdaptor_DoResponse_ResponsesUpstreamStream_Error 流式请求但上游在 SSE 开始前 +// 返回非 200：同样必须返回 upstream error。 +func TestAdaptor_DoResponse_ResponsesUpstreamStream_Error(t *testing.T) { + respBody := `{"error":{"type":"invalid_request_error","message":"model not found"}}` + resp := &http.Response{ + StatusCode: http.StatusNotFound, + Header: http.Header{"Content-Type": {"application/json"}}, + Body: io.NopCloser(strings.NewReader(respBody)), + } + + info := responsesUpstreamInfo(constant.RelayModeResponses, true) + rec := httptest.NewRecorder() + a := &Adaptor{} + _, err := a.DoResponse(context.Background(), resp, info, rec) + if err == nil { + t.Fatal("non-200 upstream response should return an upstream error, got nil") + } + if rec.Code != http.StatusNotFound { + t.Errorf("status = %d, want 404", rec.Code) + } +} + +// TestAdaptor_DoResponse_ResponsesUpstreamStream_ModelMapped 模型映射时， +// SSE 事件中的上游模型名须回写为客户端请求的模型名。 +func TestAdaptor_DoResponse_ResponsesUpstreamStream_ModelMapped(t *testing.T) { + ss := strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-4o-upstream"}}`, + "", + "event: response.completed", + `data: {"type":"response.completed","response":{"id":"resp_1","model":"gpt-4o-upstream","usage":{"input_tokens":10,"output_tokens":20,"total_tokens":30}}}`, + "", + }, "\n") + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ss)), + } + + info := responsesUpstreamInfo(constant.RelayModeResponses, true) + info.ChannelMeta.IsModelMapped = true + rec := httptest.NewRecorder() + a := &Adaptor{} + usage, err := a.DoResponse(context.Background(), resp, info, rec) + if err != nil { + t.Fatalf("DoResponse error: %v", err) + } + out := rec.Body.String() + if strings.Contains(out, "gpt-4o-upstream") { + t.Errorf("upstream model name leaked to client:\n%s", out) + } + if !strings.Contains(out, `"model":"gpt-4o"`) { + t.Errorf("client-requested model name missing in stream:\n%s", out) + } + // 模型名回写不得破坏 usage 解析 + if usage.PromptTokens != 10 || usage.CompletionTokens != 20 { + t.Errorf("usage = %+v, want prompt=10 completion=20", usage) + } +} diff --git a/relay/channel/openai/responses.go b/relay/channel/openai/responses.go index 0186512..9dc77bf 100644 --- a/relay/channel/openai/responses.go +++ b/relay/channel/openai/responses.go @@ -11,6 +11,7 @@ import ( "time" "github.com/qianfree/team-api/relay/common" + "github.com/qianfree/team-api/relay/constant" "github.com/qianfree/team-api/relay/dto" "github.com/qianfree/team-api/relay/helper" ) @@ -26,12 +27,15 @@ func (a *Adaptor) handleResponsesInboundNonStream(ctx context.Context, resp *htt return nil, fmt.Errorf("read response body failed: %w", err) } - // 非 200：转换为 Responses 格式的错误 + // 非 200：透传上游错误响应并返回上游错误（驱动重试/渠道健康上报） if resp.StatusCode != http.StatusOK { - writer.Header().Set("Content-Type", "application/json") - writer.WriteHeader(resp.StatusCode) - _, _ = writer.Write(body) - return &common.Usage{}, nil + if isUpstreamOpenAIError(body) { + writeUpstreamErrorResponse(writer, resp.StatusCode, body) + upstreamErr := constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) + upstreamErr.ResponseWritten = true + return &common.Usage{}, upstreamErr + } + return nil, constant.NewUpstreamErrorFromResponse(resp, body) } // 解析 Chat Completions 响应 @@ -161,12 +165,16 @@ func chatCompletionToResponsesResponse(chatResp *dto.ChatCompletionResponse, inf func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) { defer resp.Body.Close() + // 非 200：上游在 SSE 开始前返回错误（非 SSE 体），透传并返回上游错误驱动重试 if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(resp.Body) - writer.Header().Set("Content-Type", "application/json") - writer.WriteHeader(resp.StatusCode) - _, _ = writer.Write(body) - return &common.Usage{}, nil + if isUpstreamOpenAIError(body) { + writeUpstreamErrorResponse(writer, resp.StatusCode, body) + upstreamErr := constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) + upstreamErr.ResponseWritten = true + return &common.Usage{}, upstreamErr + } + return nil, constant.NewUpstreamErrorFromResponse(resp, body) } helper.SetEventStreamHeaders(writer) @@ -614,3 +622,197 @@ func buildResponsesUsageMap(usage *common.Usage) map[string]any { func float64Ptr(v float64) *float64 { return &v } + +// ========== 上游为 Responses 协议：Responses 响应原样透传 ========== + +// responsesUsageToCommon 将 Responses API usage 转换为 common.Usage。 +// OpenAI 原生 API 的 input_tokens 已含缓存 token（cache 是其子集），故 CacheIncludedInPrompt=true。 +func responsesUsageToCommon(u *dto.ResponsesUsage) *common.Usage { + usage := &common.Usage{CacheIncludedInPrompt: true} + if u == nil { + return usage + } + usage.PromptTokens = u.InputTokens + usage.CompletionTokens = u.OutputTokens + usage.TotalTokens = u.TotalTokens + if d := u.InputTokensDetails; d != nil { + usage.PromptTokensDetails = &common.TokenDetails{ + CachedTokens: d.CachedTokens, + TextTokens: d.TextTokens, + AudioTokens: d.AudioTokens, + ImageTokens: d.ImageTokens, + } + } + if d := u.OutputTokenDetails; d != nil { + usage.CompletionTokenDetails = &common.TokenDetails{ + TextTokens: d.TextTokens, + AudioTokens: d.AudioTokens, + ReasoningTokens: d.ReasoningTokens, + AcceptedPredictionTokens: d.AcceptedPredictionTokens, + RejectedPredictionTokens: d.RejectedPredictionTokens, + } + } + return usage +} + +// handleResponsesUpstreamNonStream 上游为 Responses 协议时的非流式响应： +// 解析 usage 后原样透传上游响应体（模型映射时回写客户端请求的模型名）。 +func (a *Adaptor) handleResponsesUpstreamNonStream(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) { + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, constant.NewUpstreamError(resp.StatusCode, "read response body failed", err).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) + } + + // 非 200：透传上游错误响应并返回上游错误（驱动重试/渠道健康上报，同 handleChatNonStreamResponse） + if resp.StatusCode != http.StatusOK { + if isUpstreamOpenAIError(body) { + writeUpstreamErrorResponse(writer, resp.StatusCode, body) + upstreamErr := constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) + upstreamErr.ResponseWritten = true + return &common.Usage{}, upstreamErr + } + return nil, constant.NewUpstreamErrorFromResponse(resp, body) + } + + if info.ChannelMeta.IsModelMapped { + body = helper.ReplaceModelName(body, info.OriginModelName) + } + + writer.Header().Set("Content-Type", "application/json") + writer.WriteHeader(http.StatusOK) + _, _ = writer.Write(body) + + // 解析 usage（计费用），响应体仍透传上游原始内容 + var responsesResp dto.OpenAIResponsesResponse + if err := json.Unmarshal(body, &responsesResp); err == nil { + return responsesUsageToCommon(responsesResp.Usage), nil + } + return &common.Usage{}, nil +} + +// handleResponsesUpstreamStream 上游为 Responses 协议时的流式响应： +// 逐行原样透传 SSE（含 event: 行），从 response.completed / response.done 事件解析 usage。 +func (a *Adaptor) handleResponsesUpstreamStream(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) { + defer resp.Body.Close() + + // 非 200：上游在 SSE 开始前返回错误（非 SSE 体），透传并返回上游错误驱动重试 + if resp.StatusCode != http.StatusOK { + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, constant.NewUpstreamError(resp.StatusCode, "read response body failed", err).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) + } + if isUpstreamOpenAIError(body) { + writeUpstreamErrorResponse(writer, resp.StatusCode, body) + upstreamErr := constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) + upstreamErr.ResponseWritten = true + return &common.Usage{}, upstreamErr + } + return nil, constant.NewUpstreamErrorFromResponse(resp, body) + } + + helper.SetEventStreamHeaders(writer) + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() + + scanner := bufio.NewScanner(resp.Body) + buf := make([]byte, 0, 64*1024) + scanner.Buffer(buf, 10*1024*1024) + + var usage common.Usage + var contentBuilder strings.Builder + + flush := func() { + if f, ok := writer.(http.Flusher); ok { + f.Flush() + } + } + + for scanner.Scan() { + select { + case <-ctx.Done(): + info.StreamStatus.SetEndReason(common.StreamEndReasonClientGone, ctx.Err()) + // 流中断计费兜底：输出缺失按已转发文本 2 字符/token 估算，输入用请求侧估算值补齐 + helper.ApplyInterruptedUsageFallback(info, &usage, contentBuilder.Len()) + return &usage, common.ErrStreamInterrupted + default: + } + + line := scanner.Text() + if line == "" { + fmt.Fprintf(writer, "\n") + flush() + continue + } + // 原样透传 event: 行 + if strings.HasPrefix(line, "event:") { + fmt.Fprintf(writer, "%s\n", line) + continue + } + if !strings.HasPrefix(line, "data:") { + continue + } + data, _ := helper.ExtractSSEData(line) + + if data != "" && data != "[DONE]" { + info.SetFirstResponseTime() + } + + // 解析 data 提取 usage（不影响原样透传） + if data != "[DONE]" { + var streamResp dto.ResponsesStreamResponse + if err := json.Unmarshal([]byte(data), &streamResp); err == nil { + switch streamResp.Type { + case "response.completed", "response.done": + if r := streamResp.Response; r != nil && r.Usage != nil { + if u := responsesUsageToCommon(r.Usage); u != nil { + usage = *u + } + } + case "response.output_text.delta": + contentBuilder.WriteString(streamResp.Delta) + } + } + } + + // 模型映射时回写客户端请求的模型名：response.created / response.completed 等事件携带上游模型名， + // 直连透传前替换（同 chat StreamHandler 的逐行替换） + outLine := line + if info.ChannelMeta.IsModelMapped && data != "" && data != "[DONE]" { + outLine = "data: " + string(helper.ReplaceModelName([]byte(data), info.OriginModelName)) + } + + fmt.Fprintf(writer, "%s\n", outLine) + flush() + + if data == "[DONE]" { + break + } + } + + if err := scanner.Err(); err != nil { + if err != io.EOF && ctx.Err() == nil { + info.StreamStatus.SetEndReason(common.StreamEndReasonError, err) + return &usage, fmt.Errorf("stream scanner error: %w", err) + } + } + + // 估算 usage（正常结束 4 字符/token；异常部分传输 2 字符/token） + if usage.CompletionTokens == 0 { + text := contentBuilder.String() + if len(text) > 0 { + usage.CompletionTokens = helper.EstimateStreamOutputTokens(info, len(text)) + } + } + if usage.TotalTokens == 0 { + usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens + } + usage.CacheIncludedInPrompt = true + + if info.StreamStatus.GetEndReason() == "" { + info.StreamStatus.SetEndReason(common.StreamEndReasonDone, nil) + } + + return &usage, nil +} diff --git a/relay/common/relay_info.go b/relay/common/relay_info.go index d118eac..1466199 100644 --- a/relay/common/relay_info.go +++ b/relay/common/relay_info.go @@ -53,6 +53,7 @@ type ChannelSettings struct { HeaderOverride map[string]any `json:"header_override,omitempty"` // Header 改写规则 ChatCompletionsViaResponses bool `json:"chat_completions_via_responses,omitempty"` // Chat Completions → Responses API 桥接 PassThroughBodyEnabled bool `json:"pass_through_body_enabled,omitempty"` // 直连转发：跳过协议转换，原始请求体直接转发上游 + UpstreamResponses bool `json:"upstream_responses,omitempty"` // 上游使用 OpenAI Responses 协议（/v1/responses），responses 入站直连转发 // System Prompt 注入 SystemPrompt string `json:"system_prompt,omitempty"` // 渠道级系统提示词 diff --git a/relay/handler/passthrough.go b/relay/handler/passthrough.go index 5981d1e..3f4f444 100644 --- a/relay/handler/passthrough.go +++ b/relay/handler/passthrough.go @@ -48,9 +48,13 @@ func canPassThrough(info *common.RelayInfo) bool { } // inboundMatchesChannelNative 判断入站格式是否与渠道原生格式匹配（匹配则可原样直连转发）。 +// - 渠道显式声明上游为 Responses 协议（upstream_responses）：Responses 入站视为原生匹配。 // - 普通渠道：入站格式须等于该渠道的唯一原生格式（helper.ProviderNativeFormat）。 // - 多协议原生透传渠道（New API / Sub2API）：OpenAI/Claude/Gemini 三种格式均视为匹配。 func inboundMatchesChannelNative(info *common.RelayInfo) bool { + if info.ChannelMeta.Settings.UpstreamResponses { + return info.InboundFormat == constant.RelayFormatResponses + } if constant.IsMultiNativeProvider(info.ChannelMeta.ChannelType) { switch info.InboundFormat { case constant.RelayFormatOpenAI, constant.RelayFormatClaude, constant.RelayFormatGemini: diff --git a/relay/handler/passthrough_test.go b/relay/handler/passthrough_test.go index 5cc5b6c..6888861 100644 --- a/relay/handler/passthrough_test.go +++ b/relay/handler/passthrough_test.go @@ -193,6 +193,60 @@ func TestCanPassThrough_MultiNative_ResponsesNotNative(t *testing.T) { } } +// TestCanPassThrough_UpstreamResponses_ResponsesInbound 渠道声明上游为 Responses 协议后， +// Responses 入站在无改写时应原样直连转发。 +func TestCanPassThrough_UpstreamResponses_ResponsesInbound(t *testing.T) { + info := &common.RelayInfo{ + InboundFormat: constant.RelayFormatResponses, + ChannelMeta: &common.ChannelMeta{ + ChannelType: int(constant.ProviderOpenAI), + IsModelMapped: false, + Settings: common.ChannelSettings{ + UpstreamResponses: true, + }, + }, + } + if !canPassThrough(info) { + t.Error("channel declaring UpstreamResponses should pass through Responses inbound natively") + } +} + +// TestCanPassThrough_UpstreamResponses_NonResponsesInbound 渠道声明上游为 Responses 协议时， +// 其它入站格式（如 chat）不属于原生匹配，不得原样直连。 +func TestCanPassThrough_UpstreamResponses_NonResponsesInbound(t *testing.T) { + info := &common.RelayInfo{ + InboundFormat: constant.RelayFormatOpenAI, + ChannelMeta: &common.ChannelMeta{ + ChannelType: int(constant.ProviderOpenAI), + IsModelMapped: false, + Settings: common.ChannelSettings{ + UpstreamResponses: true, + }, + }, + } + if canPassThrough(info) { + t.Error("UpstreamResponses channel should NOT pass through non-Responses inbound natively") + } +} + +// TestCanPassThrough_UpstreamResponses_ModelMappedForcesConversion 上游为 Responses 协议 +// 但配置了模型映射时，仍需经 ConvertRequest 替换模型名，不得原样透传。 +func TestCanPassThrough_UpstreamResponses_ModelMappedForcesConversion(t *testing.T) { + info := &common.RelayInfo{ + InboundFormat: constant.RelayFormatResponses, + ChannelMeta: &common.ChannelMeta{ + ChannelType: int(constant.ProviderOpenAI), + IsModelMapped: true, + Settings: common.ChannelSettings{ + UpstreamResponses: true, + }, + }, + } + if canPassThrough(info) { + t.Error("UpstreamResponses channel with model mapping should NOT pass through") + } +} + // TestRelaykitRequestConverterID 验证请求侧 converter ID 解析，含新供应商 // 与 Ollama 仅 chat 路径启用的 RelayMode 守卫。 func TestRelaykitRequestConverterID(t *testing.T) { diff --git a/relay/handler/relay_handler.go b/relay/handler/relay_handler.go index 5ff6f4a..ea9a741 100644 --- a/relay/handler/relay_handler.go +++ b/relay/handler/relay_handler.go @@ -215,6 +215,14 @@ func validateRelayRequest( maxTokens = mc } } + if moVal, ok := rawRequest["max_output_tokens"]; ok { + // Responses API 用 max_output_tokens 而非 max_tokens/max_completion_tokens + var mo int + _ = json.Unmarshal(moVal, &mo) + if mo > maxTokens { + maxTokens = mo + } + } return &relayValidation{ relayMode: relayMode, diff --git a/web/admin/src/views/ai/ChannelDetailPage.vue b/web/admin/src/views/ai/ChannelDetailPage.vue index f1a528d..9b0a3ee 100644 --- a/web/admin/src/views/ai/ChannelDetailPage.vue +++ b/web/admin/src/views/ai/ChannelDetailPage.vue @@ -68,6 +68,7 @@ const editForm = reactive({ status: 'active', is_vip: false, use_proxy: false, + upstream_responses: false, sharing_threshold: null as number | null, preemption_threshold: null as number | null, borrowing_cooldown_seconds: null as number | null, @@ -89,6 +90,7 @@ function openEditModal() { status: detail.value.status || 'active', is_vip: detail.value.is_vip || false, use_proxy: detail.value.use_proxy || false, + upstream_responses: detail.value.upstream_responses || false, sharing_threshold: detail.value.sharing_threshold ?? null, preemption_threshold: detail.value.preemption_threshold ?? null, borrowing_cooldown_seconds: detail.value.borrowing_cooldown_seconds ?? null, @@ -797,6 +799,12 @@ function formatHeaders(headers: Record): string { + + + + + + diff --git a/web/admin/src/views/ai/ChannelsPage.vue b/web/admin/src/views/ai/ChannelsPage.vue index 73404aa..17634d7 100644 --- a/web/admin/src/views/ai/ChannelsPage.vue +++ b/web/admin/src/views/ai/ChannelsPage.vue @@ -257,10 +257,10 @@ const providerDefaultURLs: Record = { 35: 'https://api.openai.com', } -const form = reactive({ name: '', type: 1, base_url: '', api_key: '', priority: 0, weight: 100, max_concurrency: 100, tier: 'primary', strict_capacity: false, test_model: '', remark: '', status: 'active', is_vip: false, use_proxy: false, sharing_threshold: null as number | null, preemption_threshold: null as number | null, borrowing_cooldown_seconds: null as number | null }) +const form = reactive({ name: '', type: 1, base_url: '', api_key: '', priority: 0, weight: 100, max_concurrency: 100, tier: 'primary', strict_capacity: false, test_model: '', remark: '', status: 'active', is_vip: false, use_proxy: false, upstream_responses: false, sharing_threshold: null as number | null, preemption_threshold: null as number | null, borrowing_cooldown_seconds: null as number | null }) function openCreate() { - Object.assign(form, { name: '', type: 1, base_url: '', api_key: '', priority: 0, weight: 100, max_concurrency: 100, tier: 'primary', strict_capacity: false, test_model: '', remark: '', status: 'active', is_vip: false, use_proxy: false, sharing_threshold: null, preemption_threshold: null, borrowing_cooldown_seconds: null }) + Object.assign(form, { name: '', type: 1, base_url: '', api_key: '', priority: 0, weight: 100, max_concurrency: 100, tier: 'primary', strict_capacity: false, test_model: '', remark: '', status: 'active', is_vip: false, use_proxy: false, upstream_responses: false, sharing_threshold: null, preemption_threshold: null, borrowing_cooldown_seconds: null }) showModal.value = true } @@ -603,6 +603,12 @@ const { exporting, exportFile } = useExport({ + + + + + +