diff --git a/api/admin/v1/channel.go b/api/admin/v1/channel.go index 6e9cdbd..83dd64c 100644 --- a/api/admin/v1/channel.go +++ b/api/admin/v1/channel.go @@ -40,7 +40,6 @@ 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"` @@ -66,7 +65,6 @@ 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:"被抢占后冷却时间(秒)"` @@ -95,7 +93,6 @@ 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:"被抢占后冷却时间(秒)"` @@ -130,7 +127,6 @@ 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"` @@ -177,11 +173,13 @@ type ChannelAbilityBatchReq struct { // AbilityItem 模型能力项 type AbilityItem struct { - ID int64 `json:"id"` - ModelName string `json:"model_name" v:"required" dc:"平台标准模型名"` - UpstreamModel string `json:"upstream_model" dc:"上游实际模型名"` - Enabled bool `json:"enabled" d:"true" dc:"是否启用"` - CostRatio float64 `json:"cost_ratio" d:"1" v:"between:0,100" dc:"成本比例：上游实际价/平台基准价，1.0=等价（参与调度 costFactor）"` + ID int64 `json:"id"` + ModelName string `json:"model_name" v:"required" dc:"平台标准模型名"` + UpstreamModel string `json:"upstream_model" dc:"上游实际模型名"` + Enabled bool `json:"enabled" d:"true" dc:"是否启用"` + CostRatio float64 `json:"cost_ratio" d:"1" v:"between:0,100" dc:"成本比例：上游实际价/平台基准价，1.0=等价（参与调度 costFactor）"` + SupportsResponses bool `json:"supports_responses" d:"false" dc:"支持 OpenAI Responses 协议（/v1/responses 原生直连，responses 入站软偏好）"` + ChatViaResponses bool `json:"chat_via_responses" d:"false" dc:"上游仅有 Responses 协议（responses-only），chat 入站经桥接发送 /v1/responses"` } // ProviderDefaultURLReq 获取供应商默认 URL diff --git a/internal/dao/internal/chn_abilities.go b/internal/dao/internal/chn_abilities.go index c0c4e00..87266af 100644 --- a/internal/dao/internal/chn_abilities.go +++ b/internal/dao/internal/chn_abilities.go @@ -21,26 +21,30 @@ type ChnAbilitiesDao struct { // ChnAbilitiesColumns defines and stores column names for the table chn_abilities. type ChnAbilitiesColumns struct { - Id string // 主键ID - ChannelId string // 关联渠道ID - ModelName string // 平台标准模型名（用户请求使用的模型名） - UpstreamModel string // 上游实际模型名（与平台标准名不同时需要映射，如平台名 gpt-4 → 上游名 gpt-4-0314） - Enabled string // 是否启用该模型能力 - CreatedAt string // 创建时间 - UpdatedAt string // 更新时间 - CostRatio string // 成本比例：该渠道该模型上游实际价/平台基准价，1.0=等价，0.8=八折，参与调度 costFactor 计算（无量纲比例，非金额） + Id string // 主键ID + ChannelId string // 关联渠道ID + ModelName string // 平台标准模型名（用户请求使用的模型名） + UpstreamModel string // 上游实际模型名（与平台标准名不同时需要映射，如平台名 gpt-4 → 上游名 gpt-4-0314） + Enabled string // 是否启用该模型能力 + CreatedAt string // 创建时间 + UpdatedAt string // 更新时间 + CostRatio string // 成本比例：该渠道该模型上游实际价/平台基准价，1.0=等价，0.8=八折，参与调度 costFactor 计算（无量纲比例，非金额） + SupportsResponses string // 模型在该渠道支持 OpenAI Responses 协议（/v1/responses），responses 入站直连转发并获调度软偏好 + ChatViaResponses string // 上游仅有 Responses 协议（responses-only 上游），chat 入站经桥接转换后发送 /v1/responses } // chnAbilitiesColumns holds the columns for the table chn_abilities. var chnAbilitiesColumns = ChnAbilitiesColumns{ - Id: "id", - ChannelId: "channel_id", - ModelName: "model_name", - UpstreamModel: "upstream_model", - Enabled: "enabled", - CreatedAt: "created_at", - UpdatedAt: "updated_at", - CostRatio: "cost_ratio", + Id: "id", + ChannelId: "channel_id", + ModelName: "model_name", + UpstreamModel: "upstream_model", + Enabled: "enabled", + CreatedAt: "created_at", + UpdatedAt: "updated_at", + CostRatio: "cost_ratio", + SupportsResponses: "supports_responses", + ChatViaResponses: "chat_via_responses", } // NewChnAbilitiesDao creates and returns a new DAO object for table data access. diff --git a/internal/dispatchadapter/catalog.go b/internal/dispatchadapter/catalog.go index 5973117..abb9ce7 100644 --- a/internal/dispatchadapter/catalog.go +++ b/internal/dispatchadapter/catalog.go @@ -21,33 +21,37 @@ const ( // ChannelMeta 转发所需的渠道元数据（调度核心不关心，handler 构造 ChannelSelection 用）。 type ChannelMeta struct { - ChannelID int64 - ChannelName string - ChannelType int - BaseURL string - UpstreamModel string - IsModelMapped bool - Settings string - MaxConcurrency int - StrictCapacity bool - Tier string + ChannelID int64 + ChannelName string + ChannelType int + BaseURL string + UpstreamModel string + IsModelMapped bool + Settings string + MaxConcurrency int + StrictCapacity bool + Tier string + SupportsResponses bool // 模型在该渠道支持 /v1/responses（responses 入站直连 + 调度软偏好） + ChatViaResponses bool // responses-only 上游：chat 入站经桥接发送 /v1/responses } // catalogRow 目录加载的原始行（渠道×模型）。 type catalogRow struct { - ChannelID int64 `json:"channel_id"` - ChannelName string `json:"channel_name"` - ChannelType int `json:"channel_type"` - BaseURL string `json:"base_url"` - ModelName string `json:"model_name"` - UpstreamModel string `json:"upstream_model"` - Weight int `json:"weight"` - MaxConcurrency int `json:"max_concurrency"` - Settings string `json:"settings"` - Tier string `json:"tier"` - StrictCapacity bool `json:"strict_capacity"` - CostRatio float64 `json:"cost_ratio"` - CreatedAtMs int64 `json:"created_at_ms"` + ChannelID int64 `json:"channel_id"` + ChannelName string `json:"channel_name"` + ChannelType int `json:"channel_type"` + BaseURL string `json:"base_url"` + ModelName string `json:"model_name"` + UpstreamModel string `json:"upstream_model"` + Weight int `json:"weight"` + MaxConcurrency int `json:"max_concurrency"` + Settings string `json:"settings"` + Tier string `json:"tier"` + StrictCapacity bool `json:"strict_capacity"` + CostRatio float64 `json:"cost_ratio"` + CreatedAtMs int64 `json:"created_at_ms"` + SupportsResponses bool `json:"supports_responses"` + ChatViaResponses bool `json:"chat_via_responses"` } // catalogData 一次完整加载的目录数据。 @@ -294,20 +298,22 @@ func (c *Catalog) Rebuild(ctx context.Context) { } ch := dispatch.Channel{ - ID: row.ChannelID, - Name: row.ChannelName, - Tier: dispatch.Tier(row.Tier), - BaseWeight: float64(row.Weight), - CostRatio: row.CostRatio, - SuccEwma: rt.SuccEwma, - LatEwmaMs: rt.LatEwmaMs, - Inflight: rt.Inflight, - SoftLimit: effectiveSoftLimit(row.MaxConcurrency, rt.Onset429Ewma), - Breaker: rt.Breaker, - ModelBreaker: rt.ModelBreaker, - RampElapsedMs: rampElapsed(now, rt.RecoveredMs, row.CreatedAtMs, rampWindowMs), - KeyIDs: data.keysByChannel[row.ChannelID], - StrictCapacity: row.StrictCapacity, + ID: row.ChannelID, + Name: row.ChannelName, + Tier: dispatch.Tier(row.Tier), + BaseWeight: float64(row.Weight), + CostRatio: row.CostRatio, + SuccEwma: rt.SuccEwma, + LatEwmaMs: rt.LatEwmaMs, + Inflight: rt.Inflight, + SoftLimit: effectiveSoftLimit(row.MaxConcurrency, rt.Onset429Ewma), + Breaker: rt.Breaker, + ModelBreaker: rt.ModelBreaker, + RampElapsedMs: rampElapsed(now, rt.RecoveredMs, row.CreatedAtMs, rampWindowMs), + KeyIDs: data.keysByChannel[row.ChannelID], + StrictCapacity: row.StrictCapacity, + SupportsResponses: row.SupportsResponses, + ChatViaResponses: row.ChatViaResponses, } idx.byModel[row.ModelName] = append(idx.byModel[row.ModelName], ch) idx.channelModels[row.ChannelID] = append(idx.channelModels[row.ChannelID], row.ModelName) @@ -320,16 +326,18 @@ func (c *Catalog) Rebuild(ctx context.Context) { upstream = row.ModelName } idx.meta[row.ChannelID][row.ModelName] = ChannelMeta{ - ChannelID: row.ChannelID, - ChannelName: row.ChannelName, - ChannelType: row.ChannelType, - BaseURL: row.BaseURL, - UpstreamModel: upstream, - IsModelMapped: row.UpstreamModel != "" && row.UpstreamModel != row.ModelName, - Settings: row.Settings, - MaxConcurrency: row.MaxConcurrency, - StrictCapacity: row.StrictCapacity, - Tier: row.Tier, + ChannelID: row.ChannelID, + ChannelName: row.ChannelName, + ChannelType: row.ChannelType, + BaseURL: row.BaseURL, + UpstreamModel: upstream, + IsModelMapped: row.UpstreamModel != "" && row.UpstreamModel != row.ModelName, + Settings: row.Settings, + MaxConcurrency: row.MaxConcurrency, + StrictCapacity: row.StrictCapacity, + Tier: row.Tier, + SupportsResponses: row.SupportsResponses, + ChatViaResponses: row.ChatViaResponses, } if row.StrictCapacity { idx.strict[row.ChannelID] = row.MaxConcurrency @@ -379,6 +387,7 @@ func loadCatalogFromDB(ctx context.Context) (*catalogData, error) { Fields("c.id as channel_id, c.name as channel_name, c.type as channel_type, c.base_url, " + "a.model_name, a.upstream_model, c.weight, c.max_concurrency, c.settings, " + "c.tier, c.strict_capacity, a.cost_ratio, " + + "a.supports_responses, a.chat_via_responses, " + "(EXTRACT(EPOCH FROM c.created_at) * 1000)::bigint as created_at_ms"). Scan(&rows) if err != nil { diff --git a/internal/logic/admin/channel.go b/internal/logic/admin/channel.go index 2794c4a..727a49e 100644 --- a/internal/logic/admin/channel.go +++ b/internal/logic/admin/channel.go @@ -139,7 +139,6 @@ 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, @@ -241,12 +240,15 @@ func (s *sAdmin) CloneChannel(ctx context.Context, req *v1.ChannelCloneReq) (*v1 // 克隆 abilities var abilities []struct { - ModelName string `json:"model_name"` - UpstreamModel string `json:"upstream_model"` - Enabled bool `json:"enabled"` + ModelName string `json:"model_name"` + UpstreamModel string `json:"upstream_model"` + Enabled bool `json:"enabled"` + CostRatio float64 `json:"cost_ratio"` + SupportsResponses bool `json:"supports_responses"` + ChatViaResponses bool `json:"chat_via_responses"` } err = dao.ChnAbilities.Ctx(ctx). - Fields("model_name, upstream_model, enabled"). + Fields("model_name, upstream_model, enabled, cost_ratio, supports_responses, chat_via_responses"). Where("channel_id", req.ID). Scan(&abilities) if err != nil { @@ -254,10 +256,13 @@ func (s *sAdmin) CloneChannel(ctx context.Context, req *v1.ChannelCloneReq) (*v1 } for _, ab := range abilities { _, err := dao.ChnAbilities.Ctx(ctx).Insert(do.ChnAbilities{ - ChannelId: newID, - ModelName: ab.ModelName, - UpstreamModel: ab.UpstreamModel, - Enabled: ab.Enabled, + ChannelId: newID, + ModelName: ab.ModelName, + UpstreamModel: ab.UpstreamModel, + Enabled: ab.Enabled, + CostRatio: ab.CostRatio, + SupportsResponses: ab.SupportsResponses, + ChatViaResponses: ab.ChatViaResponses, }) if err != nil { g.Log().Warningf(ctx, "clone ability %s for channel %d failed: %v", ab.ModelName, newID, err) @@ -280,15 +285,10 @@ func (s *sAdmin) CreateChannel(ctx context.Context, req *v1.ChannelCreateReq) (* baseURL = defaultProviderURL(req.Type) } - // Build settings JSON（use_proxy / upstream_responses 协议类开关） + // Build settings JSON settingsJSON := "{}" - 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) - } + if req.UseProxy { + settingsJSON = `{"use_proxy":true}` } tier := req.Tier @@ -386,17 +386,12 @@ func (s *sAdmin) UpdateChannel(ctx context.Context, req *v1.ChannelUpdateReq) (* data.BorrowingCooldownSeconds = *req.BorrowingCooldownSeconds } - // Update protocol-related settings in settings JSONB - if req.UseProxy != nil || req.UpstreamResponses != nil { + // Update use_proxy in settings JSONB + if req.UseProxy != nil { var currentSettings string _ = dao.ChnChannels.Ctx(ctx).Where("id", req.ID).Fields("settings").Scan(¤tSettings) settings := relay.ParseChannelSettings(currentSettings) - if req.UseProxy != nil { - settings.UseProxy = *req.UseProxy - } - if req.UpstreamResponses != nil { - settings.UpstreamResponses = *req.UpstreamResponses - } + settings.UseProxy = *req.UseProxy if settingsJSON, err := json.Marshal(settings); err == nil { data.Settings = string(settingsJSON) } @@ -540,7 +535,6 @@ 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, @@ -595,11 +589,13 @@ func (s *sAdmin) SetChannelAbilities(ctx context.Context, req *v1.ChannelAbility costRatio = 1.0 } if _, err := dao.ChnAbilities.Ctx(ctx).Insert(do.ChnAbilities{ - ChannelId: req.ChannelID, - ModelName: ab.ModelName, - UpstreamModel: ab.UpstreamModel, - Enabled: ab.Enabled, - CostRatio: costRatio, + ChannelId: req.ChannelID, + ModelName: ab.ModelName, + UpstreamModel: ab.UpstreamModel, + Enabled: ab.Enabled, + CostRatio: costRatio, + SupportsResponses: ab.SupportsResponses, + ChatViaResponses: ab.ChatViaResponses, }); err != nil { return err } @@ -660,11 +656,13 @@ func (s *sAdmin) GetChannelKeys(ctx context.Context, req *v1.ChannelKeyListReq) // GetChannelAbilities 获取渠道模型能力列表 func (s *sAdmin) GetChannelAbilities(ctx context.Context, req *v1.ChannelAbilitiesGetReq) (*v1.ChannelAbilitiesGetRes, error) { var abilities []struct { - ID int64 `json:"id"` - ModelName string `json:"model_name"` - UpstreamModel string `json:"upstream_model"` - Enabled bool `json:"enabled"` - CostRatio float64 `json:"cost_ratio"` + ID int64 `json:"id"` + ModelName string `json:"model_name"` + UpstreamModel string `json:"upstream_model"` + Enabled bool `json:"enabled"` + CostRatio float64 `json:"cost_ratio"` + SupportsResponses bool `json:"supports_responses"` + ChatViaResponses bool `json:"chat_via_responses"` } err := dao.ChnAbilities.Ctx(ctx). @@ -678,11 +676,13 @@ func (s *sAdmin) GetChannelAbilities(ctx context.Context, req *v1.ChannelAbiliti list := make([]v1.AbilityItem, len(abilities)) for i, a := range abilities { list[i] = v1.AbilityItem{ - ID: a.ID, - ModelName: a.ModelName, - UpstreamModel: a.UpstreamModel, - Enabled: a.Enabled, - CostRatio: a.CostRatio, + ID: a.ID, + ModelName: a.ModelName, + UpstreamModel: a.UpstreamModel, + Enabled: a.Enabled, + CostRatio: a.CostRatio, + SupportsResponses: a.SupportsResponses, + ChatViaResponses: a.ChatViaResponses, } } diff --git a/internal/logic/relay/apikey_test.go b/internal/logic/relay/apikey_test.go index bfb2b48..a4612d0 100644 --- a/internal/logic/relay/apikey_test.go +++ b/internal/logic/relay/apikey_test.go @@ -77,13 +77,11 @@ func TestParseChannelSettings_FullOverride(t *testing.T) { } } -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") +func TestParseChannelSettings_IgnoresUnknownProtoKeys(t *testing.T) { + // 协议能力已迁移到 chn_abilities（supports_responses / chat_via_responses）， + // 渠道 settings 不再有协议类字段：历史遗留的 upstream_responses 键应被静默忽略 + s := ParseChannelSettings(`{"upstream_responses":true,"chat_completions_via_responses":true}`) + if s.PassThroughBodyEnabled { + t.Error("unknown protocol keys must not map to any channel setting") } } diff --git a/internal/logic/relay/provider.go b/internal/logic/relay/provider.go index ff18f0c..180495e 100644 --- a/internal/logic/relay/provider.go +++ b/internal/logic/relay/provider.go @@ -993,6 +993,8 @@ func (p *DataProviderImpl) MaterializeSelection(ctx context.Context, channelID, IsModelMapped: meta.IsModelMapped, MaxConcurrency: meta.MaxConcurrency, Settings: ParseChannelSettings(meta.Settings), + SupportsResponses: meta.SupportsResponses, + ChatViaResponses: meta.ChatViaResponses, }, nil } diff --git a/internal/model/do/chn_abilities.go b/internal/model/do/chn_abilities.go index 762b9b9..a844dd3 100644 --- a/internal/model/do/chn_abilities.go +++ b/internal/model/do/chn_abilities.go @@ -11,13 +11,15 @@ import ( // ChnAbilities is the golang structure of table chn_abilities for DAO operations like Where/Data. type ChnAbilities struct { - g.Meta `orm:"table:chn_abilities, do:true"` - Id any // 主键ID - ChannelId any // 关联渠道ID - ModelName any // 平台标准模型名（用户请求使用的模型名） - UpstreamModel any // 上游实际模型名（与平台标准名不同时需要映射，如平台名 gpt-4 → 上游名 gpt-4-0314） - Enabled any // 是否启用该模型能力 - CreatedAt *gtime.Time // 创建时间 - UpdatedAt *gtime.Time // 更新时间 - CostRatio any // 成本比例：该渠道该模型上游实际价/平台基准价，1.0=等价，0.8=八折，参与调度 costFactor 计算（无量纲比例，非金额） + g.Meta `orm:"table:chn_abilities, do:true"` + Id any // 主键ID + ChannelId any // 关联渠道ID + ModelName any // 平台标准模型名（用户请求使用的模型名） + UpstreamModel any // 上游实际模型名（与平台标准名不同时需要映射，如平台名 gpt-4 → 上游名 gpt-4-0314） + Enabled any // 是否启用该模型能力 + CreatedAt *gtime.Time // 创建时间 + UpdatedAt *gtime.Time // 更新时间 + CostRatio any // 成本比例：该渠道该模型上游实际价/平台基准价，1.0=等价，0.8=八折，参与调度 costFactor 计算（无量纲比例，非金额） + SupportsResponses any // 模型在该渠道支持 OpenAI Responses 协议（/v1/responses），responses 入站直连转发并获调度软偏好 + ChatViaResponses any // 上游仅有 Responses 协议（responses-only 上游），chat 入站经桥接转换后发送 /v1/responses } diff --git a/internal/model/entity/chn_abilities.go b/internal/model/entity/chn_abilities.go index 7336dad..57708a2 100644 --- a/internal/model/entity/chn_abilities.go +++ b/internal/model/entity/chn_abilities.go @@ -11,12 +11,14 @@ import ( // ChnAbilities is the golang structure for table chn_abilities. type ChnAbilities struct { - Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID - ChannelId int64 `json:"channel_id" orm:"channel_id" description:"关联渠道ID"` // 关联渠道ID - ModelName string `json:"model_name" orm:"model_name" description:"平台标准模型名（用户请求使用的模型名）"` // 平台标准模型名（用户请求使用的模型名） - UpstreamModel string `json:"upstream_model" orm:"upstream_model" description:"上游实际模型名（与平台标准名不同时需要映射，如平台名 gpt-4 → 上游名 gpt-4-0314）"` // 上游实际模型名（与平台标准名不同时需要映射，如平台名 gpt-4 → 上游名 gpt-4-0314） - Enabled bool `json:"enabled" orm:"enabled" description:"是否启用该模型能力"` // 是否启用该模型能力 - CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 - UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 - CostRatio decimal.Decimal `json:"cost_ratio" orm:"cost_ratio" description:"成本比例：该渠道该模型上游实际价/平台基准价，1.0=等价，0.8=八折，参与调度 costFactor 计算（无量纲比例，非金额）"` // 成本比例：该渠道该模型上游实际价/平台基准价，1.0=等价，0.8=八折，参与调度 costFactor 计算（无量纲比例，非金额） + Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID + ChannelId int64 `json:"channel_id" orm:"channel_id" description:"关联渠道ID"` // 关联渠道ID + ModelName string `json:"model_name" orm:"model_name" description:"平台标准模型名（用户请求使用的模型名）"` // 平台标准模型名（用户请求使用的模型名） + UpstreamModel string `json:"upstream_model" orm:"upstream_model" description:"上游实际模型名（与平台标准名不同时需要映射，如平台名 gpt-4 → 上游名 gpt-4-0314）"` // 上游实际模型名（与平台标准名不同时需要映射，如平台名 gpt-4 → 上游名 gpt-4-0314） + Enabled bool `json:"enabled" orm:"enabled" description:"是否启用该模型能力"` // 是否启用该模型能力 + CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 + UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 + CostRatio decimal.Decimal `json:"cost_ratio" orm:"cost_ratio" description:"成本比例：该渠道该模型上游实际价/平台基准价，1.0=等价，0.8=八折，参与调度 costFactor 计算（无量纲比例，非金额）"` // 成本比例：该渠道该模型上游实际价/平台基准价，1.0=等价，0.8=八折，参与调度 costFactor 计算（无量纲比例，非金额） + SupportsResponses bool `json:"supports_responses" orm:"supports_responses" description:"模型在该渠道支持 OpenAI Responses 协议（/v1/responses），responses 入站直连转发并获调度软偏好"` // 模型在该渠道支持 OpenAI Responses 协议（/v1/responses），responses 入站直连转发并获调度软偏好 + ChatViaResponses bool `json:"chat_via_responses" orm:"chat_via_responses" description:"上游仅有 Responses 协议（responses-only 上游），chat 入站经桥接转换后发送 /v1/responses"` // 上游仅有 Responses 协议（responses-only 上游），chat 入站经桥接转换后发送 /v1/responses } diff --git a/migrations/000015_chn_abilities_protocol_flags.sql b/migrations/000015_chn_abilities_protocol_flags.sql new file mode 100644 index 0000000..22be6f2 --- /dev/null +++ b/migrations/000015_chn_abilities_protocol_flags.sql @@ -0,0 +1,17 @@ +-- +goose Up +-- 渠道×模型级协议能力标记（配合 Responses 端点支持）： +-- supports_responses：模型在该渠道支持 OpenAI Responses 协议（/v1/responses）。 +-- responses 入站请求命中时原样直连上游 /v1/responses；调度时对 responses 入站软偏好此类渠道。 +-- chat_via_responses：上游仅有 Responses 协议（responses-only，如 Codex 类中转）， +-- chat 入站请求经 chat→Responses 桥接发送到 /v1/responses。 +-- 两项均为渠道×模型粒度（非渠道级）：同一渠道的不同模型协议支持可能不同。 + +ALTER TABLE chn_abilities ADD COLUMN supports_responses BOOLEAN DEFAULT false NOT NULL; +COMMENT ON COLUMN chn_abilities.supports_responses IS '模型在该渠道支持 OpenAI Responses 协议（/v1/responses），responses 入站直连转发并获调度软偏好'; + +ALTER TABLE chn_abilities ADD COLUMN chat_via_responses BOOLEAN DEFAULT false NOT NULL; +COMMENT ON COLUMN chn_abilities.chat_via_responses IS '上游仅有 Responses 协议（responses-only 上游），chat 入站经桥接转换后发送 /v1/responses'; + +-- +goose Down +ALTER TABLE chn_abilities DROP COLUMN IF EXISTS supports_responses; +ALTER TABLE chn_abilities DROP COLUMN IF EXISTS chat_via_responses; diff --git a/relay/channel/openai/adaptor.go b/relay/channel/openai/adaptor.go index d277f25..6f2c3b8 100644 --- a/relay/channel/openai/adaptor.go +++ b/relay/channel/openai/adaptor.go @@ -58,8 +58,8 @@ func (a *Adaptor) GetRequestURL(info *common.RelayInfo) (string, error) { case constant.RelayModeRerank: return baseURL + "/v1/rerank", nil case constant.RelayModeResponses, constant.RelayModeResponsesCompact: - // 渠道上游为 Responses 协议：直接打 /v1/responses 端点（响应按 Responses 格式原样透传） - if info.ChannelMeta.Settings.UpstreamResponses { + // 上游原生支持 Responses 协议：直接打 /v1/responses 端点（响应按 Responses 格式原样透传） + if info.ChannelMeta.UpstreamSpeaksResponses() { if constant.RelayMode(info.RelayMode) == constant.RelayModeResponsesCompact { return baseURL + "/v1/responses/compact", nil } @@ -134,8 +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 + // responses 入站 + 上游原生支持 Responses 协议：保持 Responses 格式直连，不做 chat 转换 + responsesUpstream := info.ChannelMeta.UpstreamSpeaksResponses() && info.InboundFormat == constant.RelayFormatResponses // Audio/Rerank: OpenAI 原生格式直接透传（不做 replaceModelIfNeeded 和 injectStreamOptions） switch mode { @@ -172,6 +172,25 @@ func (a *Adaptor) ConvertRequest(ctx context.Context, info *common.RelayInfo, re default: converted = bytes.NewReader(requestBody) } + + // responses-only 上游桥接：chat 请求体转换为 Responses 格式发送 /v1/responses。 + // 转换器自行处理模型映射，并读取 chat 体的 reasoning_effort 映射为 reasoning.effort + // （thinking 后缀先注入 chat 体再进转换器），故此分支直接返回、不走后续 chat 后处理。 + if info.UseResponsesAPI && mode == constant.RelayModeChatCompletions && !responsesUpstream { + if info.ReasoningEffort != "" { + converted = injectReasoningEffort(converted, info.ReasoningEffort) + } + body, err := io.ReadAll(converted) + if err != nil { + return nil, fmt.Errorf("read chat body for responses bridge failed: %w", err) + } + r, err := ConvertOpenAIToResponses(body, info) + if err != nil { + return nil, err + } + return bytes.NewReader(r), nil + } + result := replaceModelIfNeeded(converted, info) // stream_options 是 Chat Completions 专属字段，GPT Image 使用 stream/partial_images 原生参数； // Responses 上游走原生 stream 字段，不注入 chat 专属 stream_options @@ -262,7 +281,7 @@ 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 { + if info.ChannelMeta.UpstreamSpeaksResponses() { // 上游为 Responses 协议：原样透传上游 Responses 响应 if info.IsStream { return a.handleResponsesUpstreamStream(ctx, resp, info, writer) diff --git a/relay/channel/openai/adaptor_responses_test.go b/relay/channel/openai/adaptor_responses_test.go index 24f8ef7..1077818 100644 --- a/relay/channel/openai/adaptor_responses_test.go +++ b/relay/channel/openai/adaptor_responses_test.go @@ -3,6 +3,7 @@ package openai import ( "context" "encoding/json" + "errors" "io" "net/http" "net/http/httptest" @@ -27,9 +28,8 @@ func responsesUpstreamInfo(relayMode constant.RelayMode, isStream bool) *common. BaseURL: "https://upstream.example.com", UpstreamModelName: "gpt-4o-upstream", IsModelMapped: false, - Settings: common.ChannelSettings{ - UpstreamResponses: true, - }, + + SupportsResponses: true, }, } } @@ -62,7 +62,8 @@ func TestAdaptor_GetRequestURL_ResponsesUpstream(t *testing.T) { // responses 入站仍转换到 /v1/chat/completions（chat-only 上游兜底）。 func TestAdaptor_GetRequestURL_ResponsesChatFallback(t *testing.T) { info := responsesUpstreamInfo(constant.RelayModeResponses, false) - info.ChannelMeta.Settings.UpstreamResponses = false + info.ChannelMeta.SupportsResponses = false + info.ChannelMeta.ChatViaResponses = false a := &Adaptor{} got, err := a.GetRequestURL(info) if err != nil { @@ -73,6 +74,40 @@ func TestAdaptor_GetRequestURL_ResponsesChatFallback(t *testing.T) { } } +// TestAdaptor_GetRequestURL_ChatViaResponses responses-only 渠道（chat_via_responses）： +// responses 入站同样直连 /v1/responses（上游本来就说 Responses 协议）， +// chat 入站经 UseResponsesAPI 桥接也打 /v1/responses。 +func TestAdaptor_GetRequestURL_ChatViaResponses(t *testing.T) { + a := &Adaptor{} + + // responses 入站：仅勾 chat_via_responses（未勾 supports_responses）也应直连 + info := responsesUpstreamInfo(constant.RelayModeResponses, false) + info.ChannelMeta.SupportsResponses = false + info.ChannelMeta.ChatViaResponses = true + got, err := a.GetRequestURL(info) + if err != nil { + t.Fatalf("GetRequestURL error: %v", err) + } + if want := "https://upstream.example.com/v1/responses"; got != want { + t.Errorf("responses inbound on chat_via_responses channel: URL = %q, want %q", got, want) + } + + // chat 入站 + 桥接标志：URL 走 /v1/responses + chatInfo := responsesUpstreamInfo(constant.RelayModeChatCompletions, false) + chatInfo.InboundFormat = constant.RelayFormatOpenAI + chatInfo.ClientFormat = constant.RelayFormatOpenAI + chatInfo.ChannelMeta.SupportsResponses = false + chatInfo.ChannelMeta.ChatViaResponses = true + chatInfo.UseResponsesAPI = true + got, err = a.GetRequestURL(chatInfo) + if err != nil { + t.Fatalf("GetRequestURL error: %v", err) + } + if want := "https://upstream.example.com/v1/responses"; got != want { + t.Errorf("chat inbound via bridge: URL = %q, want %q", got, want) + } +} + // TestAdaptor_ConvertRequest_ResponsesUpstream 上游声明 Responses 协议时， // responses 请求保持 Responses 格式：模型名替换、不注入 chat 专属 stream_options/messages。 func TestAdaptor_ConvertRequest_ResponsesUpstream(t *testing.T) { @@ -309,3 +344,202 @@ func TestAdaptor_DoResponse_ResponsesUpstreamStream_ModelMapped(t *testing.T) { t.Errorf("usage = %+v, want prompt=10 completion=20", usage) } } + +// TestAdaptor_ConvertRequest_ChatViaResponsesBridge responses-only 上游桥接： +// chat 请求体转换为 Responses 格式（messages→input），thinking 后缀映射 reasoning.effort， +// 不注入 chat 专属 stream_options。 +func TestAdaptor_ConvertRequest_ChatViaResponsesBridge(t *testing.T) { + info := responsesUpstreamInfo(constant.RelayModeChatCompletions, true) + info.InboundFormat = constant.RelayFormatOpenAI + info.ClientFormat = constant.RelayFormatOpenAI + info.OriginModelName = "gpt-4o" + info.ChannelMeta.SupportsResponses = false + info.ChannelMeta.ChatViaResponses = true + info.UseResponsesAPI = true + + body := []byte(`{"model":"gpt-4o","messages":[{"role":"user","content":"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 bridged json: %v\n%s", err, raw) + } + if _, ok := m["messages"]; ok { + t.Error("bridge should convert messages to responses input") + } + if _, ok := m["input"]; !ok { + t.Error("bridge output missing input field") + } + if _, ok := m["stream_options"]; ok { + t.Error("bridge should NOT inject chat stream_options") + } + if stream := string(m["stream"]); stream != "true" { + t.Errorf("stream = %s, want true", stream) + } +} + +// TestAdaptor_ConvertRequest_ChatViaResponsesBridge_ModelMapped 桥接 + 模型映射： +// 转换器应将模型名替换为上游模型名。 +func TestAdaptor_ConvertRequest_ChatViaResponsesBridge_ModelMapped(t *testing.T) { + info := responsesUpstreamInfo(constant.RelayModeChatCompletions, false) + info.InboundFormat = constant.RelayFormatOpenAI + info.ClientFormat = constant.RelayFormatOpenAI + info.ChannelMeta.IsModelMapped = true + info.ChannelMeta.SupportsResponses = false + info.ChannelMeta.ChatViaResponses = true + info.UseResponsesAPI = true + + body := []byte(`{"model":"gpt-4o","messages":[{"role":"user","content":"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 bridged json: %v", err) + } + if got := string(m["model"]); got != `"gpt-4o-upstream"` { + t.Errorf("model = %s, want gpt-4o-upstream", got) + } +} + +// responsesInboundInfo responses 入站 + 上游 chat 渠道（无 responses 协议能力）， +// 响应侧走 chat→responses 桥接（handleResponsesInboundStream）。 +func responsesInboundInfo(isStream bool) *common.RelayInfo { + info := responsesUpstreamInfo(constant.RelayModeResponses, isStream) + info.ChannelMeta.SupportsResponses = false + info.ChannelMeta.ChatViaResponses = false + return info +} + +// TestExtractStreamEmbeddedError 内嵌错误对象检测： +// "error":null 与无 error 键不算错误，对象/字符串错误都要识别。 +func TestExtractStreamEmbeddedError(t *testing.T) { + if _, ok := extractStreamEmbeddedError([]byte(`{"id":"1","choices":[]}`)); ok { + t.Error("chunk without error key should not be detected") + } + if _, ok := extractStreamEmbeddedError([]byte(`{"error":null,"choices":[]}`)); ok { + t.Error(`"error":null should not be detected`) + } + if body, ok := extractStreamEmbeddedError([]byte(`{"error":{"type":"rate_limit_error","message":"limited"}}`)); !ok { + t.Error("error object should be detected") + } else if !strings.Contains(string(body), "rate_limit_error") { + t.Errorf("error body = %s", string(body)) + } + if _, ok := extractStreamEmbeddedError([]byte(`{"error":"overloaded"}`)); !ok { + t.Error("string error should be detected") + } + if _, ok := extractStreamEmbeddedError([]byte(`not json`)); ok { + t.Error("invalid json should not be detected") + } +} + +// TestAdaptor_DoResponse_ResponsesInboundStream_EmbeddedErrorBeforeEvents +// 上游 200 + SSE 但首个 data 行即内嵌错误对象（部分聚合商的出错形态）： +// 必须返回 upstream error 并向客户端透传错误体，而非静默合成空 response.completed。 +func TestAdaptor_DoResponse_ResponsesInboundStream_EmbeddedErrorBeforeEvents(t *testing.T) { + errLine := `{"error":{"type":"rate_limit_error","message":"Rate limit reached"}}` + ss := strings.Join([]string{ + "data: " + errLine, + "", + "data: [DONE]", + "", + }, "\n") + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ss)), + } + + info := responsesInboundInfo(true) + rec := httptest.NewRecorder() + a := &Adaptor{} + _, err := a.DoResponse(context.Background(), resp, info, rec) + if err == nil { + t.Fatal("embedded upstream error should return an upstream error, got nil") + } + var relayErr *constant.RelayError + if !errors.As(err, &relayErr) { + t.Fatalf("error should be *constant.RelayError, got %T", err) + } + if !relayErr.ResponseWritten { + t.Error("ResponseWritten should be true after writing error body to client") + } + if !strings.Contains(rec.Body.String(), "rate_limit_error") { + t.Errorf("error body not passed through: %s", rec.Body.String()) + } + if strings.Contains(rec.Body.String(), "response.completed") { + t.Errorf("should not synthesize empty response.completed on upstream error: %s", rec.Body.String()) + } +} + +// TestAdaptor_DoResponse_ResponsesInboundStream_EmbeddedErrorMidStream +// 流中途出现内嵌错误：已发送的事件保留，处理终止并返回 upstream error。 +func TestAdaptor_DoResponse_ResponsesInboundStream_EmbeddedErrorMidStream(t *testing.T) { + ss := strings.Join([]string{ + `data: {"id":"chatcmpl-1","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o","choices":[{"index":0,"delta":{"role":"assistant","content":"Hi"},"finish_reason":null}]}`, + "", + `data: {"error":{"type":"server_error","message":"upstream crashed"}}`, + "", + }, "\n") + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ss)), + } + + info := responsesInboundInfo(true) + rec := httptest.NewRecorder() + a := &Adaptor{} + usage, err := a.DoResponse(context.Background(), resp, info, rec) + if err == nil { + t.Fatal("mid-stream embedded error should return an upstream error, got nil") + } + if usage == nil { + t.Fatal("usage should be non-nil for error path") + } + out := rec.Body.String() + if !strings.Contains(out, "response.created") || !strings.Contains(out, `"delta":"Hi"`) { + t.Errorf("events emitted before the error should remain:\n%s", out) + } + if strings.Contains(out, "response.completed") { + t.Errorf("should not synthesize response.completed after mid-stream error:\n%s", out) + } +} + +// TestAdaptor_DoResponse_ResponsesInboundStream_UnparseableChunks +// 上游 data 行全部无法解析为 chat chunk（如返回了非 chat 格式）： +// 保持既有兜底行为（合成空 response.completed、不报错），但不再静默无痕。 +func TestAdaptor_DoResponse_ResponsesInboundStream_UnparseableChunks(t *testing.T) { + ss := strings.Join([]string{ + "data: {invalid json", + "", + "data: [DONE]", + "", + }, "\n") + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(ss)), + } + + info := responsesInboundInfo(true) + rec := httptest.NewRecorder() + a := &Adaptor{} + usage, err := a.DoResponse(context.Background(), resp, info, rec) + if err != nil { + t.Fatalf("unparseable chunks keep fallback behavior, unexpected error: %v", err) + } + if !strings.Contains(rec.Body.String(), "response.completed") { + t.Errorf("fallback response.completed missing:\n%s", rec.Body.String()) + } + if usage == nil { + t.Fatal("usage should be non-nil") + } +} diff --git a/relay/channel/openai/converter.go b/relay/channel/openai/converter.go index 5ecc2cf..306ce7a 100644 --- a/relay/channel/openai/converter.go +++ b/relay/channel/openai/converter.go @@ -1149,6 +1149,22 @@ func extractOutputFromResponses(resp *dto.OpenAIResponsesResponse) (string, []dt func HandleResponsesStreamToChat(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) { defer resp.Body.Close() + // 非 200：上游在 SSE 开始前返回错误（非 SSE 体）。必须在 StreamScannerHandler 写出 + // 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) + } + responseID := fmt.Sprintf("chatcmpl-%s", info.RequestID) createAt := time.Now().Unix() model := info.ChannelMeta.UpstreamModelName @@ -1412,7 +1428,10 @@ func HandleResponsesNonStreamToChat(ctx context.Context, resp *http.Response, in writer.Header().Set("Content-Type", "application/json") writer.WriteHeader(resp.StatusCode) _, _ = writer.Write(body) - return nil, constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) + // 错误体已写给客户端，标记 ResponseWritten 防止调度 FSM 误判为可重试导致二次写入 + upstreamErr := constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) + upstreamErr.ResponseWritten = true + return nil, upstreamErr } var responsesResp dto.OpenAIResponsesResponse diff --git a/relay/channel/openai/responses.go b/relay/channel/openai/responses.go index 9dc77bf..748e988 100644 --- a/relay/channel/openai/responses.go +++ b/relay/channel/openai/responses.go @@ -2,6 +2,7 @@ package openai import ( "bufio" + "bytes" "context" "encoding/json" "fmt" @@ -10,6 +11,8 @@ import ( "strings" "time" + "github.com/gogf/gf/v2/frame/g" + "github.com/qianfree/team-api/relay/common" "github.com/qianfree/team-api/relay/constant" "github.com/qianfree/team-api/relay/dto" @@ -194,6 +197,7 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R var contentBuilder strings.Builder sentCreated := false sentTextDone := false + parsedChunks := 0 outputIndex := 0 contentIndex := 0 toolCallIndexByID := make(map[string]int) @@ -232,10 +236,32 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R break } + // 检测 SSE 流中内嵌的上游错误对象：部分聚合商出错时返回 HTTP 200 + SSE， + // 错误信息夹在 data 行里（{"error":{...}}）。不识别会被当作解析失败静默丢弃， + // 最终合成空的 response.completed（客户端表现为"成功但无内容"）。 + if errBody, ok := extractStreamEmbeddedError([]byte(data)); ok { + err := fmt.Errorf("upstream embedded error in SSE stream: %.500s", string(errBody)) + g.Log().Warningf(ctx, "[OpenAI.handleResponsesInboundStream] %v", err) + info.StreamStatus.SetEndReason(common.StreamEndReasonError, err) + if !sentCreated { + // 尚未向客户端发送任何事件：透传上游错误体（与 SSE 开始前的非 200 分支行为一致） + writeUpstreamErrorResponse(writer, resp.StatusCode, []byte(data)) + upstreamErr := constant.NewUpstreamError(resp.StatusCode, string(errBody), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) + upstreamErr.ResponseWritten = true + return &usage, upstreamErr + } + // 已发送部分事件：流已污染无法回写错误体，直接返回错误终止处理 + return &usage, constant.NewUpstreamError(resp.StatusCode, string(errBody), nil) + } + var chunk dto.ChatCompletionStreamResponse if err := json.Unmarshal([]byte(data), &chunk); err != nil { + // 解析失败不静默丢弃：记录日志便于定位上游返回非 chat 格式的问题 + // （如 responses 格式 SSE、纯 JSON 非流式体、空 body 等） + g.Log().Warningf(ctx, "[OpenAI.handleResponsesInboundStream] unmarshal chat stream chunk failed: %v, data: %.200s", err, data) continue } + parsedChunks++ // 第一个 chunk：发送 response.created + output_item.added + content_part.added if !sentCreated { @@ -488,6 +514,14 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R } } + // 上游零有效 chunk 时不再静默合成空响应：记录告警便于定位 + // （上游返回非 SSE 体 / 非 chat 格式 / 空 body / 纯 [DONE] 等异常都会走到这里） + if !sentCreated { + g.Log().Warningf(ctx, + "[OpenAI.handleResponsesInboundStream] upstream stream yielded no parseable chat chunks (parsed=%d), synthesizing empty response.completed: channel=%d(%s) model=%s request_id=%s", + parsedChunks, info.ChannelMeta.ChannelID, info.ChannelMeta.ChannelName, info.OriginModelName, info.RequestID) + } + // 估算 usage（正常结束 4 字符/token；scanner 异常等部分传输场景 2 字符/token） if usage.CompletionTokens == 0 { text := contentBuilder.String() @@ -544,6 +578,25 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R return &usage, nil } +// extractStreamEmbeddedError 检测 SSE data 行中内嵌的上游错误对象（存在 "error" 键且值非 null）。 +// 与 isUpstreamOpenAIError 的区别：后者用于完整响应体且不区分 null；这里对流式 chunk 逐行 +// 检测，并排除 "error":null（部分供应商的正常 chunk 会携带空 error 字段）。 +func extractStreamEmbeddedError(data []byte) (json.RawMessage, bool) { + var raw map[string]json.RawMessage + if err := json.Unmarshal(data, &raw); err != nil { + return nil, false + } + errBody, ok := raw["error"] + if !ok { + return nil, false + } + trimmed := string(bytes.TrimSpace(errBody)) + if trimmed == "" || trimmed == "null" { + return nil, false + } + return errBody, true +} + // emitResponsesSSE 发送一个 Responses API 格式的 SSE 事件 func emitResponsesSSE(w http.ResponseWriter, eventType string, data any) { jsonData, err := json.Marshal(data) diff --git a/relay/common/provider.go b/relay/common/provider.go index 0b5b463..ec7f65b 100644 --- a/relay/common/provider.go +++ b/relay/common/provider.go @@ -123,6 +123,11 @@ type ChannelSelection struct { IsModelMapped bool MaxConcurrency int // 该渠道最大并发（0/负值表示不限），供各转发入口做容量控制 Settings ChannelSettings + // 协议能力（chn_abilities 渠道×模型级）： + // SupportsResponses：responses 入站原样直连 /v1/responses； + // ChatViaResponses：responses-only 上游，chat 入站经桥接发送 /v1/responses。 + SupportsResponses bool + ChatViaResponses bool SelectionReason string // bind / hrw / overflow / probe / cred_rotate Priority int Weight int diff --git a/relay/common/relay_info.go b/relay/common/relay_info.go index 1466199..acfb6c3 100644 --- a/relay/common/relay_info.go +++ b/relay/common/relay_info.go @@ -20,6 +20,20 @@ type ChannelMeta struct { UpstreamModelName string // 上游实际模型名（可能与用户请求不同） IsModelMapped bool // 是否经过模型名映射 Settings ChannelSettings + + // 协议能力（chn_abilities 渠道×模型级）： + // SupportsResponses：模型支持 /v1/responses，responses 入站原样直连（不做 chat 转换）； + // ChatViaResponses：responses-only 上游，chat 入站经桥接转换后发送 /v1/responses。 + // ChatViaResponses 隐含上游原生说 Responses 协议（UpstreamSpeaksResponses）。 + SupportsResponses bool + ChatViaResponses bool +} + +// UpstreamSpeaksResponses 上游（针对当前模型）是否原生支持 Responses 协议： +// supports_responses 或 chat_via_responses 任一为真即成立——后者定义上就是 +// "上游只有 /v1/responses"，responses 入站在其上同样应直连而非转 chat。 +func (m *ChannelMeta) UpstreamSpeaksResponses() bool { + return m != nil && (m.SupportsResponses || m.ChatViaResponses) } const ( @@ -47,13 +61,11 @@ func (s ChannelSettings) GetTimeoutSeconds(relayMode int) int { // ChannelSettings 渠道配置（来自 chn_channels.settings JSONB） type ChannelSettings struct { - TimeoutSeconds int `json:"timeout_seconds"` // 请求超时秒数，默认 60 - RetryCount int `json:"retry_count"` // 重试次数，默认 1 - ParamOverride map[string]any `json:"param_override,omitempty"` // 请求体改写规则 - 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 入站直连转发 + TimeoutSeconds int `json:"timeout_seconds"` // 请求超时秒数，默认 60 + RetryCount int `json:"retry_count"` // 重试次数，默认 1 + ParamOverride map[string]any `json:"param_override,omitempty"` // 请求体改写规则 + HeaderOverride map[string]any `json:"header_override,omitempty"` // Header 改写规则 + PassThroughBodyEnabled bool `json:"pass_through_body_enabled,omitempty"` // 直连转发：跳过协议转换，原始请求体直接转发上游 // System Prompt 注入 SystemPrompt string `json:"system_prompt,omitempty"` // 渠道级系统提示词 diff --git a/relay/handler/passthrough.go b/relay/handler/passthrough.go index 3f4f444..d665d46 100644 --- a/relay/handler/passthrough.go +++ b/relay/handler/passthrough.go @@ -18,6 +18,13 @@ func canPassThrough(info *common.RelayInfo) bool { return false } + // responses-only 桥接渠道的 chat 入站：chat 体与上游 /v1/responses 端点不兼容， + // 必须经 ConvertOpenAIToResponses 桥接转换，即使显式开启直连也不得原样透传。 + if info.ChannelMeta.ChatViaResponses && + constant.RelayMode(info.RelayMode) == constant.RelayModeChatCompletions { + return false + } + // 显式开启：运营者明确配置直连，不做额外检查 if settings.PassThroughBodyEnabled { return true @@ -48,11 +55,14 @@ func canPassThrough(info *common.RelayInfo) bool { } // inboundMatchesChannelNative 判断入站格式是否与渠道原生格式匹配（匹配则可原样直连转发）。 -// - 渠道显式声明上游为 Responses 协议（upstream_responses）：Responses 入站视为原生匹配。 +// - 上游原生支持 Responses 协议（supports_responses / chat_via_responses）：Responses 入站视为原生匹配。 // - 普通渠道：入站格式须等于该渠道的唯一原生格式（helper.ProviderNativeFormat）。 // - 多协议原生透传渠道（New API / Sub2API）：OpenAI/Claude/Gemini 三种格式均视为匹配。 +// +// 注意：chat_via_responses（responses-only 桥接渠道）的 chat 入站由 canPassThrough +// 前置硬排除——chat 体必须经 ConvertOpenAIToResponses 转换后才能发 /v1/responses。 func inboundMatchesChannelNative(info *common.RelayInfo) bool { - if info.ChannelMeta.Settings.UpstreamResponses { + if info.ChannelMeta.UpstreamSpeaksResponses() { return info.InboundFormat == constant.RelayFormatResponses } if constant.IsMultiNativeProvider(info.ChannelMeta.ChannelType) { diff --git a/relay/handler/passthrough_test.go b/relay/handler/passthrough_test.go index 6888861..51b1da6 100644 --- a/relay/handler/passthrough_test.go +++ b/relay/handler/passthrough_test.go @@ -193,57 +193,87 @@ func TestCanPassThrough_MultiNative_ResponsesNotNative(t *testing.T) { } } -// TestCanPassThrough_UpstreamResponses_ResponsesInbound 渠道声明上游为 Responses 协议后， +// TestCanPassThrough_SupportsResponses_ResponsesInbound 模型能力声明支持 Responses 协议后， // Responses 入站在无改写时应原样直连转发。 -func TestCanPassThrough_UpstreamResponses_ResponsesInbound(t *testing.T) { +func TestCanPassThrough_SupportsResponses_ResponsesInbound(t *testing.T) { info := &common.RelayInfo{ InboundFormat: constant.RelayFormatResponses, ChannelMeta: &common.ChannelMeta{ - ChannelType: int(constant.ProviderOpenAI), - IsModelMapped: false, - Settings: common.ChannelSettings{ - UpstreamResponses: true, - }, + ChannelType: int(constant.ProviderOpenAI), + IsModelMapped: false, + SupportsResponses: true, }, } if !canPassThrough(info) { - t.Error("channel declaring UpstreamResponses should pass through Responses inbound natively") + t.Error("channel declaring SupportsResponses should pass through Responses inbound natively") } } -// TestCanPassThrough_UpstreamResponses_NonResponsesInbound 渠道声明上游为 Responses 协议时， +// TestCanPassThrough_SupportsResponses_NonResponsesInbound 渠道声明支持 Responses 协议时， // 其它入站格式（如 chat）不属于原生匹配，不得原样直连。 -func TestCanPassThrough_UpstreamResponses_NonResponsesInbound(t *testing.T) { +func TestCanPassThrough_SupportsResponses_NonResponsesInbound(t *testing.T) { info := &common.RelayInfo{ InboundFormat: constant.RelayFormatOpenAI, ChannelMeta: &common.ChannelMeta{ - ChannelType: int(constant.ProviderOpenAI), - IsModelMapped: false, - Settings: common.ChannelSettings{ - UpstreamResponses: true, - }, + ChannelType: int(constant.ProviderOpenAI), + IsModelMapped: false, + SupportsResponses: true, }, } if canPassThrough(info) { - t.Error("UpstreamResponses channel should NOT pass through non-Responses inbound natively") + t.Error("SupportsResponses channel should NOT pass through non-Responses inbound natively") } } -// TestCanPassThrough_UpstreamResponses_ModelMappedForcesConversion 上游为 Responses 协议 +// TestCanPassThrough_SupportsResponses_ModelMappedForcesConversion 上游为 Responses 协议 // 但配置了模型映射时，仍需经 ConvertRequest 替换模型名，不得原样透传。 -func TestCanPassThrough_UpstreamResponses_ModelMappedForcesConversion(t *testing.T) { +func TestCanPassThrough_SupportsResponses_ModelMappedForcesConversion(t *testing.T) { info := &common.RelayInfo{ InboundFormat: constant.RelayFormatResponses, ChannelMeta: &common.ChannelMeta{ - ChannelType: int(constant.ProviderOpenAI), - IsModelMapped: true, - Settings: common.ChannelSettings{ - UpstreamResponses: true, - }, + ChannelType: int(constant.ProviderOpenAI), + IsModelMapped: true, + SupportsResponses: true, }, } if canPassThrough(info) { - t.Error("UpstreamResponses channel with model mapping should NOT pass through") + t.Error("SupportsResponses channel with model mapping should NOT pass through") + } +} + +// TestCanPassThrough_ChatViaResponses_ChatInboundBlocked responses-only 桥接渠道的 +// chat 入站：即使显式开启直连（pass_through_body_enabled）也不得原样透传， +// chat 体必须经桥接转换后才能发 /v1/responses。 +func TestCanPassThrough_ChatViaResponses_ChatInboundBlocked(t *testing.T) { + info := &common.RelayInfo{ + InboundFormat: constant.RelayFormatOpenAI, + RelayMode: int(constant.RelayModeChatCompletions), + ChannelMeta: &common.ChannelMeta{ + ChannelType: int(constant.ProviderOpenAI), + IsModelMapped: false, + ChatViaResponses: true, + SupportsResponses: true, + }, + } + info.ChannelMeta.Settings.PassThroughBodyEnabled = true + if canPassThrough(info) { + t.Error("chat inbound on chat_via_responses channel must NOT pass through, even with explicit passthrough") + } +} + +// TestCanPassThrough_ChatViaResponses_ResponsesInbound responses-only 桥接渠道的 +// responses 入站视为原生匹配（上游本来就说 Responses 协议），可原样直连。 +func TestCanPassThrough_ChatViaResponses_ResponsesInbound(t *testing.T) { + info := &common.RelayInfo{ + InboundFormat: constant.RelayFormatResponses, + ChannelMeta: &common.ChannelMeta{ + ChannelType: int(constant.ProviderOpenAI), + IsModelMapped: false, + ChatViaResponses: true, + }, + } + if !canPassThrough(info) { + t.Error("responses inbound on chat_via_responses channel should pass through natively") } } diff --git a/relay/handler/relay_handler.go b/relay/handler/relay_handler.go index ea9a741..fe284e9 100644 --- a/relay/handler/relay_handler.go +++ b/relay/handler/relay_handler.go @@ -499,6 +499,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He Scope: channelScope, Replay: co.Policy().Replay.ReplayabilityForMode(v.relayModeStr), Signals: sig, + Proto: protoPreference(constant.RelayMode(v.relayMode)), Policy: dispatchadapter.TenantRoutingPolicy(ctx, rc.TenantID), }) // 兜底：任何未显式 Finish 的退出路径释放租约（Finish 幂等，成功路径的显式调用优先生效） @@ -763,8 +764,17 @@ func buildRelayInfo(ctx context.Context, rc *RelayContext, v *relayValidation, s UpstreamModelName: selection.UpstreamModelName, IsModelMapped: selection.IsModelMapped, Settings: selection.Settings, + + SupportsResponses: selection.SupportsResponses, + ChatViaResponses: selection.ChatViaResponses, }, } + // responses-only 上游桥接：chat 入站经 /v1/responses 发送（chat→Responses 请求转换 + + // Responses→chat 响应转换）。仅对 OpenAI chat 入站生效；claude/gemini 入站需 + // Responses→原生格式链式转换，暂不支持（此类渠道上会 404，运营应避免混用）。 + if selection.ChatViaResponses && v.relayMode == constant.RelayModeChatCompletions { + info.UseResponsesAPI = true + } // 流中断结算时上游 usage 常缺失，记录请求侧输入估算值（与预扣同源）供输入计费兜底 info.SetEstimatePromptTokens(v.estimatedTokens) return info @@ -1042,6 +1052,19 @@ func setPreResponseHeaders(w http.ResponseWriter, br *BillingResult) { } } +// protoPreference 入站端点协议 → 调度软偏好（只降权不排除）： +// responses 端点偏好声明支持 Responses 协议的渠道（避免经 chat 转换丢失有状态特性）， +// chat 端点偏好原生 chat 渠道（responses-only 桥接渠道降权但仍可服务），其余端点无偏好。 +func protoPreference(mode constant.RelayMode) dispatch.ProtoPreference { + switch mode { + case constant.RelayModeResponses, constant.RelayModeResponsesCompact: + return dispatch.ProtoResponses + case constant.RelayModeChatCompletions: + return dispatch.ProtoChat + } + return dispatch.ProtoAny +} + // requestType 根据 isStream 返回请求类型 func requestType(isStream bool) int { if isStream { diff --git a/relaykit/dispatch/coordinator.go b/relaykit/dispatch/coordinator.go index 71f0285..8e3e9af 100644 --- a/relaykit/dispatch/coordinator.go +++ b/relaykit/dispatch/coordinator.go @@ -361,7 +361,7 @@ func (s *RouteSession) buildCandidates(snapshot []Channel, roundExcluded map[int zeroTier = append(zeroTier, ch) continue } - w, bd := EffectiveWeight(ch, pol) + w, bd := EffectiveWeight(ch, pol, s.profile.Proto) scored = append(scored, ScoredChannel{Channel: ch, Weight: w, Breakdown: bd}) } @@ -374,10 +374,10 @@ func (s *RouteSession) buildCandidates(snapshot []Channel, roundExcluded map[int if ch.Tier != t { continue } - w, bd := EffectiveWeight(ch, pol) + w, bd := EffectiveWeight(ch, pol, s.profile.Proto) // tierFactor 为 0，用 epsilon 重算 bd.Tier = epsilonTier - bd.Effective = bd.Base * bd.Tier * bd.Health * bd.Headroom * bd.Cost * bd.Ramp + bd.Effective = bd.Base * bd.Tier * bd.Health * bd.Headroom * bd.Cost * bd.Ramp * bd.Proto w = bd.Effective if w > 0 { added = true diff --git a/relaykit/dispatch/policy.go b/relaykit/dispatch/policy.go index a4d5e10..9ea39e2 100644 --- a/relaykit/dispatch/policy.go +++ b/relaykit/dispatch/policy.go @@ -50,6 +50,15 @@ type RoutingPolicy struct { Degrade struct { MaxReplicas int `json:"maxReplicas"` // 实例级保守限额分母，默认 1 } `json:"degrade"` + + Proto struct { + // ResponsesMismatch responses 入站打在不支持 Responses 协议渠道上的降权因子， + // 默认 0.25（经 chat 转换丢失有状态 Responses 特性，降权较重）。 + ResponsesMismatch float64 `json:"responsesMismatch"` + // ChatBridgeMismatch chat 入站打在 responses-only 桥接渠道上的降权因子， + // 默认 0.75（桥接基本无损，降权较轻）。 + ChatBridgeMismatch float64 `json:"chatBridgeMismatch"` + } `json:"proto"` } // ReplayPolicy 可重放性策略。 @@ -157,6 +166,8 @@ func DefaultRoutingPolicy() *RoutingPolicy { SafeModes: []string{"embeddings", "rerank"}, } p.Degrade.MaxReplicas = 1 + p.Proto.ResponsesMismatch = 0.25 + p.Proto.ChatBridgeMismatch = 0.75 return p } @@ -214,6 +225,12 @@ func (p *RoutingPolicy) Validate() error { if p.Degrade.MaxReplicas <= 0 { return fmt.Errorf("degrade.maxReplicas 必须 > 0") } + // proto 偏好因子必须为 (0,1]：>1 会反向惩罚匹配渠道，=0 会把不匹配渠道硬排除 + if p.Proto.ResponsesMismatch <= 0 || p.Proto.ResponsesMismatch > 1 || + p.Proto.ChatBridgeMismatch <= 0 || p.Proto.ChatBridgeMismatch > 1 { + return fmt.Errorf("proto 因子必须在 (0,1]: responsesMismatch=%v chatBridgeMismatch=%v", + p.Proto.ResponsesMismatch, p.Proto.ChatBridgeMismatch) + } return nil } diff --git a/relaykit/dispatch/proto_test.go b/relaykit/dispatch/proto_test.go new file mode 100644 index 0000000..2bca535 --- /dev/null +++ b/relaykit/dispatch/proto_test.go @@ -0,0 +1,118 @@ +package dispatch + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestProtoFactor_ResponsesPreference protoFactor 在 responses 入站下的软偏好： +// 匹配渠道因子为 1，不匹配渠道按策略 ResponsesMismatch 降权；ProtoAny 不降权。 +func TestProtoFactor_ResponsesPreference(t *testing.T) { + pol := DefaultRoutingPolicy() + match := healthyChannel(1, TierPrimary, 10) + match.SupportsResponses = true + mismatch := healthyChannel(2, TierPrimary, 10) + + assert.Equal(t, 1.0, protoFactor(match, ProtoResponses, pol)) + assert.Equal(t, pol.Proto.ResponsesMismatch, protoFactor(mismatch, ProtoResponses, pol)) + assert.Equal(t, 1.0, protoFactor(mismatch, ProtoAny, pol)) + // chat 入站对 responses 支持与否不敏感 + assert.Equal(t, 1.0, protoFactor(match, ProtoChat, pol)) +} + +// TestProtoFactor_ChatBridgePreference protoFactor 在 chat 入站下的软偏好： +// 原生 chat 渠道因子为 1，chat_via_responses 桥接渠道按 ChatBridgeMismatch 降权。 +func TestProtoFactor_ChatBridgePreference(t *testing.T) { + pol := DefaultRoutingPolicy() + native := healthyChannel(1, TierPrimary, 10) + bridge := healthyChannel(2, TierPrimary, 10) + bridge.ChatViaResponses = true + + assert.Equal(t, 1.0, protoFactor(native, ProtoChat, pol)) + assert.Equal(t, pol.Proto.ChatBridgeMismatch, protoFactor(bridge, ProtoChat, pol)) + assert.Equal(t, 1.0, protoFactor(bridge, ProtoResponses, pol)) // 未声明 supports_responses +} + +// TestEffectiveWeight_ProtoInProduct proto 因子参与权重乘积且记录在分解明细中。 +func TestEffectiveWeight_ProtoInProduct(t *testing.T) { + pol := DefaultRoutingPolicy() + c := healthyChannel(1, TierPrimary, 10) + wNoPref, _ := EffectiveWeight(c, pol) + wPref, bd := EffectiveWeight(c, pol, ProtoResponses) + require.Equal(t, pol.Proto.ResponsesMismatch, bd.Proto) + assert.InDelta(t, wNoPref*pol.Proto.ResponsesMismatch, wPref, 1e-9) +} + +// TestCoordinator_ProtoResponsesPreference 统计意义验证：两个同权重同健康渠道， +// responses 入站时绝大多数会话键选中声明 SupportsResponses 的渠道（软偏好非硬过滤， +// HRW 哈希仍可能命中另一渠道——降权只是让匹配渠道的选中概率占优）。 +func TestCoordinator_ProtoResponsesPreference(t *testing.T) { + ctx := context.Background() + match := healthyChannel(1, TierPrimary, 10) + match.SupportsResponses = true + other := healthyChannel(2, TierPrimary, 10) + + matchWins, otherWins := 0, 0 + for i := range 200 { + state := newFakeState() + co, _ := newTestCoordinator(state, match, other) + p := testProfile() + p.RequestID = fmt.Sprintf("req-%d", i) + p.Signals = SessionSignals{HeaderSessionID: fmt.Sprintf("sess-%d", i)} + p.Proto = ProtoResponses + d := co.Route(ctx, p).Next(ctx) + require.NotNil(t, d) + if d.Channel.ID == 1 { + matchWins++ + } else { + otherWins++ + } + } + assert.Greater(t, matchWins, otherWins*3, "responses-capable channel should dominate: match=%d other=%d", matchWins, otherWins) +} + +// TestCoordinator_ProtoAnyNoPreference 对照组：无偏好时两同权重渠道选中次数同量级。 +func TestCoordinator_ProtoAnyNoPreference(t *testing.T) { + ctx := context.Background() + a := healthyChannel(1, TierPrimary, 10) + a.SupportsResponses = true + b := healthyChannel(2, TierPrimary, 10) + + aWins, bWins := 0, 0 + for i := range 200 { + state := newFakeState() + co, _ := newTestCoordinator(state, a, b) + p := testProfile() + p.RequestID = fmt.Sprintf("req-%d", i) + p.Signals = SessionSignals{HeaderSessionID: fmt.Sprintf("sess-%d", i)} + d := co.Route(ctx, p).Next(ctx) + require.NotNil(t, d) + if d.Channel.ID == 1 { + aWins++ + } else { + bWins++ + } + } + // HRW 均匀性：无偏好时两渠道选中次数应为同量级（放宽到 4 倍避免哈希偶然性导致的 flake） + assert.Less(t, aWins, bWins*4) + assert.Less(t, bWins, aWins*4) +} + +// TestPolicyValidate_ProtoFactors proto 因子越界（>1 或 <=0）必须被拒绝。 +func TestPolicyValidate_ProtoFactors(t *testing.T) { + pol := DefaultRoutingPolicy() + require.NoError(t, pol.Validate()) + + pol.Proto.ResponsesMismatch = 0 + assert.Error(t, pol.Validate()) + pol = DefaultRoutingPolicy() + pol.Proto.ResponsesMismatch = 1.5 + assert.Error(t, pol.Validate()) + pol = DefaultRoutingPolicy() + pol.Proto.ChatBridgeMismatch = 0 + assert.Error(t, pol.Validate()) +} diff --git a/relaykit/dispatch/score.go b/relaykit/dispatch/score.go index 46ff108..f55bae1 100644 --- a/relaykit/dispatch/score.go +++ b/relaykit/dispatch/score.go @@ -6,10 +6,15 @@ import ( // EffectiveWeight 权重合成函数 W(c)（纯函数）。 // -// W(c) = baseWeight × tierFactor × healthFactor × headroom^γ × costFactor × rampFactor +// W(c) = baseWeight × tierFactor × healthFactor × headroom^γ × costFactor × rampFactor × protoFactor // // 返回合成权重与分解明细（供决策日志 / ForwardingTrace）。 -func EffectiveWeight(c Channel, pol *RoutingPolicy) (float64, WeightBreakdown) { +// pref 为可选的入站协议偏好（省略 = ProtoAny，protoFactor 恒为 1）。 +func EffectiveWeight(c Channel, pol *RoutingPolicy, pref ...ProtoPreference) (float64, WeightBreakdown) { + var proto ProtoPreference + if len(pref) > 0 { + proto = pref[0] + } bd := WeightBreakdown{ Base: c.BaseWeight, Tier: tierFactor(c.Tier, pol), @@ -17,11 +22,12 @@ func EffectiveWeight(c Channel, pol *RoutingPolicy) (float64, WeightBreakdown) { Headroom: headroomFactor(c, pol), Cost: costFactor(c, pol), Ramp: rampFactor(c, pol), + Proto: protoFactor(c, proto, pol), } if bd.Base <= 0 { bd.Base = 0 } - bd.Effective = bd.Base * bd.Tier * bd.Health * bd.Headroom * bd.Cost * bd.Ramp + bd.Effective = bd.Base * bd.Tier * bd.Health * bd.Headroom * bd.Cost * bd.Ramp * bd.Proto return bd.Effective, bd } @@ -89,6 +95,29 @@ func rampFactor(c Channel, pol *RoutingPolicy) float64 { return clamp(float64(c.RampElapsedMs)/float64(window), pol.Ramp.Floor, 1.0) } +// protoFactor 协议偏好因子（软偏好，只降权不排除，恒 > 0）： +// - ProtoResponses：上游原生支持 Responses 协议（supports_responses 或 +// chat_via_responses，后者定义上即 responses-only 上游）的渠道为 1，其余降权 +// （经 chat 转换会丢失 previous_response_id 有状态多轮等 Responses 专属特性， +// 损失大故降权较重）； +// - ProtoChat：原生 chat 渠道为 1，ChatViaResponses 桥接渠道降权（桥接基本无损， +// 仅响应细节略有出入，降权较轻）。 +func protoFactor(c Channel, pref ProtoPreference, pol *RoutingPolicy) float64 { + switch pref { + case ProtoResponses: + if c.SupportsResponses || c.ChatViaResponses { + return 1 + } + return pol.Proto.ResponsesMismatch + case ProtoChat: + if !c.ChatViaResponses { + return 1 + } + return pol.Proto.ChatBridgeMismatch + } + return 1 +} + func clamp(v, lo, hi float64) float64 { if v < lo { return lo diff --git a/relaykit/dispatch/types.go b/relaykit/dispatch/types.go index 513e131..0782df0 100644 --- a/relaykit/dispatch/types.go +++ b/relaykit/dispatch/types.go @@ -67,8 +67,23 @@ type Channel struct { KeyIDs []int64 // active 状态的渠道 Key ID 列表（按 id 升序）；空表示由适配层自行取 Key StrictCapacity bool // Redis 故障时 fail-closed + + // 协议能力（chn_abilities 渠道×模型级），参与 protoFactor 软偏好： + // responses 入站偏好 SupportsResponses 渠道（避免经 chat 转换丢失有状态 Responses 特性）， + // chat 入站偏好原生 chat 渠道（ChatViaResponses 桥接渠道降权但仍可用）。 + SupportsResponses bool // 模型在该渠道支持 /v1/responses 原生协议 + ChatViaResponses bool // 上游仅有 Responses 协议，chat 需经桥接发送 } +// ProtoPreference 入站端点协议偏好（软偏好，只降权不排除）。 +type ProtoPreference uint8 + +const ( + ProtoAny ProtoPreference = iota // 非文本生成端点 / 无偏好 + ProtoResponses // responses 入站：偏好 SupportsResponses 渠道 + ProtoChat // chat 入站：偏好原生 chat 渠道 +) + // ScoredChannel 打分后的候选。 type ScoredChannel struct { Channel @@ -84,6 +99,7 @@ type WeightBreakdown struct { Headroom float64 `json:"headroom"` Cost float64 `json:"cost"` Ramp float64 `json:"ramp"` + Proto float64 `json:"proto"` Effective float64 `json:"effective"` } @@ -97,6 +113,9 @@ type RequestProfile struct { Scope []int64 // 租户渠道范围（空 = 不限制），由适配层填充 Replay Replayability Signals SessionSignals + // Proto 入站端点协议偏好（ProtoAny=无偏好）。responses 入站软偏好 + // SupportsResponses 渠道、chat 入站软偏好原生 chat 渠道，只降权不排除。 + Proto ProtoPreference // Policy 请求级策略覆盖（租户级差异化）。 // nil = 使用协调器全局策略。由适配层解析（全局 + 租户浅合并）后传入。 Policy *RoutingPolicy diff --git a/web/admin/src/views/ai/ChannelDetailPage.vue b/web/admin/src/views/ai/ChannelDetailPage.vue index 9b0a3ee..e7586f2 100644 --- a/web/admin/src/views/ai/ChannelDetailPage.vue +++ b/web/admin/src/views/ai/ChannelDetailPage.vue @@ -1,7 +1,7 @@