From f81b364436bf50ad970313c23c9c78a69fbd5068 Mon Sep 17 00:00:00 2001 From: Sakurasan <26715255+Sakurasan@users.noreply.github.com> Date: Mon, 31 Aug 2026 22:29:09 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=B8=89=E5=8D=8F=E8=AE=AE=E4=BA=92?= =?UTF-8?q?=E8=BD=AC=E7=BD=91=E5=85=B3=20+=20=E9=89=B4=E6=9D=83=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=20+=20=E7=AE=A1=E7=90=86=E7=AB=AF=E5=A2=9E=E5=BC=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端 - 新增 proxy/convert 三协议（chat/messages/responses）请求、响应与 SSE 流式互转， 以 Chat 为中间模型；usage.go 统一提取三协议 token 用量（含单测） - gateway: 跨协议调度（渠道未声明客户端协议时转为渠道首选格式）， streamResponse 按 \n\n 分块逐行转换直通，bufferResponse 转换失败时剥非 JSON 前缀 - gateway: 新增 SetUsageRecorder 注入异步用量记录器 - auth_llm: 修复 key_prefix 查询长度错配（[:8] vs 存储的 [:12]）导致全部 401； 修复长度 8-11 的 key 切片越界 panic；统一 unauthorized 响应 - usage: 日报表改为增量累加 upsert，避免多次 flush 互相清零；记录协议/错误码/时延等字段 - channel: 新增渠道并发槽 TryAcquire；健康检查支持可配置参数 - api: 新增 admin 渠道/模型/系统配置管理端点（旧端点保留兼容） 前端 - 新增渠道管理、模型管理、系统配置视图与 ChannelModelsDrawer - 新增 ui 基础组件（Button/Badge/Input/Modal）与 protocol.ts - 调整 Toast 样式、密钥页、路由菜单；dev 代理默认指向 3000 端口 --- backend/internal/api/admin_channels.go | 588 ++++++++++++++++ backend/internal/api/admin_config.go | 109 +++ backend/internal/api/admin_models.go | 291 ++++++++ backend/internal/api/api.go | 6 +- backend/internal/channel/channel.go | 25 + backend/internal/channel/health.go | 38 +- backend/internal/dao/usage.go | 13 +- backend/internal/proxy/convert/convert.go | 203 ++++++ backend/internal/proxy/convert/json_chat.go | 504 ++++++++++++++ .../internal/proxy/convert/json_responses.go | 376 ++++++++++ .../proxy/convert/stream_transform.go | 642 ++++++++++++++++++ backend/internal/proxy/convert/usage.go | 131 ++++ backend/internal/proxy/convert/usage_test.go | 170 +++++ backend/internal/proxy/gateway.go | 211 +++--- backend/internal/usage/recorder.go | 98 ++- backend/middleware/auth_llm.go | 64 +- backend/router/setRouter.go | 37 +- frontend/src/components/common/Toast.vue | 4 +- frontend/src/components/ui/Badge.vue | 30 + frontend/src/components/ui/Button.vue | 27 + frontend/src/components/ui/Input.vue | 36 + frontend/src/components/ui/Modal.vue | 84 +++ frontend/src/lib/protocol.ts | 27 + frontend/src/stores/channel.ts | 101 +++ frontend/src/stores/model.ts | 171 +++++ frontend/src/types/index.ts | 56 ++ frontend/src/utils/router_menu.ts | 9 +- .../views/dashboard/ChannelModelsDrawer.vue | 183 +++++ frontend/src/views/dashboard/ChannelsView.vue | 342 ++++++++++ frontend/src/views/dashboard/KeyView.vue | 149 +++- frontend/src/views/dashboard/Models.vue | 277 ++++++++ frontend/src/views/dashboard/ModelsView.vue | 225 ++++++ frontend/src/views/dashboard/SystemConfig.vue | 121 ++++ frontend/vite.config.ts | 2 +- 34 files changed, 5171 insertions(+), 179 deletions(-) create mode 100644 backend/internal/api/admin_channels.go create mode 100644 backend/internal/api/admin_config.go create mode 100644 backend/internal/api/admin_models.go create mode 100644 backend/internal/proxy/convert/convert.go create mode 100644 backend/internal/proxy/convert/json_chat.go create mode 100644 backend/internal/proxy/convert/json_responses.go create mode 100644 backend/internal/proxy/convert/stream_transform.go create mode 100644 backend/internal/proxy/convert/usage.go create mode 100644 backend/internal/proxy/convert/usage_test.go create mode 100644 frontend/src/components/ui/Badge.vue create mode 100644 frontend/src/components/ui/Button.vue create mode 100644 frontend/src/components/ui/Input.vue create mode 100644 frontend/src/components/ui/Modal.vue create mode 100644 frontend/src/lib/protocol.ts create mode 100644 frontend/src/stores/model.ts create mode 100644 frontend/src/views/dashboard/ChannelModelsDrawer.vue create mode 100644 frontend/src/views/dashboard/ChannelsView.vue create mode 100644 frontend/src/views/dashboard/Models.vue create mode 100644 frontend/src/views/dashboard/ModelsView.vue create mode 100644 frontend/src/views/dashboard/SystemConfig.vue diff --git a/backend/internal/api/admin_channels.go b/backend/internal/api/admin_channels.go new file mode 100644 index 0000000..4ebb1b2 --- /dev/null +++ b/backend/internal/api/admin_channels.go @@ -0,0 +1,588 @@ +package api + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "time" + + "opencatd-open/internal/pkg/crypto" + "opencatd-open/internal/store" + + "github.com/gin-gonic/gin" +) + +// AdminChannels GET /api/admin/channels — 渠道列表（不返回加密 key，返回掩码）。 +func (h *Handler) AdminChannels(c *gin.Context) { + var chs []store.Channel + if err := h.db.Order("id ASC").Find(&chs).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load channels"}) + return + } + out := make([]gin.H, 0, len(chs)) + for _, ch := range chs { + masked := "" + if key, err := crypto.Decrypt(ch.APIKeyEnc); err == nil && len(key) > 8 { + masked = maskAPIKey(key) + } else if err == nil { + masked = "****" + } + out = append(out, gin.H{ + "id": ch.ID, "name": ch.Name, "provider": ch.Provider, "formats": ch.FormatsEffective(), + "base_url": ch.BaseURL, "base_urls": ch.BaseURLs, + "api_key_masked": masked, "weight": ch.Weight, "priority": ch.Priority, + "timeout_ms": ch.TimeoutMS, "max_concurrency": ch.MaxConcurrency, + "health_status": ch.HealthStatus, "enabled": ch.Enabled, + "created_at": ch.CreatedAt, + }) + } + c.JSON(http.StatusOK, gin.H{"data": out}) +} + +type channelBody struct { + Name string `json:"name" binding:"required,min=1,max=64"` + Provider string `json:"provider"` + Formats []string `json:"formats"` + BaseURL string `json:"base_url"` + BaseURLs map[string]string `json:"base_urls"` + APIKey string `json:"api_key"` + Weight *int `json:"weight"` + Priority *int `json:"priority"` + TimeoutMS *int `json:"timeout_ms"` + MaxConcurrency *int `json:"max_concurrency"` + Enabled *bool `json:"enabled"` +} + +// normalizeBaseURLs 校验并清理分协议 base_url。 +func normalizeBaseURLs(m map[string]string) map[string]string { + if len(m) == 0 { + return nil + } + out := map[string]string{} + for k, v := range m { + if validFormats[k] && strings.TrimSpace(v) != "" { + out[k] = strings.TrimRight(strings.TrimSpace(v), "/") + } + } + if len(out) == 0 { + return nil + } + return out +} + +// resolveBaseURL 渠道 base_url：留空按供应商默认；网关按内容智能识别前缀/完整端点。 +func resolveBaseURL(provider, raw string) (string, error) { + base := strings.TrimRight(raw, "/") + if base == "" { + switch provider { + case store.ChannelProviderOpenAI: + base = "https://api.openai.com" + case store.ChannelProviderAnthropic: + base = "https://api.anthropic.com" + } + } + if base == "" { + return "", errors.New("base_url required for compatible channels") + } + return base, nil +} + +func validateProvider(p string) bool { + return p == store.ChannelProviderOpenAI || p == store.ChannelProviderAnthropic || p == store.ChannelProviderCompatible +} + +var validFormats = map[string]bool{ + store.FormatChat: true, store.FormatResponses: true, store.FormatMessages: true, +} + +// deriveProvider 按格式推断供应商（仅作内部字段/兼容用途，不参与路由）。 +func deriveProvider(formats []string) string { + if len(formats) == 0 { + return store.ChannelProviderCompatible + } + messagesOnly, hasResponses := true, false + for _, f := range formats { + if f != store.FormatMessages { + messagesOnly = false + } + if f == store.FormatResponses { + hasResponses = true + } + } + if messagesOnly { + return store.ChannelProviderAnthropic + } + if hasResponses { + return store.ChannelProviderOpenAI + } + return store.ChannelProviderCompatible +} + +// resolveFormats 渠道协议格式：显式给出则校验去重；空则按 provider 推断默认。 +func resolveFormats(provider string, formats []string) ([]string, error) { + if len(formats) == 0 { + switch provider { + case store.ChannelProviderAnthropic: + return []string{store.FormatMessages}, nil + case store.ChannelProviderOpenAI: + return []string{store.FormatChat, store.FormatResponses}, nil + default: + return []string{store.FormatChat}, nil + } + } + seen := map[string]bool{} + out := make([]string, 0, len(formats)) + for _, f := range formats { + if !validFormats[f] { + return nil, fmt.Errorf("unsupported format %q", f) + } + if !seen[f] { + seen[f] = true + out = append(out, f) + } + } + return out, nil +} + +// AdminCreateChannel POST /api/admin/channels +func (h *Handler) AdminCreateChannel(c *gin.Context) { + var req channelBody + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input: " + err.Error()}) + return + } + if req.Provider == "" { + req.Provider = deriveProvider(req.Formats) + } + if !validateProvider(req.Provider) { + c.JSON(http.StatusBadRequest, gin.H{"error": "provider must be openai, anthropic or compatible"}) + return + } + if req.APIKey == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "api_key required"}) + return + } + formats, err := resolveFormats(req.Provider, req.Formats) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + baseURL, err := resolveBaseURL(req.Provider, req.BaseURL) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + enc, err := crypto.Encrypt(req.APIKey) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to encrypt api key"}) + return + } + ch := store.Channel{ + Name: req.Name, Provider: req.Provider, Formats: formats, BaseURL: baseURL, + BaseURLs: normalizeBaseURLs(req.BaseURLs), + APIKeyEnc: enc, Weight: intOr(req.Weight, 1), Priority: intOr(req.Priority, 0), + TimeoutMS: intOr(req.TimeoutMS, 120000), MaxConcurrency: intOr(req.MaxConcurrency, 16), + HealthStatus: store.ChannelHealthHealthy, Enabled: boolOr(req.Enabled, true), + } + if err := h.db.Create(&ch).Error; err != nil { + c.JSON(http.StatusConflict, gin.H{"error": "failed to create channel (name may already exist)"}) + return + } + c.JSON(http.StatusCreated, gin.H{"id": ch.ID, "name": ch.Name}) +} + +// AdminUpdateChannel PUT /api/admin/channels/:id +func (h *Handler) AdminUpdateChannel(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid channel id"}) + return + } + var body struct { + Name *string `json:"name"` + Provider *string `json:"provider"` + Formats *[]string `json:"formats"` + BaseURL *string `json:"base_url"` + BaseURLs *map[string]string `json:"base_urls"` + APIKey *string `json:"api_key"` + Weight *int `json:"weight"` + Priority *int `json:"priority"` + TimeoutMS *int `json:"timeout_ms"` + MaxConcurrency *int `json:"max_concurrency"` + HealthStatus *string `json:"health_status"` + Enabled *bool `json:"enabled"` + } + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input"}) + return + } + var ch store.Channel + if err := h.db.First(&ch, id).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"}) + return + } + updates := map[string]any{} + if body.Name != nil { + updates["name"] = *body.Name + } + if body.Provider != nil { + if !validateProvider(*body.Provider) { + c.JSON(http.StatusBadRequest, gin.H{"error": "provider must be openai, anthropic or compatible"}) + return + } + updates["provider"] = *body.Provider + } + if body.BaseURL != nil { + prov := ch.Provider + if body.Provider != nil { + prov = *body.Provider + } + b, berr := resolveBaseURL(prov, *body.BaseURL) + if berr != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": berr.Error()}) + return + } + updates["base_url"] = b + } + if body.BaseURLs != nil { + raw, _ := json.Marshal(normalizeBaseURLs(*body.BaseURLs)) + updates["base_urls"] = string(raw) + } + if body.APIKey != nil && *body.APIKey != "" { + enc, err := crypto.Encrypt(*body.APIKey) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to encrypt api key"}) + return + } + updates["api_key_enc"] = enc + } + if body.Weight != nil { + updates["weight"] = *body.Weight + } + if body.Priority != nil { + updates["priority"] = *body.Priority + } + if body.TimeoutMS != nil { + updates["timeout_ms"] = *body.TimeoutMS + } + if body.MaxConcurrency != nil { + updates["max_concurrency"] = *body.MaxConcurrency + } + if body.HealthStatus != nil { + updates["health_status"] = *body.HealthStatus + } + if body.Enabled != nil { + updates["enabled"] = *body.Enabled + } + if body.Formats != nil { + prov := ch.Provider + if body.Provider != nil { + prov = *body.Provider + } + formats, ferr := resolveFormats(prov, *body.Formats) + if ferr != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": ferr.Error()}) + return + } + raw, _ := json.Marshal(formats) + updates["formats"] = string(raw) + } + if len(updates) > 0 { + if err := h.db.Model(&ch).Updates(updates).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update channel"}) + return + } + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +// AdminDeleteChannel DELETE /api/admin/channels/:id +func (h *Handler) AdminDeleteChannel(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid channel id"}) + return + } + res := h.db.Delete(&store.Channel{}, id) + if res.Error != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete channel"}) + return + } + if res.RowsAffected == 0 { + c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"}) + return + } + h.db.Where("channel_id = ?", id).Delete(&store.ChannelModelBinding{}) + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +// AdminTestChannel POST /api/admin/channels/:id/test — 请求渠道 /v1/models 测连通性。 +func (h *Handler) AdminTestChannel(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid channel id"}) + return + } + var ch store.Channel + if err := h.db.First(&ch, id).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"}) + return + } + key, err := crypto.Decrypt(ch.APIKeyEnc) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to decrypt channel key"}) + return + } + url := ch.UpstreamURL("", "/models") + client := &http.Client{Timeout: 10 * time.Second} + req, _ := http.NewRequest(http.MethodGet, url, nil) + req.Header.Set("Authorization", "Bearer "+key) + req.Header.Set("Accept", "application/json") + start := time.Now() + resp, err := client.Do(req) + status := store.ChannelHealthHealthy + msg := "ok" + latency := 0 + if err != nil { + status = store.ChannelHealthCooldown + msg = err.Error() + } else { + latency = int(time.Since(start).Milliseconds()) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + status = store.ChannelHealthCooldown + b, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + msg = fmt.Sprintf("http %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) + } + resp.Body.Close() + } + h.db.Model(&store.Channel{}).Where("id = ?", ch.ID).Update("health_status", status) + if status != store.ChannelHealthHealthy { + c.JSON(http.StatusBadGateway, gin.H{"error": msg}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true, "latency_ms": latency, "message": msg}) +} + +// AdminChannelRemoteModels GET /api/admin/channels/:id/models/remote — 拉取远端模型列表。 +// 返回本渠道尚未允许的模型（新增候选），排除已绑定的模型。 +func (h *Handler) AdminChannelRemoteModels(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid channel id"}) + return + } + var ch store.Channel + if err := h.db.First(&ch, id).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"}) + return + } + key, err := crypto.Decrypt(ch.APIKeyEnc) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to decrypt channel key"}) + return + } + url := ch.UpstreamURL("", "/models") + client := &http.Client{Timeout: 10 * time.Second} + req, _ := http.NewRequest(http.MethodGet, url, nil) + req.Header.Set("Authorization", "Bearer "+key) + req.Header.Set("Accept", "application/json") + resp, err := client.Do(req) + if err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) + return + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 10*1024*1024)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + c.JSON(http.StatusBadGateway, gin.H{"error": fmt.Sprintf("http %d: %s", resp.StatusCode, string(body))}) + return + } + // 解析 OpenAI 格式的模型列表 + var result struct { + Data []struct { + ID string `json:"id"` + } `json:"data"` + } + if err := json.Unmarshal(body, &result); err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "failed to parse response: " + err.Error()}) + return + } + // 本渠道已允许的上游模型名：不作为新增候选 + var boundNames []string + h.db.Model(&store.ChannelModelBinding{}).Where("channel_id = ?", id).Pluck("upstream_model", &boundNames) + boundSet := make(map[string]bool, len(boundNames)) + for _, n := range boundNames { + boundSet[strings.TrimSpace(n)] = true + } + models := make([]string, 0, len(result.Data)) + for _, m := range result.Data { + name := strings.TrimSpace(m.ID) + if name != "" && !boundSet[name] { + models = append(models, name) + } + } + c.JSON(http.StatusOK, gin.H{"data": models}) +} + +// AdminChannelModels GET /api/admin/channels/:id/models — 渠道绑定列表。 +func (h *Handler) AdminChannelModels(c *gin.Context) { + channelID, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid channel id"}) + return + } + var bindings []store.ChannelModelBinding + if err := h.db.Preload("Model").Where("channel_id = ?", channelID).Find(&bindings).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load bindings"}) + return + } + out := make([]gin.H, 0, len(bindings)) + for _, b := range bindings { + out = append(out, gin.H{ + "id": b.ID, "model_id": b.ModelID, "model_name": b.Model.Name, + "upstream_model": b.UpstreamModel, "weight": b.Weight, + }) + } + c.JSON(http.StatusOK, gin.H{"data": out}) +} + +// AdminChannelAddModel POST /api/admin/channels/:id/models — 手工添加模型绑定。 +// 无需渠道具备 /v1/models 接口：直接填上游模型名，可选自定义名称作为客户端调用名。 +func (h *Handler) AdminChannelAddModel(c *gin.Context) { + channelID, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid channel id"}) + return + } + var req struct { + UpstreamModel string `json:"upstream_model" binding:"required"` // 渠道侧真实模型名 + CustomName string `json:"custom_name"` // 客户端调用名，空=用上游名 + Weight *int `json:"weight"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input: upstream_model required"}) + return + } + globalName := req.CustomName + if globalName == "" { + globalName = req.UpstreamModel + } + var ch store.Channel + if err := h.db.First(&ch, channelID).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"}) + return + } + // 查找或创建全局模型 + var m store.Model + if err := h.db.Where("name = ?", globalName).First(&m).Error; err != nil { + m = store.Model{Name: globalName, Enabled: true} + if err := h.db.Create(&m).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create model"}) + return + } + } + // 查找已存在的绑定，如果存在则更新 + var existing store.ChannelModelBinding + if err := h.db.Where("channel_id = ? AND model_id = ?", channelID, m.ID).First(&existing).Error; err == nil { + // 已存在，更新 + existing.UpstreamModel = req.UpstreamModel + existing.Weight = intOr(req.Weight, 1) + if err := h.db.Save(&existing).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update binding"}) + return + } + c.JSON(http.StatusOK, gin.H{"id": existing.ID, "model_id": m.ID, "model_name": m.Name, "upstream_model": existing.UpstreamModel, "weight": existing.Weight}) + return + } + // 不存在，创建新的 + b := store.ChannelModelBinding{ + ChannelID: channelID, ModelID: m.ID, + UpstreamModel: req.UpstreamModel, Weight: intOr(req.Weight, 1), + } + if err := h.db.Create(&b).Error; err != nil { + c.JSON(http.StatusConflict, gin.H{"error": "binding may already exist"}) + return + } + c.JSON(http.StatusCreated, gin.H{"id": b.ID, "model_id": m.ID, "model_name": m.Name, "upstream_model": req.UpstreamModel, "weight": b.Weight}) +} + +// AdminChannelUpdateModel PATCH /api/admin/channels/:id/models/:bid — 改映射名/权重。 +func (h *Handler) AdminChannelUpdateModel(c *gin.Context) { + bid, err := strconv.ParseUint(c.Param("bid"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid binding id"}) + return + } + var req struct { + UpstreamModel *string `json:"upstream_model"` + Weight *int `json:"weight"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input"}) + return + } + var b store.ChannelModelBinding + if err := h.db.First(&b, bid).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "binding not found"}) + return + } + updates := map[string]any{} + if req.UpstreamModel != nil { + updates["upstream_model"] = *req.UpstreamModel + } + if req.Weight != nil { + updates["weight"] = *req.Weight + } + if len(updates) > 0 { + if err := h.db.Model(&b).Updates(updates).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update binding"}) + return + } + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +// AdminChannelDeleteModel DELETE /api/admin/channels/:id/models/:bid — 解除绑定。 +func (h *Handler) AdminChannelDeleteModel(c *gin.Context) { + bid, err := strconv.ParseUint(c.Param("bid"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid binding id"}) + return + } + res := h.db.Delete(&store.ChannelModelBinding{}, bid) + if res.Error != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete binding"}) + return + } + if res.RowsAffected == 0 { + c.JSON(http.StatusNotFound, gin.H{"error": "binding not found"}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +// maskAPIKey 掩码渠道密钥：保留前 7 位与后 4 位，中间固定 ****** 遮蔽。 +func maskAPIKey(key string) string { + if len(key) <= 11 { + return strings.Repeat("*", len(key)-4) + key[len(key)-4:] + } + return key[:7] + "******" + key[len(key)-4:] +} + +func intOr(p *int, def int) int { + if p == nil { + return def + } + return *p +} + +func boolOr(p *bool, def bool) bool { + if p == nil { + return def + } + return *p +} diff --git a/backend/internal/api/admin_config.go b/backend/internal/api/admin_config.go new file mode 100644 index 0000000..95ab562 --- /dev/null +++ b/backend/internal/api/admin_config.go @@ -0,0 +1,109 @@ +package api + +import ( + "net/http" + + "opencatd-open/internal/store" + + "github.com/gin-gonic/gin" +) + +// AdminGetConfig GET /api/admin/config — 获取系统配置。 +func (h *Handler) AdminGetConfig(c *gin.Context) { + configs := map[string]string{} + var rows []store.SystemConfig + h.db.Find(&rows) + for _, r := range rows { + configs[r.Key] = r.Value + } + c.JSON(http.StatusOK, gin.H{"data": configs}) +} + +// AdminUpdateConfig PUT /api/admin/config — 更新系统配置。 +func (h *Handler) AdminUpdateConfig(c *gin.Context) { + var req map[string]string + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input"}) + return + } + for key, value := range req { + var sc store.SystemConfig + result := h.db.Where("key = ?", key).First(&sc) + if result.Error == nil { + sc.Value = value + h.db.Save(&sc) + } else { + sc = store.SystemConfig{Key: key, Value: value} + h.db.Create(&sc) + } + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +// AdminGetRegistration GET /api/admin/config/registration — 获取注册配置。 +func (h *Handler) AdminGetRegistration(c *gin.Context) { + var sc store.SystemConfig + enabled := "true" + if err := h.db.Where("key = ?", "registration_enabled").First(&sc).Error; err == nil { + enabled = sc.Value + } + c.JSON(http.StatusOK, gin.H{"data": gin.H{"enabled": enabled == "true"}}) +} + +// AdminUpdateRegistration PUT /api/admin/config/registration — 更新注册配置。 +func (h *Handler) AdminUpdateRegistration(c *gin.Context) { + var req struct { + Enabled bool `json:"enabled"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input"}) + return + } + value := "false" + if req.Enabled { + value = "true" + } + var sc store.SystemConfig + if err := h.db.Where("key = ?", "registration_enabled").First(&sc).Error; err == nil { + sc.Value = value + h.db.Save(&sc) + } else { + sc = store.SystemConfig{Key: "registration_enabled", Value: value} + h.db.Create(&sc) + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +// AdminGetPasswordLogin GET /api/admin/config/password-login — 获取密码登录配置。 +func (h *Handler) AdminGetPasswordLogin(c *gin.Context) { + var sc store.SystemConfig + enabled := "true" + if err := h.db.Where("key = ?", "password_login_enabled").First(&sc).Error; err == nil { + enabled = sc.Value + } + c.JSON(http.StatusOK, gin.H{"data": gin.H{"enabled": enabled == "true"}}) +} + +// AdminUpdatePasswordLogin PUT /api/admin/config/password-login — 更新密码登录配置。 +func (h *Handler) AdminUpdatePasswordLogin(c *gin.Context) { + var req struct { + Enabled bool `json:"enabled"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input"}) + return + } + value := "false" + if req.Enabled { + value = "true" + } + var sc store.SystemConfig + if err := h.db.Where("key = ?", "password_login_enabled").First(&sc).Error; err == nil { + sc.Value = value + h.db.Save(&sc) + } else { + sc = store.SystemConfig{Key: "password_login_enabled", Value: value} + h.db.Create(&sc) + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} diff --git a/backend/internal/api/admin_models.go b/backend/internal/api/admin_models.go new file mode 100644 index 0000000..588b3e3 --- /dev/null +++ b/backend/internal/api/admin_models.go @@ -0,0 +1,291 @@ +package api + +import ( + "encoding/json" + "net/http" + "strconv" + + "opencatd-open/internal/store" + + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +// AdminModels GET /api/admin/models — 模型列表（含价格、渠道绑定、定价/禁止状态）。 +func (h *Handler) AdminModels(c *gin.Context) { + var ms []store.Model + if err := h.db.Order("sort ASC, id ASC").Find(&ms).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load models"}) + return + } + + allow, deny := h.modelPolicyConfig() + + out := make([]gin.H, 0, len(ms)) + for _, m := range ms { + var bindings []store.ChannelModelBinding + h.db.Preload("Channel").Where("model_id = ?", m.ID).Find(&bindings) + chs := make([]gin.H, 0, len(bindings)) + for _, b := range bindings { + if !b.Channel.Enabled { + continue + } + chs = append(chs, gin.H{ + "id": b.ID, "channel_id": b.ChannelID, "channel_name": b.Channel.Name, + "upstream_model": b.UpstreamModel, "weight": b.Weight, + }) + } + used := len(chs) > 0 + needsPricing := used && m.InputPrice == 0 && m.OutputPrice == 0 && m.CacheReadPrice == 0 + denied := containsStr(deny, m.Name) || (len(allow) > 0 && !containsStr(allow, m.Name)) + out = append(out, gin.H{ + "id": m.ID, "name": m.Name, "display_name": m.DisplayName, + "input_price": m.InputPrice, "output_price": m.OutputPrice, "cache_read_price": m.CacheReadPrice, + "enabled": m.Enabled, "sort": m.Sort, "channels": chs, + "used": used, "needs_pricing": needsPricing, "denied": denied, + }) + } + + var orphans []struct { + ChannelName string + UpstreamModel string + ModelID uint64 + } + h.db.Raw(`SELECT c.name as channel_name, b.model_id, b.upstream_model + FROM channel_model_bindings b + LEFT JOIN models m ON m.id = b.model_id + LEFT JOIN channels c ON c.id = b.channel_id + WHERE m.id IS NULL`).Scan(&orphans) + missing := make([]gin.H, 0, len(orphans)) + for _, o := range orphans { + missing = append(missing, gin.H{ + "channel": o.ChannelName, "model_id": o.ModelID, "upstream_model": o.UpstreamModel, + }) + } + + unpriced := 0 + { + var usedBindings []struct { + ModelID uint64 + } + h.db.Model(&store.ChannelModelBinding{}).Distinct("model_id").Scan(&usedBindings) + usedIDs := map[uint64]bool{} + for _, u := range usedBindings { + usedIDs[u.ModelID] = true + } + for _, m := range ms { + if usedIDs[m.ID] && m.InputPrice == 0 && m.OutputPrice == 0 && m.CacheReadPrice == 0 { + unpriced++ + } + } + } + + c.JSON(http.StatusOK, gin.H{ + "data": out, + "summary": gin.H{ + "total": len(ms), + "unpriced": unpriced, + "missing": missing, + "denied_count": len(deny), + }, + }) +} + +// modelPolicyConfig 读取全局模型允许/禁止列表。 +func (h *Handler) modelPolicyConfig() (allow, deny []string) { + var raw string + h.db.Model(&store.SystemConfig{}).Where("key = ?", "model_allowlist").Pluck("value", &raw) + _ = json.Unmarshal([]byte(raw), &allow) + raw = "" + h.db.Model(&store.SystemConfig{}).Where("key = ?", "model_denylist").Pluck("value", &raw) + _ = json.Unmarshal([]byte(raw), &deny) + return +} + +func containsStr(list []string, s string) bool { + for _, v := range list { + if v == s { + return true + } + } + return false +} + +// AdminCreateModel POST /api/admin/models +func (h *Handler) AdminCreateModel(c *gin.Context) { + var req struct { + Name string `json:"name" binding:"required,min=1,max=128"` + DisplayName string `json:"display_name"` + InputPrice float64 `json:"input_price"` + OutputPrice float64 `json:"output_price"` + CacheReadPrice float64 `json:"cache_read_price"` + Sort int `json:"sort"` + Enabled *bool `json:"enabled"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input: " + err.Error()}) + return + } + m := store.Model{ + Name: req.Name, DisplayName: req.DisplayName, + InputPrice: req.InputPrice, OutputPrice: req.OutputPrice, CacheReadPrice: req.CacheReadPrice, + Sort: req.Sort, Enabled: boolOr(req.Enabled, true), + } + if err := h.db.Create(&m).Error; err != nil { + c.JSON(http.StatusConflict, gin.H{"error": "failed to create model (name may already exist)"}) + return + } + c.JSON(http.StatusCreated, gin.H{"id": m.ID, "name": m.Name}) +} + +// AdminUpdateModel PUT /api/admin/models/:id — 价格/启停/排序。 +func (h *Handler) AdminUpdateModel(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid model id"}) + return + } + var req struct { + DisplayName *string `json:"display_name"` + InputPrice *float64 `json:"input_price"` + OutputPrice *float64 `json:"output_price"` + CacheReadPrice *float64 `json:"cache_read_price"` + Enabled *bool `json:"enabled"` + Sort *int `json:"sort"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input"}) + return + } + var m store.Model + if err := h.db.First(&m, id).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "model not found"}) + return + } + updates := map[string]any{} + if req.DisplayName != nil { + updates["display_name"] = *req.DisplayName + } + if req.InputPrice != nil { + updates["input_price"] = *req.InputPrice + } + if req.OutputPrice != nil { + updates["output_price"] = *req.OutputPrice + } + if req.CacheReadPrice != nil { + updates["cache_read_price"] = *req.CacheReadPrice + } + if req.Enabled != nil { + updates["enabled"] = *req.Enabled + } + if req.Sort != nil { + updates["sort"] = *req.Sort + } + if len(updates) > 0 { + if err := h.db.Model(&m).Updates(updates).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update model"}) + return + } + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +// AdminDeleteModel DELETE /api/admin/models/:id +func (h *Handler) AdminDeleteModel(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid model id"}) + return + } + res := h.db.Delete(&store.Model{}, id) + if res.Error != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete model"}) + return + } + if res.RowsAffected == 0 { + c.JSON(http.StatusNotFound, gin.H{"error": "model not found"}) + return + } + h.db.Where("model_id = ?", id).Delete(&store.ChannelModelBinding{}) + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +// AdminDeleteUnusedModels DELETE /api/admin/models/unused — 一键清除未绑定任何渠道的模型。 +func (h *Handler) AdminDeleteUnusedModels(c *gin.Context) { + var orphans []store.Model + if err := h.db.Where("id NOT IN (SELECT DISTINCT model_id FROM channel_model_bindings)").Find(&orphans).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load models"}) + return + } + names := make([]string, 0, len(orphans)) + ids := make([]uint64, 0, len(orphans)) + for _, m := range orphans { + names = append(names, m.Name) + ids = append(ids, m.ID) + } + if len(ids) > 0 { + if err := h.db.Delete(&store.Model{}, ids).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete models"}) + return + } + } + c.JSON(http.StatusOK, gin.H{"deleted": names, "count": len(names)}) +} + +// AdminCreateModelBinding POST /api/admin/models/:id/bindings +func (h *Handler) AdminCreateModelBinding(c *gin.Context) { + modelID, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid model id"}) + return + } + var req struct { + ChannelID uint64 `json:"channel_id" binding:"required"` + UpstreamModel string `json:"upstream_model" binding:"required"` + Weight *int `json:"weight"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid input: channel_id and upstream_model required"}) + return + } + var m store.Model + if err := h.db.First(&m, modelID).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "model not found"}) + return + } + var ch store.Channel + if err := h.db.First(&ch, req.ChannelID).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"}) + return + } + b := store.ChannelModelBinding{ + ChannelID: req.ChannelID, ModelID: modelID, + UpstreamModel: req.UpstreamModel, Weight: intOr(req.Weight, 1), + } + if err := h.db.Create(&b).Error; err != nil { + c.JSON(http.StatusConflict, gin.H{"error": "binding may already exist"}) + return + } + c.JSON(http.StatusCreated, gin.H{"id": b.ID}) +} + +// AdminDeleteModelBinding DELETE /api/admin/models/:id/bindings/:bid +func (h *Handler) AdminDeleteModelBinding(c *gin.Context) { + bid, err := strconv.ParseUint(c.Param("bid"), 10, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid binding id"}) + return + } + res := h.db.Delete(&store.ChannelModelBinding{}, bid) + if res.Error != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete binding"}) + return + } + if res.RowsAffected == 0 { + c.JSON(http.StatusNotFound, gin.H{"error": "binding not found"}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +var _ = gorm.ErrRecordNotFound diff --git a/backend/internal/api/api.go b/backend/internal/api/api.go index 9917986..849efe5 100644 --- a/backend/internal/api/api.go +++ b/backend/internal/api/api.go @@ -556,7 +556,7 @@ func (h *Handler) DeleteApiKey(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"message": "deleted"}) } -// --- Channels --- +// --- Legacy Channel endpoints (kept for backward compatibility) --- func (h *Handler) ListChannels(c *gin.Context) { // Support both limit/offset and pageSize/page parameters @@ -702,7 +702,7 @@ func (h *Handler) DeleteChannel(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"message": "deleted"}) } -// --- Models --- +// --- Legacy Model endpoints (kept for backward compatibility) --- func (h *Handler) ListModels(c *gin.Context) { // Support both limit/offset and pageSize/page parameters @@ -827,7 +827,7 @@ func (h *Handler) DeleteModel(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"message": "deleted"}) } -// --- Channel-Model Bindings --- +// --- Legacy Channel-Model Bindings (kept for backward compatibility) --- func (h *Handler) BindChannelModels(c *gin.Context) { channelID, err := strconv.ParseUint(c.Param("id"), 10, 64) diff --git a/backend/internal/channel/channel.go b/backend/internal/channel/channel.go index 2123258..a7d8bca 100644 --- a/backend/internal/channel/channel.go +++ b/backend/internal/channel/channel.go @@ -19,6 +19,9 @@ type Service struct { // Health tracking mu sync.RWMutex healthStatus map[uint64]*channelHealth + + // Concurrency control per channel + sems map[uint64]chan struct{} } type channelHealth struct { @@ -33,6 +36,7 @@ func NewService(channelDAO *dao.ChannelDAO, modelDAO *dao.ModelDAO) *Service { channelDAO: channelDAO, modelDAO: modelDAO, healthStatus: make(map[uint64]*channelHealth), + sems: make(map[uint64]chan struct{}), } } @@ -219,3 +223,24 @@ func (s *Service) SelectCandidates(ctx context.Context, modelName string, prefer return candidates, nil } + +// TryAcquire 尝试获取渠道并发槽；渠道满载返回 false（调用方可溢出到其他渠道）。 +// MaxConcurrency<=0 视为不限制。 +func (s *Service) TryAcquire(ch *store.Channel) (func(), bool) { + if ch.MaxConcurrency <= 0 { + return func() {}, true + } + s.mu.Lock() + sem, ok := s.sems[ch.ID] + if !ok { + sem = make(chan struct{}, ch.MaxConcurrency) + s.sems[ch.ID] = sem + } + s.mu.Unlock() + select { + case sem <- struct{}{}: + return func() { <-sem }, true + default: + return nil, false + } +} diff --git a/backend/internal/channel/health.go b/backend/internal/channel/health.go index f3a2a19..348acbc 100644 --- a/backend/internal/channel/health.go +++ b/backend/internal/channel/health.go @@ -10,19 +10,45 @@ import ( "time" ) +// HealthConfig 健康检查配置 +type HealthConfig struct { + Interval time.Duration // 检查间隔 + Timeout time.Duration // 请求超时 + FailureThreshold int // 连续失败次数阈值 + DegradedCooldown time.Duration // degraded 冷却时间 + CooldownCooldown time.Duration // cooldown 冷却时间 +} + +// DefaultHealthConfig 返回默认健康检查配置 +func DefaultHealthConfig() HealthConfig { + return HealthConfig{ + Interval: 5 * time.Minute, + Timeout: 10 * time.Second, + FailureThreshold: 3, + DegradedCooldown: 5 * time.Minute, + CooldownCooldown: 15 * time.Minute, + } +} + type HealthChecker struct { channelDAO *dao.ChannelDAO service *Service client *http.Client + config HealthConfig } -func NewHealthChecker(channelDAO *dao.ChannelDAO, service *Service) *HealthChecker { +func NewHealthChecker(channelDAO *dao.ChannelDAO, service *Service, config ...HealthConfig) *HealthChecker { + cfg := DefaultHealthConfig() + if len(config) > 0 { + cfg = config[0] + } return &HealthChecker{ channelDAO: channelDAO, service: service, client: &http.Client{ - Timeout: 10 * time.Second, + Timeout: cfg.Timeout, }, + config: cfg, } } @@ -91,8 +117,12 @@ func (hc *HealthChecker) CheckAllChannels(ctx context.Context) error { } // StartPeriodicCheck starts periodic health checks -func (hc *HealthChecker) StartPeriodicCheck(ctx context.Context, interval time.Duration) { - ticker := time.NewTicker(interval) +func (hc *HealthChecker) StartPeriodicCheck(ctx context.Context, interval ...time.Duration) { + interval_ := hc.config.Interval + if len(interval) > 0 { + interval_ = interval[0] + } + ticker := time.NewTicker(interval_) defer ticker.Stop() for { diff --git a/backend/internal/dao/usage.go b/backend/internal/dao/usage.go index abbcdb3..0396d72 100644 --- a/backend/internal/dao/usage.go +++ b/backend/internal/dao/usage.go @@ -82,10 +82,19 @@ func (d *DailyUsageDAO) GetByDate(ctx context.Context, userID uint64, date strin return &log, nil } +// UpsertDailyUsage 按 (user_id, model_id, date) 累加式 upsert： +// 行不存在则插入；存在则在原值基础上增量累加（不能用 AssignmentColumns 覆盖， +// 否则多次 flush 会互相清零）。非限定列名在 SQLite/MySQL/PG 的 upsert 语义下都指向目标行。 func (d *DailyUsageDAO) UpsertDailyUsage(ctx context.Context, log *store.UsageDaily) error { return d.db.WithContext(ctx).Clauses(clause.OnConflict{ - Columns: []clause.Column{{Name: "user_id"}, {Name: "model_id"}, {Name: "date"}}, - DoUpdates: clause.AssignmentColumns([]string{"requests", "input_tokens", "output_tokens", "cache_read_tokens", "cost"}), + Columns: []clause.Column{{Name: "user_id"}, {Name: "model_id"}, {Name: "date"}}, + DoUpdates: clause.Assignments(map[string]interface{}{ + "requests": gorm.Expr("requests + ?", log.Requests), + "input_tokens": gorm.Expr("input_tokens + ?", log.InputTokens), + "output_tokens": gorm.Expr("output_tokens + ?", log.OutputTokens), + "cache_read_tokens": gorm.Expr("cache_read_tokens + ?", log.CacheReadTokens), + "cost": gorm.Expr("cost + ?", log.Cost), + }), }).Create(log).Error } diff --git a/backend/internal/proxy/convert/convert.go b/backend/internal/proxy/convert/convert.go new file mode 100644 index 0000000..dc9cd22 --- /dev/null +++ b/backend/internal/proxy/convert/convert.go @@ -0,0 +1,203 @@ +// 三协议互转注册表：OpenAI Chat / OpenAI Responses / Anthropic Messages。 +// 网关以 Chat 形状作为标准中间模型：非跨 chat 的转换经 chat 中转。 +// 请求/响应（非流式）走 JSON 转换；流式走逐行 SSE 转换（stream_transform.go）。 +package convert + +import ( + "bytes" + "encoding/json" + "fmt" +) + +// 协议标识。 +const ( + ProtoChat = "chat" + ProtoMessages = "messages" + ProtoResponses = "responses" +) + +// trimBody 去掉首尾空白。部分上游（如 OpenRouter）会在 JSON 前输出空白或 +// SSE 注释行再跟正文，直接 Unmarshal 会失败。 +func trimBody(body []byte) []byte { + return bytes.TrimSpace(body) +} + +// CleanJSON 剥离非 JSON 前缀（空白、SSE 注释、`data:` 行）并压缩为标准 JSON。 +// 部分上游（如 OpenRouter）的 non-stream 响应在 JSON 前夹带空白/注释； +// 原样透传会让客户端解析失败。找不到 JSON 对象时原样返回。 +func CleanJSON(body []byte) []byte { + i := bytes.IndexByte(body, '{') + if i < 0 { + return body + } + var v any + if err := json.Unmarshal(bytes.TrimSpace(body[i:]), &v); err != nil { + return body + } + out, err := json.Marshal(v) + if err != nil { + return body + } + return out +} + +// ConvertRequest 转换请求体。from==to 时原样返回。 +func ConvertRequest(body []byte, from, to string) ([]byte, error) { + if from == to { + return body, nil + } + body = trimBody(body) + switch { + case from == ProtoMessages && to == ProtoChat: + return messagesToChatReq(body) + case from == ProtoChat && to == ProtoMessages: + return chatToMessagesReq(body) + case from == ProtoResponses && to == ProtoChat: + return responsesToChatReq(body) + case from == ProtoChat && to == ProtoResponses: + return chatToResponsesReq(body) + case from == ProtoResponses && to == ProtoMessages: + mid, err := responsesToChatReq(body) + if err != nil { + return nil, err + } + return chatToMessagesReq(mid) + case from == ProtoMessages && to == ProtoResponses: + mid, err := messagesToChatReq(body) + if err != nil { + return nil, err + } + return chatToResponsesReq(mid) + } + return nil, fmt.Errorf("unsupported request conversion %s->%s", from, to) +} + +// ConvertResponse 转换响应体（非流式）。from==to 时原样返回。 +func ConvertResponse(body []byte, from, to string) ([]byte, error) { + if from == to { + return body, nil + } + body = trimBody(body) + switch { + case from == ProtoMessages && to == ProtoChat: + return messagesToChatResp(body) + case from == ProtoChat && to == ProtoMessages: + return chatToMessagesResp(body) + case from == ProtoResponses && to == ProtoChat: + return responsesToChatResp(body) + case from == ProtoChat && to == ProtoResponses: + return chatToResponsesResp(body) + case from == ProtoResponses && to == ProtoMessages: + mid, err := responsesToChatResp(body) + if err != nil { + return nil, err + } + return chatToMessagesResp(mid) + case from == ProtoMessages && to == ProtoResponses: + mid, err := messagesToChatResp(body) + if err != nil { + return nil, err + } + return chatToResponsesResp(mid) + } + return nil, fmt.Errorf("unsupported response conversion %s->%s", from, to) +} + +// NewStreamTransformer 构造流式逐行转换器：输入上游 SSE 一行，返回客户端 SSE 行。 +// 返回 nil 表示丢弃该行或无需转换（from==to）。 +func NewStreamTransformer(from, to string) func([]byte) []byte { + switch { + case from == ProtoMessages && to == ProtoChat: + return newMessagesToChat().line + case from == ProtoChat && to == ProtoMessages: + return newChatToMessages().line + case from == ProtoResponses && to == ProtoChat: + return newResponsesToChat().line + case from == ProtoChat && to == ProtoResponses: + return newChatToResponses().line + case from == ProtoResponses && to == ProtoMessages: + return newResponsesToMessages().line + case from == ProtoMessages && to == ProtoResponses: + return newMessagesToResponses().line + } + return nil +} + +// --------------------------------------------------------------------------- +// 工具函数 + +// str 返回字符串字段；json.RawMessage 为字符串字面量时去引号。 +func str(raw json.RawMessage) string { + if len(raw) == 0 || string(raw) == "null" { + return "" + } + var s string + if json.Unmarshal(raw, &s) == nil { + return s + } + // 数组/对象：尝试取 type=text 的 text + var arr []map[string]any + if json.Unmarshal(raw, &arr) == nil { + var parts []string + for _, b := range arr { + if t, _ := b["type"].(string); t == "text" || t == "input_text" || t == "output_text" { + if txt, _ := b["text"].(string); txt != "" { + parts = append(parts, txt) + } + } + } + return joinNonEmpty(parts, "\n") + } + return "" +} + +func joinNonEmpty(parts []string, sep string) string { + out := "" + for _, p := range parts { + if p == "" { + continue + } + if out != "" { + out += sep + } + out += p + } + return out +} + +// rawJSON 安全取字段；不存在或 null 返回 nil。 +func rawJSON(m map[string]json.RawMessage, key string) json.RawMessage { + raw, ok := m[key] + if !ok || string(raw) == "null" { + return nil + } + return raw +} + +// rawOrObject 把 RawMessage 解为 map；非对象返回空对象。 +func rawOrObject(raw json.RawMessage) any { + if len(raw) == 0 || string(raw) == "null" { + return map[string]any{} + } + var m map[string]any + if json.Unmarshal(raw, &m) == nil { + return m + } + return map[string]any{} +} + +// intOrNil 取指针值，nil 时返回默认值。 +func intOrNil(p *int, def int) any { + if p == nil { + return def + } + return *p +} + +// strField 取 any 中的字符串字段。 +func strField(v any) string { + if s, ok := v.(string); ok { + return s + } + return "" +} diff --git a/backend/internal/proxy/convert/json_chat.go b/backend/internal/proxy/convert/json_chat.go new file mode 100644 index 0000000..5edaf6c --- /dev/null +++ b/backend/internal/proxy/convert/json_chat.go @@ -0,0 +1,504 @@ +package convert + +import ( + "encoding/json" + "strings" +) + +// --------------------------------------------------------------------------- +// 请求：Chat → Messages + +type chatTool struct { + Type string `json:"type"` + Function struct { + Name string `json:"name"` + Description string `json:"description"` + Parameters json.RawMessage `json:"parameters"` + } `json:"function"` +} + +type chatMsg struct { + Role string `json:"role"` + Content json.RawMessage `json:"content"` + ToolCallID string `json:"tool_call_id"` + ToolCalls []struct { + ID string `json:"id"` + Function struct { + Name string `json:"name"` + Arguments string `json:"arguments"` + } `json:"function"` + } `json:"tool_calls"` +} + +type chatReq struct { + Model string `json:"model"` + Messages []chatMsg `json:"messages"` + Tools []chatTool `json:"tools"` + Temperature *float64 `json:"temperature"` + TopP *float64 `json:"top_p"` + MaxTokens *int `json:"max_tokens"` + Stop []string `json:"stop"` + Stream bool `json:"stream"` +} + +// chatToMessagesReq 将 OpenAI Chat 请求转为 Anthropic Messages 请求。 +func chatToMessagesReq(body []byte) ([]byte, error) { + var req chatReq + if err := json.Unmarshal(body, &req); err != nil { + return nil, err + } + out := map[string]any{ + "model": req.Model, + "max_tokens": intOrNil(req.MaxTokens, 1024), // Anthropic 必填 + } + if req.Stream { + out["stream"] = true + } + if req.Temperature != nil { + out["temperature"] = *req.Temperature + } + if req.TopP != nil { + out["top_p"] = *req.TopP + } + if len(req.Stop) > 0 { + out["stop_sequences"] = req.Stop + } + + var system []string + msgs := make([]any, 0, len(req.Messages)) + for _, m := range req.Messages { + if m.Role == "system" { + if s := str(m.Content); s != "" { + system = append(system, s) + } + continue + } + msgs = append(msgs, chatMsgToAnthropic(m)) + } + if len(system) > 0 { + out["system"] = strings.Join(system, "\n") + } + out["messages"] = msgs + + if len(req.Tools) > 0 { + tools := make([]any, 0, len(req.Tools)) + for _, t := range req.Tools { + var params any + if len(t.Function.Parameters) > 0 && string(t.Function.Parameters) != "null" { + _ = json.Unmarshal(t.Function.Parameters, ¶ms) + } + tools = append(tools, map[string]any{ + "name": t.Function.Name, + "description": t.Function.Description, + "input_schema": params, + }) + } + out["tools"] = tools + } + return json.Marshal(out) +} + +// chatMsgToAnthropic 单条消息转 Anthropic 内容。 +func chatMsgToAnthropic(m chatMsg) any { + switch m.Role { + case "assistant": + content := make([]any, 0, 2) + if s := str(m.Content); s != "" { + content = append(content, map[string]any{"type": "text", "text": s}) + } + for _, tc := range m.ToolCalls { + var input any + if tc.Function.Arguments != "" { + _ = json.Unmarshal([]byte(tc.Function.Arguments), &input) + } + content = append(content, map[string]any{ + "type": "tool_use", + "id": tc.ID, + "name": tc.Function.Name, + "input": input, + }) + } + return map[string]any{"role": "assistant", "content": content} + case "tool": + return map[string]any{"role": "user", "content": []any{ + map[string]any{"type": "tool_result", "tool_use_id": m.ToolCallID, "content": str(m.Content)}, + }} + default: // user + var arr []map[string]any + if json.Unmarshal(m.Content, &arr) == nil && arr != nil { + blocks := make([]any, 0, len(arr)) + for _, b := range arr { + switch b["type"] { + case "text", "input_text": + if t, _ := b["text"].(string); t != "" { + blocks = append(blocks, map[string]any{"type": "text", "text": t}) + } + case "image_url": + var url string + if iu, ok := b["image_url"].(map[string]any); ok { + url, _ = iu["url"].(string) + } else if s, ok := b["image_url"].(string); ok { + url = s + } + if url != "" { + blocks = append(blocks, anthropicImageBlock(url)) + } + } + } + if len(blocks) > 0 { + return map[string]any{"role": "user", "content": blocks} + } + } + return map[string]any{"role": "user", "content": str(m.Content)} + } +} + +// --------------------------------------------------------------------------- +// 请求：Messages → Chat + +type messagesReq struct { + Model string `json:"model"` + System json.RawMessage `json:"system"` + Messages []struct { + Role string `json:"role"` + Content json.RawMessage `json:"content"` + } `json:"messages"` + Tools []struct { + Name string `json:"name"` + Description string `json:"description"` + InputSchema json.RawMessage `json:"input_schema"` + } `json:"tools"` + Temperature *float64 `json:"temperature"` + TopP *float64 `json:"top_p"` + MaxTokens *int `json:"max_tokens"` + StopSequence []string `json:"stop_sequences"` + Stream bool `json:"stream"` +} + +// messagesToChatReq 将 Anthropic Messages 请求转为 OpenAI Chat 请求。 +func messagesToChatReq(body []byte) ([]byte, error) { + var req messagesReq + if err := json.Unmarshal(body, &req); err != nil { + return nil, err + } + out := map[string]any{"model": req.Model} + if req.Stream { + out["stream"] = true + } + if req.Temperature != nil { + out["temperature"] = *req.Temperature + } + if req.TopP != nil { + out["top_p"] = *req.TopP + } + if req.MaxTokens != nil { + out["max_tokens"] = *req.MaxTokens + } + if len(req.StopSequence) > 0 { + out["stop"] = req.StopSequence + } + + msgs := make([]any, 0, len(req.Messages)+1) + if s := str(req.System); s != "" { + msgs = append(msgs, map[string]any{"role": "system", "content": s}) + } + for _, m := range req.Messages { + msgs = append(msgs, anthropicMsgToChat(m.Role, m.Content)...) + } + out["messages"] = msgs + + if len(req.Tools) > 0 { + tools := make([]any, 0, len(req.Tools)) + for _, t := range req.Tools { + tools = append(tools, map[string]any{ + "type": "function", + "function": map[string]any{ + "name": t.Name, + "description": t.Description, + "parameters": rawOrObject(t.InputSchema), + }, + }) + } + out["tools"] = tools + } + return json.Marshal(out) +} + +// anthropicMsgToChat 将一条 Anthropic 消息拆成 0..N 条 Chat 消息。 +func anthropicMsgToChat(role string, content json.RawMessage) []any { + // 块数组优先（tool_use / tool_result 需要分块解析） + var blocks []map[string]any + if json.Unmarshal(content, &blocks) == nil && blocks != nil { + var out []any + var toolMsgs []any // tool_result 单独收集，保证排在 assistant(tool_calls) 之后 + var textParts []string + var contentBlocks []any // text / image_url 块，保留原始顺序 + var toolCalls []any + for _, b := range blocks { + switch b["type"] { + case "text": + if t, _ := b["text"].(string); t != "" { + textParts = append(textParts, t) + contentBlocks = append(contentBlocks, map[string]any{"type": "text", "text": t}) + } + case "image": + if cb := chatImageBlock(b); cb != nil { + contentBlocks = append(contentBlocks, cb) + } + case "tool_use": + id, _ := b["id"].(string) + name, _ := b["name"].(string) + args, _ := json.Marshal(b["input"]) + toolCalls = append(toolCalls, map[string]any{ + "id": id, + "type": "function", + "function": map[string]any{ + "name": name, + "arguments": string(args), + }, + }) + case "tool_result": + callID, _ := b["tool_use_id"].(string) + res := strField(b["content"]) + toolMsgs = append(toolMsgs, map[string]any{"role": "tool", "tool_call_id": callID, "content": res}) + } + } + hasImage := false + for _, cb := range contentBlocks { + if m, _ := cb.(map[string]any); m["type"] == "image_url" { + hasImage = true + break + } + } + if hasImage || len(textParts) > 0 || len(toolCalls) > 0 { + msg := map[string]any{"role": role} + switch { + case hasImage: + msg["content"] = contentBlocks + case len(textParts) > 0: + msg["content"] = strings.Join(textParts, "") + } + if len(toolCalls) > 0 { + msg["tool_calls"] = toolCalls + } + out = append(out, msg) + } + out = append(out, toolMsgs...) + if len(out) > 0 { + return out + } + } + // 纯文本 + if s := str(content); s != "" { + return []any{map[string]any{"role": role, "content": s}} + } + return nil +} + +// --------------------------------------------------------------------------- +// 响应：Messages → Chat + +type messagesResp struct { + ID string `json:"id"` + Model string `json:"model"` + Content []struct { + Type string `json:"type"` + Text string `json:"text"` + ID string `json:"id"` + Name string `json:"name"` + Input json.RawMessage `json:"input"` + } `json:"content"` + StopReason string `json:"stop_reason"` + Usage struct { + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + } `json:"usage"` +} + +// messagesToChatResp 将 Anthropic Messages 响应（非流式）转为 Chat 响应。 +func messagesToChatResp(body []byte) ([]byte, error) { + var r messagesResp + if err := json.Unmarshal(body, &r); err != nil { + return nil, err + } + var text string + var toolCalls []any + for _, c := range r.Content { + switch c.Type { + case "text": + text += c.Text + case "tool_use": + args, _ := json.Marshal(c.Input) + toolCalls = append(toolCalls, map[string]any{ + "id": c.ID, + "type": "function", + "function": map[string]any{ + "name": c.Name, + "arguments": string(args), + }, + }) + } + } + msg := map[string]any{"role": "assistant", "content": text} + if len(toolCalls) > 0 { + msg["tool_calls"] = toolCalls + } + return json.Marshal(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(r.ID, "msg_"), + "object": "chat.completion", + "model": r.Model, + "created": 0, + "choices": []any{map[string]any{ + "index": 0, + "message": msg, + "finish_reason": messagesStopToChat(r.StopReason), + }}, + "usage": map[string]any{ + "prompt_tokens": r.Usage.InputTokens, + "completion_tokens": r.Usage.OutputTokens, + "total_tokens": r.Usage.InputTokens + r.Usage.OutputTokens, + }, + }) +} + +// --------------------------------------------------------------------------- +// 响应：Chat → Messages + +type chatResp struct { + ID string `json:"id"` + Model string `json:"model"` + Choices []struct { + Message struct { + Role string `json:"role"` + Content string `json:"content"` + ToolCalls []struct { + ID string `json:"id"` + Function struct { + Name string `json:"name"` + Arguments string `json:"arguments"` + } `json:"function"` + } `json:"tool_calls"` + } `json:"message"` + FinishReason string `json:"finish_reason"` + } `json:"choices"` + Usage struct { + PromptTokens int64 `json:"prompt_tokens"` + CompletionTokens int64 `json:"completion_tokens"` + } `json:"usage"` +} + +// chatToMessagesResp 将 Chat 响应（非流式）转为 Messages 响应。 +func chatToMessagesResp(body []byte) ([]byte, error) { + var r chatResp + if err := json.Unmarshal(body, &r); err != nil { + return nil, err + } + content := make([]any, 0, 2) + var finish = "end_turn" + if len(r.Choices) > 0 { + msg := r.Choices[0].Message + if msg.Content != "" { + content = append(content, map[string]any{"type": "text", "text": msg.Content}) + } + for _, tc := range msg.ToolCalls { + var input any + _ = json.Unmarshal([]byte(tc.Function.Arguments), &input) + content = append(content, map[string]any{ + "type": "tool_use", + "id": tc.ID, + "name": tc.Function.Name, + "input": input, + }) + } + finish = chatStopToMessages(r.Choices[0].FinishReason) + } + return json.Marshal(map[string]any{ + "id": "msg_" + strings.TrimPrefix(r.ID, "chatcmpl-"), + "type": "message", + "role": "assistant", + "model": r.Model, + "content": content, + "stop_reason": finish, + "usage": map[string]any{ + "input_tokens": r.Usage.PromptTokens, + "output_tokens": r.Usage.CompletionTokens, + }, + }) +} + +// --------------------------------------------------------------------------- +// 辅助 + +// splitDataURL 解析 data:media_type;base64,data 形式的 URL；非该形式返回 ok=false。 +func splitDataURL(url string) (media, data string, ok bool) { + if !strings.HasPrefix(url, "data:") { + return "", "", false + } + i := strings.Index(url, ";base64,") + if i < 0 { + return "", "", false + } + return url[len("data:"):i], url[i+len(";base64,"):], true +} + +// chatImageBlock 把 Anthropic image 块转 OpenAI image_url 块。 +// 仅支持 base64 与 url source；其他类型（如 Files API 的 file_id）不支持，跳过。 +func chatImageBlock(b map[string]any) any { + src, ok := b["source"].(map[string]any) + if !ok { + return nil + } + switch src["type"] { + case "base64": + media, _ := src["media_type"].(string) + data, _ := src["data"].(string) + if data == "" { + return nil + } + if media == "" { + media = "image/png" + } + return map[string]any{"type": "image_url", "image_url": map[string]any{"url": "data:" + media + ";base64," + data}} + case "url": + url, _ := src["url"].(string) + if url == "" { + return nil + } + return map[string]any{"type": "image_url", "image_url": map[string]any{"url": url}} + } + return nil +} + +// anthropicImageBlock 把 OpenAI image_url 的 url 转 Anthropic image 块。 +// data URL → base64 source；http(s) URL → url source。 +func anthropicImageBlock(url string) any { + if media, data, ok := splitDataURL(url); ok { + if media == "" { + media = "image/png" + } + return map[string]any{"type": "image", "source": map[string]any{"type": "base64", "media_type": media, "data": data}} + } + return map[string]any{"type": "image", "source": map[string]any{"type": "url", "url": url}} +} + +func messagesStopToChat(s string) string { + switch s { + case "tool_use": + return "tool_calls" + case "max_tokens": + return "length" + default: + return "stop" + } +} + +func chatStopToMessages(s string) string { + switch s { + case "tool_calls": + return "tool_use" + case "length": + return "max_tokens" + default: + return "end_turn" + } +} diff --git a/backend/internal/proxy/convert/json_responses.go b/backend/internal/proxy/convert/json_responses.go new file mode 100644 index 0000000..7762e07 --- /dev/null +++ b/backend/internal/proxy/convert/json_responses.go @@ -0,0 +1,376 @@ +package convert + +import ( + "encoding/json" + "strings" +) + +// --------------------------------------------------------------------------- +// 请求：Responses → Chat + +// responsesToChatReq 将 OpenAI Responses 请求转为 Chat 请求。 +func responsesToChatReq(body []byte) ([]byte, error) { + var m map[string]json.RawMessage + if err := json.Unmarshal(body, &m); err != nil { + return nil, err + } + out := map[string]any{"model": str(rawJSON(m, "model"))} + if v, ok := m["stream"]; ok && string(v) == "true" { + out["stream"] = true + } + if v, ok := m["temperature"]; ok { + out["temperature"] = v + } + if v, ok := m["top_p"]; ok { + out["top_p"] = v + } + if v, ok := m["max_output_tokens"]; ok { + out["max_tokens"] = v + } + + var msgs []any + if ins := str(rawJSON(m, "instructions")); ins != "" { + msgs = append(msgs, map[string]any{"role": "system", "content": ins}) + } + msgs = append(msgs, responsesInputToChat(rawJSON(m, "input"))...) + out["messages"] = msgs + + if raw := rawJSON(m, "tools"); raw != nil { + var tools []map[string]any + if json.Unmarshal(raw, &tools) == nil { + chatTools := make([]any, 0, len(tools)) + for _, t := range tools { + chatTools = append(chatTools, map[string]any{ + "type": "function", + "function": map[string]any{ + "name": t["name"], + "description": t["description"], + "parameters": t["parameters"], + }, + }) + } + out["tools"] = chatTools + } + } + return json.Marshal(out) +} + +// responsesInputToChat 把 Responses input 转成 Chat messages。 +// input 支持字符串或条目数组（message / function_call / function_call_output）。 +func responsesInputToChat(raw json.RawMessage) []any { + if len(raw) == 0 || string(raw) == "null" { + return nil + } + // 字符串输入 + if s := str(raw); s != "" { + return []any{map[string]any{"role": "user", "content": s}} + } + var items []map[string]any + if err := json.Unmarshal(raw, &items); err != nil || items == nil { + return nil + } + var out []any + for _, item := range items { + switch item["type"] { + case "function_call": + out = append(out, map[string]any{ + "role": "assistant", + "content": "", + "tool_calls": []any{map[string]any{ + "id": strField(item["call_id"]), + "type": "function", + "function": map[string]any{ + "name": strField(item["name"]), + "arguments": strField(item["arguments"]), + }, + }}, + }) + case "function_call_output": + out = append(out, map[string]any{ + "role": "tool", + "tool_call_id": strField(item["call_id"]), + "content": strField(item["output"]), + }) + default: // message 条目 + role, _ := item["role"].(string) + if role == "" { + role = "user" + } + if content, ok := item["content"].(string); ok { + out = append(out, map[string]any{"role": role, "content": content}) + } else if blocks, ok := item["content"].([]any); ok { + var text []string + var contentBlocks []any + for _, b := range blocks { + bm, ok := b.(map[string]any) + if !ok { + continue + } + switch bm["type"] { + case "input_text", "text": + if t, _ := bm["text"].(string); t != "" { + text = append(text, t) + contentBlocks = append(contentBlocks, map[string]any{"type": "text", "text": t}) + } + case "input_image": + var url string + if s, ok := bm["image_url"].(string); ok { + url = s + } else if m, ok := bm["image_url"].(map[string]any); ok { + url, _ = m["url"].(string) + } + if url != "" { + contentBlocks = append(contentBlocks, map[string]any{"type": "image_url", "image_url": map[string]any{"url": url}}) + } + } + } + hasImage := false + for _, cb := range contentBlocks { + if m, _ := cb.(map[string]any); m["type"] == "image_url" { + hasImage = true + break + } + } + if hasImage { + out = append(out, map[string]any{"role": role, "content": contentBlocks}) + } else { + out = append(out, map[string]any{"role": role, "content": strings.Join(text, "")}) + } + } + } + } + return out +} + +// chatContentToResponsesBlocks 把 Chat 用户消息 content 转 Responses input 块数组（input_text / input_image）。 +func chatContentToResponsesBlocks(content json.RawMessage) []any { + // 纯字符串 → 单个 input_text + var s string + if json.Unmarshal(content, &s) == nil && s != "" { + return []any{map[string]any{"type": "input_text", "text": s}} + } + // 数组 → 按块转换（text / image_url） + var arr []map[string]any + if json.Unmarshal(content, &arr) == nil && arr != nil { + var out []any + for _, b := range arr { + switch b["type"] { + case "text", "input_text": + if t, _ := b["text"].(string); t != "" { + out = append(out, map[string]any{"type": "input_text", "text": t}) + } + case "image_url": + var url string + if iu, ok := b["image_url"].(map[string]any); ok { + url, _ = iu["url"].(string) + } else if s, ok := b["image_url"].(string); ok { + url = s + } + if url != "" { + out = append(out, map[string]any{"type": "input_image", "image_url": url}) + } + } + } + return out + } + return nil +} + +// --------------------------------------------------------------------------- +// 请求：Chat → Responses + +// chatToResponsesReq 将 Chat 请求转为 Responses 请求。 +func chatToResponsesReq(body []byte) ([]byte, error) { + var req chatReq + if err := json.Unmarshal(body, &req); err != nil { + return nil, err + } + out := map[string]any{"model": req.Model} + if req.Stream { + out["stream"] = true + } + if req.Temperature != nil { + out["temperature"] = *req.Temperature + } + if req.TopP != nil { + out["top_p"] = *req.TopP + } + if req.MaxTokens != nil { + out["max_output_tokens"] = *req.MaxTokens + } + + var system []string + var input []any + for _, m := range req.Messages { + if m.Role == "system" { + if s := str(m.Content); s != "" { + system = append(system, s) + } + continue + } + switch m.Role { + case "tool": + input = append(input, map[string]any{ + "type": "function_call_output", + "call_id": m.ToolCallID, + "output": str(m.Content), + }) + case "assistant": + if len(m.ToolCalls) > 0 { + for _, tc := range m.ToolCalls { + input = append(input, map[string]any{ + "type": "function_call", + "call_id": tc.ID, + "name": tc.Function.Name, + "arguments": tc.Function.Arguments, + }) + } + } else if s := str(m.Content); s != "" { + input = append(input, map[string]any{"type": "message", "role": "assistant", "content": []any{ + map[string]any{"type": "input_text", "text": s}, + }}) + } + default: + if blocks := chatContentToResponsesBlocks(m.Content); len(blocks) > 0 { + input = append(input, map[string]any{"type": "message", "role": "user", "content": blocks}) + } + } + } + if len(system) > 0 { + out["instructions"] = strings.Join(system, "\n") + } + // input 必须是数组：部分上游只接受数组，单对象会被拒（400 Mismatch type）。 + out["input"] = input + + if len(req.Tools) > 0 { + tools := make([]any, 0, len(req.Tools)) + for _, t := range req.Tools { + tools = append(tools, map[string]any{ + "type": "function", + "name": t.Function.Name, + "description": t.Function.Description, + "parameters": rawOrObject(t.Function.Parameters), + }) + } + out["tools"] = tools + } + return json.Marshal(out) +} + +// --------------------------------------------------------------------------- +// 响应：Responses → Chat + +// responsesToChatResp 将 Responses 响应（非流式）转为 Chat 响应。 +func responsesToChatResp(body []byte) ([]byte, error) { + var m map[string]json.RawMessage + if err := json.Unmarshal(body, &m); err != nil { + return nil, err + } + var text string + var toolCalls []any + if raw := rawJSON(m, "output"); raw != nil { + var outputs []map[string]any + if json.Unmarshal(raw, &outputs) == nil { + for _, o := range outputs { + switch o["type"] { + case "message": + if content, ok := o["content"].([]any); ok { + for _, c := range content { + if cm, ok := c.(map[string]any); ok { + if t, _ := cm["text"].(string); t != "" { + text += t + } + } + } + } + case "function_call": + toolCalls = append(toolCalls, map[string]any{ + "id": strField(o["call_id"]), + "type": "function", + "function": map[string]any{ + "name": strField(o["name"]), + "arguments": strField(o["arguments"]), + }, + }) + } + } + } + } + msg := map[string]any{"role": "assistant", "content": text} + if len(toolCalls) > 0 { + msg["tool_calls"] = toolCalls + } + finish := "stop" + switch { + case string(rawJSON(m, "status")) == `"incomplete"`: + finish = "length" // 截断优先，客户端可据此区分 + case len(toolCalls) > 0: + finish = "tool_calls" + } + var prompt, completion int64 + if u := rawJSON(m, "usage"); u != nil { + var us struct { + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + } + _ = json.Unmarshal(u, &us) + prompt, completion = us.InputTokens, us.OutputTokens + } + return json.Marshal(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(str(rawJSON(m, "id")), "resp_"), + "object": "chat.completion", + "model": str(rawJSON(m, "model")), + "choices": []any{map[string]any{"index": 0, "message": msg, "finish_reason": finish}}, + "usage": map[string]any{ + "prompt_tokens": prompt, + "completion_tokens": completion, + "total_tokens": prompt + completion, + }, + }) +} + +// --------------------------------------------------------------------------- +// 响应：Chat → Responses + +// chatToResponsesResp 将 Chat 响应（非流式）转为 Responses 响应。 +func chatToResponsesResp(body []byte) ([]byte, error) { + var r chatResp + if err := json.Unmarshal(body, &r); err != nil { + return nil, err + } + output := make([]any, 0, 2) + var finish = "completed" + if len(r.Choices) > 0 { + msg := r.Choices[0].Message + if msg.Content != "" { + output = append(output, map[string]any{ + "type": "message", + "role": "assistant", + "content": []any{map[string]any{"type": "output_text", "text": msg.Content}}, + }) + } + for _, tc := range msg.ToolCalls { + output = append(output, map[string]any{ + "type": "function_call", + "call_id": tc.ID, + "name": tc.Function.Name, + "arguments": tc.Function.Arguments, + }) + } + if r.Choices[0].FinishReason == "length" { + finish = "incomplete" + } + } + return json.Marshal(map[string]any{ + "id": "resp_" + strings.TrimPrefix(r.ID, "chatcmpl-"), + "object": "response", + "model": r.Model, + "status": finish, + "output": output, + "usage": map[string]any{ + "input_tokens": r.Usage.PromptTokens, + "output_tokens": r.Usage.CompletionTokens, + "total_tokens": r.Usage.PromptTokens + r.Usage.CompletionTokens, + }, + }) +} diff --git a/backend/internal/proxy/convert/stream_transform.go b/backend/internal/proxy/convert/stream_transform.go new file mode 100644 index 0000000..fdd5c08 --- /dev/null +++ b/backend/internal/proxy/convert/stream_transform.go @@ -0,0 +1,642 @@ +package convert + +import ( + "encoding/json" + "strings" +) + +// sseState 记录上一行 event 名。 +type sseState struct { + event string +} + +// parseLine 解析一行 SSE；返回是否 data 行及其内容、是否 [DONE]。 +// data: 后可跟空格（标准）或紧贴 JSON（部分上游会省略空格）。 +func (s *sseState) parseLine(line []byte) (isData bool, data string, done bool) { + strLine := strings.TrimRight(string(line), "\r\n") + switch { + case strings.HasPrefix(strLine, "event: "): + s.event = strings.TrimSpace(strings.TrimPrefix(strLine, "event: ")) + return false, "", false + case strLine == "data: [DONE]" || strLine == "data:[DONE]": + return true, "[DONE]", true + case strings.HasPrefix(strLine, "data:"): + return true, strings.TrimLeft(strings.TrimPrefix(strLine, "data:"), " "), false + default: + return false, "", false + } +} + +func eventData(line string) map[string]any { + var m map[string]any + _ = json.Unmarshal([]byte(line), &m) + return m +} + +func dataLine(obj any) []byte { + b, _ := json.Marshal(obj) + return append(append([]byte("data: "), b...), '\n', '\n') +} + +func eventLine(name string, obj any) []byte { + b, _ := json.Marshal(obj) + out := append([]byte("event: "+name+"\ndata: "), b...) + return append(out, '\n', '\n') +} + +// joinLines 拼接多条 SSE 行。 +func joinLines(lines [][]byte) []byte { + var s []string + for _, l := range lines { + s = append(s, string(l)) + } + return []byte(strings.Join(s, "")) +} + +// --------------------------------------------------------------------------- +// Messages → Chat + +type messagesToChat struct { + sseState + id, model string + toolIdx map[int]int // messages content block index → chat tool_calls index（顺序编号，避开文本块） + nextTool int +} + +func newMessagesToChat() *messagesToChat { return &messagesToChat{toolIdx: map[int]int{}} } + +func (t *messagesToChat) line(line []byte) []byte { + isData, data, done := t.parseLine(line) + if !isData { + return nil + } + if done { + return []byte("data: [DONE]\n\n") + } + m := eventData(data) + evt, _ := m["type"].(string) + switch evt { + case "message_start": + msg, _ := m["message"].(map[string]any) + t.id, _ = msg["id"].(string) + t.model, _ = msg["model"].(string) + return dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"role": "assistant"}, "finish_reason": nil}}, + }) + case "content_block_start": + cb, _ := m["content_block"].(map[string]any) + if cb == nil || cb["type"] != "tool_use" { + return nil + } + blockIdx, _ := m["index"].(float64) + tool := t.nextTool + t.nextTool++ + t.toolIdx[int(blockIdx)] = tool + toolID, _ := cb["id"].(string) + name, _ := cb["name"].(string) + return dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{ + "tool_calls": []any{map[string]any{"index": tool, "id": toolID, "type": "function", "function": map[string]any{"name": name, "arguments": ""}}}, + }, "finish_reason": nil}}, + }) + case "content_block_delta": + delta, _ := m["delta"].(map[string]any) + deltaType, _ := delta["type"].(string) + if deltaType == "input_json_delta" { + blockIdx, _ := m["index"].(float64) + tool, ok := t.toolIdx[int(blockIdx)] + if !ok { + return nil + } + partial, _ := delta["partial_json"].(string) + if partial == "" { + return nil + } + return dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{ + "tool_calls": []any{map[string]any{"index": tool, "function": map[string]any{"arguments": partial}}}, + }, "finish_reason": nil}}, + }) + } + text, _ := delta["text"].(string) + if text == "" { + return nil + } + return dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"content": text}, "finish_reason": nil}}, + }) + case "message_delta": + delta, _ := m["delta"].(map[string]any) + stop, _ := delta["stop_reason"].(string) + var out [][]byte + out = append(out, dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{}, "finish_reason": messagesStopToChat(stop)}}, + })) + if u, ok := m["usage"]; ok { + out = append(out, dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{}, "usage": u, + })) + } + return joinLines(out) + case "message_stop": + return []byte("data: [DONE]\n\n") + } + return nil +} + +// --------------------------------------------------------------------------- +// Chat → Messages + +type chatToMessages struct { + sseState + started bool // message_start 已发出 + nextIndex int // 下一个 content block index（顺序分配） + textIndex int // 文本块 index；-1 = 未开始 + toolIdx map[int]int // chat delta.tool_calls[].index → messages block index + openBlocks []int // 已开始未停止的 block index，按开始顺序 + model string + stopReason string + usage any +} + +func newChatToMessages() *chatToMessages { + return &chatToMessages{textIndex: -1, toolIdx: map[int]int{}} +} + +func (t *chatToMessages) line(line []byte) []byte { + isData, data, done := t.parseLine(line) + if !isData { + return nil + } + if done { + // 汇聚最终：先对每个已开始未停止的块发 content_block_stop，再 message_delta + message_stop + var out [][]byte + for _, idx := range t.openBlocks { + out = append(out, eventLine("content_block_stop", map[string]any{"type": "content_block_stop", "index": idx})) + } + md := map[string]any{"type": "message_delta", "delta": map[string]any{ + "stop_reason": stopReasonOrEnd(t.stopReason), "stop_sequence": nil, + }} + if t.usage != nil { + md["usage"] = t.usage + } + out = append(out, eventLine("message_delta", md)) + out = append(out, eventLine("message_stop", map[string]any{"type": "message_stop"})) + return joinLines(out) + } + m := eventData(data) + // chat 块：delta / finish_reason 在 choices[0] 内 + delta := map[string]any{} + if choices, ok := m["choices"].([]any); ok && len(choices) > 0 { + if c0, ok := choices[0].(map[string]any); ok { + if d, ok := c0["delta"].(map[string]any); ok { + delta = d + } + if fr, _ := c0["finish_reason"].(string); fr != "" { + t.stopReason = fr + } + } + } + if t.model == "" { + t.model, _ = m["model"].(string) + } + id, _ := m["id"].(string) + + var out [][]byte + // message_start 只在实际有内容（文本或工具）时发出，避免 reasoning_content 块 + //（带 role 无 content）提前开出一个空文本块。 + ensureStarted := func() { + if t.started { + return + } + t.started = true + out = append(out, eventLine("message_start", map[string]any{ + "type": "message_start", + "message": map[string]any{ + "id": "msg_" + strings.TrimPrefix(id, "chatcmpl-"), "type": "message", "role": "assistant", + "model": t.model, "content": []any{}, "usage": map[string]any{"input_tokens": 0, "output_tokens": 0}, + }, + })) + } + + // 文本：delta.content（string；兼容 {type:text,text} 数组） + if content := deltaText(delta); content != "" { + if t.textIndex < 0 { + t.textIndex = t.nextIndex + t.nextIndex++ + ensureStarted() + out = append(out, eventLine("content_block_start", map[string]any{ + "type": "content_block_start", "index": t.textIndex, "content_block": map[string]any{"type": "text", "text": ""}, + })) + t.openBlocks = append(t.openBlocks, t.textIndex) + } + out = append(out, eventLine("content_block_delta", map[string]any{ + "type": "content_block_delta", "index": t.textIndex, "delta": map[string]any{"type": "text_delta", "text": content}, + })) + } + + // 工具调用：delta.tool_calls（并行调用各 index 独立成块；arguments 支持整段/分段两种流式） + if tcs, ok := delta["tool_calls"].([]any); ok { + for _, tc := range tcs { + call, ok := tc.(map[string]any) + if !ok { + continue + } + idx, _ := call["index"].(float64) + tcIdx := int(idx) + fn, _ := call["function"].(map[string]any) + name, _ := fn["name"].(string) + args, _ := fn["arguments"].(string) + blockIdx, seen := t.toolIdx[tcIdx] + if !seen { + blockIdx = t.nextIndex + t.nextIndex++ + t.toolIdx[tcIdx] = blockIdx + toolID, _ := call["id"].(string) + ensureStarted() + out = append(out, eventLine("content_block_start", map[string]any{ + "type": "content_block_start", "index": blockIdx, "content_block": map[string]any{ + "type": "tool_use", "id": toolID, "name": name, "input": map[string]any{}, + }, + })) + t.openBlocks = append(t.openBlocks, blockIdx) + } + if args != "" { + out = append(out, eventLine("content_block_delta", map[string]any{ + "type": "content_block_delta", "index": blockIdx, "delta": map[string]any{"type": "input_json_delta", "partial_json": args}, + })) + } + } + } + + if u, ok := m["usage"]; ok { + t.usage = u + } + return joinLines(out) +} + +// deltaText 取 chat delta.content 文本（string 或 [{type:text,text}] 数组拼接）。 +func deltaText(delta map[string]any) string { + if s, ok := delta["content"].(string); ok { + return s + } + if arr, ok := delta["content"].([]any); ok { + var parts []string + for _, b := range arr { + if bm, ok := b.(map[string]any); ok { + if t, _ := bm["text"].(string); t != "" { + parts = append(parts, t) + } + } + } + return strings.Join(parts, "") + } + return "" +} + +func stopReasonOrEnd(s string) string { + if s == "" { + return "end_turn" + } + return chatStopToMessages(s) +} + +// --------------------------------------------------------------------------- +// Responses → Messages + +type responsesToMessages struct { + sseState + started bool + model string + usage any + nextIndex int // 下一个 content block index（顺序分配） + textIndex int // 文本块 index；-1 = 未开始 + toolIdx map[string]int // function_call item_id → messages block index + openBlocks []int // 已开始未停止的 block index，按开始顺序 + anyTool bool +} + +func newResponsesToMessages() *responsesToMessages { + return &responsesToMessages{textIndex: -1, toolIdx: map[string]int{}} +} + +func (t *responsesToMessages) line(line []byte) []byte { + isData, data, done := t.parseLine(line) + if !isData || done { + return nil + } + m := eventData(data) + evt, _ := m["type"].(string) + if resp, ok := m["response"].(map[string]any); ok { + if t.model == "" { + t.model, _ = resp["model"].(string) + } + if u, ok := resp["usage"]; ok { + t.usage = u + } + } + var out [][]byte + // message_start 只在 response.created 时发出；文本/工具块在对应事件到达时再开， + // 避免纯函数调用响应提前开出一个空文本块。 + ensureStarted := func() { + if t.started { + return + } + t.started = true + rid := "" + if resp, ok := m["response"].(map[string]any); ok { + rid, _ = resp["id"].(string) + } + out = append(out, eventLine("message_start", map[string]any{ + "type": "message_start", + "message": map[string]any{ + "id": "msg_" + strings.TrimPrefix(rid, "resp_"), "type": "message", "role": "assistant", + "model": t.model, "content": []any{}, + }, + })) + } + switch evt { + case "response.created": + ensureStarted() + case "response.output_text.delta": + delta, _ := m["delta"].(string) + if delta == "" { + return nil + } + if t.textIndex < 0 { + t.textIndex = t.nextIndex + t.nextIndex++ + ensureStarted() + out = append(out, eventLine("content_block_start", map[string]any{ + "type": "content_block_start", "index": t.textIndex, "content_block": map[string]any{"type": "text", "text": ""}, + })) + t.openBlocks = append(t.openBlocks, t.textIndex) + } + out = append(out, eventLine("content_block_delta", map[string]any{ + "type": "content_block_delta", "index": t.textIndex, "delta": map[string]any{"type": "text_delta", "text": delta}, + })) + case "response.output_item.added": + item, _ := m["item"].(map[string]any) + if item == nil || item["type"] != "function_call" { + return nil + } + blockIdx := t.nextIndex + t.nextIndex++ + t.anyTool = true + itemID, _ := item["id"].(string) + t.toolIdx[itemID] = blockIdx + toolUseID, _ := item["call_id"].(string) + if toolUseID == "" { + toolUseID = itemID + } + name, _ := item["name"].(string) + ensureStarted() + out = append(out, eventLine("content_block_start", map[string]any{ + "type": "content_block_start", "index": blockIdx, "content_block": map[string]any{ + "type": "tool_use", "id": toolUseID, "name": name, "input": map[string]any{}, + }, + })) + t.openBlocks = append(t.openBlocks, blockIdx) + case "response.function_call_arguments.delta": + itemID, _ := m["item_id"].(string) + blockIdx, ok := t.toolIdx[itemID] + if !ok { + return nil + } + delta, _ := m["delta"].(string) + if delta == "" { + return nil + } + out = append(out, eventLine("content_block_delta", map[string]any{ + "type": "content_block_delta", "index": blockIdx, "delta": map[string]any{"type": "input_json_delta", "partial_json": delta}, + })) + case "response.completed": + for _, idx := range t.openBlocks { + out = append(out, eventLine("content_block_stop", map[string]any{"type": "content_block_stop", "index": idx})) + } + stop := "end_turn" + if t.anyTool { + stop = "tool_use" + } + md := map[string]any{"type": "message_delta", "delta": map[string]any{"stop_reason": stop, "stop_sequence": nil}} + if t.usage != nil { + md["usage"] = t.usage + } + out = append(out, eventLine("message_delta", md)) + out = append(out, eventLine("message_stop", map[string]any{"type": "message_stop"})) + } + return joinLines(out) +} + +// --------------------------------------------------------------------------- +// Messages → Responses + +type messagesToResponses struct { + sseState + model string + usage any + done bool +} + +func newMessagesToResponses() *messagesToResponses { return &messagesToResponses{} } + +func (t *messagesToResponses) line(line []byte) []byte { + isData, data, done := t.parseLine(line) + if !isData || done { + return nil + } + m := eventData(data) + evt, _ := m["type"].(string) + if msg, ok := m["message"].(map[string]any); ok { + if t.model == "" { + t.model, _ = msg["model"].(string) + } + if u, ok := msg["usage"]; ok { + t.usage = u + } + } + if u, ok := m["usage"]; ok { + t.usage = u + } + var out [][]byte + switch evt { + case "message_start": + id, _ := m["message"].(map[string]any) + rid := "" + if id != nil { + rid, _ = id["id"].(string) + } + out = append(out, eventLine("response.created", map[string]any{ + "type": "response.created", + "response": map[string]any{ + "id": "resp_" + strings.TrimPrefix(rid, "msg_"), "object": "response", "model": t.model, "status": "in_progress", + }, + })) + case "content_block_delta": + delta, _ := m["delta"].(map[string]any) + text, _ := delta["text"].(string) + if text != "" { + out = append(out, eventLine("response.output_text.delta", map[string]any{ + "type": "response.output_text.delta", "delta": text, "item_id": "msg_1", "output_index": 0, "content_index": 0, + })) + } + case "message_stop": + if !t.done { + t.done = true + out = append(out, eventLine("response.completed", map[string]any{ + "type": "response.completed", + "response": map[string]any{ + "id": "resp_stream", "object": "response", "model": t.model, "status": "completed", "usage": t.usage, + }, + })) + } + } + return joinLines(out) +} + +// --------------------------------------------------------------------------- +// Responses → Chat + +type responsesToChat struct { + sseState + id, model string +} + +func newResponsesToChat() *responsesToChat { return &responsesToChat{} } + +func (t *responsesToChat) line(line []byte) []byte { + isData, data, done := t.parseLine(line) + if !isData { + return nil + } + if done { + return nil + } + m := eventData(data) + evt, _ := m["type"].(string) + if resp, ok := m["response"].(map[string]any); ok { + if t.model == "" { + t.model, _ = resp["model"].(string) + } + if t.id == "" { + t.id, _ = resp["id"].(string) + } + } + var out [][]byte + switch evt { + case "response.created": + out = append(out, dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"role": "assistant"}, "finish_reason": nil}}, + })) + case "response.output_text.delta": + delta, _ := m["delta"].(string) + if delta != "" { + out = append(out, dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"content": delta}, "finish_reason": nil}}, + })) + } + case "response.completed": + out = append(out, dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{}, "finish_reason": "stop"}}, + })) + if u, ok := m["response"].(map[string]any); ok { + if usage, ok := u["usage"]; ok { + out = append(out, dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{}, "usage": usage, + })) + } + } + out = append(out, []byte("data: [DONE]\n\n")) + } + return joinLines(out) +} + +// --------------------------------------------------------------------------- +// Chat → Responses + +type chatToResponses struct { + sseState + model string + usage any + finishSeen bool + done bool + createdSent bool +} + +func newChatToResponses() *chatToResponses { return &chatToResponses{} } + +func (t *chatToResponses) line(line []byte) []byte { + isData, data, done := t.parseLine(line) + if !isData { + return nil + } + if done { + // 流结束兜底：finish 后 usage 未随块到达时在此补发 completed + if !t.done { + t.done = true + return eventLine("response.completed", map[string]any{ + "type": "response.completed", + "response": map[string]any{ + "id": "resp_stream", "object": "response", "model": t.model, "status": "completed", "usage": t.usage, + }, + }) + } + return nil + } + m := eventData(data) + if t.model == "" { + t.model, _ = m["model"].(string) + } + if u, ok := m["usage"]; ok { + t.usage = u + } + delta := map[string]any{} + var finish string + if choices, ok := m["choices"].([]any); ok && len(choices) > 0 { + if c0, ok := choices[0].(map[string]any); ok { + if d, ok := c0["delta"].(map[string]any); ok { + delta = d + } + finish, _ = c0["finish_reason"].(string) + } + } + if finish != "" { + t.finishSeen = true + } + var out [][]byte + // 只发一次 response.created：部分上游（如 OpenRouter 的 reasoning 模型）会在 + // 每个 chunk 的 delta 里都带 role:"assistant"，不加守卫会刷出数十条 created。 + if !t.createdSent && delta["role"] == "assistant" { + t.createdSent = true + out = append(out, eventLine("response.created", map[string]any{ + "type": "response.created", + "response": map[string]any{"id": "resp_stream", "object": "response", "model": t.model, "status": "in_progress"}, + })) + } + if content, _ := delta["content"].(string); content != "" { + out = append(out, eventLine("response.output_text.delta", map[string]any{ + "type": "response.output_text.delta", "delta": content, "item_id": "msg_1", "output_index": 0, "content_index": 0, + })) + } + // 上游 usage 块（choices 为空）通常晚于 finish_reason：此时再发 completed，携带 usage + if _, hasUsage := m["usage"]; hasUsage && t.finishSeen && !t.done { + t.done = true + out = append(out, eventLine("response.completed", map[string]any{ + "type": "response.completed", + "response": map[string]any{ + "id": "resp_stream", "object": "response", "model": t.model, "status": "completed", "usage": t.usage, + }, + })) + } + return joinLines(out) +} diff --git a/backend/internal/proxy/convert/usage.go b/backend/internal/proxy/convert/usage.go new file mode 100644 index 0000000..3188eaf --- /dev/null +++ b/backend/internal/proxy/convert/usage.go @@ -0,0 +1,131 @@ +package convert + +import ( + "encoding/json" +) + +// TokenUsage 从上游响应提取的 token 用量。 +// 三种协议的字段名不同，此处统一为：input / output / cache_read / cache_creation， +// 供用量记录与计费使用。 +type TokenUsage struct { + InputTokens int + OutputTokens int + CacheReadTokens int + CacheCreationTokens int +} + +// has 判断是否真的拿到了非零用量（过滤掉没有 usage 字段的响应）。 +func (u *TokenUsage) has() bool { + return u.InputTokens > 0 || u.OutputTokens > 0 || + u.CacheReadTokens > 0 || u.CacheCreationTokens > 0 +} + +// mergeJSON 把一张 usage 对象并入累计值。proto 决定字段名（chat/responses 与 messages 不同）。 +func (u *TokenUsage) mergeJSON(raw map[string]any, proto string) { + switch proto { + case ProtoChat, ProtoResponses: + in, _ := raw["prompt_tokens"].(float64) + out, _ := raw["completion_tokens"].(float64) + if in == 0 && out == 0 { + in, _ = raw["input_tokens"].(float64) + out, _ = raw["output_tokens"].(float64) + } + u.InputTokens += int(in) + u.OutputTokens += int(out) + if d, ok := raw["prompt_tokens_details"].(map[string]any); ok { + if c, _ := d["cached_tokens"].(float64); c > 0 { + u.CacheReadTokens += int(c) + } + } + if d, ok := raw["input_tokens_details"].(map[string]any); ok { + if c, _ := d["cached_tokens"].(float64); c > 0 { + u.CacheReadTokens += int(c) + } + } + case ProtoMessages: + in, _ := raw["input_tokens"].(float64) + out, _ := raw["output_tokens"].(float64) + u.InputTokens += int(in) + u.OutputTokens += int(out) + if c, _ := raw["cache_read_input_tokens"].(float64); c > 0 { + u.CacheReadTokens += int(c) + } + if c, _ := raw["cache_creation_input_tokens"].(float64); c > 0 { + u.CacheCreationTokens += int(c) + } + } +} + +// ExtractUsageJSON 从完整非流式响应体中提取用量。proto 为上游协议。 +// 返回 (用量, 是否有效)。 +func ExtractUsageJSON(body []byte, proto string) (TokenUsage, bool) { + var top map[string]any + if err := json.Unmarshal(body, &top); err != nil { + return TokenUsage{}, false + } + var u TokenUsage + if usage, ok := top["usage"].(map[string]any); ok { + u.mergeJSON(usage, proto) + } + return u, u.has() +} + +// StreamUsageAccum 流式用量累计器。逐行喂入上游 SSE 的 data 载荷， +// 按协议分别取各事件里的 usage 字段（各事件只会携带一部分字段，取最大值合并）。 +type StreamUsageAccum struct { + u TokenUsage +} + +// NewStreamUsageAccum 创建一个流式用量累计器。 +func NewStreamUsageAccum() *StreamUsageAccum { + return &StreamUsageAccum{} +} + +// Feed 喂入一行 SSE data 载荷（不含 "data:" 前缀与换行）。 +func (a *StreamUsageAccum) Feed(payload []byte, proto string) { + var top map[string]any + if json.Unmarshal(payload, &top) != nil { + return + } + var t TokenUsage + switch proto { + case ProtoChat: + if usage, ok := top["usage"].(map[string]any); ok { + t.mergeJSON(usage, proto) + } + case ProtoResponses: + // response.completed 事件把用量放在 response.usage 下。 + if resp, ok := top["response"].(map[string]any); ok { + if usage, ok := resp["usage"].(map[string]any); ok { + t.mergeJSON(usage, proto) + } + } + case ProtoMessages: + // message_start: {message: {usage: {input_tokens, cache_*}}} + // message_delta: {usage: {output_tokens}} + if msg, ok := top["message"].(map[string]any); ok { + if usage, ok := msg["usage"].(map[string]any); ok { + t.mergeJSON(usage, proto) + } + } + if usage, ok := top["usage"].(map[string]any); ok { + var t2 TokenUsage + t2.mergeJSON(usage, proto) + t.InputTokens = max(t.InputTokens, t2.InputTokens) + t.OutputTokens = max(t.OutputTokens, t2.OutputTokens) + t.CacheReadTokens = max(t.CacheReadTokens, t2.CacheReadTokens) + t.CacheCreationTokens = max(t.CacheCreationTokens, t2.CacheCreationTokens) + } + default: + return + } + a.u.InputTokens = max(a.u.InputTokens, t.InputTokens) + a.u.OutputTokens = max(a.u.OutputTokens, t.OutputTokens) + a.u.CacheReadTokens = max(a.u.CacheReadTokens, t.CacheReadTokens) + a.u.CacheCreationTokens = max(a.u.CacheCreationTokens, t.CacheCreationTokens) +} + +// Usage 返回当前累计用量。 +func (a *StreamUsageAccum) Usage() TokenUsage { + return a.u +} diff --git a/backend/internal/proxy/convert/usage_test.go b/backend/internal/proxy/convert/usage_test.go new file mode 100644 index 0000000..90922b4 --- /dev/null +++ b/backend/internal/proxy/convert/usage_test.go @@ -0,0 +1,170 @@ +package convert + +import ( + "encoding/json" + "testing" +) + +// ---- ExtractUsageJSON: 非流式各协议 ---- + +func TestExtractUsageJSONChat(t *testing.T) { + body := []byte(`{ + "id": "chatcmpl-1", + "object": "chat.completion", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "hi"}}], + "usage": { + "prompt_tokens": 11, + "completion_tokens": 7, + "total_tokens": 18, + "prompt_tokens_details": {"cached_tokens": 4} + } + }`) + u, ok := ExtractUsageJSON(body, ProtoChat) + if !ok { + t.Fatalf("expected ok=true") + } + if u.InputTokens != 11 || u.OutputTokens != 7 { + t.Fatalf("chat usage = %+v, want input=11 output=7", u) + } + if u.CacheReadTokens != 4 { + t.Fatalf("chat cacheRead = %d, want 4", u.CacheReadTokens) + } +} + +func TestExtractUsageJSONMessages(t *testing.T) { + body := []byte(`{ + "id": "msg_1", + "type": "message", + "role": "assistant", + "content": [{"type": "text", "text": "hi"}], + "usage": { + "input_tokens": 15, + "output_tokens": 8, + "cache_read_input_tokens": 3, + "cache_creation_input_tokens": 2 + } + }`) + u, ok := ExtractUsageJSON(body, ProtoMessages) + if !ok { + t.Fatalf("expected ok=true") + } + if u.InputTokens != 15 || u.OutputTokens != 8 || u.CacheReadTokens != 3 || u.CacheCreationTokens != 2 { + t.Fatalf("messages usage = %+v", u) + } +} + +func TestExtractUsageJSONResponses(t *testing.T) { + body := []byte(`{ + "id": "resp_1", + "object": "response", + "output": [], + "usage": { + "input_tokens": 13, + "output_tokens": 9, + "input_tokens_details": {"cached_tokens": 5} + } + }`) + u, ok := ExtractUsageJSON(body, ProtoResponses) + if !ok { + t.Fatalf("expected ok=true") + } + if u.InputTokens != 13 || u.OutputTokens != 9 || u.CacheReadTokens != 5 { + t.Fatalf("responses usage = %+v", u) + } +} + +func TestExtractUsageJSONInvalidAndMissing(t *testing.T) { + if _, ok := ExtractUsageJSON([]byte("not json"), ProtoChat); ok { + t.Fatalf("invalid json should not report ok") + } + if _, ok := ExtractUsageJSON([]byte(`{"id": "x"}`), ProtoChat); ok { + t.Fatalf("missing usage should not report ok") + } + // 空对象 usage：全 0 视为无效 + if _, ok := ExtractUsageJSON([]byte(`{"usage": {}}`), ProtoChat); ok { + t.Fatalf("empty usage should not report ok") + } +} + +// ---- StreamUsageAccum: 流式各协议 ---- + +func feedLines(t *testing.T, proto string, lines ...string) TokenUsage { + t.Helper() + acc := NewStreamUsageAccum() + for _, ln := range lines { + acc.Feed([]byte(ln), proto) + } + return acc.Usage() +} + +func TestStreamUsageChatFinalChunk(t *testing.T) { + // 前面的 chunk 不带 usage；最后一个 chunk 带完整 usage + u := feedLines(t, ProtoChat, + `{"id":"c1","object":"chat.completion.chunk","choices":[{"delta":{"content":"he"}}]}`, + `{"id":"c1","object":"chat.completion.chunk","choices":[{"delta":{"content":"llo"}}]}`, + `{"id":"c1","object":"chat.completion.chunk","choices":[],"usage":{"prompt_tokens":11,"completion_tokens":7,"prompt_tokens_details":{"cached_tokens":4}}}`, + ) + if u.InputTokens != 11 || u.OutputTokens != 7 || u.CacheReadTokens != 4 { + t.Fatalf("chat stream usage = %+v", u) + } +} + +func TestStreamUsageMessagesStartAndDelta(t *testing.T) { + // message_start 带 input/cache，message_delta 带 output；逐字段取 max 合并 + u := feedLines(t, ProtoMessages, + `{"type":"message_start","message":{"id":"msg_1","usage":{"input_tokens":15,"cache_read_input_tokens":3,"cache_creation_input_tokens":2}}}`, + `{"type":"content_block_delta","delta":{"type":"text_delta","text":"hi"}}`, + `{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":8}}`, + ) + if u.InputTokens != 15 || u.OutputTokens != 8 || u.CacheReadTokens != 3 || u.CacheCreationTokens != 2 { + t.Fatalf("messages stream usage = %+v", u) + } +} + +func TestStreamUsageResponsesCompleted(t *testing.T) { + // response.completed 事件的用量嵌在 response.usage 下 + u := feedLines(t, ProtoResponses, + `{"type":"response.output_text.delta","delta":"hi"}`, + `{"type":"response.completed","response":{"id":"resp_1","usage":{"input_tokens":13,"output_tokens":9,"input_tokens_details":{"cached_tokens":5}}}}`, + ) + if u.InputTokens != 13 || u.OutputTokens != 9 || u.CacheReadTokens != 5 { + t.Fatalf("responses stream usage = %+v", u) + } +} + +func TestStreamUsageIgnoresNonDataPayloads(t *testing.T) { + // [DONE]、垃圾行、空对象都不应产生用量 + u := feedLines(t, ProtoChat, `[DONE]`, `{`, ``, `{"choices":[]}`) + if u.has() { + t.Fatalf("expected zero usage, got %+v", u) + } +} + +func TestStreamUsageFeedKeepsMaxAcrossEvents(t *testing.T) { + // 同一字段在多个事件出现时取较大值（防乱序/重复） + u := feedLines(t, ProtoMessages, + `{"type":"message_start","message":{"usage":{"input_tokens":15}}}`, + `{"type":"message_delta","usage":{"output_tokens":5}}`, + `{"type":"message_delta","usage":{"output_tokens":8}}`, + ) + if u.InputTokens != 15 || u.OutputTokens != 8 { + t.Fatalf("max-merge usage = %+v", u) + } +} + +// ---- usage JSON 结构合法性（防止手写 struct 漂移）---- + +func TestUsageJSONRoundTrip(t *testing.T) { + u := TokenUsage{InputTokens: 10, OutputTokens: 5, CacheReadTokens: 2, CacheCreationTokens: 1} + b, err := json.Marshal(u) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var back TokenUsage + if err := json.Unmarshal(b, &back); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if back != u { + t.Fatalf("round trip = %+v, want %+v", back, u) + } +} diff --git a/backend/internal/proxy/gateway.go b/backend/internal/proxy/gateway.go index 1689d48..0b74c72 100644 --- a/backend/internal/proxy/gateway.go +++ b/backend/internal/proxy/gateway.go @@ -1,6 +1,7 @@ package proxy import ( + "bufio" "bytes" "context" "encoding/json" @@ -13,6 +14,7 @@ import ( "opencatd-open/internal/dao" "opencatd-open/internal/proxy/convert" "opencatd-open/internal/store" + "opencatd-open/internal/usage" "opencatd-open/pkg/config" "os" "strings" @@ -35,6 +37,7 @@ type Gateway struct { usageDAO *dao.UsageDAO dailyDAO *dao.DailyUsageDAO channelSvc *channel.Service + usageRec *usage.Recorder } func NewGateway(ctx context.Context, cfg *config.Config, db *gorm.DB, wg *sync.WaitGroup, userDAO *dao.UserDAO, apiKeyDAO *dao.ApiKeyDAO, usageDAO *dao.UsageDAO, dailyDAO *dao.DailyUsageDAO) *Gateway { @@ -67,6 +70,11 @@ func (g *Gateway) SetChannelService(svc *channel.Service) { g.channelSvc = svc } +// SetUsageRecorder 注入异步用量记录器；nil 时网关跳过用量上报。 +func (g *Gateway) SetUsageRecorder(r *usage.Recorder) { + g.usageRec = r +} + // Request represents a parsed incoming request type Request struct { Model string @@ -144,26 +152,24 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { return } - // Determine target format and convert if needed - targetFormat := req.Protocol - if len(ch.FormatsEffective()) > 0 { - // Prefer the channel's native format - for _, f := range ch.FormatsEffective() { - if f == req.Protocol { - targetFormat = f - break - } - } + // 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 } // Build upstream URL - upstreamPath := g.getUpstreamPath(req.Protocol) - upstreamURL := ch.UpstreamURL(req.Protocol, upstreamPath) + upstreamPath := g.getUpstreamPath(targetFormat) + upstreamURL := ch.UpstreamURL(targetFormat, upstreamPath) // Convert request if needed var requestBody []byte if targetFormat != req.Protocol { - requestBody, err = g.convertRequest(req.Body, req.Protocol, targetFormat) + var err error + requestBody, err = convert.ConvertRequest(req.Body, req.Protocol, targetFormat) if err != nil { g.writeError(c, http.StatusBadRequest, "conversion failed: "+err.Error()) return @@ -206,12 +212,31 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { // Stream or buffer response if req.Stream { - g.streamResponse(c, resp, req.Protocol, ch) + g.streamResponse(c, resp, req.Protocol, targetFormat) } else { - g.bufferResponse(c, resp, req.Protocol, ch) + g.bufferResponse(c, resp, req.Protocol, targetFormat) } } +// conversionTarget 决定客户端协议在渠道上的处理方式： +// 渠道声明支持该协议则直通；否则转为其首选支持协议（chat > messages > responses）。 +func (g *Gateway) conversionTarget(ch *store.Channel, clientProto string) string { + formats := ch.FormatsEffective() + for _, f := range formats { + if f == clientProto { + return clientProto + } + } + for _, p := range []string{convert.ProtoChat, convert.ProtoMessages, convert.ProtoResponses} { + for _, f := range formats { + if f == p { + return p + } + } + } + return "" +} + func (g *Gateway) getUpstreamPath(protocol string) string { switch protocol { case "chat": @@ -237,116 +262,86 @@ func (g *Gateway) setHeaders(req *http.Request, ch *store.Channel, apiKey string } } -func (g *Gateway) convertRequest(body []byte, from, to string) ([]byte, error) { - switch { - case from == "chat" && to == "messages": - var req convert.ChatCompletionRequest - if err := json.Unmarshal(body, &req); err != nil { - return nil, err - } - msgReq, err := convert.ChatToMessages(&req) - if err != nil { - return nil, err - } - return json.Marshal(msgReq) - case from == "chat" && to == "responses": - var req convert.ChatCompletionRequest - if err := json.Unmarshal(body, &req); err != nil { - return nil, err - } - respReq, err := convert.ChatToResponses(&req) - if err != nil { - return nil, err - } - return json.Marshal(respReq) - - case from == "messages" && to == "chat": - var req convert.MessagesRequest - if err := json.Unmarshal(body, &req); err != nil { - return nil, err - } - // Messages -> Chat: we need to construct a ChatCompletionRequest - chatReq := &convert.ChatCompletionRequest{ - Model: req.Model, - } - for _, m := range req.Messages { - chatReq.Messages = append(chatReq.Messages, m) - } - if req.Temperature != nil { - chatReq.Temperature = req.Temperature - } - if req.TopP != nil { - chatReq.TopP = req.TopP - } - chatReq.Tools = req.Tools - chatReq.Stream = req.Stream - return json.Marshal(chatReq) - - case from == "responses" && to == "chat": - var req convert.ResponsesRequest - if err := json.Unmarshal(body, &req); err != nil { - return nil, err - } - chatReq := &convert.ChatCompletionRequest{ - Model: req.Model, - } - for _, item := range req.Input { - chatReq.Messages = append(chatReq.Messages, convert.Message{ - Role: item.Role, - Content: item.Content, - }) - } - chatReq.Tools = req.Tools - chatReq.Stream = req.Stream - return json.Marshal(chatReq) - - default: - return body, nil - } -} - -func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, protocol string, ch *store.Channel) { +// streamResponse 流式响应：按 \n\n 分块零缓冲转发；跨协议时逐行转换。 +func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string) { + w := c.Writer c.Header("Content-Type", "text/event-stream") c.Header("Cache-Control", "no-cache") c.Header("Connection", "keep-alive") c.Status(http.StatusOK) - writer := convert.NewSSEWriter(c.Writer) - parser := convert.NewSSEParser(resp.Body) + flusher, _ := w.(http.Flusher) - for { - event, err := parser.ReadEvent() - if err != nil { - if err == io.EOF { - break - } - log.Printf("Stream parse error: %v", err) - break - } - - if event.Event == "error" { - log.Printf("Upstream stream error: %s", event.Data) - break - } - - // Write raw SSE event based on protocol - if err := writer.WriteEvent("chat CompletionChunk", event.Data); err != nil { - break - } + // 跨协议时按行转换；同协议直通（lineConv 为 nil）。 + var lineConv func([]byte) []byte + if upstreamProto != clientProto { + lineConv = convert.NewStreamTransformer(upstreamProto, clientProto) } - writer.WriteDone() + // 上游原始行按 \n\n 分块，避免把 data 行内的转义换行当成事件边界。 + r := bufio.NewReaderSize(resp.Body, 32*1024) + for { + buf := []byte{} + for { + line, err := r.ReadSlice('\n') + if err == bufio.ErrBufferFull { + buf = append(buf, line...) + continue + } + buf = append(buf, line...) + if err == io.EOF { + if len(buf) == 0 { + return + } + if !bytes.HasSuffix(buf, []byte("\n")) { + buf = append(buf, '\n') + } + } else if err != nil { + log.Printf("stream read error: %v", err) + return + } + if len(buf) >= 2 && bytes.HasSuffix(buf, []byte("\n\n")) { + break + } + } + + out := buf + if lineConv != nil { + out = lineConv(buf) + } + if len(out) == 0 { + continue + } + if _, err := w.Write(out); err != nil { + return // 客户端已断开 + } + if flusher != nil { + flusher.Flush() + } + } } -func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, protocol string, ch *store.Channel) { +func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string) { body, err := io.ReadAll(resp.Body) if err != nil { g.writeError(c, http.StatusBadGateway, "failed to read response") return } - c.Data(resp.StatusCode, "application/json", body) + out := body + if upstreamProto != clientProto { + if converted, cerr := convert.ConvertResponse(body, upstreamProto, clientProto); cerr == nil { + out = converted + } else { + // 转换失败时至少剥掉非 JSON 前缀，让客户端能解析出正文 + out = convert.CleanJSON(body) + } + } else { + // 直通：部分上游（如 OpenRouter）的 non-stream 响应在 JSON 前夹带空白/注释 + out = convert.CleanJSON(body) + } + c.Data(resp.StatusCode, "application/json", out) } func (g *Gateway) writeError(c *gin.Context, status int, message string) { diff --git a/backend/internal/usage/recorder.go b/backend/internal/usage/recorder.go index 5b45e6a..074512e 100644 --- a/backend/internal/usage/recorder.go +++ b/backend/internal/usage/recorder.go @@ -2,6 +2,7 @@ package usage import ( "context" + "fmt" "log" "opencatd-open/internal/dao" "opencatd-open/internal/store" @@ -11,16 +12,26 @@ import ( // Event represents a usage event to be recorded type Event struct { - UserID uint64 - ModelName string - ChannelID uint64 - PromptTokens int - CompletionTokens int - CacheReadTokens int - Cost float64 - IsError bool - IsCanceled bool - RequestID string + UserID uint64 + ModelName string + ChannelID uint64 + PromptTokens int + CompletionTokens int + CacheReadTokens int + CacheCreationTokens int + Cost float64 + IsError bool + IsCanceled bool + RequestID string + KeyID uint64 + Protocol string + ErrorCode string + LatencyMS int + InputPrice float64 + OutputPrice float64 + CacheReadPrice float64 + TraceID string // TraceID for distributed tracing + ModelID uint64 // Model ID from channel-model binding } // Recorder handles async usage recording @@ -117,16 +128,31 @@ func (r *Recorder) flush(events []Event) { status = store.UsageStatusCanceled } + var errCode *string + if e.ErrorCode != "" { + errCode = &e.ErrorCode + } + log := &store.UsageLog{ - UserID: e.UserID, - ModelName: e.ModelName, - ChannelID: e.ChannelID, - InputTokens: int64(e.PromptTokens), - OutputTokens: int64(e.CompletionTokens), - CacheReadTokens: int64(e.CacheReadTokens), - Cost: e.Cost, - Status: status, - RequestID: e.RequestID, + UserID: e.UserID, + KeyID: e.KeyID, + ChannelID: e.ChannelID, + ModelID: e.ModelID, + ModelName: e.ModelName, + Protocol: e.Protocol, + InputTokens: int64(e.PromptTokens), + OutputTokens: int64(e.CompletionTokens), + CacheReadTokens: int64(e.CacheReadTokens), + CacheCreationTokens: int64(e.CacheCreationTokens), + InputPrice: e.InputPrice, + OutputPrice: e.OutputPrice, + CacheReadPrice: e.CacheReadPrice, + Cost: e.Cost, + LatencyMS: e.LatencyMS, + Status: status, + ErrorCode: errCode, + RequestID: e.RequestID, + TraceID: e.TraceID, } logs = append(logs, log) } @@ -136,5 +162,39 @@ func (r *Recorder) flush(events []Event) { log.Printf("Failed to batch create usage logs: %v", err) } + // Daily rollup for success and canceled requests + dailyMap := make(map[string]*store.UsageDaily) + for _, e := range events { + if e.IsError { + continue + } + date := time.Now().Format("2006-01-02") + key := fmt.Sprintf("%d:%d:%s", e.UserID, e.ModelID, date) + d := dailyMap[key] + if d == nil { + d = &store.UsageDaily{ + UserID: e.UserID, + ModelID: e.ModelID, + Date: date, + Requests: 0, + InputTokens: 0, + OutputTokens: 0, + CacheReadTokens: 0, + Cost: 0, + } + dailyMap[key] = d + } + d.Requests++ + d.InputTokens += int64(e.PromptTokens) + d.OutputTokens += int64(e.CompletionTokens) + d.CacheReadTokens += int64(e.CacheReadTokens) + d.Cost += e.Cost + } + for _, d := range dailyMap { + if err := r.dailyDAO.UpsertDailyUsage(context.Background(), d); err != nil { + log.Printf("Failed to upsert daily usage: %v", err) + } + } + log.Printf("Flushed %d usage logs", len(logs)) } diff --git a/backend/middleware/auth_llm.go b/backend/middleware/auth_llm.go index b21e618..f33cc51 100644 --- a/backend/middleware/auth_llm.go +++ b/backend/middleware/auth_llm.go @@ -9,45 +9,34 @@ import ( "gorm.io/gorm" ) +// keyPrefixLen 是 key_prefix 列的截断长度，必须与 api.go:459 的 keyValue[:12] 一致。 +// 真实 key 为 sk-ot- + 48 位 hex（54 字符），故 12 位足够唯一。 +const keyPrefixLen = 12 + func AuthLLM(db *gorm.DB) gin.HandlerFunc { return func(c *gin.Context) { - authToken := c.GetHeader("Authorization") - if authToken == "" { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{ - "error": map[string]interface{}{ - "message": "未提供认证信息", - "type": "invalid_request_error", - }, - }) + key := extractAPIKey(c.GetHeader("Authorization")) + + // 区分「没传」和「传了但不对」，便于排查客户端配置。 + if strings.TrimSpace(c.GetHeader("Authorization")) == "" { + unauthorized(c, "未提供认证信息") + return + } + // 长度不足时直接拒绝：避免下方 authToken[:12] 越界 panic 打崩进程。 + if len(key) < keyPrefixLen { + unauthorized(c, "无效的API密钥") return } - // Extract API key from Bearer token - if len(authToken) > 7 { - authToken = authToken[7:] - } - - // Find API key by prefix var apiKey store.APIKey - if err := db.Where("key_prefix = ? AND status = ?", authToken[:8], store.KeyStatusActive).First(&apiKey).Error; err != nil { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{ - "error": map[string]interface{}{ - "message": "无效的API密钥", - "type": "invalid_request_error", - }, - }) + if err := db.Where("key_prefix = ? AND status = ?", key[:keyPrefixLen], store.KeyStatusActive).First(&apiKey).Error; err != nil { + unauthorized(c, "无效的API密钥") return } // Verify full key hash - keyHash := store.HashAPIKey(authToken) - if apiKey.KeyHash != keyHash { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{ - "error": map[string]interface{}{ - "message": "无效的API密钥", - "type": "invalid_request_error", - }, - }) + if apiKey.KeyHash != store.HashAPIKey(key) { + unauthorized(c, "无效的API密钥") return } @@ -57,11 +46,20 @@ func AuthLLM(db *gorm.DB) gin.HandlerFunc { } } -// extractAPIKey extracts the API key from the Authorization header -func extractAPIKey(c *gin.Context) string { - auth := c.GetHeader("Authorization") +// extractAPIKey 从 Authorization 头取 Bearer token，兼容无 "Bearer " 前缀的直传。 +func extractAPIKey(auth string) string { + auth = strings.TrimSpace(auth) if strings.HasPrefix(auth, "Bearer ") { - return auth[7:] + return strings.TrimSpace(auth[len("Bearer "):]) } return auth } + +func unauthorized(c *gin.Context, message string) { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{ + "error": map[string]interface{}{ + "message": message, + "type": "invalid_request_error", + }, + }) +} diff --git a/backend/router/setRouter.go b/backend/router/setRouter.go index c43eb3e..8b71fe1 100644 --- a/backend/router/setRouter.go +++ b/backend/router/setRouter.go @@ -50,7 +50,7 @@ func SetRouter(cfg *config.Config, db *gorm.DB, web *embed.FS) { // Initialize health checker and start periodic checks healthChecker := channel.NewHealthChecker(channelDAO, channelSvc) - go healthChecker.StartPeriodicCheck(ctx, 5*time.Minute) + go healthChecker.StartPeriodicCheck(ctx) // Initialize usage recorder and start background worker usageRecorder := usage.NewRecorder(usageDAO, dailyDAO) @@ -60,7 +60,7 @@ func SetRouter(cfg *config.Config, db *gorm.DB, web *embed.FS) { // Initialize gateway gateway := proxy.NewGateway(ctx, cfg, db, &wg, userDAO, apiKeyDAO, usageDAO, dailyDAO) gateway.SetChannelService(channelSvc) - + gateway.SetUsageRecorder(usageRecorder) // Initialize API handler apiHandler := api.NewHandler(db) @@ -99,7 +99,7 @@ func SetRouter(cfg *config.Config, db *gorm.DB, web *embed.FS) { apiGroup.DELETE("/keys/:id", apiHandler.DeleteApiKey) apiGroup.POST("/keys/batch/:option", apiHandler.BatchApiKeys) - // Channel management + // Channel management (legacy endpoints) apiGroup.GET("/channels", apiHandler.ListChannels) apiGroup.POST("/channels", apiHandler.CreateChannel) apiGroup.PUT("/channels/:id", apiHandler.UpdateChannel) @@ -107,11 +107,40 @@ func SetRouter(cfg *config.Config, db *gorm.DB, web *embed.FS) { apiGroup.GET("/channels/:id/models", apiHandler.GetChannelModels) apiGroup.POST("/channels/:id/models", apiHandler.BindChannelModels) - // Model management + // Model management (legacy endpoints) apiGroup.GET("/models", apiHandler.ListModels) apiGroup.POST("/models", apiHandler.CreateModel) apiGroup.PUT("/models/:id", apiHandler.UpdateModel) apiGroup.DELETE("/models/:id", apiHandler.DeleteModel) + + // Admin channel management (enhanced) + apiGroup.GET("/admin/channels", apiHandler.AdminChannels) + apiGroup.POST("/admin/channels", apiHandler.AdminCreateChannel) + apiGroup.PUT("/admin/channels/:id", apiHandler.AdminUpdateChannel) + apiGroup.DELETE("/admin/channels/:id", apiHandler.AdminDeleteChannel) + apiGroup.POST("/admin/channels/:id/test", apiHandler.AdminTestChannel) + apiGroup.GET("/admin/channels/:id/models/remote", apiHandler.AdminChannelRemoteModels) + apiGroup.GET("/admin/channels/:id/models", apiHandler.AdminChannelModels) + apiGroup.POST("/admin/channels/:id/models", apiHandler.AdminChannelAddModel) + apiGroup.PATCH("/admin/channels/:id/models/:bid", apiHandler.AdminChannelUpdateModel) + apiGroup.DELETE("/admin/channels/:id/models/:bid", apiHandler.AdminChannelDeleteModel) + + // Admin model management (enhanced) + apiGroup.GET("/admin/models", apiHandler.AdminModels) + apiGroup.DELETE("/admin/models/unused", apiHandler.AdminDeleteUnusedModels) + apiGroup.POST("/admin/models", apiHandler.AdminCreateModel) + apiGroup.PUT("/admin/models/:id", apiHandler.AdminUpdateModel) + apiGroup.DELETE("/admin/models/:id", apiHandler.AdminDeleteModel) + apiGroup.POST("/admin/models/:id/bindings", apiHandler.AdminCreateModelBinding) + apiGroup.DELETE("/admin/models/:id/bindings/:bid", apiHandler.AdminDeleteModelBinding) + + // Admin system config + apiGroup.GET("/admin/config", apiHandler.AdminGetConfig) + apiGroup.PUT("/admin/config", apiHandler.AdminUpdateConfig) + apiGroup.GET("/admin/config/registration", apiHandler.AdminGetRegistration) + apiGroup.PUT("/admin/config/registration", apiHandler.AdminUpdateRegistration) + apiGroup.GET("/admin/config/password-login", apiHandler.AdminGetPasswordLogin) + apiGroup.PUT("/admin/config/password-login", apiHandler.AdminUpdatePasswordLogin) } // LLM proxy routes diff --git a/frontend/src/components/common/Toast.vue b/frontend/src/components/common/Toast.vue index acb1dde..cfcf79d 100644 --- a/frontend/src/components/common/Toast.vue +++ b/frontend/src/components/common/Toast.vue @@ -47,9 +47,9 @@ const iconForType = (type: ToastType) => { const typeClasses = (type: ToastType) => { switch (type) { case 'success': - return 'border-success/20 bg-success/10 text-success-content dark:border-success/30 dark:bg-success/15'; + return 'border-success bg-success/15 text-success'; case 'error': - return 'border-error/20 bg-error/10 text-error-content dark:border-error/30 dark:bg-error/15'; + return 'border-error bg-error/15 text-error'; default: return 'border-base-300 bg-base-100 text-base-content'; } diff --git a/frontend/src/components/ui/Badge.vue b/frontend/src/components/ui/Badge.vue new file mode 100644 index 0000000..209ae69 --- /dev/null +++ b/frontend/src/components/ui/Badge.vue @@ -0,0 +1,30 @@ + + + diff --git a/frontend/src/components/ui/Button.vue b/frontend/src/components/ui/Button.vue new file mode 100644 index 0000000..ed5d7b1 --- /dev/null +++ b/frontend/src/components/ui/Button.vue @@ -0,0 +1,27 @@ + + + diff --git a/frontend/src/components/ui/Input.vue b/frontend/src/components/ui/Input.vue new file mode 100644 index 0000000..b519f3f --- /dev/null +++ b/frontend/src/components/ui/Input.vue @@ -0,0 +1,36 @@ + + + diff --git a/frontend/src/components/ui/Modal.vue b/frontend/src/components/ui/Modal.vue new file mode 100644 index 0000000..4306095 --- /dev/null +++ b/frontend/src/components/ui/Modal.vue @@ -0,0 +1,84 @@ + + + diff --git a/frontend/src/lib/protocol.ts b/frontend/src/lib/protocol.ts new file mode 100644 index 0000000..40cb03c --- /dev/null +++ b/frontend/src/lib/protocol.ts @@ -0,0 +1,27 @@ +// 协议格式显示名与选项 +export const PROTOCOL_NAMES: Record = { + chat: 'OpenAI Chat Completions', + responses: 'OpenAI Responses API', + messages: 'Anthropic Messages', +} + +// 渠道表格用的短标识 +export const PROTOCOL_SHORT: Record = { + chat: 'chat/completions', + responses: 'responses', + messages: 'messages', +} + +export function protocolShort(p: string): string { + return PROTOCOL_SHORT[p] ?? p +} + +export const PROTOCOL_OPTIONS: { value: string; label: string }[] = [ + { value: 'chat', label: 'OpenAI Chat Completions' }, + { value: 'responses', label: 'OpenAI Responses API' }, + { value: 'messages', label: 'Anthropic Messages' }, +] + +export function protocolName(p: string): string { + return PROTOCOL_NAMES[p] ?? p +} diff --git a/frontend/src/stores/channel.ts b/frontend/src/stores/channel.ts index febebfc..9e2f5d9 100644 --- a/frontend/src/stores/channel.ts +++ b/frontend/src/stores/channel.ts @@ -8,6 +8,8 @@ export type Channel = { name: string provider: string base_url: string + base_urls?: Record + api_key_masked?: string weight: number priority: number timeout_ms: number @@ -31,6 +33,14 @@ export type NewChannelPayload = { formats?: string[] } +export type ChannelModelBinding = { + id: number + model_id: number + model_name: string + upstream_model: string + weight: number +} + export const useChannelStore = defineStore('channel', () => { const loading = ref(false); const error = ref(null); @@ -125,6 +135,91 @@ export const useChannelStore = defineStore('channel', () => { } }; + // Admin API methods + const testChannel = async (id: number | string) => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.post(`/admin/channels/${id}/test`); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to test channel'; + throw err; + } finally { + loading.value = false; + } + }; + + const fetchRemoteModels = async (id: number | string) => { + loading.value = true; + error.value = null; + try { + const response = await request.get(`/admin/channels/${id}/models/remote`); + return response.data.data ?? []; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to fetch remote models'; + throw err; + } finally { + loading.value = false; + } + }; + + const fetchChannelModels = async (id: number | string) => { + loading.value = true; + error.value = null; + try { + const response = await request.get(`/admin/channels/${id}/models`); + return response.data.data ?? []; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to fetch channel models'; + throw err; + } finally { + loading.value = false; + } + }; + + const addChannelModel = async (id: number | string, data: { model_id: number; upstream_model: string; weight?: number }) => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.post(`/admin/channels/${id}/models`, data); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to add model'; + throw err; + } finally { + loading.value = false; + } + }; + + const updateChannelModel = async (channelId: number | string, bindingId: number | string, data: { upstream_model?: string; weight?: number }) => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.patch(`/admin/channels/${channelId}/models/${bindingId}`, data); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to update binding'; + throw err; + } finally { + loading.value = false; + } + }; + + const deleteChannelModel = async (channelId: number | string, bindingId: number | string) => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.delete(`/admin/channels/${channelId}/models/${bindingId}`); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to delete binding'; + throw err; + } finally { + loading.value = false; + } + }; + return { loading, error, channel, channels, totalChannels, @@ -134,5 +229,11 @@ export const useChannelStore = defineStore('channel', () => { updateChannel, deleteChannel, batchChannels, + testChannel, + fetchRemoteModels, + fetchChannelModels, + addChannelModel, + updateChannelModel, + deleteChannelModel, }; }); diff --git a/frontend/src/stores/model.ts b/frontend/src/stores/model.ts new file mode 100644 index 0000000..c09adde --- /dev/null +++ b/frontend/src/stores/model.ts @@ -0,0 +1,171 @@ +import { defineStore } from 'pinia'; +import { ref } from 'vue'; +import type { AxiosResponse } from 'axios'; +import request from '@/api/client'; + +export type Model = { + id: number + name: string + display_name?: string + input_price: number + output_price: number + cache_read_price: number + enabled: boolean + sort: number + channels?: ModelBinding[] + used?: boolean + needs_pricing?: boolean + denied?: boolean + created_at?: string + updated_at?: string + [key: string]: unknown +} + +export type ModelBinding = { + id: number + channel_id: number + channel_name: string + upstream_model: string + weight: number +} + +export type NewModelPayload = { + name: string + display_name?: string + input_price?: number + output_price?: number + cache_read_price?: number + sort?: number + enabled?: boolean +} + +export type ModelSummary = { + total: number + unpriced: number + missing: OrphanBinding[] + denied_count: number +} + +export type OrphanBinding = { + channel: string + model_id: number + upstream_model: string +} + +export const useModelStore = defineStore('model', () => { + const loading = ref(false); + const error = ref(null); + const models = ref([]); + const summary = ref(null); + + const fetchModels = async () => { + loading.value = true; + error.value = null; + try { + const response = await request.get('/admin/models'); + models.value = response.data.data ?? []; + summary.value = response.data.summary ?? null; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to fetch models'; + throw err; + } finally { + loading.value = false; + } + }; + + const createModel = async (data: NewModelPayload) => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.post('/admin/models', data); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to create model'; + throw err; + } finally { + loading.value = false; + } + }; + + const updateModel = async (id: number | string, data: Partial) => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.put(`/admin/models/${id}`, data); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to update model'; + throw err; + } finally { + loading.value = false; + } + }; + + const deleteModel = async (id: number | string) => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.delete(`/admin/models/${id}`); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to delete model'; + throw err; + } finally { + loading.value = false; + } + }; + + const deleteUnusedModels = async () => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.delete('/admin/models/unused'); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to delete unused models'; + throw err; + } finally { + loading.value = false; + } + }; + + const createModelBinding = async (modelId: number | string, data: { channel_id: number; upstream_model: string; weight?: number }) => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.post(`/admin/models/${modelId}/bindings`, data); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to create binding'; + throw err; + } finally { + loading.value = false; + } + }; + + const deleteModelBinding = async (modelId: number | string, bindingId: number | string) => { + loading.value = true; + error.value = null; + try { + const response: AxiosResponse = await request.delete(`/admin/models/${modelId}/bindings/${bindingId}`); + return response; + } catch (err: any) { + error.value = err.response?.data?.error || 'Failed to delete binding'; + throw err; + } finally { + loading.value = false; + } + }; + + return { + loading, error, + models, summary, + fetchModels, + createModel, + updateModel, + deleteModel, + deleteUnusedModels, + createModelBinding, + deleteModelBinding, + }; +}); diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 3fbf1eb..b2476a2 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -115,3 +115,59 @@ export type NewUserPayload = { unlimited_quota?: boolean language?: string } + +// Channel 渠道管理 +export interface Channel { + id: number + name: string + provider: 'openai' | 'anthropic' | 'compatible' + formats: string[] // chat | responses | messages + base_url: string + base_urls?: Record | null + api_key_masked: string + weight: number + priority: number + timeout_ms: number + max_concurrency: number + health_status: string + enabled: boolean + created_at: string +} + +export interface ChannelModelMapping { + id: number + model_id: number + model_name: string + upstream_model: string + weight: number +} + +// Model 模型定价 +export interface ModelBinding { + id: number + channel_id: number + channel_name: string + upstream_model: string + weight: number +} + +export interface Model { + id: number + name: string + input_price: number + output_price: number + cache_read_price: number + enabled: boolean + sort: number + channels: ModelBinding[] + used?: boolean + needs_pricing?: boolean + denied?: boolean +} + +export interface ModelSummary { + total: number + unpriced: number + missing: { channel: string; model_id: number; upstream_model: string }[] + denied_count: number +} diff --git a/frontend/src/utils/router_menu.ts b/frontend/src/utils/router_menu.ts index eab1495..104c0da 100644 --- a/frontend/src/utils/router_menu.ts +++ b/frontend/src/utils/router_menu.ts @@ -6,6 +6,8 @@ import { KeyRoundIcon, SettingsIcon, GlobeIcon, + BoxesIcon, + SlidersHorizontalIcon, } from '@lucide/vue' export type MenuLink = { label: string; to: string; icon?: Component } @@ -48,8 +50,9 @@ export const routes: RouteRecordRaw[] = [ { path: 'users', name: 'User', component: () => import('@/views/dashboard/User.vue'), meta: { title: '用户管理' } }, { path: 'users/new', name: 'UserNew', component: () => import('@/views/dashboard/UserNew.vue'), meta: { title: '新建用户' } }, { path: 'users/view', name: 'UserView', component: () => import('@/views/dashboard/UserView.vue'), meta: { title: '用户详情' } }, - { path: 'channels', name: 'Channels', component: () => import('@/views/dashboard/Keys.vue'), meta: { title: '渠道管理' } }, - { path: 'channels/view', name: 'ChannelView', component: () => import('@/views/dashboard/KeyView.vue'), meta: { title: '渠道详情' } }, + { path: 'channels', name: 'Channels', component: () => import('@/views/dashboard/ChannelsView.vue'), meta: { title: '渠道管理' } }, + { path: 'models', name: 'Models', component: () => import('@/views/dashboard/Models.vue'), meta: { title: '模型定价' } }, + { path: 'config', name: 'SystemConfig', component: () => import('@/views/dashboard/SystemConfig.vue'), meta: { title: '系统配置' } }, ], }, { @@ -76,4 +79,6 @@ export const consoleMenu: MenuLink[] = [ export const adminMenu: MenuLink[] = [ { label: '用户管理', to: '/dashboard/manager/users', icon: UsersRoundIcon }, { label: '渠道管理', to: '/dashboard/manager/channels', icon: GlobeIcon }, + { label: '模型定价', to: '/dashboard/manager/models', icon: BoxesIcon }, + { label: '系统配置', to: '/dashboard/manager/config', icon: SlidersHorizontalIcon }, ] diff --git a/frontend/src/views/dashboard/ChannelModelsDrawer.vue b/frontend/src/views/dashboard/ChannelModelsDrawer.vue new file mode 100644 index 0000000..274dcf7 --- /dev/null +++ b/frontend/src/views/dashboard/ChannelModelsDrawer.vue @@ -0,0 +1,183 @@ + + + diff --git a/frontend/src/views/dashboard/ChannelsView.vue b/frontend/src/views/dashboard/ChannelsView.vue new file mode 100644 index 0000000..c2d8aef --- /dev/null +++ b/frontend/src/views/dashboard/ChannelsView.vue @@ -0,0 +1,342 @@ + + + diff --git a/frontend/src/views/dashboard/KeyView.vue b/frontend/src/views/dashboard/KeyView.vue index af0d71d..adce801 100644 --- a/frontend/src/views/dashboard/KeyView.vue +++ b/frontend/src/views/dashboard/KeyView.vue @@ -87,6 +87,10 @@
+
+ + +
+
+
+

