feat(model):支持responses协议

This commit is contained in:
chengqian committed 2026-08-14 16:24:58 +08:00
1 parent ebd2664d67
commit b5a1770018
12 files changed
+696 -26

No files matched your search

+4
View File
@@ -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"`
+18 -6
View File
@@ -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(&currentSettings)
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,
+11
View File
@@ -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")
}
}
+58 -9
View File
@@ -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 {
@@ -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)
}
}
+211 -9
View File
@@ -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
}
+1
View File
@@ -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"` // 渠道级系统提示词
+4
View File
@@ -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:
+54
View File
@@ -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) {
+8
View File
@@ -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,
@@ -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, string>): string {
<ASwitch v-model="editForm.use_proxy" />
</AFormItem>
</ACol>
<ACol :span="12">
<AFormItem label="上游使用 Responses 协议">
<ASwitch v-model="editForm.upstream_responses" />
<template #extra><span class="field-help">上游为 OpenAI Responses API(/v1/responses),responses 入站直连转发</span></template>
</AFormItem>
</ACol>
</ARow>
<AFormItem label="备注"><AInput v-model="editForm.remark" type="textarea" :auto-size="{ minRows: 2, maxRows: 4 }" /></AFormItem>
+8 -2
View File
@@ -257,10 +257,10 @@ const providerDefaultURLs: Record<number, string> = {
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({
<template #extra><span class="field-help">需先在系统设置中配置代理</span></template>
</AFormItem>
</ACol>
<ACol :span="12">
<AFormItem label="上游使用 Responses 协议">
<ASwitch v-model="form.upstream_responses" />
<template #extra><span class="field-help">上游为 OpenAI Responses API(/v1/responses),responses 入站直连转发</span></template>
</AFormItem>
</ACol>
</ARow>
<AFormItem label="备注"><AInput v-model="form.remark" type="textarea" :auto-size="{ minRows: 2, maxRows: 4 }" placeholder="备注(选填)" /></AFormItem>
<!-- VIP 设置 -->