1352 lines
60 KiB
Go
1352 lines
60 KiB
Go
package handler
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
|
||
"github.com/qianfree/team-api/internal/dispatchadapter"
|
||
commonlogic "github.com/qianfree/team-api/internal/logic/common"
|
||
"github.com/qianfree/team-api/internal/logic/monitor"
|
||
tenantlogic "github.com/qianfree/team-api/internal/logic/tenant"
|
||
"github.com/qianfree/team-api/relay/channel"
|
||
"github.com/qianfree/team-api/relay/common"
|
||
"github.com/qianfree/team-api/relay/constant"
|
||
"github.com/qianfree/team-api/relay/helper"
|
||
"github.com/qianfree/team-api/relay/override"
|
||
"github.com/qianfree/team-api/relaykit/dispatch"
|
||
)
|
||
|
||
// RelayContext relay 请求上下文（从 GoFrame handler 传入）
|
||
type RelayContext struct {
|
||
TenantID int64
|
||
UserID int64
|
||
ApiKeyID int64
|
||
ProjectID int64 // 通过 API Key 关联的项目 ID
|
||
RequestID string
|
||
Writer http.ResponseWriter
|
||
Scope string // API Key scope
|
||
ClientIP string // 客户端 IP
|
||
KeyRateLimitQps int // API Key QPS 覆盖值，0 表示使用全局默认
|
||
KeyConcurrency int // API Key 并发限制，0 表示不限制
|
||
KeyIpWhitelist string // API Key IP 白名单
|
||
KeyTotalQuota float64 // API Key 总额度，0 表示不限制
|
||
KeyUsedQuota float64 // API Key 已用额度（仅展示/审计，额度检查会鲜读 DB）
|
||
ForwardingTrace *common.ForwardingTrace // 转发路径追踪（仅管理员可见）
|
||
Debug *common.DebugSession // 渠道调试日志会话（仅调试开关开启的请求非 nil）
|
||
}
|
||
|
||
// BillingResult 计费结果（返回给调用方用于设置响应头）
|
||
type BillingResult struct {
|
||
PreDeductAmount float64
|
||
ActualCost float64
|
||
RateLimitInfo *common.RateLimitInfo
|
||
Deprecation *common.DeprecationInfo
|
||
FirstTokenMs int
|
||
}
|
||
|
||
// relayValidation 请求校验结果，供重试循环使用
|
||
type relayValidation struct {
|
||
relayMode constant.RelayMode
|
||
relayModeStr string
|
||
modelName string
|
||
lookupModel string
|
||
isStream bool
|
||
estimatedTokens int
|
||
maxTokens int
|
||
depInfo *common.DeprecationInfo
|
||
billingResult *BillingResult
|
||
sessionSignals dispatch.SessionSignals // 请求体中的会话信号（影子模式/新调度用）
|
||
}
|
||
|
||
type modelMappingProvider interface {
|
||
GetModelMapping(ctx context.Context, modelName string) (standardName string, category string, err error)
|
||
}
|
||
|
||
// resolveRelayModel 校验模型在目录中存在，返回计费/调度用的目录模型名。
|
||
// 历史上此处还承担 thinking/effort 模型名后缀（-thinking/-high 等虚拟模型语法）的
|
||
// 解析回退；该语法已于 2026-09 移除（无存量使用，思考参数一律由客户端请求参数表达），
|
||
// 目录名即请求名。
|
||
func resolveRelayModel(ctx context.Context, provider modelMappingProvider, modelName string) (string, error) {
|
||
if _, _, err := provider.GetModelMapping(ctx, modelName); err != nil {
|
||
return "", err
|
||
}
|
||
return modelName, nil
|
||
}
|
||
|
||
// validateRelayRequest 校验请求合法性：relay mode、QPS 限流（前置）、模型存在性、弃用状态、成员/API Key 模型范围。
|
||
// 纯校验逻辑，无 defer 副作用。
|
||
func validateRelayRequest(
|
||
ctx context.Context,
|
||
body []byte,
|
||
path string,
|
||
rc *RelayContext,
|
||
provider common.DataProvider,
|
||
billing common.BillingProvider,
|
||
) (*relayValidation, error) {
|
||
// 1. 确定 relay mode
|
||
relayMode := constant.Path2RelayMode(path)
|
||
if relayMode == constant.RelayModeUnknown {
|
||
g.Log().Errorf(ctx, "[RelayHandler] Unknown relay mode for path: %s", path)
|
||
return nil, constant.NewRequestError("unsupported endpoint: "+path, nil)
|
||
}
|
||
|
||
// 1.5 QPS 限流检查（尽可能前置：只依赖认证阶段就已就绪的 tenant/user/key ID，
|
||
// 提前到请求体解析和模型权限检查之前，让超限请求以 1 次 Redis EVAL 的成本被拒绝，
|
||
// 不再为其解析大 JSON、查模型缓存。副作用：无效请求（坏 body/无权限模型）也计入
|
||
// QPS 计数——这是期望行为，探测型流量同样应被限流兜住。
|
||
billingResult := &BillingResult{}
|
||
if billing != nil {
|
||
allowed, limitLevel, limit, remaining, resetAt := billing.CheckRateLimit(ctx, rc.TenantID, rc.UserID, rc.ApiKeyID, rc.KeyRateLimitQps)
|
||
if !allowed {
|
||
return nil, &RelayErrorWithRateLimit{
|
||
StatusCode: 429,
|
||
Message: fmt.Sprintf("rate limit exceeded at %s level", limitLevel),
|
||
LimitLevel: limitLevel,
|
||
Remaining: remaining,
|
||
ResetAt: resetAt,
|
||
}
|
||
}
|
||
billingResult.RateLimitInfo = &common.RateLimitInfo{
|
||
Limit: limit,
|
||
Remaining: remaining,
|
||
ResetAt: resetAt,
|
||
}
|
||
}
|
||
|
||
relayModeStr := relayModeString(relayMode)
|
||
if billing != nil {
|
||
if !billing.CheckScope(rc.Scope, relayModeStr) {
|
||
return nil, constant.NewAuthError("API key scope denied")
|
||
}
|
||
if !billing.CheckIPWhitelist(rc.KeyIpWhitelist, rc.ClientIP) {
|
||
return nil, constant.NewAuthError("IP address is not allowed")
|
||
}
|
||
}
|
||
|
||
// 2. 解析请求体获取模型名
|
||
var rawRequest map[string]json.RawMessage
|
||
if err := json.Unmarshal(body, &rawRequest); err != nil {
|
||
return nil, constant.NewRequestError("invalid request body", err)
|
||
}
|
||
|
||
modelName := ""
|
||
if v, ok := rawRequest["model"]; ok {
|
||
var m string
|
||
if err := json.Unmarshal(v, &m); err == nil {
|
||
modelName = m
|
||
}
|
||
}
|
||
if modelName == "" {
|
||
return nil, constant.NewRequestError("model is required", nil)
|
||
}
|
||
|
||
// 2.5 验证模型在目录中存在
|
||
lookupModel, modelErr := resolveRelayModel(ctx, provider, modelName)
|
||
if modelErr != nil {
|
||
if modelErr == common.ErrModelNotFound {
|
||
return nil, constant.NewRequestError("model not found: "+modelName, modelErr)
|
||
}
|
||
return nil, modelErr
|
||
}
|
||
|
||
// 3.5 检查模型弃用状态
|
||
depInfo, _ := provider.GetModelDeprecationInfo(ctx, lookupModel)
|
||
if depInfo != nil && depInfo.Deprecated && depInfo.SunsetDate != "" {
|
||
sunsetTime, _ := time.Parse("2006-01-02", depInfo.SunsetDate)
|
||
if !sunsetTime.IsZero() && time.Now().After(sunsetTime) {
|
||
return nil, constant.NewModelGoneError(lookupModel, depInfo.SunsetDate)
|
||
}
|
||
}
|
||
billingResult.Deprecation = depInfo
|
||
|
||
// 3.7 检查成员模型范围
|
||
if allowed, err := provider.CheckMemberModelAccess(ctx, rc.TenantID, rc.UserID, lookupModel); err != nil {
|
||
return nil, err
|
||
} else if !allowed {
|
||
return nil, constant.NewAuthError("model not allowed for this member")
|
||
}
|
||
|
||
// 4. API Key 模型范围校验
|
||
if allowed, err := provider.CheckApiKeyModelAccess(ctx, rc.ApiKeyID, lookupModel); err != nil {
|
||
return nil, err
|
||
} else if !allowed {
|
||
return nil, constant.NewAuthError("model not allowed for this API key")
|
||
}
|
||
|
||
// 5. QPS 限流已前置到步骤 1.5（见上），此处不再检查
|
||
|
||
// 7. 估算输入 token 数
|
||
estimatedInputTokens := estimateInputTokens(body)
|
||
|
||
var isStream bool
|
||
if streamVal, ok := rawRequest["stream"]; ok {
|
||
_ = json.Unmarshal(streamVal, &isStream)
|
||
}
|
||
|
||
var maxTokens int
|
||
if mtVal, ok := rawRequest["max_tokens"]; ok {
|
||
_ = json.Unmarshal(mtVal, &maxTokens)
|
||
}
|
||
if mcVal, ok := rawRequest["max_completion_tokens"]; ok {
|
||
var mc int
|
||
_ = json.Unmarshal(mcVal, &mc)
|
||
if mc > maxTokens {
|
||
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,
|
||
relayModeStr: relayModeStr,
|
||
modelName: modelName,
|
||
lookupModel: lookupModel,
|
||
isStream: isStream,
|
||
estimatedTokens: estimatedInputTokens,
|
||
maxTokens: maxTokens,
|
||
depInfo: depInfo,
|
||
billingResult: billingResult,
|
||
sessionSignals: extractSessionSignals(rawRequest),
|
||
}, nil
|
||
}
|
||
|
||
// extractSessionSignals 从已解析的请求体提取会话信号（解析链的协议内信号部分）。
|
||
// 复用 rawRequest 的一次解析，不额外反序列化整个 body。
|
||
func extractSessionSignals(rawRequest map[string]json.RawMessage) dispatch.SessionSignals {
|
||
var sig dispatch.SessionSignals
|
||
if v, ok := rawRequest["metadata"]; ok {
|
||
var meta struct {
|
||
UserID string `json:"user_id"`
|
||
}
|
||
if json.Unmarshal(v, &meta) == nil {
|
||
sig.AnthropicUserID = meta.UserID
|
||
}
|
||
}
|
||
if v, ok := rawRequest["previous_response_id"]; ok {
|
||
_ = json.Unmarshal(v, &sig.PreviousResponseID)
|
||
}
|
||
if v, ok := rawRequest["conversation_id"]; ok {
|
||
_ = json.Unmarshal(v, &sig.ConversationID)
|
||
}
|
||
return sig
|
||
}
|
||
|
||
// newSettleContext 创建结算/退款/调度上报专用上下文：脱离请求取消（客户端断开不得影响结算），
|
||
// 且超时预算从调用时刻起算。不可在请求发出前创建后跨 DoRequest/DoResponse 复用——
|
||
// 上游请求与流式转发全程可能远超 30s，提前起算的预算会在结算前耗尽，
|
||
// 导致计费记录写入、预扣退款、渠道健康上报拿到已过期的 ctx 而失败。
|
||
func newSettleContext(parent context.Context) (context.Context, context.CancelFunc) {
|
||
return context.WithTimeout(context.WithoutCancel(parent), 30*time.Second)
|
||
}
|
||
|
||
// settleSuccessfulRequest 成功路径的计费结算、健康度更新和用量记录。
|
||
func settleSuccessfulRequest(
|
||
rc *RelayContext,
|
||
v *relayValidation,
|
||
usage *common.Usage,
|
||
info *common.RelayInfo,
|
||
selection *common.ChannelSelection,
|
||
preDeductAmount float64,
|
||
provider common.DataProvider,
|
||
billing common.BillingProvider,
|
||
headers http.Header,
|
||
path string,
|
||
) *BillingResult {
|
||
// 16. 结算费用（使用完整 Usage 含 cache token）
|
||
// 重新创建 context，上游 DoResponse 可能耗时很长（长文本流式输出）
|
||
postCtx, postCancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||
defer postCancel()
|
||
|
||
var settleResult *common.SettlementResult
|
||
if billing != nil && preDeductAmount > 0 {
|
||
settleResult = billing.SettleWithUsage(postCtx, rc.TenantID, rc.UserID, rc.ApiKeyID, selection.ChannelID,
|
||
v.modelName, rc.RequestID, v.relayModeStr,
|
||
usage, preDeductAmount, info)
|
||
if settleResult != nil {
|
||
v.billingResult.ActualCost = settleResult.ActualCost
|
||
} else {
|
||
g.Log().Warningf(postCtx, "[RelayHandler] Settlement failed for request=%s model=%s, refunding pre-deduct amount=%.6f",
|
||
rc.RequestID, v.modelName, preDeductAmount)
|
||
if rErr := billing.SettleFailed(postCtx, rc.TenantID, rc.RequestID, preDeductAmount); rErr != nil {
|
||
g.Log().Errorf(postCtx, "[RelayHandler] Refund pre-deduct failed after settlement failure (needs reconciliation): request=%s tenant=%d err=%v",
|
||
rc.RequestID, rc.TenantID, rErr)
|
||
}
|
||
}
|
||
}
|
||
|
||
// 16.5 累加成员已用额度（幂等重复结算 DuplicateSkip 时不得重复累加）
|
||
if billing != nil && settleResult != nil && settleResult.ActualCost > 0 && !settleResult.DuplicateSkip {
|
||
billing.IncrMemberQuotaUsed(postCtx, rc.TenantID, rc.UserID, settleResult.ActualCost)
|
||
billing.IncrApiKeyQuotaUsed(postCtx, rc.ApiKeyID, settleResult.ActualCost)
|
||
}
|
||
if settleResult != nil && settleResult.ActualCost > 0 && rc.ProjectID > 0 {
|
||
if err := tenantlogic.CheckProjectBudget(postCtx, rc.TenantID, rc.ProjectID); err != nil {
|
||
g.Log().Warningf(postCtx, "[RelayHandler] Check project budget failed: tenant=%d project=%d request=%s err=%v",
|
||
rc.TenantID, rc.ProjectID, rc.RequestID, err)
|
||
}
|
||
}
|
||
|
||
// 17. 记录用量（健康上报与绑定续期已由调度会话 Finish 完成）
|
||
firstTokenMs := 0
|
||
if !info.FirstResponseTime.IsZero() {
|
||
firstTokenMs = int(info.FirstResponseTime.Sub(info.StartTime).Milliseconds())
|
||
}
|
||
|
||
// 构建用量记录。input_tokens 统一按「含缓存总输入」口径入库（Claude 口径未含缓存时补加），
|
||
// 保证跨渠道 SUM 聚合语义一致；total_tokens 同步按 totalInput+output 重算
|
||
totalInput := usage.TotalInputTokens()
|
||
totalTokens := totalInput + usage.CompletionTokens
|
||
if totalTokens == 0 {
|
||
totalTokens = usage.TotalTokens
|
||
}
|
||
usageRecord := &common.UsageRecord{
|
||
TenantID: rc.TenantID,
|
||
UserID: rc.UserID,
|
||
ApiKeyID: rc.ApiKeyID,
|
||
ProjectID: rc.ProjectID,
|
||
ChannelID: selection.ChannelID,
|
||
ModelName: v.modelName,
|
||
RelayMode: int(v.relayMode),
|
||
PromptTokens: totalInput,
|
||
CompletionTokens: usage.CompletionTokens,
|
||
TotalTokens: totalTokens,
|
||
CachedTokens: tokenDetailField(usage.PromptTokensDetails, func(d *common.TokenDetails) int { return d.CachedTokens }),
|
||
AudioTokens: tokenDetailField(usage.PromptTokensDetails, func(d *common.TokenDetails) int { return d.AudioTokens }) +
|
||
tokenDetailField(usage.CompletionTokenDetails, func(d *common.TokenDetails) int { return d.AudioTokens }),
|
||
ImageTokens: tokenDetailField(usage.PromptTokensDetails, func(d *common.TokenDetails) int { return d.ImageTokens }),
|
||
ReasoningTokens: tokenDetailField(usage.CompletionTokenDetails, func(d *common.TokenDetails) int { return d.ReasoningTokens }),
|
||
LatencyMs: info.TotalLatencyMs(),
|
||
IsStream: v.isStream,
|
||
Success: true,
|
||
RequestID: rc.RequestID,
|
||
Status: "success",
|
||
|
||
// Cache token 明细。cache_creation_tokens 列统一记录「写入缓存的 token」：
|
||
// Claude 为 cache_creation_input_tokens，OpenAI Responses 为 cache_write_tokens（两者物理语义相同，
|
||
// 仅计价不同——后者按普通输入价计费且已含于 input_tokens，不参与计费扣减，仅作观测）
|
||
CacheCreationTokens: usage.CacheCreationTokens +
|
||
tokenDetailField(usage.PromptTokensDetails, func(d *common.TokenDetails) int { return d.CacheWriteTokens }),
|
||
CacheCreation5mTokens: tokenDetailField(usage.PromptTokensDetails, func(d *common.TokenDetails) int { return d.CachedCreation5mTokens }),
|
||
CacheCreation1hTokens: tokenDetailField(usage.PromptTokensDetails, func(d *common.TokenDetails) int { return d.CachedCreation1hTokens }),
|
||
CacheReadTokens: tokenDetailField(usage.PromptTokensDetails, func(d *common.TokenDetails) int { return d.CachedTokens }),
|
||
|
||
// 音频 token 分离
|
||
AudioInputTokens: tokenDetailField(usage.PromptTokensDetails, func(d *common.TokenDetails) int { return d.AudioTokens }),
|
||
AudioOutputTokens: tokenDetailField(usage.CompletionTokenDetails, func(d *common.TokenDetails) int { return d.AudioTokens }),
|
||
|
||
// 其他 token
|
||
ImageOutputTokens: tokenDetailField(usage.CompletionTokenDetails, func(d *common.TokenDetails) int { return d.ImageTokens }),
|
||
|
||
// 请求元数据
|
||
RequestedModel: v.modelName,
|
||
UpstreamModel: selection.UpstreamModelName,
|
||
RequestType: requestType(v.isStream),
|
||
UserAgent: headers.Get("User-Agent"),
|
||
ClientIP: rc.ClientIP,
|
||
FirstTokenMs: firstTokenMs,
|
||
InboundEndpoint: path,
|
||
UpstreamRequestID: info.UpstreamRequestID,
|
||
|
||
// 渠道详情
|
||
ChannelName: selection.ChannelName,
|
||
ChannelType: selection.ChannelType,
|
||
|
||
// 重试
|
||
RetryIndex: info.RetryIndex,
|
||
}
|
||
|
||
// 流式追踪：正常结束也落库（done / eof）。eof（上游未发 [DONE] 即结束）按正常结束
|
||
// 结算但属上游侧异常信号，记录后可在用量日志直接区分，无需翻应用日志。
|
||
if info.StreamStatus != nil {
|
||
usageRecord.StreamEndReason = string(info.StreamStatus.GetEndReason())
|
||
}
|
||
|
||
// 填充结算费用数据
|
||
if settleResult != nil {
|
||
usageRecord.TotalCost = settleResult.BaseCost
|
||
usageRecord.ActualCost = settleResult.ActualCost
|
||
usageRecord.Currency = "USD"
|
||
usageRecord.PreDeductAmount = settleResult.PreDeductAmount
|
||
usageRecord.RefundAmount = settleResult.RefundAmount
|
||
usageRecord.SupplementAmount = settleResult.SupplementAmount
|
||
usageRecord.BillingSnapshot = settleResult.BillingSnapshot
|
||
usageRecord.BillingSummary = settleResult.BillingSummary
|
||
usageRecord.BillingMode = settleResult.BillingMode
|
||
usageRecord.BillingSource = settleResult.BillingSource
|
||
usageRecord.RateMultiplier = settleResult.RateMultiplier
|
||
usageRecord.InputCost = settleResult.InputCost
|
||
usageRecord.OutputCost = settleResult.OutputCost
|
||
usageRecord.CacheCreationCost = settleResult.CacheCreationCost
|
||
usageRecord.CacheReadCost = settleResult.CacheReadCost
|
||
}
|
||
|
||
v.billingResult.FirstTokenMs = firstTokenMs
|
||
provider.RecordUsage(postCtx, usageRecord)
|
||
|
||
return v.billingResult
|
||
}
|
||
|
||
// RelayHandler 共享的 relay 请求编排逻辑（带重试 + 计费）
|
||
func RelayHandler(ctx context.Context, body []byte, path string, headers http.Header, rc *RelayContext, provider common.DataProvider, billing common.BillingProvider) (*common.Usage, *BillingResult, error) {
|
||
// 客户端已断开的请求直接终止（fail-fast）：断开风暴下若继续空转流水线，
|
||
// 额度检查会误报 "load failed ... skipping check"、调度器会把 canceled ctx
|
||
// 误报成"无可用渠道"，还平白消耗限流/并发/DB 资源。
|
||
if err := ctx.Err(); err != nil {
|
||
return nil, nil, err
|
||
}
|
||
|
||
// 校验请求（relay mode、模型、权限、限流）
|
||
v, err := validateRelayRequest(ctx, body, path, rc, provider, billing)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
|
||
// 在写入任何响应体之前设置限流/弃用 header（流式响应中 WriteHeader 后无法追加）
|
||
setPreResponseHeaders(rc.Writer, v.billingResult)
|
||
|
||
// 资源准备（并发控制 + 监控注册 + 预扣）
|
||
if billing != nil {
|
||
if !billing.AcquireConcurrent(ctx, rc.TenantID, rc.UserID, rc.ApiKeyID, v.modelName) {
|
||
return nil, nil, constant.NewRateLimitError("concurrent request limit exceeded")
|
||
}
|
||
defer billing.ReleaseConcurrent(ctx, rc.TenantID, rc.UserID, rc.ApiKeyID, v.modelName)
|
||
|
||
if !billing.AcquireApiKeyConcurrent(ctx, rc.ApiKeyID, rc.KeyConcurrency) {
|
||
return nil, nil, constant.NewRateLimitError("API key concurrent request limit exceeded")
|
||
}
|
||
defer billing.ReleaseApiKeyConcurrent(ctx, rc.ApiKeyID)
|
||
}
|
||
|
||
monitor.RegisterRequest(&monitor.TrackedRequest{
|
||
RequestID: rc.RequestID,
|
||
TenantID: rc.TenantID,
|
||
UserID: rc.UserID,
|
||
ProjectID: rc.ProjectID,
|
||
ModelName: v.modelName,
|
||
IsStream: v.isStream,
|
||
StartTime: time.Now(),
|
||
Path: path,
|
||
})
|
||
defer monitor.UnregisterRequest(rc.RequestID)
|
||
|
||
var preDeductAmount float64
|
||
if billing != nil {
|
||
amt, err := billing.PreDeduct(ctx, rc.TenantID, v.modelName, v.estimatedTokens, v.maxTokens, v.isStream, rc.RequestID)
|
||
if err != nil {
|
||
// 未配价模型 fail-closed：返回明确的请求错误，而非误导性的"余额不足"
|
||
if errors.Is(err, common.ErrModelPricingNotConfigured) {
|
||
return nil, nil, constant.NewRequestError("model pricing not configured: "+v.modelName, err)
|
||
}
|
||
return nil, nil, constant.NewQuotaError("insufficient balance", err)
|
||
}
|
||
preDeductAmount = amt
|
||
v.billingResult.PreDeductAmount = amt
|
||
// 成员/Key 额度检查只做这一次（带实际预扣额）：额度是"控制线"计数器而非资源冻结，
|
||
// 预扣前后检查的竞态语义相同。原先预扣前额外的 Check(0) 快速闸门对全量请求
|
||
// 各多付一次 Redis 往返，只为让"额度已耗尽"的请求少走一步预扣+退款，得不偿失
|
||
//（该场景已被上游 QPS 限流兜底），故移除。
|
||
if err := billing.CheckApiKeyQuota(ctx, rc.ApiKeyID, amt); err != nil {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, amt)
|
||
return nil, nil, constant.NewQuotaError("API key quota exceeded", err)
|
||
}
|
||
if err := billing.CheckMemberQuota(ctx, rc.TenantID, rc.UserID, amt); err != nil {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, amt)
|
||
return nil, nil, constant.NewQuotaError("member quota exceeded", err)
|
||
}
|
||
}
|
||
|
||
// 渠道调度与请求执行（新调度引擎：Route / Next / Report / Finish）
|
||
// 重试语义由调度核心的 FSM 决定（三预算：原地/凭证轮换/failover），handler 只执行决策。
|
||
channelErrors := make([]string, 0)
|
||
// responses 有状态协议不匹配的失败次数：与 channelErrors 总数相等说明本次请求的全部失败
|
||
// 均为协议约束（无其他渠道故障），无渠道出口应返回端点切换指引而非"模型不可用"
|
||
statefulMismatchCount := 0
|
||
|
||
trace := &common.ForwardingTrace{
|
||
EntryPath: path,
|
||
EntryFormat: string(relayModeToInboundFormat(v.relayMode)),
|
||
RequestedModel: v.modelName,
|
||
}
|
||
|
||
// 租户模型权限与渠道范围校验（原 GetChannelForModel 的第一步，不属于调度模块）
|
||
modelEnabled, channelScope, err := provider.CheckTenantModelAccess(ctx, rc.TenantID, v.lookupModel)
|
||
if err != nil {
|
||
// 客户端已断开：权限查询被中断不是渠道/权限问题，静默退款退出，
|
||
// 不记失败用量、不打无可用渠道告警（与下方 MaterializeSelection 的处理一致）
|
||
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(context.WithoutCancel(ctx), rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
return nil, v.billingResult, err
|
||
}
|
||
result := handleChannelUnavailable(ctx, billing, provider, rc, v, preDeductAmount, channelErrors, err, "")
|
||
return result.usage, result.billingResult, result.err
|
||
}
|
||
if !modelEnabled {
|
||
result := handleChannelUnavailable(ctx, billing, provider, rc, v, preDeductAmount, channelErrors, common.ErrTenantModelNotEnabled, "")
|
||
return result.usage, result.billingResult, result.err
|
||
}
|
||
|
||
sig := v.sessionSignals
|
||
sig.HeaderSessionID = headers.Get("X-Session-Id")
|
||
co := dispatchadapter.Coordinator(ctx)
|
||
sess := co.Route(ctx, dispatch.RequestProfile{
|
||
RequestID: rc.RequestID,
|
||
TenantID: rc.TenantID,
|
||
UserID: rc.UserID,
|
||
APIKeyID: rc.ApiKeyID,
|
||
Model: v.lookupModel,
|
||
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 幂等，成功路径的显式调用优先生效）
|
||
defer sess.Finish(context.WithoutCancel(ctx), false, 0)
|
||
|
||
for attempt := 0; ; attempt++ {
|
||
d := sess.Next(ctx)
|
||
if d == nil {
|
||
// 客户端已断开：调度器内部的 Redis 操作（快照/租约/绑定）被 canceled ctx
|
||
// 中断会返回空决策，这不是"无可用渠道"——静默退款退出，不记失败用量、
|
||
// 不打无可用渠道告警、不污染渠道健康统计
|
||
if ctxErr := ctx.Err(); ctxErr != nil {
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(context.WithoutCancel(ctx), rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
return nil, v.billingResult, ctxErr
|
||
}
|
||
// 无可用渠道：附带调度器的排除原因摘要（各原因独立计数，仅列非零项）。
|
||
// 例：熔断OPEN×1 / 容量租约满×1（并发超 softLimit） / 半开探测限流×1（恢复期每窗口仅 1 个探测）
|
||
// 全部失败均为 responses 有状态协议不匹配时，属客户端可自救的协议约束而非运营故障，
|
||
// 静默返回端点切换指引（与上方 Abort 路径同文案），不误导为"当前模型暂时不可用"，
|
||
// 也不按全部渠道失败记 ERROR 告警/失败用量；事件仍由审计日志与调度监控计数覆盖
|
||
if len(channelErrors) > 0 && statefulMismatchCount == len(channelErrors) {
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
return nil, v.billingResult, constant.NewRequestError(
|
||
constant.StatefulResponsesSwitchEndpointHint, nil)
|
||
}
|
||
result := handleChannelUnavailable(ctx, billing, provider, rc, v, preDeductAmount, channelErrors, common.ErrChannelUnavailable, sess.NoChannelDiagnosis().Summary())
|
||
return result.usage, result.billingResult, result.err
|
||
}
|
||
appendSchedulerDecision(trace, d, attempt)
|
||
|
||
selection, mErr := provider.MaterializeSelection(ctx, d.Channel.ID, d.KeyID, v.lookupModel)
|
||
if mErr != nil {
|
||
// 客户端已断开（context.Canceled / DeadlineExceeded）：Key 查询被中断不代表渠道有问题，
|
||
// 静默退出，不记入 channelErrors，不触发渠道健康惩罚，退还预扣款项后返回
|
||
if errors.Is(mErr, context.Canceled) || errors.Is(mErr, context.DeadlineExceeded) {
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(context.WithoutCancel(ctx), rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
return nil, v.billingResult, mErr
|
||
}
|
||
// Key 解密失败 / 目录元数据缺失：按渠道级致命上报换渠道，不发起上游请求
|
||
channelErrors = append(channelErrors, fmt.Sprintf("attempt=%d channel=%d materialize_error=[%v]", attempt, d.Channel.ID, mErr))
|
||
if decision := reportMaterializeFailure(ctx, sess, mErr); decision == dispatch.DecisionAbort {
|
||
result := handleChannelUnavailable(ctx, billing, provider, rc, v, preDeductAmount, channelErrors, common.ErrChannelUnavailable, "")
|
||
return result.usage, result.billingResult, result.err
|
||
}
|
||
continue
|
||
}
|
||
selection.SelectionReason = string(d.Reason)
|
||
selection.Weight = int(d.Channel.BaseWeight)
|
||
selection.HealthScore = d.Breakdown.Health * 100
|
||
|
||
// 阿里云百炼（DashScope）异步图片模型（wanx*、qwen-image-plus 等）上游为异步任务式
|
||
// （提交拿 task_id → 轮询取图），其 image-synthesis 端点强制异步，无法经同步
|
||
// /v1/images/generations 一次性返回图片。命中时直接引导客户端改用异步端点。
|
||
// 判定收敛到 constant.IsAsyncImageModel（按 provider + 模型），qwen-image-2.x 等
|
||
// 同步 multimodal 模型不在此列，会继续走下方同步转发。租户模型列表 async_image 标记同源。
|
||
if v.relayMode == constant.RelayModeImagesGenerations &&
|
||
constant.IsAsyncImageModel(constant.ProviderType(selection.ChannelType), selection.UpstreamModelName) {
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
return nil, v.billingResult, constant.NewRequestError(
|
||
"this image model uses asynchronous generation on Alibaba DashScope; submit via POST /v1/images/generations/async and poll for the result", nil)
|
||
}
|
||
|
||
// 渠道调试日志：首个开启调试且匹配目标过滤（租户/成员/密钥，AND 组合，0=不限）的尝试
|
||
// 创建会话（捕获段1客户端请求 + 包装段4 writer，writer 保持到请求结束——中间失败尝试
|
||
// 不向客户端写字节，累积字节即最终响应）；每轮尝试独立捕获器经 attemptCtx 传递给传输层。
|
||
// 必须用独立变量 attemptCtx，不可重赋值 ctx：否则陈旧捕获器会泄漏进下一轮 sess.Next
|
||
// 选中无调试渠道的请求。
|
||
var dbgAttempt *common.DebugAttempt
|
||
if selection.Settings.DebugLogEnabled &&
|
||
selection.Settings.DebugTargetMatch(rc.TenantID, rc.UserID, rc.ApiKeyID) {
|
||
if rc.Debug == nil {
|
||
rc.Debug = common.NewDebugSession(rc.RequestID, rc.TenantID, rc.UserID, rc.ApiKeyID, path)
|
||
rc.Debug.CaptureClientRequest(headers, body)
|
||
dw := common.NewDebugClientWriter(rc.Writer)
|
||
rc.Debug.SetClientWriter(dw)
|
||
rc.Writer = dw
|
||
}
|
||
dbgAttempt = rc.Debug.BeginAttempt(selection, v.modelName, v.relayModeStr, v.isStream, attempt)
|
||
}
|
||
attemptCtx := ctx
|
||
if dbgAttempt != nil {
|
||
attemptCtx = common.WithDebugAttempt(ctx, dbgAttempt.Capture)
|
||
}
|
||
|
||
info := buildRelayInfo(attemptCtx, rc, v, selection, path, headers, attempt)
|
||
dbgAttempt.CaptureProtocol(info)
|
||
|
||
if tr := monitor.GetTrackedRequest(rc.RequestID); tr != nil {
|
||
tr.ChannelID = selection.ChannelID
|
||
tr.ChannelName = selection.ChannelName
|
||
}
|
||
|
||
adaptor := channel.GetAdaptor(selection.ChannelType)
|
||
if adaptor == nil {
|
||
g.Log().Errorf(ctx, "[RelayHandler] No adaptor found for channelType: %d", selection.ChannelType)
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
dbgAttempt.MarkFinal(fmt.Errorf("unsupported channel type: %d", selection.ChannelType))
|
||
return nil, v.billingResult, fmt.Errorf("unsupported channel type: %d", selection.ChannelType)
|
||
}
|
||
adaptor.Init(info)
|
||
|
||
hop := buildTraceHop(attempt, selection, adaptor, info)
|
||
|
||
if info.ClientFormat == "" {
|
||
info.ClientFormat = info.InboundFormat
|
||
}
|
||
|
||
// 转换请求（直连模式跳过协议转换和参数改写）
|
||
convertedBody, err := convertRequestBody(ctx, info, body, adaptor)
|
||
if err != nil {
|
||
// responses 有状态请求（previous_response_id）落在 chat-only 渠道：协议能力不匹配
|
||
// 而非请求错误，换渠道可能命中 Responses 原生渠道——按渠道级致命上报驱动 failover，
|
||
// 而非直接返回（直接返回会让有能力渠道存在时也失败）
|
||
if errors.Is(err, constant.ErrStatefulResponsesUnsupported) {
|
||
statefulMismatchCount++
|
||
channelErrors = append(channelErrors, fmt.Sprintf("attempt=%d channel=%d(%s) model=%s convert_error=[%v]",
|
||
attempt, selection.ChannelID, selection.ChannelName, v.modelName, err))
|
||
if reportProtocolMismatch(ctx, sess, err) == dispatch.DecisionAbort {
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
dbgAttempt.MarkFinal(err)
|
||
return nil, v.billingResult, constant.NewRequestError(
|
||
constant.StatefulResponsesSwitchEndpointHint, nil)
|
||
}
|
||
dbgAttempt.Submit(err)
|
||
continue
|
||
}
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
dbgAttempt.MarkFinal(err)
|
||
return nil, v.billingResult, err
|
||
}
|
||
|
||
// 容量租约已由调度器在 Next 内获取；长流式请求由续期器保活
|
||
leaseRefresh := startDispatchLeaseRefresher(d.Channel.ID, rc.RequestID)
|
||
|
||
upstreamCtx := context.WithoutCancel(attemptCtx)
|
||
|
||
// 发送请求到上游
|
||
resp, err := adaptor.DoRequest(upstreamCtx, info, convertedBody)
|
||
if err != nil {
|
||
leaseRefresh.Stop()
|
||
// 送达状态标注：连接拒绝/DNS/TLS 建连失败 = 确定未送达；
|
||
// 写出后 EOF/RST/读超时 = 可能已送达。ReplayUnsafe+MaybeSent 由 FSM 硬规则终止。
|
||
delivery := deliveryStateOfRequestErr(err)
|
||
// 结算/上报 ctx 此刻新建：上游挂死可耗时数分钟，请求开始时起算的预算早已耗尽
|
||
settleCtx, settleCancel := newSettleContext(ctx)
|
||
defer settleCancel()
|
||
decision, backoff := sess.Report(settleCtx, dispatchStatusCode(err), err, delivery, info.LatencyMs(), 0)
|
||
trackRetryDecision(dispatchStatusCode(err), err, delivery, decision)
|
||
|
||
failReason := fmt.Sprintf("attempt=%d channel=%d(%s) model=%s upstreamModel=%s error=[%v] latency=%.0fms decision=%s",
|
||
attempt, selection.ChannelID, selection.ChannelName, v.modelName, selection.UpstreamModelName, err, info.LatencyMs(), decision)
|
||
channelErrors = append(channelErrors, failReason)
|
||
g.Log().Warningf(ctx, "[RelayHandler] Upstream request failed: %s", failReason)
|
||
|
||
if decision != dispatch.DecisionAbort {
|
||
recordChannelError(rc, selection, v.modelName, attempt, false, err, info.LatencyMs())
|
||
appendHop(trace, hop, false, err.Error(), info.LatencyMs())
|
||
settleCancel()
|
||
sleepBackoff(ctx, backoff)
|
||
dbgAttempt.Submit(err)
|
||
continue
|
||
}
|
||
|
||
if billing != nil && preDeductAmount > 0 {
|
||
// 退款失败意味着预扣冻结需等 2h TTL 清扫才释放，留痕供对账
|
||
if rErr := billing.SettleFailed(settleCtx, rc.TenantID, rc.RequestID, preDeductAmount); rErr != nil {
|
||
g.Log().Errorf(settleCtx, "[RelayHandler] Refund pre-deduct failed after upstream request failure (needs reconciliation): request=%s tenant=%d err=%v",
|
||
rc.RequestID, rc.TenantID, rErr)
|
||
}
|
||
}
|
||
recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, info.StreamStatus, info.UpstreamRequestID)
|
||
recordChannelError(rc, selection, v.modelName, attempt, true, err, info.LatencyMs())
|
||
finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs())
|
||
dbgAttempt.MarkFinal(err)
|
||
return nil, v.billingResult, helper.RemapStatusCode(constant.NewUpstreamError(502, "请求处理失败", err), info.ChannelMeta.Settings.StatusCodeMapping)
|
||
}
|
||
|
||
if !v.isStream {
|
||
info.SetFirstResponseTime()
|
||
}
|
||
|
||
// 捕获上游请求 ID（所有对话类渠道适配器的响应统一经过此处）：
|
||
// 流式/非流式响应头在首帧/响应体之前就已就位。存入 info 使失败路径
|
||
// （DoResponse 出错、流中断）的用量记录同样携带，排障错误请求往往更有价值；
|
||
// 重试时每次尝试覆盖，终值 = 最后一次尝试。
|
||
info.UpstreamRequestID = helper.ExtractUpstreamRequestID(resp.Header)
|
||
|
||
// 处理上游响应
|
||
usage, err := adaptor.DoResponse(ctx, resp, info, rc.Writer)
|
||
leaseRefresh.Stop()
|
||
if err != nil {
|
||
err = helper.RemapStatusCode(err, info.ChannelMeta.Settings.StatusCodeMapping)
|
||
|
||
if info.StreamStatus != nil && info.StreamStatus.IsPartialStreamEnd() {
|
||
g.Log().Warningf(ctx, "[RelayHandler] Stream interrupted: request=%s tenant=%d channel=%d(%s) adaptor=%s model=%s attempt=%d reason=%s ctx_err=%v usage=%v pre_deduct=%.6f",
|
||
rc.RequestID, rc.TenantID, selection.ChannelID, selection.ChannelName, adaptor.GetChannelName(), v.modelName,
|
||
attempt, info.StreamStatus.Summary(), ctx.Err(), usage, preDeductAmount)
|
||
// 客户端已收到部分流：不可重试（响应已污染），上报健康后按流中断结算
|
||
// 结算 ctx 此刻新建：流式转发全程可能远超 30s，且请求 ctx 已随客户端断开而取消
|
||
settleCtx, settleCancel := newSettleContext(ctx)
|
||
defer settleCancel()
|
||
_, _ = sess.Report(settleCtx, dispatchStatusCode(err), err, dispatch.DeliveryResponseStarted, info.LatencyMs(), 0)
|
||
streamUsage := usage
|
||
if streamUsage == nil {
|
||
streamUsage = &common.Usage{}
|
||
}
|
||
var interruptSettle *common.SettlementResult
|
||
if billing != nil && preDeductAmount > 0 {
|
||
settleResult, err := billing.SettleStreamInterrupted(settleCtx, rc.TenantID, rc.UserID, rc.ApiKeyID, selection.ChannelID,
|
||
v.modelName, rc.RequestID, v.relayModeStr, streamUsage, preDeductAmount, rc.ProjectID, info.StartTime)
|
||
if err != nil {
|
||
g.Log().Warningf(settleCtx, "[RelayHandler] Stream interrupted settlement failed: tenant=%d project=%d request=%s err=%v",
|
||
rc.TenantID, rc.ProjectID, rc.RequestID, err)
|
||
} else if rc.ProjectID > 0 {
|
||
if err := tenantlogic.CheckProjectBudget(settleCtx, rc.TenantID, rc.ProjectID); err != nil {
|
||
g.Log().Warningf(settleCtx, "[RelayHandler] Check project budget failed: tenant=%d project=%d request=%s err=%v",
|
||
rc.TenantID, rc.ProjectID, rc.RequestID, err)
|
||
}
|
||
}
|
||
if settleResult != nil && settleResult.ActualCost > 0 && !settleResult.DuplicateSkip {
|
||
billing.IncrMemberQuotaUsed(settleCtx, rc.TenantID, rc.UserID, settleResult.ActualCost)
|
||
billing.IncrApiKeyQuotaUsed(settleCtx, rc.ApiKeyID, settleResult.ActualCost)
|
||
}
|
||
interruptSettle = settleResult
|
||
}
|
||
recordFailedUsageWithTokens(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, streamUsage, interruptSettle, info.StreamStatus, info.UpstreamRequestID)
|
||
finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs())
|
||
dbgAttempt.MarkFinal(err)
|
||
return usage, v.billingResult, err
|
||
}
|
||
|
||
// 送达状态判定：
|
||
// 1. adaptor 显式标记 ResponseWritten → 已写出，必须终止；
|
||
// 2. 流式请求且流已实际开始（StreamStatus 已有结束原因）→ 已写出 SSE 数据，必须终止；
|
||
// 3. 流式请求但流尚未开始（上游在 SSE 头之前返回错误，StreamStatus 无结束原因）
|
||
// → 未向客户端写出任何字节，允许换渠道重试；
|
||
// 4. 非流式请求 → 已收到完整错误响应，可按分类决定原地重试/换渠道。
|
||
delivery := dispatch.DeliveryResponseReceived
|
||
if constant.IsResponseWritten(err) {
|
||
delivery = dispatch.DeliveryResponseStarted
|
||
} else if v.isStream && (info.StreamStatus == nil || info.StreamStatus.GetEndReason() != "") {
|
||
// StreamStatus 为 nil：保守兜底，按旧行为终止（遗留流式处理器可能未初始化 StreamStatus）
|
||
// GetEndReason() != ""：流已开始传输后出错，字节已提交给客户端，必须终止
|
||
delivery = dispatch.DeliveryResponseStarted
|
||
}
|
||
// 结算/上报 ctx 此刻新建：DoResponse（含长流式转发）可能已消耗任意时长
|
||
settleCtx, settleCancel := newSettleContext(ctx)
|
||
defer settleCancel()
|
||
decision, backoff := sess.Report(settleCtx, dispatchStatusCode(err), err, delivery, info.LatencyMs(), retryAfterOf(err))
|
||
trackRetryDecision(dispatchStatusCode(err), err, delivery, decision)
|
||
|
||
if decision != dispatch.DecisionAbort {
|
||
g.Log().Warningf(ctx, "[RelayHandler] DoResponse failed (will retry): adaptor=%s, inboundFormat=%s, channel=%d(%s) model=%s attempt=%d error=%v latency=%.0fms decision=%s",
|
||
adaptor.GetChannelName(), info.InboundFormat, selection.ChannelID, selection.ChannelName, v.modelName, attempt, err, info.LatencyMs(), decision)
|
||
} else {
|
||
// 上游响应错误属于预期内的运营事件（4xx/5xx/超时等），非代码 bug：
|
||
// 用 Warningf 避免触发 glog 默认对 ERROR+ 级别自动打印调用栈（StStatus=1），
|
||
// 堆栈只会复述对所有请求都一样的中间件链，污染日志且无诊断价值。
|
||
g.Log().Warningf(ctx, "[RelayHandler] DoResponse failed (abort): adaptor=%s, inboundFormat=%s, channel=%d(%s) model=%s attempt=%d error=%v latency=%.0fms decision=%s",
|
||
adaptor.GetChannelName(), info.InboundFormat, selection.ChannelID, selection.ChannelName, v.modelName, attempt, err, info.LatencyMs(), decision)
|
||
}
|
||
channelErrors = append(channelErrors, fmt.Sprintf("attempt=%d channel=%d(%s) model=%s doResponse_error=[%v] latency=%.0fms",
|
||
attempt, selection.ChannelID, selection.ChannelName, v.modelName, err, info.LatencyMs()))
|
||
|
||
if decision != dispatch.DecisionAbort {
|
||
recordChannelError(rc, selection, v.modelName, attempt, false, err, info.LatencyMs())
|
||
appendHop(trace, hop, false, err.Error(), info.LatencyMs())
|
||
settleCancel()
|
||
sleepBackoff(ctx, backoff)
|
||
dbgAttempt.Submit(err)
|
||
continue
|
||
}
|
||
|
||
if billing != nil && preDeductAmount > 0 {
|
||
// 退款失败意味着预扣冻结需等 2h TTL 清扫才释放，留痕供对账
|
||
if rErr := billing.SettleFailed(settleCtx, rc.TenantID, rc.RequestID, preDeductAmount); rErr != nil {
|
||
g.Log().Errorf(settleCtx, "[RelayHandler] Refund pre-deduct failed after response failure (needs reconciliation): request=%s tenant=%d err=%v",
|
||
rc.RequestID, rc.TenantID, rErr)
|
||
}
|
||
}
|
||
recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, info.StreamStatus, info.UpstreamRequestID)
|
||
recordChannelError(rc, selection, v.modelName, attempt, true, err, info.LatencyMs())
|
||
finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs())
|
||
dbgAttempt.MarkFinal(err)
|
||
return nil, v.billingResult, err
|
||
}
|
||
|
||
// 成功路径：绑定续期 + 健康上报由 Finish 完成
|
||
dbgAttempt.MarkFinal(nil)
|
||
finishCtx, finishCancel := newSettleContext(ctx)
|
||
defer finishCancel()
|
||
sess.Finish(finishCtx, true, info.LatencyMs())
|
||
appendHop(trace, hop, true, "", info.LatencyMs())
|
||
trace.TotalAttempts = attempt + 1
|
||
trace.UpstreamModel = selection.UpstreamModelName
|
||
trace.ModelMapped = selection.IsModelMapped
|
||
rc.ForwardingTrace = trace
|
||
|
||
result := settleSuccessfulRequest(rc, v, usage, info, selection, preDeductAmount, provider, billing, headers, path)
|
||
return usage, result, nil
|
||
}
|
||
}
|
||
|
||
// buildRelayInfo 从渠道选择结果构建 RelayInfo
|
||
// attempt 为当前重试轮次（0=首次），写入 RetryIndex 供 ParamOverride「是否重试」规则与 bil_usage_logs.retry_index 使用（C3）。
|
||
func buildRelayInfo(ctx context.Context, rc *RelayContext, v *relayValidation, selection *common.ChannelSelection, path string, headers http.Header, attempt int) *common.RelayInfo {
|
||
info := &common.RelayInfo{
|
||
Context: ctx,
|
||
TenantID: rc.TenantID,
|
||
UserID: rc.UserID,
|
||
ApiKeyID: rc.ApiKeyID,
|
||
ProjectID: rc.ProjectID,
|
||
RequestID: rc.RequestID,
|
||
RetryIndex: attempt,
|
||
RelayMode: int(v.relayMode),
|
||
IsStream: v.isStream,
|
||
OriginModelName: v.modelName,
|
||
BaseModelName: v.lookupModel,
|
||
RequestURLPath: path,
|
||
RequestHeaders: headers,
|
||
StartTime: time.Now(),
|
||
StreamStatus: common.NewStreamStatus(),
|
||
InboundFormat: relayModeToInboundFormat(v.relayMode),
|
||
ChannelMeta: &common.ChannelMeta{
|
||
ChannelID: selection.ChannelID,
|
||
ChannelType: selection.ChannelType,
|
||
ChannelName: selection.ChannelName,
|
||
BaseURL: selection.BaseURL,
|
||
ApiKey: selection.ApiKey,
|
||
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
|
||
}
|
||
|
||
// convertRequestBody 根据是否直连模式转换请求体
|
||
func convertRequestBody(ctx context.Context, info *common.RelayInfo, body []byte, adaptor common.Adaptor) (io.Reader, error) {
|
||
if canPassThrough(info) {
|
||
if info.InboundFormat == constant.RelayFormatGemini {
|
||
body = helper.StripStreamField(body)
|
||
}
|
||
return bytes.NewReader(body), nil
|
||
}
|
||
|
||
// relaykit 转换器路径（唯一路径，hard-fail）：方向命中后解析/转换失败直接拒绝请求；
|
||
// 未命中方向（同格式直连、非文本模式等）走 adaptor 的原生后处理路径。
|
||
var convertedBody io.Reader
|
||
relaykitBody, handled, relaykitErr := tryConvertRequestViaRelaykit(ctx, info, body)
|
||
switch {
|
||
case handled && relaykitErr != nil:
|
||
if !errors.Is(relaykitErr, constant.ErrStatefulResponsesUnsupported) {
|
||
g.Log().Errorf(ctx, "[RelayHandler] relaykit ConvertRequest failed: inboundFormat=%s, error=%v",
|
||
info.InboundFormat, relaykitErr)
|
||
}
|
||
return nil, relaykitErr
|
||
case handled:
|
||
convertedBody = relaykitBody
|
||
// 供应商私有后处理：矩阵按格式裁决、与供应商无关，命中后 adaptor.ConvertRequest
|
||
// 不再被调用，其中的私有请求适配（zhipu GLM 参数兼容、ali DashScope 参数裁剪）
|
||
// 会随之丢失。此处按可选接口接回，仅 relaykit 路径调用。
|
||
if pp, ok := adaptor.(common.RequestPostProcessor); ok {
|
||
relaykitBytes, err := io.ReadAll(convertedBody)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("read relaykit converted body: %w", err)
|
||
}
|
||
processed, err := pp.PostProcessConvertedRequest(ctx, info, relaykitBytes)
|
||
if err != nil {
|
||
g.Log().Errorf(ctx, "[RelayHandler] provider request post-process failed: adaptor=%s, inboundFormat=%s, error=%v",
|
||
adaptor.GetChannelName(), info.InboundFormat, err)
|
||
return nil, err
|
||
}
|
||
convertedBody = bytes.NewReader(processed)
|
||
}
|
||
default:
|
||
legacyBody, err := adaptor.ConvertRequest(ctx, info, body)
|
||
if err != nil {
|
||
// responses 有状态协议不匹配为预期内错误（上层按哨兵驱动换渠道并向客户端返回
|
||
// 端点切换指引），不打 ERRO 噪音；其余转换失败仍按内部错误记录
|
||
if !errors.Is(err, constant.ErrStatefulResponsesUnsupported) {
|
||
g.Log().Errorf(ctx, "[RelayHandler] ConvertRequest failed: adaptor=%s, inboundFormat=%s, error=%v",
|
||
adaptor.GetChannelName(), info.InboundFormat, err)
|
||
}
|
||
return nil, err
|
||
}
|
||
convertedBody = legacyBody
|
||
}
|
||
|
||
// 注入渠道系统提示词
|
||
if info.ChannelMeta.Settings.SystemPrompt != "" {
|
||
bodyBytes, _ := io.ReadAll(convertedBody)
|
||
bodyBytes = helper.InjectSystemPrompt(bodyBytes, info)
|
||
convertedBody = bytes.NewReader(bodyBytes)
|
||
}
|
||
|
||
// 应用请求体改写（ParamOverride）
|
||
if info.ChannelMeta.Settings.ParamOverride != nil {
|
||
bodyBytes, err := io.ReadAll(convertedBody)
|
||
if err != nil {
|
||
g.Log().Errorf(ctx, "[RelayHandler] Read converted body for param override failed: %v", err)
|
||
return nil, err
|
||
}
|
||
bodyBytes, err = override.ApplyParamOverride(bodyBytes, info)
|
||
if err != nil {
|
||
if retErr, ok := override.AsReturnError(err); ok {
|
||
return nil, constant.NewUpstreamError(retErr.StatusCode, retErr.Message, retErr)
|
||
}
|
||
return nil, err
|
||
}
|
||
convertedBody = bytes.NewReader(bodyBytes)
|
||
}
|
||
|
||
// 字段清理
|
||
sanitized, _ := io.ReadAll(convertedBody)
|
||
sanitized = helper.SanitizeFields(sanitized, info.ChannelMeta.Settings)
|
||
return bytes.NewReader(sanitized), nil
|
||
}
|
||
|
||
// channelUnavailableResult 渠道不可用时的返回值
|
||
type channelUnavailableResult struct {
|
||
usage *common.Usage
|
||
billingResult *BillingResult
|
||
err error
|
||
}
|
||
|
||
// handleChannelUnavailable 处理渠道选择失败：退还预扣、记录错误
|
||
func handleChannelUnavailable(
|
||
ctx context.Context,
|
||
billing common.BillingProvider,
|
||
provider common.DataProvider,
|
||
rc *RelayContext,
|
||
v *relayValidation,
|
||
preDeductAmount float64,
|
||
channelErrors []string,
|
||
err error,
|
||
noChannelDiag string, // 调度器无可用渠道时的排除原因摘要（NoChannelDiag.Summary()），无诊断信息传 ""
|
||
) *channelUnavailableResult {
|
||
if err != common.ErrChannelUnavailable {
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
return &channelUnavailableResult{nil, v.billingResult, err}
|
||
}
|
||
|
||
diagSuffix := ""
|
||
if noChannelDiag != "" {
|
||
diagSuffix = " 原因: " + noChannelDiag
|
||
}
|
||
|
||
if len(channelErrors) > 0 {
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
// 全部渠道失败是真实运营告警，保留 ERROR 级别；但调用栈固定无意义，禁用堆栈打印
|
||
g.Log().Stack(false).Errorf(ctx, "[RelayHandler] All %d channels failed for model=%s tenant=%d user=%d request=%s.%s Failure details: %s",
|
||
len(channelErrors), v.modelName, rc.TenantID, rc.UserID, rc.RequestID, diagSuffix, strings.Join(channelErrors, "\n"))
|
||
// 用户侧只返回通用错误，不暴露渠道数量和详情
|
||
allFailedErr := constant.NewChannelError(
|
||
"当前模型暂时不可用",
|
||
constant.ErrAllChannelsFailed,
|
||
)
|
||
recordFailedUsage(provider, rc, nil, v.modelName, v.relayMode, v.isStream, allFailedErr, nil, "")
|
||
return &channelUnavailableResult{nil, v.billingResult, allFailedErr}
|
||
}
|
||
|
||
if billing != nil && preDeductAmount > 0 {
|
||
_ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount)
|
||
}
|
||
// 无可用渠道属于正常业务条件（模型未配渠道/容量满/熔断中等），降级为 Warning 并禁用堆栈打印；
|
||
// 具体原因由调度器的排除明细摘要给出（熔断OPEN/半开探测限流/容量租约满/凭证冷却/目录为空）
|
||
g.Log().Stack(false).Warningf(ctx, "[RelayHandler] 无可用渠道: model=%s tenant=%d user=%d request=%s%s",
|
||
v.modelName, rc.TenantID, rc.UserID, rc.RequestID, diagSuffix)
|
||
// 用户侧返回通用错误，不暴露 "channel" 字眼
|
||
noChErr := constant.NewChannelError("当前模型暂时不可用", err)
|
||
// 无可用渠道时请求未转发到任何上游（与 QPS 限流、权限/额度拒绝等前置拦截同理），
|
||
// 不记用量日志，避免零消耗的 error 记录污染请求日志；运营侧的“被拒请求量”已由
|
||
// 上方 Warning 日志、调度器排除明细指标、chn_error_events 覆盖。
|
||
return &channelUnavailableResult{nil, v.billingResult, noChErr}
|
||
}
|
||
|
||
// buildTraceHop 构建转发追踪的单跳记录
|
||
func buildTraceHop(attempt int, selection *common.ChannelSelection, adaptor common.Adaptor, info *common.RelayInfo) common.ForwardingHop {
|
||
var upstreamURL string
|
||
if u, err := adaptor.GetRequestURL(info); err == nil {
|
||
upstreamURL = u
|
||
}
|
||
return common.ForwardingHop{
|
||
Attempt: attempt,
|
||
ChannelID: selection.ChannelID,
|
||
ChannelName: selection.ChannelName,
|
||
ChannelType: selection.ChannelType,
|
||
Provider: constant.ProviderType(selection.ChannelType).String(),
|
||
BaseURL: selection.BaseURL,
|
||
UpstreamURL: upstreamURL,
|
||
UpstreamModel: selection.UpstreamModelName,
|
||
ModelMapped: selection.IsModelMapped,
|
||
SelectionReason: selection.SelectionReason,
|
||
Priority: selection.Priority,
|
||
Weight: selection.Weight,
|
||
HealthScore: selection.HealthScore,
|
||
}
|
||
}
|
||
|
||
// appendHop 追加一条 hop 到 trace
|
||
func appendHop(trace *common.ForwardingTrace, hop common.ForwardingHop, success bool, errMsg string, latencyMs float64) {
|
||
hop.Success = success
|
||
hop.Error = errMsg
|
||
hop.LatencyMs = latencyMs
|
||
trace.Hops = append(trace.Hops, hop)
|
||
}
|
||
|
||
// finalizeTrace 设置 trace 的最终状态并写入 rc
|
||
func finalizeTrace(trace *common.ForwardingTrace, rc *RelayContext, hop common.ForwardingHop, success bool, attempt int, selection *common.ChannelSelection, errMsg string, latencyMs float64) {
|
||
hop.Success = success
|
||
hop.Error = errMsg
|
||
hop.LatencyMs = latencyMs
|
||
trace.Hops = append(trace.Hops, hop)
|
||
trace.TotalAttempts = attempt + 1
|
||
trace.UpstreamModel = selection.UpstreamModelName
|
||
trace.ModelMapped = selection.IsModelMapped
|
||
rc.ForwardingTrace = trace
|
||
}
|
||
|
||
// RelayErrorWithRateLimit 带 429 限流信息的错误
|
||
type RelayErrorWithRateLimit struct {
|
||
StatusCode int
|
||
Message string
|
||
LimitLevel string
|
||
Remaining int
|
||
ResetAt int64
|
||
}
|
||
|
||
func (e *RelayErrorWithRateLimit) Error() string {
|
||
return e.Message
|
||
}
|
||
|
||
// applyStreamEndDiag 把流式结束原因落进失败用量记录，替代无归因价值的哨兵文案。
|
||
// stream_end_reason 区分 client_gone / handler_stop / scanner_err / timeout / ping_fail / eof，
|
||
// error_message 保留真实底层错误（ctx 取消原因、写客户端失败的 TCP 错误等）；
|
||
// 哨兵错误（ErrStreamInterrupted）或状态里已带底层错误时直接用摘要，其余追加在原始错误后。
|
||
// error_message 对租户端可见，全部经 RedactMessage/SafeUpstreamErrorMessage 脱敏：
|
||
// 摘要中的本地连接地址抹 IP，外层错误按传输层归一化，禁止上游域名/URL/IP 落入用量日志。
|
||
func applyStreamEndDiag(record *common.UsageRecord, streamStatus *common.StreamStatus, err error) {
|
||
if streamStatus == nil {
|
||
return
|
||
}
|
||
reason := streamStatus.GetEndReason()
|
||
if reason == "" {
|
||
return
|
||
}
|
||
record.StreamEndReason = string(reason)
|
||
summary := helper.RedactMessage(streamStatus.Summary())
|
||
if err == nil || errors.Is(err, common.ErrStreamInterrupted) || streamStatus.Error() != nil {
|
||
record.ErrorMessage = summary
|
||
return
|
||
}
|
||
record.ErrorMessage = helper.SafeUpstreamErrorMessage(err) + " [stream: " + summary + "]"
|
||
}
|
||
|
||
// recordFailedUsage 记录失败用量。
|
||
// selection 非 nil 时记录具体失败渠道（ID/名称/类型/上游模型），便于在用量日志定位失败渠道；
|
||
// nil 表示无单一渠道（全部渠道失败），渠道字段留空，失败详情见 error_message。
|
||
// streamStatus 非 nil 且已有结束原因时，同步落 stream_end_reason 并修正 error_message。
|
||
func recordFailedUsage(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error, streamStatus *common.StreamStatus, upstreamRequestID string) {
|
||
record := &common.UsageRecord{
|
||
TenantID: rc.TenantID,
|
||
UserID: rc.UserID,
|
||
ApiKeyID: rc.ApiKeyID,
|
||
ProjectID: rc.ProjectID,
|
||
ModelName: modelName,
|
||
RequestedModel: modelName,
|
||
RelayMode: int(relayMode),
|
||
IsStream: isStream,
|
||
Success: false,
|
||
RequestID: rc.RequestID,
|
||
Status: "error",
|
||
// 租户端用量日志可见 error_message，须经脱敏（上游域名/URL/IP 禁止落库）；
|
||
// 完整原始错误保留在应用日志与审计 ForwardingTrace（仅管理员可见）
|
||
ErrorMessage: helper.SafeUpstreamErrorMessage(err),
|
||
}
|
||
applyStreamEndDiag(record, streamStatus, err)
|
||
record.UpstreamRequestID = upstreamRequestID
|
||
if selection != nil {
|
||
record.ChannelID = selection.ChannelID
|
||
record.ChannelName = selection.ChannelName
|
||
record.ChannelType = selection.ChannelType
|
||
record.UpstreamModel = selection.UpstreamModelName
|
||
}
|
||
provider.RecordUsage(context.Background(), record)
|
||
}
|
||
|
||
// recordFailedUsageWithTokens 记录失败用量（含 token 明细，用于流中断等已有部分 usage 的场景）。
|
||
// 与 recordFailedUsage 的区别：此函数会填充 token 字段，避免报表中流中断记录的 token 全为 0。
|
||
func recordFailedUsageWithTokens(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error, usage *common.Usage, settleResult *common.SettlementResult, streamStatus *common.StreamStatus, upstreamRequestID string) {
|
||
record := &common.UsageRecord{
|
||
TenantID: rc.TenantID,
|
||
UserID: rc.UserID,
|
||
ApiKeyID: rc.ApiKeyID,
|
||
ProjectID: rc.ProjectID,
|
||
ModelName: modelName,
|
||
RequestedModel: modelName,
|
||
RelayMode: int(relayMode),
|
||
IsStream: isStream,
|
||
Success: false,
|
||
RequestID: rc.RequestID,
|
||
Status: "error",
|
||
// 租户端用量日志可见 error_message，须经脱敏（上游域名/URL/IP 禁止落库）；
|
||
// 完整原始错误保留在应用日志与审计 ForwardingTrace（仅管理员可见）
|
||
ErrorMessage: helper.SafeUpstreamErrorMessage(err),
|
||
}
|
||
applyStreamEndDiag(record, streamStatus, err)
|
||
record.UpstreamRequestID = upstreamRequestID
|
||
if selection != nil {
|
||
record.ChannelID = selection.ChannelID
|
||
record.ChannelName = selection.ChannelName
|
||
record.ChannelType = selection.ChannelType
|
||
record.UpstreamModel = selection.UpstreamModelName
|
||
}
|
||
if usage != nil {
|
||
record.PromptTokens = usage.PromptTokens
|
||
record.CompletionTokens = usage.CompletionTokens
|
||
record.TotalTokens = usage.TotalTokens
|
||
}
|
||
// 流中断结算成功时补充费用字段：中断已按已传输部分实际扣费，用量记录须与账本一致，
|
||
// 否则后台看到的中断请求永远是 0 费用（结算与用量记录是两条独立写入链路）
|
||
if settleResult != nil {
|
||
record.TotalCost = settleResult.BaseCost
|
||
record.ActualCost = settleResult.ActualCost
|
||
record.PreDeductAmount = settleResult.PreDeductAmount
|
||
record.RefundAmount = settleResult.RefundAmount
|
||
record.SupplementAmount = settleResult.SupplementAmount
|
||
record.BillingMode = settleResult.BillingMode
|
||
record.BillingSource = settleResult.BillingSource
|
||
}
|
||
provider.RecordUsage(context.Background(), record)
|
||
}
|
||
|
||
// recordChannelError 记录渠道错误事件到 chn_error_events（异步，不阻塞请求）
|
||
func recordChannelError(rc *RelayContext, selection *common.ChannelSelection, modelName string, attempt int, isFinal bool, err error, latencyMs float64) {
|
||
if commonlogic.DefaultChannelErrorWriter == nil {
|
||
return
|
||
}
|
||
|
||
errMsg := err.Error()
|
||
if len(errMsg) > 500 {
|
||
errMsg = errMsg[:500]
|
||
}
|
||
|
||
statusCode := 0
|
||
errType := "unknown"
|
||
var relayErr *constant.RelayError
|
||
if errors.As(err, &relayErr) {
|
||
statusCode = relayErr.StatusCode
|
||
errType = relayErr.Type
|
||
}
|
||
|
||
event := map[string]any{
|
||
"channel_id": selection.ChannelID,
|
||
"channel_name": selection.ChannelName,
|
||
"channel_type": selection.ChannelType,
|
||
"provider": constant.ProviderType(selection.ChannelType).String(),
|
||
"model_name": modelName,
|
||
"request_id": rc.RequestID,
|
||
"tenant_id": rc.TenantID,
|
||
"error_category": constant.ClassifyError(err),
|
||
"status_code": statusCode,
|
||
"error_type": errType,
|
||
"error_message": errMsg,
|
||
"is_retryable": constant.IsRetryable(err),
|
||
"attempt": attempt,
|
||
"is_final": isFinal,
|
||
"latency_ms": latencyMs,
|
||
}
|
||
if selection.UpstreamModelName != "" {
|
||
event["upstream_model"] = selection.UpstreamModelName
|
||
}
|
||
commonlogic.DefaultChannelErrorWriter.Submit(event)
|
||
}
|
||
|
||
// inlineMediaTokenCost 单个内联媒体（base64 图片/音频）在预扣估算中的固定 token 成本。
|
||
// 视觉输入的真实 token 数与分辨率/切片策略强相关（OpenAI 1024px 约 765~1105，Gemini 约 258），
|
||
// 这里取保守上界即可——预扣只是闸门，最终以上游 usage 结算，多退少补。
|
||
const inlineMediaTokenCost = 1500
|
||
|
||
// estimateInputTokens 粗略估算输入 token 数（仅用于预扣，结算以上游 usage 为准）。
|
||
//
|
||
// 文本部分按字符数 / 4；base64 内联媒体不能按字节折算——一张 2MB 的图按 len/4
|
||
// 会被估成约 70 万 token，预扣直接冻穿余额，余额略少的租户根本发不出请求。
|
||
// 故先扫出内联媒体载荷的字节数从长度中扣除，再按每个媒体的固定成本计入。
|
||
func estimateInputTokens(body []byte) int {
|
||
mediaBytes, mediaCount := commonlogic.ScanInlineMediaBytes(body)
|
||
textBytes := len(body) - mediaBytes
|
||
if textBytes < 0 {
|
||
textBytes = 0
|
||
}
|
||
return textBytes/4 + mediaCount*inlineMediaTokenCost
|
||
}
|
||
|
||
// tokenDetailField 安全提取 TokenDetails 中的字段值
|
||
func tokenDetailField(details *common.TokenDetails, getter func(*common.TokenDetails) int) int {
|
||
if details == nil {
|
||
return 0
|
||
}
|
||
return getter(details)
|
||
}
|
||
|
||
// setPreResponseHeaders 在写入响应体之前设置限流和弃用 header
|
||
func setPreResponseHeaders(w http.ResponseWriter, br *BillingResult) {
|
||
if br == nil {
|
||
return
|
||
}
|
||
if info := br.RateLimitInfo; info != nil {
|
||
w.Header().Set("X-RateLimit-Limit", fmt.Sprintf("%d", info.Limit))
|
||
w.Header().Set("X-RateLimit-Remaining", fmt.Sprintf("%d", info.Remaining))
|
||
w.Header().Set("X-RateLimit-Reset", fmt.Sprintf("%d", info.ResetAt))
|
||
}
|
||
if dep := br.Deprecation; dep != nil {
|
||
w.Header().Set("Deprecation", "true")
|
||
if dep.SunsetDate != "" {
|
||
w.Header().Set("Sunset", dep.SunsetDate)
|
||
}
|
||
if dep.ReplacementModel != "" {
|
||
w.Header().Set("Link", fmt.Sprintf("</v1/models/%s>; rel=\"successor-version\"", dep.ReplacementModel))
|
||
}
|
||
}
|
||
}
|
||
|
||
// 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 {
|
||
return 2
|
||
}
|
||
return 1
|
||
}
|
||
|
||
// relayModeToInboundFormat 根据 relay mode 推断入站请求格式
|
||
func relayModeToInboundFormat(mode constant.RelayMode) constant.RelayFormat {
|
||
switch mode {
|
||
case constant.RelayModeClaudeMessages:
|
||
return constant.RelayFormatClaude
|
||
case constant.RelayModeGeminiChat:
|
||
return constant.RelayFormatGemini
|
||
case constant.RelayModeResponses, constant.RelayModeResponsesCompact:
|
||
return constant.RelayFormatResponses
|
||
default:
|
||
return constant.RelayFormatOpenAI
|
||
}
|
||
}
|
||
|
||
// relayModeString 转换 relay mode 为字符串
|
||
func relayModeString(mode constant.RelayMode) string {
|
||
switch mode {
|
||
case constant.RelayModeChatCompletions:
|
||
return "chat_completions"
|
||
case constant.RelayModeCompletions:
|
||
return "completions"
|
||
case constant.RelayModeEmbeddings:
|
||
return "embeddings"
|
||
case constant.RelayModeImagesGenerations:
|
||
return "images_generations"
|
||
case constant.RelayModeAudioSpeech:
|
||
return "audio_speech"
|
||
case constant.RelayModeAudioTranscription:
|
||
return "audio_transcriptions"
|
||
case constant.RelayModeAudioTranslation:
|
||
return "audio_translations"
|
||
case constant.RelayModeRerank:
|
||
return "rerank"
|
||
case constant.RelayModeResponses, constant.RelayModeResponsesCompact:
|
||
return "responses"
|
||
case constant.RelayModeRealtime:
|
||
return "realtime"
|
||
case constant.RelayModeClaudeMessages:
|
||
return "claude_messages"
|
||
case constant.RelayModeGeminiChat:
|
||
return "gemini_generate_content"
|
||
case constant.RelayModeModerations:
|
||
return "moderations"
|
||
case constant.RelayModeImagesEdits:
|
||
return "images_edits"
|
||
case constant.RelayModeMjSubmit:
|
||
return "mj_submit"
|
||
case constant.RelayModeMjFetch:
|
||
return "mj_fetch"
|
||
case constant.RelayModeMjImage:
|
||
return "mj_image"
|
||
case constant.RelayModeVideoGenerations:
|
||
return "video_generations"
|
||
case constant.RelayModeVideos:
|
||
return "videos"
|
||
default:
|
||
return ""
|
||
}
|
||
}
|