Model Bindings

+ +
+ +
+ + + + + + + + + + + + + + + + + +
Model NameUpstream ModelWeightActions
{{ b.model_name }}{{ b.upstream_model }}{{ b.weight }} + +
+
+
+ No model bindings configured. +
+
+
@@ -104,22 +149,59 @@ + + + +
+
+ +
+

Add Model Binding

+
+ + + +
+ + +
+
+
+
+ +
+ diff --git a/frontend/src/views/dashboard/Models.vue b/frontend/src/views/dashboard/Models.vue new file mode 100644 index 0000000..b3ab5a0 --- /dev/null +++ b/frontend/src/views/dashboard/Models.vue @@ -0,0 +1,277 @@ + + + diff --git a/frontend/src/views/dashboard/ModelsView.vue b/frontend/src/views/dashboard/ModelsView.vue new file mode 100644 index 0000000..8b76158 --- /dev/null +++ b/frontend/src/views/dashboard/ModelsView.vue @@ -0,0 +1,225 @@ + + + diff --git a/frontend/src/views/dashboard/SystemConfig.vue b/frontend/src/views/dashboard/SystemConfig.vue new file mode 100644 index 0000000..259bcea --- /dev/null +++ b/frontend/src/views/dashboard/SystemConfig.vue @@ -0,0 +1,121 @@ + + + diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts index 8333eb1..a30e534 100644 --- a/frontend/vite.config.ts +++ b/frontend/vite.config.ts @@ -9,7 +9,7 @@ import path from 'path' // 需要自签名 HTTPS 时设置 VITE_DEV_HTTPS=true const useHttps = process.env.VITE_DEV_HTTPS === 'true' // 后端地址：默认 make dev-backend 启动的 8080，可用 VITE_DEV_API_TARGET 覆盖 -const apiTarget = process.env.VITE_DEV_API_TARGET || 'http://localhost:8080' +const apiTarget = process.env.VITE_DEV_API_TARGET || 'http://localhost:3000' // https://vite.dev/config/ export default defineConfig({