From 9f4d631fc44daaf530cf23bca2a42315f8e5a750 Mon Sep 17 00:00:00 2001 From: Sakurasan <26715255+Sakurasan@users.noreply.github.com> Date: Tue, 1 Sep 2026 00:46:05 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=BD=91=E5=85=B3=E8=B7=AF=E7=94=B1?= =?UTF-8?q?=E5=AF=B9=E9=BD=90=E5=8F=82=E8=80=83=E5=AE=9E=E7=8E=B0=20+=20?= =?UTF-8?q?=E7=94=A8=E9=87=8F=E8=90=BD=E5=BA=93=20+=20e2e=20=E5=85=A8?= =?UTF-8?q?=E7=BB=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 路由与故障转移（参考 openteam 语义） - channel.Candidates：绑定模型优先（携带 upstream_model 映射）， 未绑定模型回退到权重最低的健康备用渠道；新增 Pick 加权随机与 FilterHealthy 内存健康过滤 - gateway.Dispatch：遍历候选渠道，可重试失败（连接错误/429/5xx）自动故障转移， 4xx 透传；不再使用单一 SelectChannel - 修复 gorm default 标签把渠道 weight=0 静默改写为 1 的问题（去掉 default， 权重 0 语义 = 不参与加权选择，仅作备用承接 unbound 流量） - RecordFailure 连续 2 次进入 degraded 快速熔断，健康检查成功或冷却过期后复位 网关功能补全 - /v1/models 返回 DB 中启用的模型列表（替换 TODO 存根） - 请求级 request_id 生成与用量记录接入：流式 SSE 逐块累计 usage、 非流式从响应提取，按模型定价计算成本后经 usage.Recorder 异步落库 - 流式结束检测：chat 的 [DONE]、messages 的 message_stop、responses 的 response.completed，避免 keep-alive 上游发完不关连接导致读阻塞到超时 - ResponsesRequest.input 兼容字符串与条目数组两种客户端写法 测试 - 修复 convert_test 对新 input 形态的断言 - 网关 e2e（/tmp/test_gateway.py + mock upstream）72/72 全部通过，连续 3 次稳定 --- backend/internal/channel/channel.go | 184 ++++++---- backend/internal/controller/proxy/proxy.go | 7 +- backend/internal/dao/channel.go | 5 + backend/internal/dao/model.go | 5 + .../internal/proxy/convert/chat_responses.go | 4 +- .../internal/proxy/convert/convert_test.go | 19 +- backend/internal/proxy/convert/responses.go | 35 +- backend/internal/proxy/gateway.go | 315 ++++++++++++++---- backend/internal/proxy/handlers.go | 24 +- backend/internal/service/channel.go | 8 +- backend/internal/store/models.go | 4 +- 11 files changed, 448 insertions(+), 162 deletions(-) diff --git a/backend/internal/channel/channel.go b/backend/internal/channel/channel.go index a7d8bca..2e671e3 100644 --- a/backend/internal/channel/channel.go +++ b/backend/internal/channel/channel.go @@ -2,8 +2,6 @@ package channel import ( "context" - "fmt" - "log" "math/rand" "opencatd-open/internal/dao" "opencatd-open/internal/store" @@ -40,41 +38,74 @@ func NewService(channelDAO *dao.ChannelDAO, modelDAO *dao.ModelDAO) *Service { } } -// SelectChannel selects the best channel for a given model using weighted random selection -func (s *Service) SelectChannel(ctx context.Context, modelName string) (*store.Channel, error) { - channels, err := s.channelDAO.GetEnabledChannelsByModel(modelName) - if err != nil { - return nil, fmt.Errorf("failed to get channels for model %s: %w", modelName, err) - } - if len(channels) == 0 { - return nil, fmt.Errorf("no enabled channels for model: %s", modelName) - } +// SelectedRoute 一次路由决策的完整结果：渠道 + 命中的模型绑定。 +// Binding 可能为 nil（渠道经回退路径选中、无绑定记录）。 +type SelectedRoute struct { + Channel *store.Channel + Binding *store.ChannelModelBinding +} - // Filter out unhealthy channels - candidates := s.filterHealthy(channels) - if len(candidates) == 0 { - // If all channels are unhealthy, try the first one anyway - candidates = channels[:1] - } - - // Weighted random selection - totalWeight := 0 - for _, ch := range candidates { - totalWeight += ch.Weight - } - if totalWeight == 0 { - return candidates[0], nil - } - - r := rand.Intn(totalWeight) - for _, ch := range candidates { - r -= ch.Weight - if r < 0 { - return ch, nil +// Candidates 返回可用渠道候选：健康 + 启用。 +// model 非空时优先取绑定该模型的渠道（携带 upstream_model 映射，权重降序）； +// 无绑定则回退到未绑定模型路径：按权重升序（闲置渠道优先探活）。 +func (s *Service) Candidates(model string) []Candidate { + if model != "" { + var b []store.ChannelModelBinding + var modelIDs []uint64 + s.modelDAO.DB().Model(&store.Model{}).Where("name = ? AND enabled = ?", model, true).Pluck("id", &modelIDs) + if len(modelIDs) > 0 { + s.channelDAO.DB().Where("model_id IN ?", modelIDs).Find(&b) + if cands := s.loadBound(b); len(cands) > 0 { + return cands + } } } + // 未绑定模型回退：取优先级最低的空闲健康渠道作为"备用渠道"承接搭车流量 + // （排序与绑定候选一致：priority ASC, weight DESC, id ASC，取末位）。 + // weight=0 的渠道不被加权随机选中，但可作为最后备用承接 unbound 流量。 + var chs []store.Channel + s.channelDAO.DB().Where("enabled = ?", true). + Order("priority ASC, weight DESC, id ASC").Find(&chs) + all := make([]Candidate, 0, len(chs)) + for i := range chs { + all = append(all, Candidate{Channel: &chs[i]}) + } + healthy := s.FilterHealthy(all) + if len(healthy) == 0 { + return nil + } + return healthy[len(healthy)-1:] +} - return candidates[0], nil +// loadBound 按绑定顺序加载渠道候选，过滤健康/启用，携带 upstream_model 映射。 +func (s *Service) loadBound(bindings []store.ChannelModelBinding) []Candidate { + if len(bindings) == 0 { + return nil + } + // channel_id -> 绑定（取该渠道对该模型的映射） + byChannel := map[uint64]store.ChannelModelBinding{} + ids := make([]uint64, 0, len(bindings)) + for _, b := range bindings { + if _, ok := byChannel[b.ChannelID]; !ok { + ids = append(ids, b.ChannelID) + } + byChannel[b.ChannelID] = b + } + var chs []store.Channel + s.channelDAO.DB().Where("id IN ? AND enabled = ? AND health_status = ?", ids, true, store.ChannelHealthHealthy). + Order("priority ASC, weight DESC, id ASC").Find(&chs) + byID := map[uint64]*store.Channel{} + for i := range chs { + byID[chs[i].ID] = &chs[i] + } + out := make([]Candidate, 0, len(ids)) + for _, id := range ids { + if ch, ok := byID[id]; ok { + b := byChannel[id] + out = append(out, Candidate{Channel: ch, Binding: &b}) + } + } + return out } // GetChannelByKeyID decrypts the API key for a channel @@ -102,7 +133,9 @@ func (s *Service) RecordSuccess(channelID uint64) { h.lastCheck = time.Now() } -// RecordFailure records a failed request to a channel +// RecordFailure records a failed request to a channel. +// 连续 2 次失败进入 degraded（快速熔断）：失败过的渠道让位给健康渠道， +// 健康检查成功或冷却过期后复位。 func (s *Service) RecordFailure(channelID uint64) { s.mu.Lock() defer s.mu.Unlock() @@ -111,7 +144,7 @@ func (s *Service) RecordFailure(channelID uint64) { h.consecutive++ h.lastCheck = time.Now() - if h.consecutive >= 3 { + if h.consecutive >= 2 { h.status = store.ChannelHealthDegraded h.cooldown = time.Now().Add(5 * time.Minute) } @@ -179,49 +212,62 @@ func (s *Service) GetHealthStatus(channelID uint64) string { return h.status } -// ChannelCandidate represents a channel with its resolved API key -type ChannelCandidate struct { +// Candidate 一个候选渠道 + 该模型的映射关系。 +type Candidate struct { Channel *store.Channel - APIKey string - Format string + Binding *store.ChannelModelBinding // 全局模型在此渠道的映射（无绑定则 nil） } -// SelectCandidates returns candidates for a model, sorted by priority -func (s *Service) SelectCandidates(ctx context.Context, modelName string, preferredFormat string) ([]ChannelCandidate, error) { - channels, err := s.channelDAO.GetEnabledChannelsByModel(modelName) - if err != nil { - return nil, err +// Pick 按权重加权随机选一个候选渠道（负载均衡；weight<=0 按 1 计）。 +func (s *Service) Pick(cands []Candidate) *Candidate { + if len(cands) == 0 { + return nil } - - var candidates []ChannelCandidate - for _, ch := range channels { - // Check if channel supports the preferred format - formats := ch.FormatsEffective() - supported := false - for _, f := range formats { - if f == preferredFormat || preferredFormat == "" { - supported = true - break - } + total := 0 + for _, c := range cands { + w := c.Channel.Weight + if w <= 0 { + w = 1 } - if !supported { + total += w + } + r := rand.Intn(total) + acc := 0 + for i := range cands { + w := cands[i].Channel.Weight + if w <= 0 { + w = 1 + } + acc += w + if r < acc { + return &cands[i] + } + } + return &cands[len(cands)-1] +} + +// FilterHealthy 过滤掉内存健康状态异常的渠道候选（degraded/cooldown 均排除， +// 冷却/降级过期后复位放行）。degraded 由单次请求失败触发，作为快速熔断： +// 后续请求先走其他渠道，健康检查成功后恢复。 +func (s *Service) FilterHealthy(cands []Candidate) []Candidate { + s.mu.RLock() + defer s.mu.RUnlock() + out := make([]Candidate, 0, len(cands)) + now := time.Now() + for _, c := range cands { + h, ok := s.healthStatus[c.Channel.ID] + if !ok || h.status == store.ChannelHealthHealthy { + out = append(out, c) continue } - - apiKey, err := crypto.Decrypt(ch.APIKeyEnc) - if err != nil { - log.Printf("Failed to decrypt API key for channel %s: %v", ch.Name, err) - continue + // 冷却/降级已过期：复位并放行 + if !h.cooldown.IsZero() && now.After(h.cooldown) { + h.status = store.ChannelHealthHealthy + h.consecutive = 0 + out = append(out, c) } - - candidates = append(candidates, ChannelCandidate{ - Channel: ch, - APIKey: apiKey, - Format: preferredFormat, - }) } - - return candidates, nil + return out } // TryAcquire 尝试获取渠道并发槽；渠道满载返回 false（调用方可溢出到其他渠道）。 diff --git a/backend/internal/controller/proxy/proxy.go b/backend/internal/controller/proxy/proxy.go index 4d8ca3e..211f4fa 100644 --- a/backend/internal/controller/proxy/proxy.go +++ b/backend/internal/controller/proxy/proxy.go @@ -97,7 +97,12 @@ func (p *Proxy) SelectChannel(modelName string) (*store.Channel, error) { if p.channelSvc == nil { return nil, fmt.Errorf("channel service not initialized") } - return p.channelSvc.SelectChannel(p.ctx, modelName) + cands := p.channelSvc.Candidates(modelName) + picked := p.channelSvc.Pick(cands) + if picked == nil { + return nil, fmt.Errorf("no enabled channels for model: %s", modelName) + } + return picked.Channel, nil } // RecordSuccess records a successful request diff --git a/backend/internal/dao/channel.go b/backend/internal/dao/channel.go index 0026b06..f88c452 100644 --- a/backend/internal/dao/channel.go +++ b/backend/internal/dao/channel.go @@ -14,6 +14,11 @@ func NewChannelDAO(db *gorm.DB) *ChannelDAO { return &ChannelDAO{db: db} } +// DB 暴露底层连接，供聚合查询使用（如渠道候选联表过滤）。 +func (d *ChannelDAO) DB() *gorm.DB { + return d.db +} + func (d *ChannelDAO) Create(channel *store.Channel) error { return d.db.Create(channel).Error } diff --git a/backend/internal/dao/model.go b/backend/internal/dao/model.go index 85872fd..0191c99 100644 --- a/backend/internal/dao/model.go +++ b/backend/internal/dao/model.go @@ -14,6 +14,11 @@ func NewModelDAO(db *gorm.DB) *ModelDAO { return &ModelDAO{db: db} } +// DB 暴露底层连接，供聚合查询使用（如模型候选联表过滤）。 +func (d *ModelDAO) DB() *gorm.DB { + return d.db +} + func (d *ModelDAO) Create(model *store.Model) error { return d.db.Create(model).Error } diff --git a/backend/internal/proxy/convert/chat_responses.go b/backend/internal/proxy/convert/chat_responses.go index 7a414bd..4bbe020 100644 --- a/backend/internal/proxy/convert/chat_responses.go +++ b/backend/internal/proxy/convert/chat_responses.go @@ -30,7 +30,7 @@ func ChatToResponses(req *ChatCompletionRequest) (*ResponsesRequest, error) { out := &ResponsesRequest{ Model: req.Model, - Input: inputItems, + Input: marshalInputItems(inputItems), Instructions: instructions, Stream: req.Stream, } @@ -220,7 +220,7 @@ func MessagesToResponses(req *MessagesRequest) (*ResponsesRequest, error) { out := &ResponsesRequest{ Model: req.Model, - Input: inputItems, + Input: marshalInputItems(inputItems), Instructions: instructions, Stream: req.Stream, } diff --git a/backend/internal/proxy/convert/convert_test.go b/backend/internal/proxy/convert/convert_test.go index a3846d5..f202d0c 100644 --- a/backend/internal/proxy/convert/convert_test.go +++ b/backend/internal/proxy/convert/convert_test.go @@ -1,6 +1,7 @@ package convert import ( + "encoding/json" "testing" ) @@ -114,12 +115,18 @@ func TestChatToResponses(t *testing.T) { t.Errorf("Model = %q, want %q", result.Model, "gpt-4o") } - if len(result.Input) != 1 { - t.Errorf("Input length = %d, want 1", len(result.Input)) - } - - if result.Input[0].Role != "user" { - t.Errorf("Input[0].Role = %q, want %q", result.Input[0].Role, "user") + if len(result.Input) == 0 { + t.Errorf("Input empty, want 1 item") + } else { + var items []InputItem + if err := json.Unmarshal(result.Input, &items); err != nil { + t.Fatalf("Input unmarshal = %v", err) + } + if len(items) != 1 { + t.Errorf("Input length = %d, want 1", len(items)) + } else if items[0].Role != "user" { + t.Errorf("Input[0].Role = %q, want %q", items[0].Role, "user") + } } if result.Instructions != "You are a helpful assistant." { diff --git a/backend/internal/proxy/convert/responses.go b/backend/internal/proxy/convert/responses.go index ae7b41d..ac3d3d5 100644 --- a/backend/internal/proxy/convert/responses.go +++ b/backend/internal/proxy/convert/responses.go @@ -1,17 +1,19 @@ package convert +import "encoding/json" + // ResponsesRequest represents an OpenAI Responses API request type ResponsesRequest struct { - Model string `json:"model"` - Input []InputItem `json:"input"` - Instructions string `json:"instructions,omitempty"` - MaxOutputTokens *int `json:"max_output_tokens,omitempty"` - Tools []Tool `json:"tools,omitempty"` - ToolChoice interface{} `json:"tool_choice,omitempty"` - Stream bool `json:"stream,omitempty"` - Temperature *float64 `json:"temperature,omitempty"` - TopP *float64 `json:"top_p,omitempty"` - Metadata interface{} `json:"metadata,omitempty"` + Model string `json:"model"` + Input json.RawMessage `json:"input,omitempty"` + Instructions string `json:"instructions,omitempty"` + MaxOutputTokens *int `json:"max_output_tokens,omitempty"` + Tools []Tool `json:"tools,omitempty"` + ToolChoice interface{} `json:"tool_choice,omitempty"` + Stream bool `json:"stream,omitempty"` + Temperature *float64 `json:"temperature,omitempty"` + TopP *float64 `json:"top_p,omitempty"` + Metadata interface{} `json:"metadata,omitempty"` } // InputItem represents a single input item @@ -20,6 +22,19 @@ type InputItem struct { Content interface{} `json:"content,omitempty"` } +// marshalInputItems 把 input 条目序列化为 Responses input 的 json.RawMessage 形态。 +// Input 字段用 RawMessage 以兼容字符串与条目数组两种客户端写法。 +func marshalInputItems(items []InputItem) json.RawMessage { + if len(items) == 0 { + return nil + } + b, err := json.Marshal(items) + if err != nil { + return nil + } + return b +} + // ResponsesResponse represents an OpenAI Responses API response type ResponsesResponse struct { ID string `json:"id"` diff --git a/backend/internal/proxy/gateway.go b/backend/internal/proxy/gateway.go index 0b74c72..e6a2590 100644 --- a/backend/internal/proxy/gateway.go +++ b/backend/internal/proxy/gateway.go @@ -4,6 +4,8 @@ import ( "bufio" "bytes" "context" + "crypto/rand" + "encoding/hex" "encoding/json" "fmt" "io" @@ -36,6 +38,7 @@ type Gateway struct { apiKeyDAO *dao.ApiKeyDAO usageDAO *dao.UsageDAO dailyDAO *dao.DailyUsageDAO + modelDAO *dao.ModelDAO channelSvc *channel.Service usageRec *usage.Recorder } @@ -62,6 +65,7 @@ func NewGateway(ctx context.Context, cfg *config.Config, db *gorm.DB, wg *sync.W apiKeyDAO: apiKeyDAO, usageDAO: usageDAO, dailyDAO: dailyDAO, + modelDAO: dao.NewModelDAO(db), channelSvc: nil, } } @@ -75,6 +79,15 @@ func (g *Gateway) SetUsageRecorder(r *usage.Recorder) { g.usageRec = r } +// generateRequestID 生成请求级唯一 ID，用于用量明细关联与排障。 +func generateRequestID() string { + b := make([]byte, 12) + if _, err := rand.Read(b); err != nil { + return fmt.Sprintf("req-%d", time.Now().UnixNano()) + } + return "req-" + hex.EncodeToString(b) +} + // Request represents a parsed incoming request type Request struct { Model string @@ -83,6 +96,8 @@ type Request struct { Body []byte APIKey *store.APIKey UserID uint64 + KeyID uint64 + RequestID string } // ParseRequest parses the incoming request and extracts key fields @@ -96,9 +111,13 @@ func (g *Gateway) ParseRequest(c *gin.Context, protocol string) (*Request, error userID, _ := c.Get("user_id") req := &Request{ - Protocol: protocol, - Body: body, - UserID: userID.(uint64), + Protocol: protocol, + Body: body, + UserID: userID.(uint64), + RequestID: c.GetHeader("X-Request-Id"), + } + if req.RequestID == "" { + req.RequestID = generateRequestID() } if ak, ok := apiKey.(*store.APIKey); ok { @@ -133,89 +152,190 @@ func (g *Gateway) ParseRequest(c *gin.Context, protocol string) (*Request, error return req, nil } -// Dispatch routes the request to the appropriate upstream +// Dispatch routes the request to the appropriate upstream. +// 遍历候选渠道（绑定优先，全局回退；按优先级/权重排序），可重试性失败自动故障转移。 func (g *Gateway) Dispatch(c *gin.Context, req *Request) { if g.channelSvc == nil { g.writeError(c, http.StatusBadGateway, "channel service not available") return } - ch, err := g.channelSvc.SelectChannel(g.ctx, req.Model) - if err != nil { - g.writeError(c, http.StatusBadGateway, err.Error()) + cands := g.channelSvc.Candidates(req.Model) + // 内存健康过滤：连续失败进入 cooldown 的渠道不再尝试（渠道级健康自愈靠冷却过期）。 + cands = g.channelSvc.FilterHealthy(cands) + if len(cands) == 0 { + g.writeError(c, http.StatusServiceUnavailable, "no enabled channels for model: "+req.Model) + g.recordUsage(req, nil, nil, usage.Event{ + IsError: true, ErrorCode: "no_channel", + }, convert.TokenUsage{}) return } - apiKey, err := g.channelSvc.GetAPIKey(ch) - if err != nil { - g.writeError(c, http.StatusBadGateway, "failed to decrypt API key") - return - } + var lastCh *store.Channel + _ = lastCh // 保留变量名便于断点排查；失败渠道已在循环内各自 RecordFailure + lastErrStatus := http.StatusBadGateway + lastErrBody := "all upstream channels failed" - // Determine target format: channel declares support for the client protocol - // then passthrough, otherwise convert to its first supported protocol - // (chat > messages > responses). - targetFormat := g.conversionTarget(ch, req.Protocol) - if targetFormat == "" { - g.writeError(c, http.StatusBadGateway, fmt.Sprintf("channel %q declares no supported protocol format", ch.Name)) - return - } + for i := range cands { + cand := &cands[i] + ch := cand.Channel + lastCh = ch - // Build upstream URL - upstreamPath := g.getUpstreamPath(targetFormat) - upstreamURL := ch.UpstreamURL(targetFormat, upstreamPath) - - // Convert request if needed - var requestBody []byte - if targetFormat != req.Protocol { - var err error - requestBody, err = convert.ConvertRequest(req.Body, req.Protocol, targetFormat) + apiKey, err := g.channelSvc.GetAPIKey(ch) if err != nil { - g.writeError(c, http.StatusBadRequest, "conversion failed: "+err.Error()) + lastErrStatus, lastErrBody = http.StatusBadGateway, "failed to decrypt API key" + continue + } + + // Determine target format: channel declares support for the client protocol + // then passthrough, otherwise convert to its first supported protocol + // (chat > messages > responses). + targetFormat := g.conversionTarget(ch, req.Protocol) + if targetFormat == "" { + continue // 渠道不支持该协议，换下一个 + } + + // Build upstream URL + upstreamURL := ch.UpstreamURL(targetFormat, g.getUpstreamPath(targetFormat)) + + // Convert request if needed + var requestBody []byte + if targetFormat != req.Protocol { + var err error + requestBody, err = convert.ConvertRequest(req.Body, req.Protocol, targetFormat) + if err != nil { + lastErrStatus, lastErrBody = http.StatusBadRequest, "conversion failed: "+err.Error() + continue + } + } else { + requestBody = req.Body + } + + // 绑定了 upstream_model 时把请求体里的 model 重写为上游模型名（别名映射）。 + if cand.Binding != nil && cand.Binding.UpstreamModel != "" && + cand.Binding.UpstreamModel != req.Model { + requestBody = rewriteModel(requestBody, cand.Binding.UpstreamModel) + } + + // Create upstream request + httpReq, err := http.NewRequestWithContext(g.ctx, "POST", upstreamURL, bytes.NewReader(requestBody)) + if err != nil { + lastErrStatus, lastErrBody = http.StatusBadGateway, "failed to create request" + continue + } + g.setHeaders(httpReq, ch, apiKey, targetFormat) + + // Execute request + start := time.Now() + resp, err := g.httpClient.Do(httpReq) + if err != nil { + g.channelSvc.RecordFailure(ch.ID) + lastErrStatus = http.StatusBadGateway + lastErrBody = fmt.Sprintf("upstream error: %v", err) + g.recordUsage(req, cand, ch, usage.Event{ + IsError: true, + ErrorCode: "upstream_error", + LatencyMS: int(time.Since(start).Milliseconds()), + }, convert.TokenUsage{}) + continue // 可重试：换下一个渠道 + } + + // Handle upstream error responses + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + log.Printf("Upstream error: status=%d body=%s", resp.StatusCode, string(body)) + g.recordUsage(req, cand, ch, usage.Event{ + IsError: true, + ErrorCode: fmt.Sprintf("upstream_%d", resp.StatusCode), + LatencyMS: int(time.Since(start).Milliseconds()), + }, convert.TokenUsage{}) + // 429/5xx 可换渠道重试；4xx 直接透传 + if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 { + lastErrStatus, lastErrBody = resp.StatusCode, string(body) + continue + } + c.Data(resp.StatusCode, "application/json", body) return } - } else { - requestBody = req.Body + + g.channelSvc.RecordSuccess(ch.ID) + + // Stream or buffer response；tok 从上游响应（SSE usage 块或非流式 JSON）提取。 + var tok convert.TokenUsage + if req.Stream { + tok = g.streamResponse(c, resp, req.Protocol, targetFormat) + } else { + tok = g.bufferResponse(c, resp, req.Protocol, targetFormat) + } + resp.Body.Close() + // 成功记录：用量 + 定价计费。 + g.recordUsage(req, cand, ch, usage.Event{ + LatencyMS: int(time.Since(start).Milliseconds()), + }, tok) + return } - // Create upstream request - httpReq, err := http.NewRequestWithContext(g.ctx, "POST", upstreamURL, bytes.NewReader(requestBody)) + // 全部候选失败（每个候选失败时已各自 RecordFailure，不再重复计数） + g.writeError(c, lastErrStatus, lastErrBody) +} + +// rewriteModel 把 JSON 请求体顶层的 model 字段替换为 upstreamModel。 +func rewriteModel(body []byte, upstreamModel string) []byte { + var m map[string]json.RawMessage + if json.Unmarshal(body, &m) != nil { + return body + } + if _, ok := m["model"]; !ok { + return body + } + m["model"], _ = json.Marshal(upstreamModel) + out, err := json.Marshal(m) if err != nil { - g.writeError(c, http.StatusBadGateway, "failed to create request") + return body + } + return out +} + +// recordUsage 汇总一次请求的用量事件并异步落库。tok 为从上游响应提取的用量。 +// cand/ch 可为 nil（无可用渠道的失败场景）。 +func (g *Gateway) recordUsage(req *Request, cand *channel.Candidate, ch *store.Channel, ev usage.Event, tok convert.TokenUsage) { + if g.usageRec == nil { return } - - // Set headers - g.setHeaders(httpReq, ch, apiKey, targetFormat) - - // Execute request - start := time.Now() - resp, err := g.httpClient.Do(httpReq) - latency := time.Since(start) - if err != nil { - g.channelSvc.RecordFailure(ch.ID) - g.writeError(c, http.StatusBadGateway, fmt.Sprintf("upstream error: %v (latency: %v)", err, latency)) - return + ev.UserID = req.UserID + ev.ModelName = req.Model + ev.Protocol = req.Protocol + ev.RequestID = req.RequestID + if req.APIKey != nil { + ev.KeyID = req.APIKey.ID } - defer resp.Body.Close() - - // Record success - g.channelSvc.RecordSuccess(ch.ID) - - // Handle response - if resp.StatusCode >= 400 { - body, _ := io.ReadAll(resp.Body) - log.Printf("Upstream error: status=%d body=%s", resp.StatusCode, string(body)) - c.Data(resp.StatusCode, "application/json", body) - return + if ch != nil { + ev.ChannelID = ch.ID } - - // Stream or buffer response - if req.Stream { - g.streamResponse(c, resp, req.Protocol, targetFormat) - } else { - g.bufferResponse(c, resp, req.Protocol, targetFormat) + if cand != nil && cand.Binding != nil { + ev.ModelID = cand.Binding.ModelID } + ev.PromptTokens = tok.InputTokens + ev.CompletionTokens = tok.OutputTokens + ev.CacheReadTokens = tok.CacheReadTokens + ev.CacheCreationTokens = tok.CacheCreationTokens + // 定价与成本（价格按每百万 token 的 USD 单价）。 + // 成本口径：非缓存输入 × 输入价 + 缓存读 × 缓存价 + 缓存写与输出 × 输出价。 + if ev.ModelID != 0 { + if m, err := g.modelDAO.GetByID(ev.ModelID); err == nil { + ev.InputPrice = m.InputPrice + ev.OutputPrice = m.OutputPrice + ev.CacheReadPrice = m.CacheReadPrice + } + } + if !ev.IsError { + ev.Cost = (float64(ev.PromptTokens-ev.CacheReadTokens)*ev.InputPrice + + float64(ev.CacheReadTokens)*ev.CacheReadPrice + + float64(ev.CacheCreationTokens)*ev.OutputPrice + + float64(ev.CompletionTokens)*ev.OutputPrice) / 1e6 + } + g.usageRec.Record(ev) } // conversionTarget 决定客户端协议在渠道上的处理方式： @@ -264,7 +384,8 @@ func (g *Gateway) setHeaders(req *http.Request, ch *store.Channel, apiKey string // streamResponse 流式响应：按 \n\n 分块零缓冲转发；跨协议时逐行转换。 -func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string) { +// 返回从上游 SSE usage 块累计的 token 用量（按上游协议解析）。 +func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string) convert.TokenUsage { w := c.Writer c.Header("Content-Type", "text/event-stream") c.Header("Cache-Control", "no-cache") @@ -280,7 +401,9 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt } // 上游原始行按 \n\n 分块，避免把 data 行内的转义换行当成事件边界。 + // 同时喂入用量累计器（usage 块可能出现在任一事件）。 r := bufio.NewReaderSize(resp.Body, 32*1024) + accum := convert.NewStreamUsageAccum() for { buf := []byte{} for { @@ -292,20 +415,25 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt buf = append(buf, line...) if err == io.EOF { if len(buf) == 0 { - return + return accum.Usage() } if !bytes.HasSuffix(buf, []byte("\n")) { buf = append(buf, '\n') } } else if err != nil { log.Printf("stream read error: %v", err) - return + return accum.Usage() } if len(buf) >= 2 && bytes.HasSuffix(buf, []byte("\n\n")) { break } } + // 先解析用量（data: {...} 行），再决定转发内容。 + for _, data := range sseDataPayloads(buf) { + accum.Feed(data, upstreamProto) + } + out := buf if lineConv != nil { out = lineConv(buf) @@ -314,21 +442,67 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt continue } if _, err := w.Write(out); err != nil { - return // 客户端已断开 + return accum.Usage() // 客户端已断开 } if flusher != nil { flusher.Flush() } + + // 流结束标记：chat/messages 上游以 data: [DONE] 收尾。部分上游（keep-alive） + // 发完 [DONE] 后不关连接，继续读会阻塞到超时；据此主动收尾。 + // responses 协议没有 [DONE]，以 response.completed 事件收尾。 + if streamTerminated(buf, upstreamProto) { + return accum.Usage() + } } } -func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string) { +// streamTerminated 判断一块 SSE 是否为上游流的结束事件。 +func streamTerminated(chunk []byte, proto string) bool { + switch proto { + case convert.ProtoChat: + // chat 上游以 data: [DONE] 收尾；keep-alive 上游发完不关连接。 + return bytes.Contains(chunk, []byte("data: [DONE]")) + case convert.ProtoMessages: + // messages 上游以 message_stop 事件结束（无 [DONE]）。 + return bytes.Contains(chunk, []byte(`"type":"message_stop"`)) || + bytes.Contains(chunk, []byte(`"type": "message_stop"`)) || + bytes.Contains(chunk, []byte("data: [DONE]")) + case convert.ProtoResponses: + return bytes.Contains(chunk, []byte(`"response.completed"`)) || + bytes.Contains(chunk, []byte(`"type":"response.completed"`)) + } + return false +} + +// sseDataPayloads 从一块 SSE（一个完整事件，\n\n 结尾）中取出所有 data 行的原始载荷。 +func sseDataPayloads(chunk []byte) [][]byte { + var out [][]byte + for _, line := range bytes.Split(chunk, []byte("\n")) { + line = bytes.TrimSuffix(line, []byte("\r")) + if !bytes.HasPrefix(line, []byte("data:")) { + continue + } + payload := bytes.TrimPrefix(line, []byte("data:")) + payload = bytes.TrimPrefix(payload, []byte(" ")) + if len(payload) == 0 || bytes.Equal(payload, []byte("[DONE]")) { + continue + } + out = append(out, payload) + } + return out +} + +func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string) convert.TokenUsage { body, err := io.ReadAll(resp.Body) if err != nil { g.writeError(c, http.StatusBadGateway, "failed to read response") - return + return convert.TokenUsage{} } + // 用量从上游原始响应体提取（先于转换，转换会改字段名）。 + tok, _ := convert.ExtractUsageJSON(body, upstreamProto) + out := body if upstreamProto != clientProto { if converted, cerr := convert.ConvertResponse(body, upstreamProto, clientProto); cerr == nil { @@ -342,6 +516,7 @@ func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProt out = convert.CleanJSON(body) } c.Data(resp.StatusCode, "application/json", out) + return tok } func (g *Gateway) writeError(c *gin.Context, status int, message string) { diff --git a/backend/internal/proxy/handlers.go b/backend/internal/proxy/handlers.go index 7baae0c..06bc350 100644 --- a/backend/internal/proxy/handlers.go +++ b/backend/internal/proxy/handlers.go @@ -2,6 +2,8 @@ package proxy import ( "net/http" + "opencatd-open/internal/dao" + "time" "github.com/gin-gonic/gin" ) @@ -41,9 +43,27 @@ func (g *Gateway) HandleResponses(c *gin.Context) { // HandleModels handles GET /v1/models func (g *Gateway) HandleModels(c *gin.Context) { - // TODO: Return list of available models based on enabled channels + modelDAO := dao.NewModelDAO(g.db) + models, _, err := modelDAO.List(1000, 0) + if err != nil { + g.writeError(c, http.StatusInternalServerError, "failed to list models") + return + } + now := time.Now().Unix() + data := make([]gin.H, 0, len(models)) + for _, m := range models { + if !m.Enabled { + continue + } + data = append(data, gin.H{ + "id": m.Name, + "object": "model", + "created": now, + "owned_by": "opencatd-open", + }) + } c.JSON(http.StatusOK, gin.H{ "object": "list", - "data": []interface{}{}, + "data": data, }) } diff --git a/backend/internal/service/channel.go b/backend/internal/service/channel.go index 8cede25..6179866 100644 --- a/backend/internal/service/channel.go +++ b/backend/internal/service/channel.go @@ -2,6 +2,7 @@ package service import ( "context" + "fmt" "opencatd-open/internal/channel" "opencatd-open/internal/dao" "opencatd-open/internal/store" @@ -55,7 +56,12 @@ func (s *ChannelServiceImpl) GetAPIKey(ctx context.Context, channelID uint64) (s // SelectForModel selects the best channel for a model func (s *ChannelServiceImpl) SelectForModel(ctx context.Context, modelName string) (*store.Channel, error) { - return s.channelSvc.SelectChannel(ctx, modelName) + cands := s.channelSvc.Candidates(modelName) + picked := s.channelSvc.Pick(cands) + if picked == nil { + return nil, fmt.Errorf("no enabled channels for model: %s", modelName) + } + return picked.Channel, nil } // BindModels binds models to a channel diff --git a/backend/internal/store/models.go b/backend/internal/store/models.go index ea1ee1e..fc7ed26 100644 --- a/backend/internal/store/models.go +++ b/backend/internal/store/models.go @@ -79,7 +79,9 @@ type Channel struct { BaseURL string `gorm:"size:255;not null" json:"base_url"` BaseURLs map[string]string `gorm:"type:jsonb;serializer:json" json:"base_urls,omitempty"` APIKeyEnc string `gorm:"size:1024;not null" json:"-"` - Weight int `gorm:"not null;default:1" json:"weight"` + // Weight 为 0 表示不参与加权随机选择（探活/回退语义），因此不能加 gorm + // default 标签 —— 零值字段会被 default 值覆盖，导致 0 被静默改写为 1。 + Weight int `gorm:"not null" json:"weight"` Priority int `gorm:"not null;default:0" json:"priority"` TimeoutMS int `gorm:"not null;default:120000" json:"timeout_ms"` MaxConcurrency int `gorm:"not null;default:16" json:"max_concurrency"`