渠道: 模型名称映射(网关生效 + 渠道侧管理)
- 候选渠道携带 upstream_model, prepareUpstream 改写请求体 model 字段为上游名 - 新增渠道视角绑定 CRUD: GET/POST/PATCH/DELETE /admin/channels/:id/models - 前端渠道页新增"模型映射": 列出绑定、内联改上游名、解除、添加 - rewriteModel 三种协议通用(model 均在顶层) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,122 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/openteam/server/internal/pkg/resp"
|
||||
"github.com/openteam/server/internal/store"
|
||||
)
|
||||
|
||||
// AdminChannelModels GET /api/v1/admin/channels/:id/models — 渠道的模型绑定列表(含上游映射名)。
|
||||
func (h *Handler) AdminChannelModels(c *gin.Context) {
|
||||
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
|
||||
if err != nil {
|
||||
resp.Fail(c, http.StatusBadRequest, "invalid channel id")
|
||||
return
|
||||
}
|
||||
var bindings []store.ChannelModelBinding
|
||||
h.a.DB.Preload("Model").Where("channel_id = ?", id).Order("id ASC").Find(&bindings)
|
||||
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,
|
||||
})
|
||||
}
|
||||
resp.OK(c, gin.H{"items": out})
|
||||
}
|
||||
|
||||
// AdminChannelAddModel POST /api/v1/admin/channels/:id/models — 绑定模型到渠道。
|
||||
func (h *Handler) AdminChannelAddModel(c *gin.Context) {
|
||||
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
|
||||
if err != nil {
|
||||
resp.Fail(c, http.StatusBadRequest, "invalid channel id")
|
||||
return
|
||||
}
|
||||
var req struct {
|
||||
ModelName string `json:"model_name" binding:"required"`
|
||||
UpstreamModel string `json:"upstream_model"`
|
||||
Weight *int `json:"weight"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
resp.Fail(c, http.StatusBadRequest, "invalid input")
|
||||
return
|
||||
}
|
||||
var m store.Model
|
||||
if err := h.a.DB.Where("name = ?", req.ModelName).First(&m).Error; err != nil {
|
||||
resp.Fail(c, http.StatusNotFound, "model not found")
|
||||
return
|
||||
}
|
||||
upstream := req.UpstreamModel
|
||||
if upstream == "" {
|
||||
upstream = m.Name
|
||||
}
|
||||
b := store.ChannelModelBinding{
|
||||
ChannelID: id, ModelID: m.ID, UpstreamModel: upstream, Weight: intOr(req.Weight, 1),
|
||||
}
|
||||
if err := h.a.DB.Create(&b).Error; err != nil {
|
||||
resp.Fail(c, http.StatusConflict, "binding may already exist")
|
||||
return
|
||||
}
|
||||
resp.Created(c, gin.H{"id": b.ID, "model_id": m.ID, "model_name": m.Name, "upstream_model": upstream, "weight": b.Weight})
|
||||
}
|
||||
|
||||
// AdminChannelUpdateModel PATCH /api/v1/admin/channels/:id/models/:bid — 改映射名/权重。
|
||||
func (h *Handler) AdminChannelUpdateModel(c *gin.Context) {
|
||||
bid, err := strconv.ParseUint(c.Param("bid"), 10, 64)
|
||||
if err != nil {
|
||||
resp.Fail(c, http.StatusBadRequest, "invalid binding id")
|
||||
return
|
||||
}
|
||||
var req struct {
|
||||
UpstreamModel *string `json:"upstream_model"`
|
||||
Weight *int `json:"weight"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
resp.Fail(c, http.StatusBadRequest, "invalid input")
|
||||
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 {
|
||||
res := h.a.DB.Model(&store.ChannelModelBinding{}).Where("id = ?", bid).Updates(updates)
|
||||
if res.Error != nil {
|
||||
resp.Fail(c, http.StatusInternalServerError, "failed to update binding")
|
||||
return
|
||||
}
|
||||
if res.RowsAffected == 0 {
|
||||
resp.Fail(c, http.StatusNotFound, "binding not found")
|
||||
return
|
||||
}
|
||||
}
|
||||
resp.OK(c, gin.H{"ok": true})
|
||||
}
|
||||
|
||||
// AdminChannelDeleteModel DELETE /api/v1/admin/channels/:id/models/:bid — 解除绑定。
|
||||
func (h *Handler) AdminChannelDeleteModel(c *gin.Context) {
|
||||
bid, err := strconv.ParseUint(c.Param("bid"), 10, 64)
|
||||
if err != nil {
|
||||
resp.Fail(c, http.StatusBadRequest, "invalid binding id")
|
||||
return
|
||||
}
|
||||
res := h.a.DB.Delete(&store.ChannelModelBinding{}, bid)
|
||||
if res.Error != nil {
|
||||
resp.Fail(c, http.StatusInternalServerError, "failed to delete binding")
|
||||
return
|
||||
}
|
||||
if res.RowsAffected == 0 {
|
||||
resp.Fail(c, http.StatusNotFound, "binding not found")
|
||||
return
|
||||
}
|
||||
resp.OK(c, gin.H{"ok": true})
|
||||
}
|
||||
@@ -94,6 +94,10 @@ func NewRouter(a *app.App, gw *proxy.Gateway) *gin.Engine {
|
||||
admin.DELETE("/channels/:id", h.AdminDeleteChannel)
|
||||
admin.POST("/channels/:id/test", h.AdminTestChannel)
|
||||
admin.POST("/channels/:id/models/import", h.AdminImportChannelModels)
|
||||
admin.GET("/channels/:id/models", h.AdminChannelModels)
|
||||
admin.POST("/channels/:id/models", h.AdminChannelAddModel)
|
||||
admin.PATCH("/channels/:id/models/:bid", h.AdminChannelUpdateModel)
|
||||
admin.DELETE("/channels/:id/models/:bid", h.AdminChannelDeleteModel)
|
||||
// 模型与定价
|
||||
admin.GET("/models", h.AdminModels)
|
||||
admin.POST("/models", h.AdminCreateModel)
|
||||
|
||||
@@ -27,69 +27,74 @@ func NewService(db *gorm.DB, enc *crypto.Encryptor) *Service {
|
||||
return &Service{db: db, enc: enc, sems: map[uint64]chan struct{}{}}
|
||||
}
|
||||
|
||||
// Candidate 一个候选渠道 + 该模型的映射关系。
|
||||
type Candidate struct {
|
||||
Channel *store.Channel
|
||||
UpstreamModel string // 全局模型在此渠道的映射名(无绑定则为空,用客户端模型名)
|
||||
}
|
||||
|
||||
// Candidates 返回可用渠道候选:健康 + 启用,按优先级、权重降序、id 升序排列。
|
||||
// model 非空时优先取绑定该模型的渠道;无绑定则退回全局。
|
||||
func (s *Service) Candidates(model string) []*store.Channel {
|
||||
// model 非空时优先取绑定该模型的渠道(携带 upstream_model 映射);无绑定则退回全局。
|
||||
func (s *Service) Candidates(model string) []Candidate {
|
||||
if model != "" {
|
||||
var b []store.ChannelModelBinding
|
||||
var modelIDs []uint64
|
||||
s.db.Model(&store.Model{}).Where("name = ? AND enabled = ?", model, true).Pluck("id", &modelIDs)
|
||||
if len(modelIDs) > 0 {
|
||||
s.db.Where("model_id IN ?", modelIDs).Find(&b)
|
||||
chs := s.loadBound(b)
|
||||
if len(chs) > 0 {
|
||||
return chs
|
||||
if cands := s.loadBound(b); len(cands) > 0 {
|
||||
return cands
|
||||
}
|
||||
}
|
||||
}
|
||||
var chs []store.Channel
|
||||
s.db.Where("enabled = ? AND health_status = ?", true, store.ChannelHealthHealthy).
|
||||
Order("priority ASC, weight DESC, id ASC").Find(&chs)
|
||||
out := make([]*store.Channel, 0, len(chs))
|
||||
out := make([]Candidate, 0, len(chs))
|
||||
for i := range chs {
|
||||
out = append(out, &chs[i])
|
||||
out = append(out, Candidate{Channel: &chs[i]})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// loadBound 按绑定顺序加载渠道,过滤健康/启用。
|
||||
func (s *Service) loadBound(bindings []store.ChannelModelBinding) []*store.Channel {
|
||||
// loadBound 按绑定顺序加载渠道候选,过滤健康/启用,携带 upstream_model 映射。
|
||||
func (s *Service) loadBound(bindings []store.ChannelModelBinding) []Candidate {
|
||||
if len(bindings) == 0 {
|
||||
return nil
|
||||
}
|
||||
// channel_id -> 绑定(取该渠道对该模型的映射)
|
||||
byChannel := map[uint64]store.ChannelModelBinding{}
|
||||
ids := make([]uint64, 0, len(bindings))
|
||||
seen := map[uint64]bool{}
|
||||
for _, b := range bindings {
|
||||
if !seen[b.ChannelID] {
|
||||
seen[b.ChannelID] = true
|
||||
if _, ok := byChannel[b.ChannelID]; !ok {
|
||||
ids = append(ids, b.ChannelID)
|
||||
}
|
||||
byChannel[b.ChannelID] = b
|
||||
}
|
||||
var chs []store.Channel
|
||||
s.db.Where("id IN ? AND enabled = ? AND health_status = ?", ids, true, store.ChannelHealthHealthy).
|
||||
Order("priority ASC, weight DESC, id ASC").Find(&chs)
|
||||
// 保持绑定顺序
|
||||
byID := map[uint64]*store.Channel{}
|
||||
for i := range chs {
|
||||
byID[chs[i].ID] = &chs[i]
|
||||
}
|
||||
out := make([]*store.Channel, 0, len(ids))
|
||||
out := make([]Candidate, 0, len(ids))
|
||||
for _, id := range ids {
|
||||
if ch, ok := byID[id]; ok {
|
||||
out = append(out, ch)
|
||||
out = append(out, Candidate{Channel: ch, UpstreamModel: byChannel[id].UpstreamModel})
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Pick 按权重加权随机选一个候选(负载均衡)。
|
||||
func (s *Service) Pick(cands []*store.Channel) *store.Channel {
|
||||
// Pick 按权重加权随机选一个候选渠道(负载均衡)。
|
||||
func (s *Service) Pick(cands []Candidate) *store.Channel {
|
||||
if len(cands) == 0 {
|
||||
return nil
|
||||
}
|
||||
total := 0
|
||||
for _, c := range cands {
|
||||
w := c.Weight
|
||||
w := c.Channel.Weight
|
||||
if w <= 0 {
|
||||
w = 1
|
||||
}
|
||||
@@ -98,16 +103,16 @@ func (s *Service) Pick(cands []*store.Channel) *store.Channel {
|
||||
n, _ := rand.Int(rand.Reader, big.NewInt(int64(total)))
|
||||
acc := 0
|
||||
for _, c := range cands {
|
||||
w := c.Weight
|
||||
w := c.Channel.Weight
|
||||
if w <= 0 {
|
||||
w = 1
|
||||
}
|
||||
acc += w
|
||||
if int(n.Int64()) < acc {
|
||||
return c
|
||||
return c.Channel
|
||||
}
|
||||
}
|
||||
return cands[len(cands)-1]
|
||||
return cands[len(cands)-1].Channel
|
||||
}
|
||||
|
||||
// TryAcquire 尝试获取渠道并发槽;渠道满载返回 false(调用方可溢出到其他渠道)。
|
||||
|
||||
@@ -54,8 +54,8 @@ func TestCandidatesFiltersUnhealthy(t *testing.T) {
|
||||
if len(cands) != 2 {
|
||||
t.Fatalf("candidates = %d, want 2", len(cands))
|
||||
}
|
||||
if cands[0].Name != "a" {
|
||||
t.Fatalf("first by priority should be a, got %s", cands[0].Name)
|
||||
if cands[0].Channel.Name != "a" {
|
||||
t.Fatalf("first by priority should be a, got %s", cands[0].Channel.Name)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
@@ -148,8 +149,8 @@ func (g *Gateway) models(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"object": "list", "data": data})
|
||||
}
|
||||
|
||||
// candidateChannels 返回可用渠道候选(按模型绑定优先,退化全局)。
|
||||
func (g *Gateway) candidateChannels(model string) []*store.Channel {
|
||||
// candidateChannels 返回可用渠道候选(按模型绑定优先,退化全局;携带模型映射)。
|
||||
func (g *Gateway) candidateChannels(model string) []channel.Candidate {
|
||||
return g.ch.Candidates(model)
|
||||
}
|
||||
|
||||
@@ -213,8 +214,9 @@ func conversionTarget(formats []string, clientProto string) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// prepareUpstream 计算上游访问计划:渠道声明支持客户端协议则直通,否则转换。
|
||||
func prepareUpstream(ch *store.Channel, clientProto string, body []byte) (*upstreamPlan, error) {
|
||||
// prepareUpstream 计算上游访问计划:渠道声明支持客户端协议则直通,否则转换;
|
||||
// 应用模型名称映射(upstream_model)。
|
||||
func prepareUpstream(ch *store.Channel, clientProto string, body []byte, upstreamModel string) (*upstreamPlan, error) {
|
||||
target := conversionTarget(ch.FormatsEffective(), clientProto)
|
||||
if target == "" {
|
||||
return nil, fmt.Errorf("channel %q declares no supported protocol format", ch.Name)
|
||||
@@ -229,7 +231,26 @@ func prepareUpstream(ch *store.Channel, clientProto string, body []byte) (*upstr
|
||||
plan.lineConv = convert.NewStreamTransformer(target, clientProto)
|
||||
plan.bodyConv = func(b []byte) ([]byte, error) { return convert.ConvertResponse(b, target, clientProto) }
|
||||
}
|
||||
// 模型名称映射:把请求体 model 字段改写为渠道侧的 upstream_model
|
||||
if upstreamModel != "" {
|
||||
if out, err := rewriteModel(plan.body, upstreamModel); err == nil {
|
||||
plan.body = out
|
||||
}
|
||||
}
|
||||
return plan, nil
|
||||
}
|
||||
|
||||
// rewriteModel 改写请求体中的 model 字段(三种协议 model 都在顶层)。
|
||||
func rewriteModel(body []byte, model string) ([]byte, error) {
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(body, &m); err != nil {
|
||||
return body, nil
|
||||
}
|
||||
if cur, _ := m["model"].(string); cur == model {
|
||||
return body, nil
|
||||
}
|
||||
m["model"] = model
|
||||
return json.Marshal(m)
|
||||
}
|
||||
|
||||
var errNoChannel = errors.New("no available channel")
|
||||
|
||||
@@ -44,19 +44,28 @@ func parseBody(c *gin.Context) (*bodyReq, []byte, error) {
|
||||
return br, body, nil
|
||||
}
|
||||
|
||||
// upstreamURL 组装上游地址:base_url + 路径。
|
||||
// 兼容用户填完整 base(含 /v1):去掉尾部 /v1,避免与请求路径重复。
|
||||
// upstreamURL 组装上游地址,智能识别用户填写的 Base URL 形式:
|
||||
// - 完整端点(以目标资源路径结尾) → 直接使用
|
||||
// - 已含版本前缀(如 /v1) → 只拼资源路径(/chat/completions 等)
|
||||
// - 纯域名/地址前缀 → 拼完整路径(/v1/chat/completions 等)
|
||||
func upstreamURL(ch *store.Channel, path string) string {
|
||||
base := strings.TrimRight(ch.BaseURL, "/")
|
||||
return strings.TrimSuffix(base, "/v1") + path
|
||||
if strings.HasSuffix(base, path) {
|
||||
return base
|
||||
}
|
||||
if strings.HasSuffix(base, "/v1") {
|
||||
return base + strings.TrimPrefix(path, "/v1")
|
||||
}
|
||||
return base + path
|
||||
}
|
||||
|
||||
// doProxy 通用代理(M5):遍历候选渠道,按需转换;可安全重试的失败自动故障转移。
|
||||
func (g *Gateway) doProxy(c *gin.Context, cands []*store.Channel, clientProto string, body []byte, stream bool, sink *usageSink) {
|
||||
func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto string, body []byte, stream bool, sink *usageSink) {
|
||||
var lastStatus = http.StatusBadGateway
|
||||
var lastBody = []byte("all upstream channels failed")
|
||||
for _, ch := range cands {
|
||||
plan, err := prepareUpstream(ch, clientProto, body)
|
||||
for _, cand := range cands {
|
||||
ch := cand.Channel
|
||||
plan, err := prepareUpstream(ch, clientProto, body, cand.UpstreamModel)
|
||||
if err != nil {
|
||||
lastStatus, lastBody = http.StatusInternalServerError, []byte("conversion error: "+err.Error())
|
||||
continue
|
||||
|
||||
Reference in New Issue
Block a user