From 4a4b3f8676e1b69b8b62305dd864d9c6b1ec0264 Mon Sep 17 00:00:00 2001 From: Zexi Li Date: Mon, 22 Jun 2026 20:14:01 +0800 Subject: [PATCH] feat(llm): sync llm deployment with aiproxy catalog and routing (#25057) Add auto_register_aiproxy on llm_deployment creation, background sync task, and register/unregister CLI actions. Link aiproxy providers, routings, and models to llm replicas via llm_id and llm_deployment_id fields. --- cmd/climc/shell/llm/llm_deployment.go | 2 + docs/aiproxy/functional-test-climc.md | 9 +- docs/llm/llm-deployment-aiproxy.md | 95 ++ pkg/aiproxy/models/ai_providers.go | 28 + pkg/aiproxy/models/ai_proxy_nodes.go | 80 +- pkg/aiproxy/models/ai_routing_models.go | 13 +- pkg/aiproxy/models/ai_routings.go | 34 +- .../models/aiproxy_catalog_validate.go | 19 + pkg/aiproxy/models/chat_upstream.go | 38 +- pkg/aiproxy/models/list_models.go | 54 +- pkg/aiproxy/models/list_models_test.go | 62 +- pkg/aiproxy/models/proxy_node_local.go | 7 + pkg/apis/aiproxy/ai_provider.go | 24 +- pkg/apis/aiproxy/ai_proxy_node.go | 28 +- pkg/apis/aiproxy/ai_routing.go | 4 + pkg/apis/llm/aiproxy_bindings.go | 25 + pkg/apis/llm/llm_deployment.go | 38 + pkg/llm/models/instantmodel.go | 118 ++- pkg/llm/models/llm.go | 14 +- pkg/llm/models/llm_aiproxy_sync.go | 935 ++++++++++++++++++ pkg/llm/models/llm_aiproxy_sync_test.go | 166 ++++ pkg/llm/models/llm_deployment.go | 101 +- pkg/llm/tasks/llm/llm_aiproxy_sync_task.go | 69 ++ pkg/llm/tasks/llm/llm_delete_task.go | 11 + .../tasks/llm/llm_deployment_create_task.go | 21 +- .../tasks/llm/llm_deployment_delete_task.go | 5 + pkg/mcclient/options/aiproxy/resources.go | 52 +- pkg/mcclient/options/llm/llm_deployment.go | 28 + 28 files changed, 1959 insertions(+), 121 deletions(-) create mode 100644 docs/llm/llm-deployment-aiproxy.md create mode 100644 pkg/apis/llm/aiproxy_bindings.go create mode 100644 pkg/llm/models/llm_aiproxy_sync.go create mode 100644 pkg/llm/models/llm_aiproxy_sync_test.go create mode 100644 pkg/llm/tasks/llm/llm_aiproxy_sync_task.go diff --git a/cmd/climc/shell/llm/llm_deployment.go b/cmd/climc/shell/llm/llm_deployment.go index 56773d78ee..f189991ff5 100644 --- a/cmd/climc/shell/llm/llm_deployment.go +++ b/cmd/climc/shell/llm/llm_deployment.go @@ -13,4 +13,6 @@ func init() { cmd.Create(new(options.LLMDeploymentCreateOptions)) cmd.Update(new(options.LLMDeploymentUpdateOptions)) cmd.Delete(new(options.LLMDeploymentDeleteOptions)) + cmd.Perform("register-aiproxy", new(options.LLMDeploymentRegisterAiproxyOptions)) + cmd.Perform("unregister-aiproxy", new(options.LLMDeploymentUnregisterAiproxyOptions)) } diff --git a/docs/aiproxy/functional-test-climc.md b/docs/aiproxy/functional-test-climc.md index 82c864e12c..9f59426c89 100644 --- a/docs/aiproxy/functional-test-climc.md +++ b/docs/aiproxy/functional-test-climc.md @@ -109,11 +109,11 @@ climc ai-proxy-node-register --address https://standby-host:30938 --hb-timeout 1 ```bash climc ai-proxy-node-create standby-1 \ --address https://standby-host:30938 \ - --domain aiproxy-standby.example.com \ + --access-address https://aiproxy-standby.example.com:443 \ --hb-timeout 120 \ --enabled -climc ai-proxy-node-update primary --address https://primary-host:30938 --domain aiproxy.example.com +climc ai-proxy-node-update primary --address https://primary-host:30938 --access-address https://aiproxy.example.com:443 climc ai-proxy-node-enable primary climc ai-proxy-node-disable ``` @@ -124,6 +124,8 @@ climc ai-proxy-node-disable climc ai-routing-update aiproxy-ft-routing --ai-proxy-node-id primary ``` +创建 `ai_routing` 时若省略 `--ai-proxy-node-id`，默认绑定 `primary` 节点；更新时显式传空值仍可清空绑定（任意 aiproxy 节点均可匹配）。 + ## 1. 检查 Keystone endpoint ```bash @@ -192,11 +194,14 @@ Virtual key 归属当前 climc 用户的 **项目**；后续 `ai_routing` 须在 将项目内请求 `model=qwen-turbo` 指到 catalog 的 aliyun/qwen-turbo。 +**精确匹配**：`--model-key qwen-turbo` 与请求 body 中 `model` 完全一致时命中（优先于 `--model-pattern` 通配规则）。 + `models` 里 `ai_model_id` 使用 catalog 固定 id（与 name 相同，如 `aliyun-qwen-turbo`），或在指定 `ai_provider_id` 时也可填 **model_key**（如 `qwen-turbo`）： ```bash climc ai-routing-create aiproxy-ft-routing \ --priority 10 \ + --model-key qwen-turbo \ --models '[{"ai_provider_id":"aliyun","ai_model_id":"qwen-turbo","priority":1}]' ``` diff --git a/docs/llm/llm-deployment-aiproxy.md b/docs/llm/llm-deployment-aiproxy.md new file mode 100644 index 0000000000..2f487c584b --- /dev/null +++ b/docs/llm/llm-deployment-aiproxy.md @@ -0,0 +1,95 @@ +# llm_deployment 与 aiproxy 自动关联 + +创建 `llm_deployment` 时**默认开启** `auto_register_aiproxy`。创建任务完成后会调度 `LLMAiproxySyncTask`；副本变为 running 后自动在 aiproxy 中创建 catalog 与路由资源。 + +## 创建 deployment（默认已关联 aiproxy） + +```bash +climc llm-deployment-create my-qwen \ + --llm-sku-id \ + --net \ + --replicas 2 +``` + +关闭自动关联： + +```bash +climc llm-deployment-create my-qwen \ + --auto-register-aiproxy=false \ + --llm-sku-id \ + --net +``` + +## 手动重同步 / 取消注册 + +```bash +climc llm-deployment-register-aiproxy my-qwen +climc llm-deployment-unregister-aiproxy my-qwen +``` + +`unregister-aiproxy` 会清理 aiproxy 侧资源，并将 `auto_register_aiproxy` 置为 false，同时清空 `aiproxy_bindings`。deployment `status` 恢复为副本健康态（`ready` / `partial` / `deploying` 等）。 + +网关同步阶段通过 deployment `status` 表达：`aiproxy_pending`、`aiproxy_syncing`、`aiproxy_partial`、`aiproxy_sync_failed`；全部同步成功后恢复为 `ready` 或 `partial`。 + +## 查看对应关系 + +deployment 详情中的 `aiproxy_bindings` 列出每个副本的客户端 model 别名与 provider id： + +```bash +climc llm-deployment-show my-qwen +``` + +aiproxy 侧也可按来源 id 反查 provider： + +```bash +climc ai-provider-list --llm-deployment-id +climc ai-provider-list --llm-id +``` + +deployment 与 routing 的关联保存在 `llm_deployment.aiproxy_routing_id`。 + +## 客户端访问 + +需先创建项目级 `ai_virtual_key`，再通过 aiproxy OpenAI 兼容接口访问： + +```bash +curl -k "$AIPROXY/openai/v1/chat/completions" \ + -H "Authorization: Bearer $VK" \ + -d '{"model":"-","messages":[{"role":"user","content":"hi"}]}' +``` + +`model` 填 deployment 详情里 `aiproxy_bindings[].client_model_alias`（格式 `{deployment_name}-{upstream_model_key}`，同一 deployment 下各副本相同），与 `ai_routing.model_key` 同值。LLM 自动注册/重同步会写入 `model_key` 并清空 `model_pattern`。 + +## 删除 deployment + +删除 `llm_deployment` 时会自动清理关联的 aiproxy 资源（`ai_provider`、`ai_model`、`ai_routing`），provider 按 `llm_deployment_id` / `llm_id` 查找，routing 按 `llm_deployment.aiproxy_routing_id` 删除，无需 `auto_register_aiproxy` 为 true。 + +## 字段对应关系 + +| llm 侧 | aiproxy 侧 | +|--------|-----------| +| `llm_deployment.aiproxy_routing_id` | `ai_routing.id` | +| `llm.id` | `ai_provider.llm_id` | +| `llm_deployment.id` | `ai_provider.llm_deployment_id` | +| upstream served model name（vLLM/SGLang 为 model 目录 basename，Ollama 为 `name:tag`） | `ai_model.model_key` | +| `{deployment_name}-{upstream_model_key}` | **`ai_routing.model_key`** 与 **`aiproxy_bindings[].client_model_alias`**（同值；客户端 model 精确匹配，选路与列表优先） | +| （手工通配/前缀规则） | `ai_routing.model_pattern` | +| （LLM 自动注册时留空） | `ai_routing_model.model_pattern` | + +`ai_routing` / `ai_routing_model` 不再保存 `llm_deployment_id` / `llm_id`；通过 `ai_provider` 与 `ai_routing_model.ai_provider_id` 间接关联推理实例。 + +## 命名规则 + +自动注册时 aiproxy catalog 资源名称基于推理部署名与副本实例名（`{deployment_name}-{index}`），不再使用 UUID 后缀： + +| aiproxy 资源 | 命名规则 | 示例（部署 `my-qwen`，副本 `my-qwen-0`） | +|--------------|----------|------------------------------------------| +| `ai_routing.name` | `llm-dep-{slug(deployment.name)}` | `llm-dep-my-qwen` | +| `ai_provider.name` | `llm-{slug(llm.name)}` | `llm-my-qwen-0` | +| `ai_model.name` | `llm-{slug(llm.name)}-{slug(model_key)}` | `llm-my-qwen-0-qwen3-0-6b` | + +`slug(...)` 将名称规范为小写并将 `/`、`.` 等特殊字符转为 `-`。部署名或副本名为空时回退为对应 id。 + +若 deployment 尚未记录 `aiproxy_routing_id`，同步时会按 `llm-dep-{slug(name)}` 查找已有 `ai_routing` 并更新，避免同名 deployment 重建或并发同步时的 409 重名错误。 + +对已注册资源执行 `climc llm-deployment-register-aiproxy` 会同步更新上述名称。 diff --git a/pkg/aiproxy/models/ai_providers.go b/pkg/aiproxy/models/ai_providers.go index e09a0492cd..6d8da98c70 100644 --- a/pkg/aiproxy/models/ai_providers.go +++ b/pkg/aiproxy/models/ai_providers.go @@ -37,6 +37,9 @@ type SAiProvider struct { ProviderKey string `width:"64" charset:"ascii" nullable:"false" list:"user" create:"required" update:"user"` // Config is a JSON snapshot of provider connectivity (base_url, optional api_key). Config *api.SAiProviderConfig `length:"long" charset:"utf8" list:"user" create:"optional" update:"user"` + // LlmDeploymentId and LlmId link this provider to an llm_deployment replica (set by llm sync). + LlmDeploymentId string `width:"128" charset:"ascii" nullable:"true" list:"user" create:"optional" update:"user" index:"true"` + LlmId string `width:"128" charset:"ascii" nullable:"true" list:"user" create:"optional" update:"user" index:"true"` } type SAiProviderManager struct { @@ -74,6 +77,12 @@ func (manager *SAiProviderManager) ListItemFilter( if key := strings.TrimSpace(query.ProviderKey); key != "" { q = q.Equals("provider_key", key) } + if v := strings.TrimSpace(query.LlmDeploymentId); v != "" { + q = q.Equals("llm_deployment_id", v) + } + if v := strings.TrimSpace(query.LlmId); v != "" { + q = q.Equals("llm_id", v) + } return q, nil } @@ -89,6 +98,9 @@ func (manager *SAiProviderManager) FetchCustomizeColumns( baseRows := manager.SEnabledStatusStandaloneResourceBaseManager.FetchCustomizeColumns(ctx, userCred, query, objs, fields, isList) for i := range objs { rows[i].EnabledStatusStandaloneResourceDetails = baseRows[i] + prov := objs[i].(*SAiProvider) + rows[i].LlmDeploymentId = prov.LlmDeploymentId + rows[i].LlmId = prov.LlmId } return rows } @@ -121,6 +133,12 @@ func (manager *SAiProviderManager) ValidateCreateData( input.Name = pk } + input.LlmDeploymentId = strings.TrimSpace(input.LlmDeploymentId) + input.LlmId = strings.TrimSpace(input.LlmId) + if err := ensureUniqueAiProviderLlmId(ctx, input.LlmId, ""); err != nil { + return input, err + } + return input, nil } @@ -151,5 +169,15 @@ func (p *SAiProvider) ValidateUpdateData( } } + if query.Contains("llm_deployment_id") { + input.LlmDeploymentId = strings.TrimSpace(input.LlmDeploymentId) + } + if query.Contains("llm_id") { + input.LlmId = strings.TrimSpace(input.LlmId) + if err := ensureUniqueAiProviderLlmId(ctx, input.LlmId, p.Id); err != nil { + return input, err + } + } + return input, nil } diff --git a/pkg/aiproxy/models/ai_proxy_nodes.go b/pkg/aiproxy/models/ai_proxy_nodes.go index ea6ca94ea9..1915a9e111 100644 --- a/pkg/aiproxy/models/ai_proxy_nodes.go +++ b/pkg/aiproxy/models/ai_proxy_nodes.go @@ -38,19 +38,19 @@ import ( ) const ( - defaultAiProxyNodeHbTimeout = 120 - defaultPrimaryAiProxyNodeId = "primary" - maxAiProxyNodeDomainLen = 256 + defaultAiProxyNodeHbTimeout = 120 + defaultPrimaryAiProxyNodeId = "primary" + maxAiProxyNodeAccessAddressLen = 256 ) -// SAiProxyNode records an aiproxy instance reachable address and optional domain name. +// SAiProxyNode records an aiproxy instance reachable address and optional public access URL. type SAiProxyNode struct { db.SEnabledStatusStandaloneResourceBase - Address string `width:"256" charset:"ascii" nullable:"false" list:"user" create:"required" update:"user" index:"true"` - Domain string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` - LastSeen time.Time `nullable:"true" list:"user"` - HbTimeout int `nullable:"false" default:"120" list:"user" create:"optional" update:"user"` + Address string `width:"256" charset:"ascii" nullable:"false" list:"user" create:"required" update:"user" index:"true"` + AccessAddress string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` + LastSeen time.Time `nullable:"true" list:"user"` + HbTimeout int `nullable:"false" default:"120" list:"user" create:"optional" update:"user"` } type SAiProxyNodeManager struct { @@ -87,18 +87,18 @@ func (manager *SAiProxyNodeManager) InitializeData() error { node.Name = defaultPrimaryAiProxyNodeId node.Description = "Default primary aiproxy node" node.Address = addr - domain := "" + accessAddress := "" if existing, err := manager.FetchById(defaultPrimaryAiProxyNodeId); err == nil { - domain = strings.TrimSpace(existing.(*SAiProxyNode).Domain) + accessAddress = strings.TrimSpace(existing.(*SAiProxyNode).AccessAddress) } - if domain == "" { - d, err := DomainFromApiServer(nil) + if accessAddress == "" { + a, err := AccessAddressFromApiServer(nil) if err != nil { return err } - domain = d + accessAddress = a } - node.Domain = domain + node.AccessAddress = accessAddress node.HbTimeout = defaultAiProxyNodeHbTimeout node.LastSeen = time.Now() node.SetEnabled(true) @@ -138,18 +138,15 @@ func normalizeAiProxyNodeAddress(raw string) (string, error) { return fmt.Sprintf("http://%s", address), nil } -func normalizeAiProxyNodeDomain(domain string) (string, error) { - domain = strings.TrimSpace(domain) - if domain == "" { +func normalizeAiProxyNodeAccessAddress(raw string) (string, error) { + accessAddress := strings.TrimSpace(raw) + if accessAddress == "" { return "", nil } - if len(domain) > maxAiProxyNodeDomainLen { - return "", errors.Wrapf(httperrors.ErrInputParameter, "domain too long (max %d)", maxAiProxyNodeDomainLen) + if len(accessAddress) > maxAiProxyNodeAccessAddressLen { + return "", errors.Wrapf(httperrors.ErrInputParameter, "access_address too long (max %d)", maxAiProxyNodeAccessAddressLen) } - if strings.Contains(domain, "://") || strings.ContainsAny(domain, "/:") { - return "", errors.Wrap(httperrors.ErrInputParameter, "domain must be a hostname without scheme or port") - } - return domain, nil + return normalizeAiProxyNodeAddress(accessAddress) } func aiProxyNodeDisplayName(address string) string { @@ -179,8 +176,8 @@ func AdvertiseAddressFromOptions(opts *options.SAiProxyOptions) (string, error) return normalizeAiProxyNodeAddress(fmt.Sprintf("%s://%s:%d", scheme, host, opts.Port)) } -// DomainFromApiServer derives ai_proxy_node.domain from --api-server (hostname only). -func DomainFromApiServer(opts *options.SAiProxyOptions) (string, error) { +// AccessAddressFromApiServer derives ai_proxy_node.access_address from --api-server. +func AccessAddressFromApiServer(opts *options.SAiProxyOptions) (string, error) { if opts == nil { opts = &options.Options } @@ -188,22 +185,7 @@ func DomainFromApiServer(opts *options.SAiProxyOptions) (string, error) { if raw == "" { return "", nil } - host := raw - if strings.Contains(raw, "://") { - u, err := url.Parse(raw) - if err != nil { - return "", errors.Wrapf(err, "parse api_server %q", raw) - } - host = strings.TrimSpace(u.Hostname()) - } else if strings.ContainsAny(raw, "/:") { - if h, _, err := net.SplitHostPort(raw); err == nil { - host = h - } - } - if host == "" { - return "", nil - } - return normalizeAiProxyNodeDomain(host) + return normalizeAiProxyNodeAccessAddress(raw) } func (manager *SAiProxyNodeManager) ListItemFilter( @@ -219,8 +201,8 @@ func (manager *SAiProxyNodeManager) ListItemFilter( if addr := strings.TrimSpace(query.Address); addr != "" { q = q.Equals("address", addr) } - if domain := strings.TrimSpace(query.Domain); domain != "" { - q = q.Equals("domain", domain) + if accessAddress := strings.TrimSpace(query.AccessAddress); accessAddress != "" { + q = q.Equals("access_address", accessAddress) } return q, nil } @@ -259,7 +241,7 @@ func (manager *SAiProxyNodeManager) ValidateCreateData( if err != nil { return input, err } - input.Domain, err = normalizeAiProxyNodeDomain(input.Domain) + input.AccessAddress, err = normalizeAiProxyNodeAccessAddress(input.AccessAddress) if err != nil { return input, err } @@ -287,9 +269,9 @@ func (manager *SAiProxyNodeManager) PerformRegister( hbTimeout = defaultAiProxyNodeHbTimeout } nodeId := aiProxyNodeId(addr) - domain := "" + accessAddress := "" if existing, err := manager.FetchById(nodeId); err == nil { - domain = existing.(*SAiProxyNode).Domain + accessAddress = existing.(*SAiProxyNode).AccessAddress } else if errors.Cause(err) != sql.ErrNoRows { return nil, errors.Wrap(err, "fetch ai_proxy_node") } @@ -298,7 +280,7 @@ func (manager *SAiProxyNodeManager) PerformRegister( node.Id = nodeId node.Name = aiProxyNodeDisplayName(addr) node.Address = addr - node.Domain = domain + node.AccessAddress = accessAddress node.HbTimeout = hbTimeout node.LastSeen = time.Now() node.SetEnabled(true) @@ -327,8 +309,8 @@ func (node *SAiProxyNode) ValidateUpdateData( return input, err } } - if query.Contains("domain") { - input.Domain, err = normalizeAiProxyNodeDomain(input.Domain) + if query.Contains("access_address") { + input.AccessAddress, err = normalizeAiProxyNodeAccessAddress(input.AccessAddress) if err != nil { return input, err } diff --git a/pkg/aiproxy/models/ai_routing_models.go b/pkg/aiproxy/models/ai_routing_models.go index c6e9fe3d37..9c339c7cb9 100644 --- a/pkg/aiproxy/models/ai_routing_models.go +++ b/pkg/aiproxy/models/ai_routing_models.go @@ -108,6 +108,13 @@ func (manager *SAiRoutingModelManager) FetchCustomizeColumns( baseRows := manager.SStandaloneResourceBaseManager.FetchCustomizeColumns(ctx, userCred, query, objs, fields, isList) for i := range objs { rows[i].StandaloneResourceDetails = baseRows[i] + rm := objs[i].(*SAiRoutingModel) + rows[i].AiRoutingId = rm.AiRoutingId + rows[i].AiProviderId = rm.AiProviderId + rows[i].AiModelId = rm.AiModelId + rows[i].Priority = rm.Priority + rows[i].ModelPattern = rm.ModelPattern + rows[i].Enabled = rm.Enabled.IsTrue() } return rows } @@ -372,7 +379,7 @@ func pickAiRoutingModelFromEntries( } if routing != nil && routing.RouterEnabled { - candidates := buildAiRoutingModelCandidates(allEntries, modelsById) + candidates := buildAiRoutingModelCandidates(routing, allEntries, modelsById) if len(candidates) == 0 { return routerFallbackPick(routing, allEntries[0], errors.Wrap(httperrors.ErrInvalidStatus, "router has no candidate models")) } @@ -412,13 +419,13 @@ func pickAiRoutingModelFromEntries( return matches[0], nil } -func buildAiRoutingModelCandidates(entries []*SAiRoutingModel, modelsById map[string]*SAiModel) []aiRoutingModelCandidate { +func buildAiRoutingModelCandidates(routing *SAiRouting, entries []*SAiRoutingModel, modelsById map[string]*SAiModel) []aiRoutingModelCandidate { candidates := make([]aiRoutingModelCandidate, 0, len(entries)) for _, entry := range entries { if entry == nil { continue } - name := clientFacingModelID(entry, modelsById[entry.AiModelId]) + name := clientFacingModelID(routing, entry, modelsById[entry.AiModelId]) if name == "" { continue } diff --git a/pkg/aiproxy/models/ai_routings.go b/pkg/aiproxy/models/ai_routings.go index 2e96d0fc6e..064b012454 100644 --- a/pkg/aiproxy/models/ai_routings.go +++ b/pkg/aiproxy/models/ai_routings.go @@ -37,6 +37,9 @@ type SAiRouting struct { db.SEnabledResourceBase Priority int `default:"100" nullable:"false" list:"user" create:"optional" update:"user"` + // ModelKey exactly matches the client request "model" (case-insensitive). + // When non-empty and matched, takes precedence over ModelPattern during routing. + ModelKey string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` // ModelPattern optionally matches the requested model id (implementation-specific glob/prefix). ModelPattern string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` // AiProxyNodeId optionally binds the rule to one aiproxy instance (ai_proxy_node id). @@ -82,6 +85,9 @@ func (manager *SAiRoutingManager) ListItemFilter( if v := strings.TrimSpace(query.ModelPattern); v != "" { q = q.Equals("model_pattern", v) } + if v := strings.TrimSpace(query.ModelKey); v != "" { + q = q.Equals("model_key", v) + } if v := strings.TrimSpace(query.AiProxyNodeId); v != "" { q = q.Equals("ai_proxy_node_id", v) } @@ -110,6 +116,7 @@ func (manager *SAiRoutingManager) FetchCustomizeColumns( routing := objs[i].(*SAiRouting) rows[i].SharableVirtualResourceDetails = sharableRows[i] rows[i].Priority = routing.Priority + rows[i].ModelKey = routing.ModelKey rows[i].ModelPattern = routing.ModelPattern rows[i].AiProxyNodeId = routing.AiProxyNodeId rows[i].RouterEnabled = routing.RouterEnabled @@ -183,6 +190,20 @@ func normalizeAiRoutingRouterTimeoutSeconds(timeout int) int { return timeout } +func normalizeAiRoutingModelKey(key string) (string, error) { + key = strings.TrimSpace(key) + if key == "" { + return "", nil + } + if strings.Contains(key, "*") { + return "", errors.Wrap(httperrors.ErrInputParameter, "model_key must not contain '*'") + } + if len(key) > 256 { + return "", errors.Wrap(httperrors.ErrInputParameter, "model_key too long") + } + return key, nil +} + func normalizeAiRoutingRouterCreate(input *api.AiRoutingCreateInput) error { input.RouterUrl = strings.TrimSpace(input.RouterUrl) input.RouterRoutePath = normalizeAiRoutingRouterRoutePath(input.RouterRoutePath) @@ -253,6 +274,13 @@ func (routing *SAiRouting) ValidateUpdateData( } else if query.Contains("ai_proxy_node_id") { input.AiProxyNodeId = "" } + if query.Contains("model_key") { + var err error + input.ModelKey, err = normalizeAiRoutingModelKey(input.ModelKey) + if err != nil { + return input, err + } + } if err := normalizeAiRoutingRouterUpdate(routing, query, input); err != nil { return input, err } @@ -278,7 +306,11 @@ func (manager *SAiRoutingManager) ValidateCreateData( } input.Models = validatedModels - input.AiProxyNodeId, err = validateAiProxyNodeId(ctx, userCred, input.AiProxyNodeId) + input.AiProxyNodeId, err = resolveAiProxyNodeIdForCreate(ctx, userCred, input.AiProxyNodeId) + if err != nil { + return input, err + } + input.ModelKey, err = normalizeAiRoutingModelKey(input.ModelKey) if err != nil { return input, err } diff --git a/pkg/aiproxy/models/aiproxy_catalog_validate.go b/pkg/aiproxy/models/aiproxy_catalog_validate.go index f1dc62a297..34861d204c 100644 --- a/pkg/aiproxy/models/aiproxy_catalog_validate.go +++ b/pkg/aiproxy/models/aiproxy_catalog_validate.go @@ -179,3 +179,22 @@ func fetchEnabledAiProvider(ctx context.Context, userCred mcclient.TokenCredenti func defaultAiModelName(providerName, modelKey string) string { return catalogModelId(providerName, modelKey) } + +func ensureUniqueAiProviderLlmId(ctx context.Context, llmId, excludeId string) error { + llmId = strings.TrimSpace(llmId) + if llmId == "" { + return nil + } + q := AiProviderManager.Query().Equals("llm_id", llmId) + if excludeId != "" { + q = q.NotEquals("id", excludeId) + } + cnt, err := q.CountWithError() + if err != nil { + return errors.Wrap(err, "count ai_provider by llm_id") + } + if cnt > 0 { + return errors.Wrapf(httperrors.ErrConflict, "ai_provider for llm_id %q already exists", llmId) + } + return nil +} diff --git a/pkg/aiproxy/models/chat_upstream.go b/pkg/aiproxy/models/chat_upstream.go index 8653eca036..11e052b6c7 100644 --- a/pkg/aiproxy/models/chat_upstream.go +++ b/pkg/aiproxy/models/chat_upstream.go @@ -21,6 +21,7 @@ import ( "strings" "yunion.io/x/jsonutils" + "yunion.io/x/log" "yunion.io/x/pkg/errors" "yunion.io/x/pkg/util/rbacscope" @@ -43,6 +44,14 @@ type ChatUpstream struct { RequestsPerMinute int } +func modelKeyMatches(key, requestedModel string) bool { + key = strings.TrimSpace(key) + if key == "" { + return false + } + return strings.EqualFold(key, strings.TrimSpace(requestedModel)) +} + func modelPatternMatches(pattern, requestedModel string) bool { pattern = strings.TrimSpace(pattern) if pattern == "" { @@ -109,13 +118,26 @@ func listProjectRoutingsForVirtualKey(ctx context.Context, userCred mcclient.Tok return routings, nil } -// pickRoutingForRequest chooses the first matching ai_routing (lowest priority value wins) -// on the current aiproxy instance. When a matched rule is bound to another node, returns forbidden. +// pickRoutingForRequest chooses the best matching ai_routing on the current aiproxy instance. +// Phase 1: exact ai_routing.model_key match (lowest priority wins). +// Phase 2: ai_routing.model_pattern match (lowest priority wins). func pickRoutingForRequest(routings []SAiRouting, reqModel, currentNodeId string) (*SAiRouting, error) { + if picked, err := pickRoutingByMatch(routings, reqModel, currentNodeId, func(r *SAiRouting, reqModel string) bool { + return modelKeyMatches(r.ModelKey, reqModel) + }); picked != nil || err != nil { + return picked, err + } + return pickRoutingByMatch(routings, reqModel, currentNodeId, func(r *SAiRouting, reqModel string) bool { + return modelPatternMatches(r.ModelPattern, reqModel) + }) +} + +func pickRoutingByMatch(routings []SAiRouting, reqModel, currentNodeId string, match func(*SAiRouting, string) bool) (*SAiRouting, error) { var boundElsewhere *SAiRouting + var best *SAiRouting for i := range routings { r := &routings[i] - if !modelPatternMatches(r.ModelPattern, reqModel) { + if !match(r, reqModel) { continue } if !proxyNodeScopeMatches(r.AiProxyNodeId, currentNodeId) { @@ -124,7 +146,12 @@ func pickRoutingForRequest(routings []SAiRouting, reqModel, currentNodeId string } continue } - return r, nil + if best == nil || r.Priority < best.Priority { + best = r + } + } + if best != nil { + return best, nil } if boundElsewhere != nil { return nil, errors.Wrapf(httperrors.ErrForbidden, @@ -184,7 +211,7 @@ func resolveCatalogModelFromRouting( // ResolveChatUpstream resolves upstream URL, API key, and catalog model_key for a chat request: // 1. ai_virtual_key (auth + project scope) -// 2. ai_routing in that project (model_pattern / optional proxy-node scope, priority) +// 2. ai_routing in that project (model_key exact match first, then model_pattern / optional proxy-node scope, priority) // 3. ai_routing_model -> ai_provider + ai_model // 4. ai_key rows for that provider matching the catalog model_key (weight), else provider.config api_key func ResolveChatUpstream(ctx context.Context, userCred mcclient.TokenCredential, virtualKey string, body *jsonutils.JSONDict) (*ChatUpstream, error) { @@ -206,6 +233,7 @@ func ResolveChatUpstream(ctx context.Context, userCred mcclient.TokenCredential, if err != nil { return nil, err } + log.Infof("=========request model: %s, pick routing: %s", reqModel, jsonutils.Marshal(routing).PrettyString()) if routing == nil { return nil, errors.Wrap(httperrors.ErrNotFound, "no ai_routing matched for virtual key project on this aiproxy node") } diff --git a/pkg/aiproxy/models/list_models.go b/pkg/aiproxy/models/list_models.go index 0e2c2459a5..77e02edd0e 100644 --- a/pkg/aiproxy/models/list_models.go +++ b/pkg/aiproxy/models/list_models.go @@ -52,8 +52,10 @@ func ListModelsForVirtualKey(ctx context.Context, userCred mcclient.TokenCredent return nil, err } currentNode := CurrentProxyNodeId() + routingById := make(map[string]*SAiRouting, len(routings)) routingIds := make([]string, 0, len(routings)) for i := range routings { + routingById[routings[i].Id] = &routings[i] if proxyNodeScopeMatches(routings[i].AiProxyNodeId, currentNode) { routingIds = append(routingIds, routings[i].Id) } @@ -67,9 +69,6 @@ func ListModelsForVirtualKey(ctx context.Context, userCred mcclient.TokenCredent if err := q.All(&entries); err != nil { return nil, errors.Wrap(err, "list ai_routing_models") } - if len(entries) == 0 { - return nil, nil - } providerIds := make([]string, 0, len(entries)) modelIds := make([]string, 0, len(entries)) @@ -87,10 +86,48 @@ func ListModelsForVirtualKey(ctx context.Context, userCred mcclient.TokenCredent return nil, err } - seen := make(map[string]ModelsListEntry, len(entries)) + firstProviderByRouting := make(map[string]string, len(routingIds)) + for i := range entries { + routingId := entries[i].AiRoutingId + if _, ok := firstProviderByRouting[routingId]; ok { + continue + } + if prov := providers[entries[i].AiProviderId]; prov != nil { + firstProviderByRouting[routingId] = strings.TrimSpace(prov.ProviderKey) + } + } + + seen := make(map[string]ModelsListEntry, len(entries)+len(routingIds)) created := time.Now().Unix() + + // Pass 1: ai_routing.model_key is itself a client-facing model id. + for i := range routings { + routing := &routings[i] + if !proxyNodeScopeMatches(routing.AiProxyNodeId, currentNode) { + continue + } + id := strings.TrimSpace(routing.ModelKey) + if id == "" { + continue + } + if _, ok := seen[id]; ok { + continue + } + seen[id] = ModelsListEntry{ + ID: id, + Object: "model", + Created: created, + OwnedBy: firstProviderByRouting[routing.Id], + } + } + + // Pass 2: routes without model_key use ai_routing_model / pattern fallbacks. for i := range entries { e := &entries[i] + routing := routingById[e.AiRoutingId] + if routing == nil || strings.TrimSpace(routing.ModelKey) != "" { + continue + } prov := providers[e.AiProviderId] mdl := modelsById[e.AiModelId] if prov == nil || mdl == nil { @@ -99,7 +136,7 @@ func ListModelsForVirtualKey(ctx context.Context, userCred mcclient.TokenCredent if !virtualKeyAllowsProvider(vk, prov) { continue } - id := clientFacingModelID(e, mdl) + id := clientFacingModelID(routing, e, mdl) if id == "" { continue } @@ -126,12 +163,17 @@ func ListModelsForVirtualKey(ctx context.Context, userCred mcclient.TokenCredent return out, nil } -func clientFacingModelID(entry *SAiRoutingModel, mdl *SAiModel) string { +func clientFacingModelID(routing *SAiRouting, entry *SAiRoutingModel, mdl *SAiModel) string { if entry != nil { if mp := strings.TrimSpace(entry.ModelPattern); mp != "" && !strings.Contains(mp, "*") { return mp } } + if routing != nil { + if mp := strings.TrimSpace(routing.ModelPattern); mp != "" && !strings.Contains(mp, "*") { + return mp + } + } if mdl != nil { return strings.TrimSpace(mdl.ModelKey) } diff --git a/pkg/aiproxy/models/list_models_test.go b/pkg/aiproxy/models/list_models_test.go index 5c0b74a897..4e921d1f8d 100644 --- a/pkg/aiproxy/models/list_models_test.go +++ b/pkg/aiproxy/models/list_models_test.go @@ -18,13 +18,17 @@ import "testing" func TestClientFacingModelID(t *testing.T) { mdl := &SAiModel{ModelKey: "gpt-4o-mini"} - if got := clientFacingModelID(&SAiRoutingModel{ModelPattern: "fast"}, mdl); got != "fast" { + routing := &SAiRouting{ModelPattern: "dep-gpt-4o-mini"} + if got := clientFacingModelID(routing, &SAiRoutingModel{ModelPattern: "fast"}, mdl); got != "fast" { t.Fatalf("expected alias fast, got %q", got) } - if got := clientFacingModelID(&SAiRoutingModel{ModelPattern: "gpt-*"}, mdl); got != "gpt-4o-mini" { - t.Fatalf("expected catalog model_key for wildcard pattern, got %q", got) + if got := clientFacingModelID(routing, &SAiRoutingModel{ModelPattern: "gpt-*"}, mdl); got != "dep-gpt-4o-mini" { + t.Fatalf("expected routing model_pattern for wildcard pattern, got %q", got) } - if got := clientFacingModelID(&SAiRoutingModel{}, mdl); got != "gpt-4o-mini" { + if got := clientFacingModelID(routing, &SAiRoutingModel{}, mdl); got != "dep-gpt-4o-mini" { + t.Fatalf("expected routing model_pattern, got %q", got) + } + if got := clientFacingModelID(nil, &SAiRoutingModel{}, mdl); got != "gpt-4o-mini" { t.Fatalf("expected catalog model_key, got %q", got) } } @@ -35,3 +39,53 @@ func TestUniqueNonEmptyStrings(t *testing.T) { t.Fatalf("unexpected dedupe result: %#v", out) } } + +func TestPickRoutingForRequestModelKeyPriority(t *testing.T) { + routings := []SAiRouting{ + {Priority: 10, ModelPattern: ""}, + {Priority: 100, ModelKey: "lzx-test-Qwen3-0.6B"}, + } + picked, err := pickRoutingForRequest(routings, "lzx-test-Qwen3-0.6B", "primary") + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if picked == nil || picked.ModelKey != "lzx-test-Qwen3-0.6B" { + t.Fatalf("expected model_key routing, got %#v", picked) + } +} + +func TestPickRoutingForRequestModelKeyBeforePattern(t *testing.T) { + routings := []SAiRouting{ + {Priority: 10, ModelPattern: "lzx-test-Qwen3-0.6B"}, + {Priority: 100, ModelKey: "lzx-test-Qwen3-0.6B"}, + } + picked, err := pickRoutingForRequest(routings, "lzx-test-Qwen3-0.6B", "primary") + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if picked == nil || picked.Priority != 100 { + t.Fatalf("expected model_key routing with priority 100, got %#v", picked) + } +} + +func TestPickRoutingForRequestFallbackPattern(t *testing.T) { + routings := []SAiRouting{ + {Priority: 20, ModelPattern: "qwen-*"}, + } + picked, err := pickRoutingForRequest(routings, "qwen-turbo", "primary") + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if picked == nil || picked.ModelPattern != "qwen-*" { + t.Fatalf("expected pattern routing, got %#v", picked) + } +} + +func TestModelKeyMatches(t *testing.T) { + if !modelKeyMatches("Foo", "foo") { + t.Fatal("expected case-insensitive match") + } + if modelKeyMatches("", "foo") { + t.Fatal("empty key should not match") + } +} diff --git a/pkg/aiproxy/models/proxy_node_local.go b/pkg/aiproxy/models/proxy_node_local.go index f3bde16b2d..1f8d0dad86 100644 --- a/pkg/aiproxy/models/proxy_node_local.go +++ b/pkg/aiproxy/models/proxy_node_local.go @@ -62,6 +62,13 @@ func validateAiProxyNodeId(ctx context.Context, userCred mcclient.TokenCredentia return node.Id, nil } +func resolveAiProxyNodeIdForCreate(ctx context.Context, userCred mcclient.TokenCredential, idOrName string) (string, error) { + if strings.TrimSpace(idOrName) == "" { + idOrName = defaultPrimaryAiProxyNodeId + } + return validateAiProxyNodeId(ctx, userCred, idOrName) +} + func proxyNodeScopeMatches(routingNodeId, currentNodeId string) bool { routingNodeId = strings.TrimSpace(routingNodeId) if routingNodeId == "" { diff --git a/pkg/apis/aiproxy/ai_provider.go b/pkg/apis/aiproxy/ai_provider.go index 9a2bd0b0e2..5bf24b9e92 100644 --- a/pkg/apis/aiproxy/ai_provider.go +++ b/pkg/apis/aiproxy/ai_provider.go @@ -66,27 +66,35 @@ func (c *SAiProviderConfig) IsZero() bool { type AiProviderListInput struct { apis.EnabledStatusStandaloneResourceListInput - ProviderKey string `json:"provider_key"` + ProviderKey string `json:"provider_key"` + LlmDeploymentId string `json:"llm_deployment_id"` + LlmId string `json:"llm_id"` } type AiProviderCreateInput struct { apis.EnabledStatusStandaloneResourceCreateInput - ProviderKey string `json:"provider_key"` - Config *SAiProviderConfig `json:"config"` + ProviderKey string `json:"provider_key"` + Config *SAiProviderConfig `json:"config"` + LlmDeploymentId string `json:"llm_deployment_id"` + LlmId string `json:"llm_id"` } type AiProviderUpdateInput struct { apis.EnabledStatusStandaloneResourceBaseUpdateInput - ProviderKey string `json:"provider_key"` - Config *SAiProviderConfig `json:"config"` - Enabled *bool `json:"enabled"` + ProviderKey string `json:"provider_key"` + Config *SAiProviderConfig `json:"config"` + LlmDeploymentId string `json:"llm_deployment_id"` + LlmId string `json:"llm_id"` + Enabled *bool `json:"enabled"` } type AiProviderDetails struct { apis.EnabledStatusStandaloneResourceDetails - ProviderKey string `json:"provider_key"` - Config *SAiProviderConfig `json:"config"` + ProviderKey string `json:"provider_key"` + Config *SAiProviderConfig `json:"config"` + LlmDeploymentId string `json:"llm_deployment_id"` + LlmId string `json:"llm_id"` } diff --git a/pkg/apis/aiproxy/ai_proxy_node.go b/pkg/apis/aiproxy/ai_proxy_node.go index 974b0efed5..4d127edff8 100644 --- a/pkg/apis/aiproxy/ai_proxy_node.go +++ b/pkg/apis/aiproxy/ai_proxy_node.go @@ -23,35 +23,35 @@ import ( type AiProxyNodeListInput struct { apis.EnabledStatusStandaloneResourceListInput - Address string `json:"address"` - Domain string `json:"domain"` + Address string `json:"address"` + AccessAddress string `json:"access_address"` } type AiProxyNodeCreateInput struct { apis.EnabledStatusStandaloneResourceCreateInput - Address string `json:"address"` - Domain string `json:"domain"` - HbTimeout int `json:"hb_timeout"` + Address string `json:"address"` + AccessAddress string `json:"access_address"` + HbTimeout int `json:"hb_timeout"` } type AiProxyNodeUpdateInput struct { apis.EnabledStatusStandaloneResourceBaseUpdateInput - Address string `json:"address"` - Domain string `json:"domain"` - HbTimeout int `json:"hb_timeout"` - Enabled *bool `json:"enabled"` + Address string `json:"address"` + AccessAddress string `json:"access_address"` + HbTimeout int `json:"hb_timeout"` + Enabled *bool `json:"enabled"` } type AiProxyNodeDetails struct { apis.EnabledStatusStandaloneResourceDetails - Address string `json:"address"` - Domain string `json:"domain"` - LastSeen time.Time `json:"last_seen"` - HbTimeout int `json:"hb_timeout"` - IsActive bool `json:"is_active"` + Address string `json:"address"` + AccessAddress string `json:"access_address"` + LastSeen time.Time `json:"last_seen"` + HbTimeout int `json:"hb_timeout"` + IsActive bool `json:"is_active"` } // AiProxyNodeRegisterInput is sent by standby instances to the primary on startup and heartbeat. diff --git a/pkg/apis/aiproxy/ai_routing.go b/pkg/apis/aiproxy/ai_routing.go index 1bf745c32b..85916aada0 100644 --- a/pkg/apis/aiproxy/ai_routing.go +++ b/pkg/apis/aiproxy/ai_routing.go @@ -31,6 +31,7 @@ type AiRoutingListInput struct { apis.EnabledResourceBaseListInput ModelPattern string `json:"model_pattern"` + ModelKey string `json:"model_key"` AiProxyNodeId string `json:"ai_proxy_node_id"` RouterEnabled *bool `json:"router_enabled"` } @@ -51,6 +52,7 @@ type AiRoutingCreateInput struct { apis.EnabledBaseResourceCreateInput Priority int `json:"priority"` + ModelKey string `json:"model_key"` ModelPattern string `json:"model_pattern"` AiProxyNodeId string `json:"ai_proxy_node_id"` RouterEnabled bool `json:"router_enabled"` @@ -65,6 +67,7 @@ type AiRoutingUpdateInput struct { apis.SharableVirtualResourceBaseUpdateInput Priority int `json:"priority"` + ModelKey string `json:"model_key"` ModelPattern string `json:"model_pattern"` AiProxyNodeId string `json:"ai_proxy_node_id"` RouterEnabled *bool `json:"router_enabled"` @@ -79,6 +82,7 @@ type AiRoutingDetails struct { apis.SharableVirtualResourceDetails Priority int `json:"priority"` + ModelKey string `json:"model_key"` ModelPattern string `json:"model_pattern"` AiProxyNodeId string `json:"ai_proxy_node_id"` RouterEnabled bool `json:"router_enabled"` diff --git a/pkg/apis/llm/aiproxy_bindings.go b/pkg/apis/llm/aiproxy_bindings.go new file mode 100644 index 0000000000..0b5b493ca7 --- /dev/null +++ b/pkg/apis/llm/aiproxy_bindings.go @@ -0,0 +1,25 @@ +package llm + +import ( + "reflect" + + "yunion.io/x/jsonutils" + "yunion.io/x/pkg/gotypes" +) + +func init() { + gotypes.RegisterSerializable(reflect.TypeOf(&AiproxyBindings{}), func() gotypes.ISerializable { + return &AiproxyBindings{} + }) +} + +// AiproxyBindings records per-replica aiproxy catalog bindings for a deployment. +type AiproxyBindings []AiproxyInstanceBinding + +func (s AiproxyBindings) String() string { + return jsonutils.Marshal(s).String() +} + +func (s AiproxyBindings) IsZero() bool { + return len(s) == 0 +} diff --git a/pkg/apis/llm/llm_deployment.go b/pkg/apis/llm/llm_deployment.go index a101845e54..2f9fabceba 100644 --- a/pkg/apis/llm/llm_deployment.go +++ b/pkg/apis/llm/llm_deployment.go @@ -85,6 +85,12 @@ const ( LLM_DEPLOYMENT_STATUS_DEPLOYING = "deploying" // Some replicas running but not all (e.g., one died, scale-up in progress). LLM_DEPLOYMENT_STATUS_PARTIAL = "partial" + + // Aiproxy gateway sync phases (stored in deployment status, not a separate column). + LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING = "aiproxy_pending" + LLM_DEPLOYMENT_STATUS_AIPROXY_SYNCING = "aiproxy_syncing" + LLM_DEPLOYMENT_STATUS_AIPROXY_PARTIAL = "aiproxy_partial" + LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED = "aiproxy_sync_failed" ) // LLMDeploymentCreateInput is the input for creating a new LLMDeployment deployment. @@ -140,6 +146,11 @@ type LLMDeploymentCreateInput struct { SpeculativeConfig *SpeculativeDecodingConfig `json:"speculative_config"` // Access policy: public, authed, allowed_users AccessPolicy string `json:"access_policy"` + + // AutoRegisterAiproxy registers running replicas with aiproxy catalog when true (default true; pass false to disable). + AutoRegisterAiproxy *bool `json:"auto_register_aiproxy"` + // AiproxyModelPrefix is deprecated and no longer affects client model alias. + AiproxyModelPrefix string `json:"aiproxy_model_prefix"` } // LLMDeploymentUpdateInput is the input for updating an existing LLMModel. @@ -158,6 +169,8 @@ type LLMDeploymentUpdateInput struct { ExtendedKVCache *ExtendedKVCacheConfig `json:"extended_kv_cache,omitempty"` SpeculativeConfig *SpeculativeDecodingConfig `json:"speculative_config,omitempty"` AccessPolicy *string `json:"access_policy,omitempty"` + AutoRegisterAiproxy *bool `json:"auto_register_aiproxy,omitempty"` + AiproxyModelPrefix *string `json:"aiproxy_model_prefix,omitempty"` } // Model source types @@ -208,6 +221,31 @@ type LLMDeploymentDetails struct { // Computed: count of running SLLM instances RunningInstances int `json:"running_instances"` + + LLMSkuId string `json:"llm_sku_id"` + LLMSku string `json:"llm_sku"` + + AutoRegisterAiproxy bool `json:"auto_register_aiproxy"` + AiproxyModelPrefix string `json:"aiproxy_model_prefix"` + AiproxyRoutingId string `json:"aiproxy_routing_id"` +} + +// Per-replica aiproxy binding sync status (AiproxyInstanceBinding.sync_status). +const ( + AIPROXY_BINDING_SYNC_PENDING = "pending" + AIPROXY_BINDING_SYNC_SYNCED = "synced" + AIPROXY_BINDING_SYNC_FAILED = "failed" +) + +// AiproxyInstanceBinding records one llm replica registered in aiproxy. +type AiproxyInstanceBinding struct { + LlmId string `json:"llm_id"` + ClientModelAlias string `json:"client_model_alias"` + AiProviderId string `json:"ai_provider_id"` + AiProviderName string `json:"ai_provider_name"` + BaseURL string `json:"base_url"` + SyncStatus string `json:"sync_status"` + LastError string `json:"last_error,omitempty"` } // GpuSelector defines manual GPU selection for scheduling. diff --git a/pkg/llm/models/instantmodel.go b/pkg/llm/models/instantmodel.go index a523379dc8..4d73bbca34 100644 --- a/pkg/llm/models/instantmodel.go +++ b/pkg/llm/models/instantmodel.go @@ -934,6 +934,118 @@ func (man *SInstantModelManager) PerformBackfillVram( return out, nil } +const instantModelImportNameMaxLen = 128 +const instantModelImportNameSuffixLen = 4 + +func buildInstantModelProvisionalName(llmType, modelName, modelTag, suffix string) string { + llmType = strings.TrimSpace(llmType) + suffix = normalizeInstantModelImportNameSuffix(suffix) + parts := []string{llmType} + nameSlug := sanitizeInstantModelImportCacheComponent(modelName) + tagSlug := sanitizeInstantModelImportCacheComponent(modelTag) + switch { + case nameSlug != "" && tagSlug != "": + parts = append(parts, nameSlug, tagSlug) + case nameSlug != "": + parts = append(parts, nameSlug) + case tagSlug != "": + parts = append(parts, tagSlug) + default: + parts = append(parts, "model") + } + parts = append(parts, suffix) + return truncateInstantModelImportName(strings.Join(parts, "-"), instantModelImportNameMaxLen) +} + +func buildInstantModelFinalName(llmType, modelId, modelTag, suffix string) string { + llmType = strings.TrimSpace(llmType) + suffix = normalizeInstantModelImportNameSuffix(suffix) + slug := sanitizeInstantModelImportCacheComponent(modelId) + if slug == "" { + slug = "model" + } + parts := []string{llmType, slug} + if shouldAppendInstantModelImportTag(llmType, modelTag) { + tagSlug := sanitizeInstantModelImportCacheComponent(modelTag) + if tagSlug != "" { + parts = append(parts, tagSlug) + } + } + parts = append(parts, suffix) + return truncateInstantModelImportName(strings.Join(parts, "-"), instantModelImportNameMaxLen) +} + +func shouldAppendInstantModelImportTag(llmType, modelTag string) bool { + if strings.TrimSpace(modelTag) == "" { + return false + } + return apis.LLMContainerType(llmType) != apis.LLM_CONTAINER_OLLAMA +} + +func normalizeInstantModelImportNameSuffix(suffix string) string { + suffix = strings.TrimSpace(suffix) + if len(suffix) == instantModelImportNameSuffixLen && isInstantModelImportNameSuffix(suffix) { + return suffix + } + return utils.GenRequestId(instantModelImportNameSuffixLen / 2) +} + +func isInstantModelImportNameSuffix(suffix string) bool { + if len(suffix) != instantModelImportNameSuffixLen { + return false + } + for _, r := range suffix { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') { + continue + } + return false + } + return true +} + +func extractInstantModelImportNameSuffix(name string) string { + name = strings.TrimSpace(name) + idx := strings.LastIndex(name, "-") + if idx < 0 || idx >= len(name)-1 { + return "" + } + suffix := name[idx+1:] + if !isInstantModelImportNameSuffix(suffix) { + return "" + } + return suffix +} + +func truncateInstantModelImportName(name string, maxLen int) string { + if maxLen <= 0 { + maxLen = instantModelImportNameMaxLen + } + if len(name) <= maxLen { + return name + } + idx := strings.LastIndex(name, "-") + if idx <= 0 || len(name)-idx-1 != instantModelImportNameSuffixLen { + return name[:maxLen] + } + suffixPart := name[idx:] + headMax := maxLen - len(suffixPart) + if headMax <= 0 { + return name[len(name)-maxLen:] + } + return name[:headMax] + suffixPart +} + +func shouldAutoRenameInstantModelImportName(name, llmType, modelName, modelTag string) bool { + if strings.HasPrefix(name, "tmp-instant-model-") { + return true + } + suffix := extractInstantModelImportNameSuffix(name) + if suffix == "" { + return false + } + return name == buildInstantModelProvisionalName(llmType, modelName, modelTag, suffix) +} + // DoImportWithParent creates a temporary InstantModel and starts an import task, // optionally chaining it to a parent task. When parentTaskId is non-empty, the // parent task will be notified when the import task completes (via subtask @@ -947,7 +1059,7 @@ func (man *SInstantModelManager) DoImportWithParent( ) (*SInstantModel, error) { tempModel := &SInstantModel{} tempModel.SetModelManager(man, &SInstantModel{}) - tempModel.Name = fmt.Sprintf("tmp-instant-model-%s.%s", time.Now().Format("060102"), utils.GenRequestId(6)) + tempModel.Name = buildInstantModelProvisionalName(string(input.LlmType), input.ModelName, input.ModelTag, utils.GenRequestId(instantModelImportNameSuffixLen/2)) tempModel.ModelName = input.ModelName tempModel.ModelTag = input.ModelTag tempModel.LlmType = string(input.LlmType) @@ -1127,6 +1239,10 @@ func (model *SInstantModel) DoImport(ctx context.Context, userCred mcclient.Toke model.ImageId = imageId model.Mounts = mounts model.Status = imageapi.IMAGE_STATUS_SAVING + if shouldAutoRenameInstantModelImportName(model.Name, model.LlmType, input.ModelName, input.ModelTag) { + suffix := extractInstantModelImportNameSuffix(model.Name) + model.Name = buildInstantModelFinalName(model.LlmType, modelId, input.ModelTag, suffix) + } return nil }) if err != nil { diff --git a/pkg/llm/models/llm.go b/pkg/llm/models/llm.go index cccdc858e2..526dd37dcf 100644 --- a/pkg/llm/models/llm.go +++ b/pkg/llm/models/llm.go @@ -507,9 +507,21 @@ func (llm *SLLM) SetStatus(ctx context.Context, userCred mcclient.TokenCredentia log.Warningf("SLLM.SetStatus: fetch deployment %s: %s", llm.LLMDeploymentId, err) return nil } - if err := depObj.(*SLLMDeployment).SyncReadyReplicas(ctx, userCred); err != nil { + dep := depObj.(*SLLMDeployment) + if err := dep.SyncReadyReplicas(ctx, userCred); err != nil { log.Warningf("SLLM.SetStatus: SyncReadyReplicas for deployment %s: %s", llm.LLMDeploymentId, err) } + if dep.AutoRegisterAiproxy { + if status == api.LLM_STATUS_RUNNING { + if err := dep.StartAiproxySyncTask(ctx, userCred, llm.Id, ""); err != nil { + log.Warningf("SLLM.SetStatus: start aiproxy sync for llm %s: %v", llm.Id, err) + } + } else if oldStatus == api.LLM_STATUS_RUNNING && status != api.LLM_STATUS_RUNNING { + if err := UnsyncLlmInstance(ctx, userCred, dep, llm.Id); err != nil { + log.Warningf("SLLM.SetStatus: unsync aiproxy for llm %s: %v", llm.Id, err) + } + } + } return nil } diff --git a/pkg/llm/models/llm_aiproxy_sync.go b/pkg/llm/models/llm_aiproxy_sync.go new file mode 100644 index 0000000000..06af294a24 --- /dev/null +++ b/pkg/llm/models/llm_aiproxy_sync.go @@ -0,0 +1,935 @@ +package models + +import ( + "context" + "fmt" + "strings" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + + apapi "yunion.io/x/onecloud/pkg/apis/aiproxy" + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" + "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/llm/options" + "yunion.io/x/onecloud/pkg/mcclient" + "yunion.io/x/onecloud/pkg/mcclient/auth" + apmodules "yunion.io/x/onecloud/pkg/mcclient/modules/aiproxy" + "yunion.io/x/pkg/util/printutils" +) + +const aiproxyPlaceholderAPIKey = "unused" + +func aiproxyAdminSession(ctx context.Context) *mcclient.ClientSession { + return auth.GetAdminSession(ctx, options.Options.Region) +} + +func mapLLMTypeToProviderKey(llmType string) (string, bool) { + switch strings.ToLower(strings.TrimSpace(llmType)) { + case string(api.LLM_CONTAINER_VLLM): + return "vllm", true + case string(api.LLM_CONTAINER_OLLAMA): + return "ollama", true + case string(api.LLM_CONTAINER_SGLANG): + return "sgl", true + default: + return "", false + } +} + +func slugAiproxyName(raw string) string { + var b strings.Builder + lastDash := false + for _, r := range strings.TrimSpace(raw) { + switch { + case r >= 'a' && r <= 'z', r >= '0' && r <= '9': + b.WriteRune(r) + lastDash = false + case r >= 'A' && r <= 'Z': + b.WriteRune(r + ('a' - 'A')) + lastDash = false + case r == '-', r == '_', r == '.', r == '/', r == ':': + if b.Len() > 0 && !lastDash { + b.WriteByte('-') + lastDash = true + } + } + } + return strings.Trim(b.String(), "-") +} + +func slugModelKey(modelKey string) string { + s := slugAiproxyName(modelKey) + if s == "" { + return "model" + } + return s +} + +func aiProviderNameForLlm(llm *SLLM) string { + suffix := strings.TrimSpace(llm.Name) + if suffix == "" { + suffix = llm.Id + } else { + suffix = slugAiproxyName(suffix) + if suffix == "" { + suffix = llm.Id + } + } + return "llm-" + suffix +} + +func aiRoutingNameForDeployment(dep *SLLMDeployment) string { + suffix := strings.TrimSpace(dep.Name) + if suffix == "" { + suffix = dep.Id + } else { + suffix = slugAiproxyName(suffix) + if suffix == "" { + suffix = dep.Id + } + } + return "llm-dep-" + suffix +} + +func aiModelNameForLlm(llm *SLLM, modelKey string) string { + return aiProviderNameForLlm(llm) + "-" + slugModelKey(modelKey) +} + +func deploymentClientModelAlias(dep *SLLMDeployment, modelKey string) string { + name := strings.TrimSpace(dep.Name) + if name == "" { + name = dep.Id + } + modelKey = strings.TrimSpace(modelKey) + if modelKey == "" { + return name + } + return name + "-" + modelKey +} + +func deploymentRoutingModelKey(dep *SLLMDeployment, modelKey string) string { + return deploymentClientModelAlias(dep, modelKey) +} + +func primaryUpstreamModelKeyFromBindings(ctx context.Context, userCred mcclient.TokenCredential, bindings []api.AiproxyInstanceBinding) string { + for i := range bindings { + b := &bindings[i] + if b.SyncStatus != api.AIPROXY_BINDING_SYNC_SYNCED || b.LlmId == "" { + continue + } + llmObj, err := GetLLMManager().FetchById(b.LlmId) + if err != nil { + continue + } + llm := llmObj.(*SLLM) + modelKeys, err := collectUpstreamModelKeys(ctx, userCred, llm) + if err != nil || len(modelKeys) == 0 { + continue + } + return modelKeys[0] + } + return "" +} + +func resolveLlmAccessBaseURL(ctx context.Context, userCred mcclient.TokenCredential, llm *SLLM) (string, error) { + if llm.CmpId == "" { + return "", errors.Wrap(httperrors.ErrInvalidStatus, "llm instance has no compute server") + } + info, err := llm.GetLLMAccessUrlInfo(ctx, userCred, jsonutils.NewDict()) + if err != nil { + return "", errors.Wrap(err, "GetLLMAccessUrlInfo") + } + if info == nil { + return "", errors.Wrap(httperrors.ErrInvalidStatus, "empty llm access url") + } + if u := strings.TrimSpace(info.InternalUrl); u != "" { + return strings.TrimRight(u, "/"), nil + } + if u := strings.TrimSpace(info.LoginUrl); u != "" { + return strings.TrimRight(u, "/"), nil + } + return "", errors.Wrap(httperrors.ErrInvalidStatus, "llm access url is empty") +} + +// upstreamModelKeyForBackend maps mounted instant-model metadata to the model id +// the inference backend actually serves (e.g. vLLM --served-model-name basename). +func upstreamModelKeyForBackend(llmType, modelName, modelTag string) string { + modelName = strings.TrimSpace(modelName) + modelTag = strings.TrimSpace(modelTag) + switch strings.ToLower(strings.TrimSpace(llmType)) { + case string(api.LLM_CONTAINER_VLLM), string(api.LLM_CONTAINER_SGLANG): + if modelName == "" { + return "" + } + if idx := strings.LastIndex(modelName, "/"); idx >= 0 { + return modelName[idx+1:] + } + return modelName + case string(api.LLM_CONTAINER_OLLAMA): + if modelName != "" && modelTag != "" { + return modelName + ":" + modelTag + } + return modelName + default: + if modelName != "" && modelTag != "" { + return modelName + ":" + modelTag + } + return modelName + } +} + +func upstreamModelKeyFromInstantModel(llmType string, instMdl *SInstantModel) string { + if instMdl == nil { + return "" + } + return upstreamModelKeyForBackend(llmType, instMdl.ModelName, instMdl.ModelTag) +} + +func upstreamModelKeyFromMountedInfo(llmType string, info *api.MountedModelInfo) string { + if info == nil { + return "" + } + if info.Id != "" { + if instMdl, _ := GetInstantModelManager().GetInstantModelById(info.Id); instMdl != nil { + if key := upstreamModelKeyFromInstantModel(llmType, instMdl); key != "" { + return key + } + } + } + if info.FullName != "" { + parts := strings.SplitN(strings.TrimSpace(info.FullName), ":", 2) + modelName := parts[0] + modelTag := "" + if len(parts) > 1 { + modelTag = parts[1] + } + if key := upstreamModelKeyForBackend(llmType, modelName, modelTag); key != "" { + return key + } + } + return strings.TrimSpace(info.ModelId) +} + +func collectUpstreamModelKeys(ctx context.Context, userCred mcclient.TokenCredential, llm *SLLM) ([]string, error) { + sku, err := llm.GetLLMSku(llm.LLMSkuId) + if err != nil { + return nil, errors.Wrap(err, "GetLLMSku") + } + llmType := sku.LLMType + + infos, err := llm.FetchMountedModelInfo() + if err != nil { + return nil, errors.Wrap(err, "FetchMountedModelInfo") + } + keys := make([]string, 0, len(infos)) + seen := make(map[string]struct{}, len(infos)) + for i := range infos { + key := upstreamModelKeyFromMountedInfo(llmType, &infos[i]) + if key == "" { + continue + } + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + keys = append(keys, key) + } + if len(keys) > 0 { + return keys, nil + } + for _, m := range GetEffectiveMountedModels(llm, sku) { + m = strings.TrimSpace(m) + if m == "" { + continue + } + key := "" + if instMdl, _ := GetInstantModelManager().GetInstantModelById(m); instMdl != nil { + key = upstreamModelKeyFromInstantModel(llmType, instMdl) + } + if key == "" { + key = m + } + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + keys = append(keys, key) + } + if len(keys) == 0 { + return nil, errors.Wrap(httperrors.ErrInvalidStatus, "no mounted models on llm instance") + } + return keys, nil +} + +func listAiproxyResources(session *mcclient.ClientSession, man interface { + List(*mcclient.ClientSession, jsonutils.JSONObject) (*printutils.ListResult, error) +}, filter jsonutils.JSONObject) ([]jsonutils.JSONObject, error) { + result, err := man.List(session, filter) + if err != nil { + return nil, err + } + if result == nil || len(result.Data) == 0 { + return nil, nil + } + return result.Data, nil +} + +func firstResourceID(rows []jsonutils.JSONObject) string { + if len(rows) == 0 { + return "" + } + id, _ := rows[0].GetString("id") + return strings.TrimSpace(id) +} + +func upsertAiProvider( + session *mcclient.ClientSession, + name, providerKey, baseURL, llmDeploymentId, llmId string, +) (string, error) { + filter := jsonutils.NewDict() + if llmId != "" { + filter.Set("llm_id", jsonutils.NewString(llmId)) + } + rows, err := listAiproxyResources(session, &apmodules.AiProviders, filter) + if err != nil { + return "", errors.Wrap(err, "list ai_providers") + } + cfg := jsonutils.Marshal(&apapi.SAiProviderConfig{ + BaseURL: baseURL, + APIKey: aiproxyPlaceholderAPIKey, + }) + params := jsonutils.NewDict() + params.Set("provider_key", jsonutils.NewString(providerKey)) + params.Set("config", cfg) + params.Set("llm_deployment_id", jsonutils.NewString(llmDeploymentId)) + params.Set("llm_id", jsonutils.NewString(llmId)) + params.Set("enabled", jsonutils.JSONTrue) + params.Set("name", jsonutils.NewString(name)) + + existingId := firstResourceID(rows) + if existingId == "" { + resp, err := apmodules.AiProviders.Create(session, params) + if err != nil { + return "", errors.Wrap(err, "create ai_provider") + } + id, _ := resp.GetString("id") + return id, nil + } + _, err = apmodules.AiProviders.Update(session, existingId, params) + if err != nil { + return "", errors.Wrap(err, "update ai_provider") + } + return existingId, nil +} + +func upsertAiModel(session *mcclient.ClientSession, name, providerId, modelKey string) (string, error) { + filter := jsonutils.NewDict() + filter.Set("ai_provider_id", jsonutils.NewString(providerId)) + filter.Set("model_key", jsonutils.NewString(modelKey)) + rows, err := listAiproxyResources(session, &apmodules.AiModels, filter) + if err != nil { + return "", errors.Wrap(err, "list ai_models") + } + params := jsonutils.NewDict() + params.Set("ai_provider_id", jsonutils.NewString(providerId)) + params.Set("model_key", jsonutils.NewString(modelKey)) + params.Set("enabled", jsonutils.JSONTrue) + params.Set("name", jsonutils.NewString(name)) + + existingId := firstResourceID(rows) + if existingId == "" { + resp, err := apmodules.AiModels.Create(session, params) + if err != nil { + return "", errors.Wrap(err, "create ai_model") + } + id, _ := resp.GetString("id") + return id, nil + } + _, err = apmodules.AiModels.Update(session, existingId, params) + if err != nil { + return "", errors.Wrap(err, "update ai_model") + } + return existingId, nil +} + +func findAiRoutingIdByName(session *mcclient.ClientSession, name string) (string, error) { + name = strings.TrimSpace(name) + if name == "" { + return "", nil + } + filter := jsonutils.NewDict() + filter.Set("name", jsonutils.NewString(name)) + rows, err := listAiproxyResources(session, &apmodules.AiRoutings, filter) + if err != nil { + return "", errors.Wrap(err, "list ai_routings by name") + } + return firstResourceID(rows), nil +} + +func ensureAiRouting( + session *mcclient.ClientSession, + name string, + dep *SLLMDeployment, + routingModelKey string, +) (string, error) { + if dep == nil { + return "", errors.Wrap(httperrors.ErrInvalidStatus, "nil deployment") + } + params := jsonutils.NewDict() + params.Set("name", jsonutils.NewString(name)) + params.Set("model_pattern", jsonutils.NewString("")) + if mk := strings.TrimSpace(routingModelKey); mk != "" { + params.Set("model_key", jsonutils.NewString(mk)) + } + + routingId := strings.TrimSpace(dep.AiproxyRoutingId) + if routingId == "" { + existingId, err := findAiRoutingIdByName(session, name) + if err != nil { + return "", err + } + routingId = existingId + } + if routingId != "" { + if _, err := apmodules.AiRoutings.Update(session, routingId, params); err != nil { + return "", errors.Wrap(err, "update ai_routing") + } + return routingId, nil + } + + params.Set("priority", jsonutils.NewInt(100)) + params.Set("enabled", jsonutils.JSONTrue) + params.Set("project_id", jsonutils.NewString(dep.ProjectId)) + params.Set("domain_id", jsonutils.NewString(dep.DomainId)) + resp, err := apmodules.AiRoutings.Create(session, params) + if err != nil { + return "", errors.Wrap(err, "create ai_routing") + } + id, _ := resp.GetString("id") + return id, nil +} + +// SyncLlmInstance registers or updates one running llm replica in aiproxy. +func SyncLlmInstance(ctx context.Context, userCred mcclient.TokenCredential, dep *SLLMDeployment, llm *SLLM) (*api.AiproxyInstanceBinding, error) { + if dep == nil || llm == nil { + return nil, errors.Wrap(httperrors.ErrInvalidStatus, "nil deployment or llm") + } + if llm.Status != api.LLM_STATUS_RUNNING { + return nil, errors.Wrapf(httperrors.ErrInvalidStatus, "llm %s is not running", llm.Id) + } + sku, err := llm.GetLLMSku(llm.LLMSkuId) + if err != nil { + return nil, errors.Wrap(err, "GetLLMSku") + } + providerKey, ok := mapLLMTypeToProviderKey(sku.LLMType) + if !ok { + return nil, errors.Wrapf(httperrors.ErrNotSupported, "llm_type %q is not supported for aiproxy sync", sku.LLMType) + } + baseURL, err := resolveLlmAccessBaseURL(ctx, userCred, llm) + if err != nil { + return nil, err + } + modelKeys, err := collectUpstreamModelKeys(ctx, userCred, llm) + if err != nil { + return nil, err + } + + session := aiproxyAdminSession(ctx) + providerName := aiProviderNameForLlm(llm) + providerId, err := upsertAiProvider(session, providerName, providerKey, baseURL, dep.Id, llm.Id) + if err != nil { + return nil, err + } + + primaryModelId := "" + for _, mk := range modelKeys { + modelName := aiModelNameForLlm(llm, mk) + modelId, err := upsertAiModel(session, modelName, providerId, mk) + if err != nil { + return nil, err + } + if primaryModelId == "" { + primaryModelId = modelId + } + } + if primaryModelId == "" { + return nil, errors.Wrap(httperrors.ErrInvalidStatus, "no ai_model created") + } + + primaryModelKey := "" + if len(modelKeys) > 0 { + primaryModelKey = modelKeys[0] + } + alias := deploymentClientModelAlias(dep, primaryModelKey) + binding := api.AiproxyInstanceBinding{ + LlmId: llm.Id, + ClientModelAlias: alias, + AiProviderId: providerId, + AiProviderName: providerName, + BaseURL: baseURL, + SyncStatus: api.AIPROXY_BINDING_SYNC_SYNCED, + } + return &binding, nil +} + +func listRunningDeploymentLlms(depId string) ([]SLLM, error) { + rows := make([]SLLM, 0, 8) + q := GetLLMManager().Query().Equals("llm_deployment_id", depId).Equals("status", api.LLM_STATUS_RUNNING) + if err := q.All(&rows); err != nil { + return nil, errors.Wrap(err, "list running llms") + } + return rows, nil +} + +func buildRoutingModelItems( + ctx context.Context, + userCred mcclient.TokenCredential, + dep *SLLMDeployment, + llms []SLLM, +) ([]apapi.AiRoutingModelItem, []api.AiproxyInstanceBinding, error) { + items := make([]apapi.AiRoutingModelItem, 0, len(llms)) + bindings := make([]api.AiproxyInstanceBinding, 0, len(llms)) + priority := 10 + for i := range llms { + llm := &llms[i] + binding, err := SyncLlmInstance(ctx, userCred, dep, llm) + if err != nil { + log.Warningf("SyncLlmInstance deployment=%s llm=%s: %v", dep.Name, llm.Id, err) + bindings = append(bindings, api.AiproxyInstanceBinding{ + LlmId: llm.Id, + SyncStatus: api.AIPROXY_BINDING_SYNC_FAILED, + LastError: err.Error(), + }) + continue + } + modelKeys, err := collectUpstreamModelKeys(ctx, userCred, llm) + if err != nil { + bindings = append(bindings, api.AiproxyInstanceBinding{ + LlmId: llm.Id, + SyncStatus: api.AIPROXY_BINDING_SYNC_FAILED, + LastError: err.Error(), + }) + continue + } + filter := jsonutils.NewDict() + filter.Set("ai_provider_id", jsonutils.NewString(binding.AiProviderId)) + filter.Set("model_key", jsonutils.NewString(modelKeys[0])) + session := aiproxyAdminSession(ctx) + rows, err := listAiproxyResources(session, &apmodules.AiModels, filter) + if err != nil { + return nil, nil, err + } + modelId := firstResourceID(rows) + if modelId == "" { + return nil, nil, errors.Wrapf(httperrors.ErrNotFound, "ai_model for llm %s", llm.Id) + } + enabled := true + items = append(items, apapi.AiRoutingModelItem{ + AiProviderId: binding.AiProviderId, + AiModelId: modelId, + Priority: priority, + Enabled: &enabled, + }) + priority += 10 + bindings = append(bindings, *binding) + } + return items, bindings, nil +} + +func applyRoutingModels(session *mcclient.ClientSession, routingId string, items []apapi.AiRoutingModelItem) error { + params := jsonutils.NewDict() + params.Set("models", jsonutils.Marshal(items)) + _, err := apmodules.AiRoutings.PerformAction(session, routingId, "set-models", params) + if err != nil { + return errors.Wrap(err, "ai_routing set-models") + } + return nil +} + +func persistDeploymentAiproxyBindings(dep *SLLMDeployment, routingId string, bindings []api.AiproxyInstanceBinding) error { + var stored *api.AiproxyBindings + if len(bindings) > 0 { + copied := api.AiproxyBindings(bindings) + stored = &copied + } + _, err := db.Update(dep, func() error { + dep.AiproxyRoutingId = routingId + dep.AiproxyBindings = stored + return nil + }) + return err +} + +func deploymentAiproxyBindings(dep *SLLMDeployment) api.AiproxyBindings { + if dep == nil || dep.AiproxyBindings == nil { + return nil + } + return *dep.AiproxyBindings +} + +// aiproxyBindingSyncResult summarizes per-replica binding outcomes (internal, not deployment status). +type aiproxyBindingSyncResult string + +const ( + aiproxyBindingSyncPending aiproxyBindingSyncResult = "pending" + aiproxyBindingSyncSynced aiproxyBindingSyncResult = "synced" + aiproxyBindingSyncPartial aiproxyBindingSyncResult = "partial" + aiproxyBindingSyncFailed aiproxyBindingSyncResult = "failed" +) + +func computeAiproxyBindingSyncResult(bindings []api.AiproxyInstanceBinding, running int) aiproxyBindingSyncResult { + if len(bindings) == 0 { + return aiproxyBindingSyncPending + } + synced := 0 + failed := 0 + for i := range bindings { + switch bindings[i].SyncStatus { + case api.AIPROXY_BINDING_SYNC_SYNCED: + synced++ + case api.AIPROXY_BINDING_SYNC_FAILED: + failed++ + } + } + if failed > 0 && synced == 0 { + return aiproxyBindingSyncFailed + } + if synced < running || failed > 0 { + return aiproxyBindingSyncPartial + } + return aiproxyBindingSyncSynced +} + +func resolveDeploymentStatusAfterAiproxySync(dep *SLLMDeployment, result aiproxyBindingSyncResult) string { + switch result { + case aiproxyBindingSyncSynced: + if dep.Replicas > 0 && dep.ReadyReplicas >= dep.Replicas { + return api.STATUS_READY + } + if dep.ReadyReplicas > 0 { + return api.LLM_DEPLOYMENT_STATUS_PARTIAL + } + return api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING + case aiproxyBindingSyncPartial: + return api.LLM_DEPLOYMENT_STATUS_AIPROXY_PARTIAL + case aiproxyBindingSyncFailed: + return api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED + default: + return api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING + } +} + +func deploymentStatusMessageAfterAiproxySync(dep *SLLMDeployment, result aiproxyBindingSyncResult) string { + switch result { + case aiproxyBindingSyncSynced: + return fmt.Sprintf("aiproxy synced, ready_replicas=%d/%d", dep.ReadyReplicas, dep.Replicas) + case aiproxyBindingSyncPartial: + return fmt.Sprintf("aiproxy partially synced, ready_replicas=%d/%d", dep.ReadyReplicas, dep.Replicas) + case aiproxyBindingSyncFailed: + return "aiproxy sync failed" + default: + return "waiting for running replicas" + } +} + +// ReconcileDeploymentAiproxy syncs all running replicas and refreshes routing bindings. +func ReconcileDeploymentAiproxy(ctx context.Context, userCred mcclient.TokenCredential, dep *SLLMDeployment) error { + if dep == nil { + return errors.Wrap(httperrors.ErrInvalidStatus, "nil deployment") + } + if !dep.AutoRegisterAiproxy { + return nil + } + + if err := dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNCING, "aiproxy sync in progress"); err != nil { + return errors.Wrap(err, "set aiproxy syncing status") + } + + llms, err := listRunningDeploymentLlms(dep.Id) + if err != nil { + return err + } + + if len(llms) == 0 { + if err := persistDeploymentAiproxyBindings(dep, dep.AiproxyRoutingId, nil); err != nil { + return err + } + return dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING, "waiting for running replicas") + } + + session := aiproxyAdminSession(ctx) + routingName := aiRoutingNameForDeployment(dep) + + items, bindings, err := buildRoutingModelItems(ctx, userCred, dep, llms) + if err != nil { + _ = persistDeploymentAiproxyBindings(dep, dep.AiproxyRoutingId, bindings) + _ = dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, err.Error()) + return err + } + primaryModelKey := primaryUpstreamModelKeyFromBindings(ctx, userCred, bindings) + routingId, err := ensureAiRouting(session, routingName, dep, deploymentRoutingModelKey(dep, primaryModelKey)) + if err != nil { + _ = persistDeploymentAiproxyBindings(dep, dep.AiproxyRoutingId, bindings) + _ = dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, err.Error()) + return err + } + if err := applyRoutingModels(session, routingId, items); err != nil { + _ = persistDeploymentAiproxyBindings(dep, routingId, bindings) + _ = dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, err.Error()) + return err + } + + result := computeAiproxyBindingSyncResult(bindings, len(llms)) + if err := persistDeploymentAiproxyBindings(dep, routingId, bindings); err != nil { + return err + } + status := resolveDeploymentStatusAfterAiproxySync(dep, result) + return dep.SetStatus(ctx, userCred, status, deploymentStatusMessageAfterAiproxySync(dep, result)) +} + +func deleteAiProviderById(session *mcclient.ClientSession, providerId string) error { + providerId = strings.TrimSpace(providerId) + if providerId == "" { + return nil + } + filterModels := jsonutils.NewDict() + filterModels.Set("ai_provider_id", jsonutils.NewString(providerId)) + modelRows, err := listAiproxyResources(session, &apmodules.AiModels, filterModels) + if err != nil { + return errors.Wrap(err, "list ai_models for delete") + } + for _, mrow := range modelRows { + mid, _ := mrow.GetString("id") + if mid != "" { + if _, err := apmodules.AiModels.Delete(session, mid, nil); err != nil { + log.Warningf("delete ai_model %s: %v", mid, err) + } + } + } + if _, err := apmodules.AiProviders.Delete(session, providerId, nil); err != nil { + return errors.Wrapf(err, "delete ai_provider %s", providerId) + } + return nil +} + +func deleteAiProviderByLlmId(session *mcclient.ClientSession, llmId string) error { + filter := jsonutils.NewDict() + filter.Set("llm_id", jsonutils.NewString(llmId)) + rows, err := listAiproxyResources(session, &apmodules.AiProviders, filter) + if err != nil { + return errors.Wrap(err, "list ai_providers for delete") + } + for _, row := range rows { + id, _ := row.GetString("id") + if err := deleteAiProviderById(session, id); err != nil { + return err + } + } + return nil +} + +func deleteAiRoutingById(session *mcclient.ClientSession, routingId string) error { + routingId = strings.TrimSpace(routingId) + if routingId == "" { + return nil + } + if _, err := apmodules.AiRoutings.Delete(session, routingId, nil); err != nil { + return errors.Wrapf(err, "delete ai_routing %s", routingId) + } + return nil +} + +// DeleteDeploymentAiproxyResources removes all aiproxy catalog/routing rows linked to a deployment. +func DeleteDeploymentAiproxyResources(ctx context.Context, deploymentId string) error { + deploymentId = strings.TrimSpace(deploymentId) + if deploymentId == "" { + return nil + } + session := aiproxyAdminSession(ctx) + + filter := jsonutils.NewDict() + filter.Set("llm_deployment_id", jsonutils.NewString(deploymentId)) + provRows, err := listAiproxyResources(session, &apmodules.AiProviders, filter) + if err != nil { + return errors.Wrap(err, "list ai_providers by llm_deployment_id") + } + for _, row := range provRows { + id, _ := row.GetString("id") + if err := deleteAiProviderById(session, id); err != nil { + log.Warningf("delete ai_provider %s for deployment %s: %v", id, deploymentId, err) + } + } + + llmRows := make([]SLLM, 0, 8) + q := GetLLMManager().Query("id").Equals("llm_deployment_id", deploymentId) + if err := q.All(&llmRows); err != nil { + return errors.Wrap(err, "list llms for aiproxy cleanup") + } + for i := range llmRows { + if err := deleteAiProviderByLlmId(session, llmRows[i].Id); err != nil { + log.Warningf("delete ai_provider for llm %s: %v", llmRows[i].Id, err) + } + } + + depObj, err := GetLLMDeploymentManager().FetchById(deploymentId) + if err != nil { + return errors.Wrap(err, "fetch llm_deployment for aiproxy cleanup") + } + dep := depObj.(*SLLMDeployment) + if err := deleteAiRoutingById(session, dep.AiproxyRoutingId); err != nil { + log.Warningf("delete ai_routing for deployment %s: %v", deploymentId, err) + } + return nil +} + +// UnsyncLlmInstance removes aiproxy resources for one llm replica. +func UnsyncLlmInstance(ctx context.Context, userCred mcclient.TokenCredential, dep *SLLMDeployment, llmId string) error { + if dep == nil || strings.TrimSpace(llmId) == "" { + return nil + } + session := aiproxyAdminSession(ctx) + if err := deleteAiProviderByLlmId(session, llmId); err != nil { + return err + } + + llms, err := listRunningDeploymentLlms(dep.Id) + if err != nil { + return err + } + remaining := make([]SLLM, 0, len(llms)) + for i := range llms { + if llms[i].Id != llmId { + remaining = append(remaining, llms[i]) + } + } + + routingId := strings.TrimSpace(dep.AiproxyRoutingId) + if routingId == "" { + if err := persistDeploymentAiproxyBindings(dep, "", nil); err != nil { + return err + } + return dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING, "waiting for running replicas") + } + + if len(remaining) == 0 { + if _, err := apmodules.AiRoutings.Delete(session, routingId, nil); err != nil { + log.Warningf("delete ai_routing %s: %v", routingId, err) + } + if err := persistDeploymentAiproxyBindings(dep, "", nil); err != nil { + return err + } + return dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING, "waiting for running replicas") + } + + items := make([]apapi.AiRoutingModelItem, 0, len(remaining)) + bindings := make([]api.AiproxyInstanceBinding, 0, len(remaining)) + priority := 10 + for i := range remaining { + llm := &remaining[i] + b, err := parseBindingForLlm(dep, llm.Id) + if err != nil { + continue + } + modelKeys, err := collectUpstreamModelKeys(ctx, userCred, llm) + if err != nil { + continue + } + filter := jsonutils.NewDict() + filter.Set("ai_provider_id", jsonutils.NewString(b.AiProviderId)) + filter.Set("model_key", jsonutils.NewString(modelKeys[0])) + rows, err := listAiproxyResources(session, &apmodules.AiModels, filter) + if err != nil { + return err + } + modelId := firstResourceID(rows) + if modelId == "" { + continue + } + enabled := true + items = append(items, apapi.AiRoutingModelItem{ + AiProviderId: b.AiProviderId, + AiModelId: modelId, + Priority: priority, + Enabled: &enabled, + }) + priority += 10 + bindings = append(bindings, b) + } + if err := applyRoutingModels(session, routingId, items); err != nil { + return err + } + primaryModelKey := primaryUpstreamModelKeyFromBindings(ctx, userCred, bindings) + if _, err := ensureAiRouting(session, aiRoutingNameForDeployment(dep), dep, deploymentRoutingModelKey(dep, primaryModelKey)); err != nil { + return err + } + result := computeAiproxyBindingSyncResult(bindings, len(remaining)) + if err := persistDeploymentAiproxyBindings(dep, routingId, bindings); err != nil { + return err + } + status := resolveDeploymentStatusAfterAiproxySync(dep, result) + return dep.SetStatus(ctx, userCred, status, deploymentStatusMessageAfterAiproxySync(dep, result)) +} + +func parseBindingForLlm(dep *SLLMDeployment, llmId string) (api.AiproxyInstanceBinding, error) { + for _, b := range deploymentAiproxyBindings(dep) { + if b.LlmId == llmId && b.SyncStatus == api.AIPROXY_BINDING_SYNC_SYNCED { + return b, nil + } + } + return api.AiproxyInstanceBinding{}, errors.Wrapf(httperrors.ErrNotFound, "binding for llm %s", llmId) +} + +// UnregisterDeploymentAiproxy removes all aiproxy resources linked to a deployment. +func UnregisterDeploymentAiproxy(ctx context.Context, userCred mcclient.TokenCredential, dep *SLLMDeployment) error { + if dep == nil { + return nil + } + if err := DeleteDeploymentAiproxyResources(ctx, dep.Id); err != nil { + return err + } + + _, err := db.Update(dep, func() error { + clearDeploymentAiproxyRegistrationState(dep) + return nil + }) + if err != nil { + return err + } + return dep.SyncReadyReplicas(ctx, userCred) +} + +func clearDeploymentAiproxyRegistrationState(dep *SLLMDeployment) { + if dep == nil { + return + } + dep.AutoRegisterAiproxy = false + dep.AiproxyRoutingId = "" + dep.AiproxyBindings = nil +} + +func (dep *SLLMDeployment) StartAiproxySyncTask(ctx context.Context, userCred mcclient.TokenCredential, llmId, parentTaskId string) error { + if dep.Status == api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNCING && parentTaskId == "" { + return nil + } + params := jsonutils.NewDict() + if llmId != "" { + params.Set("llm_id", jsonutils.NewString(llmId)) + } + return dep.startAiproxySyncTaskWithParams(ctx, userCred, params, parentTaskId) +} + +func (dep *SLLMDeployment) startAiproxySyncTaskWithParams(ctx context.Context, userCred mcclient.TokenCredential, params jsonutils.JSONObject, parentTaskId string) error { + pdict, _ := params.(*jsonutils.JSONDict) + if pdict == nil { + pdict = jsonutils.NewDict() + } + task, err := taskman.TaskManager.NewTask(ctx, "LLMAiproxySyncTask", dep, userCred, pdict, parentTaskId, "", nil) + if err != nil { + return errors.Wrap(err, "NewTask LLMAiproxySyncTask") + } + return task.ScheduleRun(nil) +} diff --git a/pkg/llm/models/llm_aiproxy_sync_test.go b/pkg/llm/models/llm_aiproxy_sync_test.go new file mode 100644 index 0000000000..b95dfe66bc --- /dev/null +++ b/pkg/llm/models/llm_aiproxy_sync_test.go @@ -0,0 +1,166 @@ +package models + +import ( + "testing" + + api "yunion.io/x/onecloud/pkg/apis/llm" +) + +func TestMapLLMTypeToProviderKey(t *testing.T) { + cases := []struct { + in string + key string + ok bool + }{ + {string(api.LLM_CONTAINER_VLLM), "vllm", true}, + {string(api.LLM_CONTAINER_OLLAMA), "ollama", true}, + {string(api.LLM_CONTAINER_SGLANG), "sgl", true}, + {"dify", "", false}, + } + for _, c := range cases { + key, ok := mapLLMTypeToProviderKey(c.in) + if ok != c.ok || key != c.key { + t.Fatalf("mapLLMTypeToProviderKey(%q) = (%q, %v), want (%q, %v)", c.in, key, ok, c.key, c.ok) + } + } +} + +func TestSlugModelKey(t *testing.T) { + if got := slugModelKey("Qwen/Qwen2.5-7B-Instruct"); got != "qwen-qwen2-5-7b-instruct" { + t.Fatalf("slugModelKey got %q", got) + } +} + +func TestDeploymentClientModelAlias(t *testing.T) { + dep := &SLLMDeployment{} + dep.Name = "my-qwen" + if got := deploymentClientModelAlias(dep, "Qwen3-0.6B"); got != "my-qwen-Qwen3-0.6B" { + t.Fatalf("deploymentClientModelAlias got %q", got) + } + depEmpty := &SLLMDeployment{} + depEmpty.Id = "dep-id-1" + if got := deploymentClientModelAlias(depEmpty, ""); got != "dep-id-1" { + t.Fatalf("deploymentClientModelAlias without model_key got %q", got) + } +} + +func TestDeploymentRoutingModelKey(t *testing.T) { + dep := &SLLMDeployment{} + dep.Name = "my-qwen" + if got := deploymentRoutingModelKey(dep, "Qwen3-0.6B"); got != "my-qwen-Qwen3-0.6B" { + t.Fatalf("deploymentRoutingModelKey got %q", got) + } +} + +func TestAiproxyResourceNames(t *testing.T) { + dep := &SLLMDeployment{} + dep.Name = "My-Qwen" + dep.Id = "dep-id-1" + if got := aiRoutingNameForDeployment(dep); got != "llm-dep-my-qwen" { + t.Fatalf("aiRoutingNameForDeployment got %q", got) + } + + depEmpty := &SLLMDeployment{} + depEmpty.Id = "dep-id-2" + if got := aiRoutingNameForDeployment(depEmpty); got != "llm-dep-dep-id-2" { + t.Fatalf("aiRoutingNameForDeployment empty name got %q", got) + } + + llm := &SLLM{} + llm.Name = "my-qwen-0" + llm.Id = "llm-id-1" + if got := aiProviderNameForLlm(llm); got != "llm-my-qwen-0" { + t.Fatalf("aiProviderNameForLlm got %q", got) + } + + llmEmpty := &SLLM{} + llmEmpty.Id = "llm-id-2" + if got := aiProviderNameForLlm(llmEmpty); got != "llm-llm-id-2" { + t.Fatalf("aiProviderNameForLlm empty name got %q", got) + } + + if got := aiModelNameForLlm(llm, "Qwen/Qwen3-0.6B"); got != "llm-my-qwen-0-qwen-qwen3-0-6b" { + t.Fatalf("aiModelNameForLlm got %q", got) + } +} + +func TestClearDeploymentAiproxyRegistrationState(t *testing.T) { + dep := &SLLMDeployment{} + dep.AutoRegisterAiproxy = true + dep.AiproxyRoutingId = "routing-1" + dep.AiproxyBindings = &api.AiproxyBindings{{LlmId: "llm-1"}} + + clearDeploymentAiproxyRegistrationState(dep) + + if dep.AutoRegisterAiproxy { + t.Fatal("AutoRegisterAiproxy should be false") + } + if dep.AiproxyRoutingId != "" { + t.Fatalf("AiproxyRoutingId should be empty, got %q", dep.AiproxyRoutingId) + } + if dep.AiproxyBindings != nil { + t.Fatal("AiproxyBindings should be nil") + } +} + +func TestResolveDeploymentStatusAfterAiproxySync(t *testing.T) { + cases := []struct { + name string + dep SLLMDeployment + result aiproxyBindingSyncResult + wantStat string + }{ + { + name: "fully synced all replicas", + dep: SLLMDeployment{Replicas: 2, ReadyReplicas: 2}, + result: aiproxyBindingSyncSynced, + wantStat: api.STATUS_READY, + }, + { + name: "synced partial replicas", + dep: SLLMDeployment{Replicas: 2, ReadyReplicas: 1}, + result: aiproxyBindingSyncSynced, + wantStat: api.LLM_DEPLOYMENT_STATUS_PARTIAL, + }, + { + name: "binding partial failure", + dep: SLLMDeployment{Replicas: 2, ReadyReplicas: 2}, + result: aiproxyBindingSyncPartial, + wantStat: api.LLM_DEPLOYMENT_STATUS_AIPROXY_PARTIAL, + }, + { + name: "all bindings failed", + dep: SLLMDeployment{Replicas: 1, ReadyReplicas: 1}, + result: aiproxyBindingSyncFailed, + wantStat: api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, + }, + } + for _, c := range cases { + got := resolveDeploymentStatusAfterAiproxySync(&c.dep, c.result) + if got != c.wantStat { + t.Fatalf("%s: got %q want %q", c.name, got, c.wantStat) + } + } +} + +func TestUpstreamModelKeyForBackend(t *testing.T) { + cases := []struct { + llmType string + modelName string + modelTag string + want string + }{ + {string(api.LLM_CONTAINER_VLLM), "Qwen/Qwen3-0.6B", "main", "Qwen3-0.6B"}, + {string(api.LLM_CONTAINER_SGLANG), "Qwen/Qwen2.5-7B-Instruct", "main", "Qwen2.5-7B-Instruct"}, + {string(api.LLM_CONTAINER_VLLM), "Qwen3-0.6B", "main", "Qwen3-0.6B"}, + {string(api.LLM_CONTAINER_OLLAMA), "qwen3", "8b", "qwen3:8b"}, + {string(api.LLM_CONTAINER_OLLAMA), "qwen3", "", "qwen3"}, + } + for _, c := range cases { + got := upstreamModelKeyForBackend(c.llmType, c.modelName, c.modelTag) + if got != c.want { + t.Fatalf("upstreamModelKeyForBackend(%q, %q, %q) = %q, want %q", + c.llmType, c.modelName, c.modelTag, got, c.want) + } + } +} diff --git a/pkg/llm/models/llm_deployment.go b/pkg/llm/models/llm_deployment.go index 2616cf18fd..c492554569 100644 --- a/pkg/llm/models/llm_deployment.go +++ b/pkg/llm/models/llm_deployment.go @@ -102,6 +102,12 @@ type SLLMDeployment struct { Nets *api.LLMDeploymentNets `charset:"utf8" length:"medium" nullable:"true" list:"user"` AutoStart bool `nullable:"false" default:"false" list:"user"` HostPaths *api.HostPaths `charset:"utf8" length:"medium" nullable:"true" list:"user"` + + // Aiproxy integration (llm sync writes aiproxy catalog via mcclient). + AutoRegisterAiproxy bool `nullable:"false" default:"false" list:"user" create:"optional" update:"user"` + AiproxyModelPrefix string `width:"128" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` + AiproxyRoutingId string `width:"128" charset:"ascii" nullable:"true" list:"user"` + AiproxyBindings *api.AiproxyBindings `charset:"utf8" length:"long" nullable:"true" list:"user"` } func (man *SLLMDeploymentManager) ValidateCreateData( @@ -194,6 +200,12 @@ func (man *SLLMDeploymentManager) ValidateCreateData( input.Replicas = 1 } + if input.AutoRegisterAiproxy == nil { + t := true + input.AutoRegisterAiproxy = &t + } + input.AiproxyModelPrefix = strings.TrimSpace(input.AiproxyModelPrefix) + input.Status = api.STATUS_READY return input, nil } @@ -227,6 +239,9 @@ func (model *SLLMDeployment) ValidateUpdateData( if input.Replicas != nil && *input.Replicas < 0 { return input, errors.Wrap(httperrors.ErrInputParameter, "replicas must be >= 0") } + if input.AiproxyModelPrefix != nil { + *input.AiproxyModelPrefix = strings.TrimSpace(*input.AiproxyModelPrefix) + } if input.GpuMemoryUtilization != nil || input.AutoGpuMemoryUtilization != nil { llmType := "" @@ -301,6 +316,9 @@ func (man *SLLMDeploymentManager) FetchCustomizeColumns( res[i].AutoGpuMemoryUtilization = models[i].AutoGpuMemoryUtilization res[i].RestartOnError = models[i].RestartOnError res[i].AccessPolicy = models[i].AccessPolicy + res[i].AutoRegisterAiproxy = models[i].AutoRegisterAiproxy + res[i].AiproxyModelPrefix = models[i].AiproxyModelPrefix + res[i].AiproxyRoutingId = models[i].AiproxyRoutingId } // Batch fetch SKU data for source/backend/categories info @@ -311,7 +329,9 @@ func (man *SLLMDeploymentManager) FetchCustomizeColumns( skuMap := make(map[string]SLLMSku) if err := db.FetchModelObjectsByIds(GetLLMSkuManager(), "id", skuIds, &skuMap); err == nil { for i, m := range models { + res[i].LLMSkuId = m.LLMSkuId if sku, ok := skuMap[m.LLMSkuId]; ok { + res[i].LLMSku = sku.Name res[i].Source = sku.Source res[i].HuggingfaceRepoId = sku.HuggingfaceRepoId res[i].HuggingfaceFilename = sku.HuggingfaceFilename @@ -378,6 +398,12 @@ func (model *SLLMDeployment) RealDelete(ctx context.Context, userCred mcclient.T return model.SVirtualResourceBase.Delete(ctx, userCred) } +// SyncReadyReplicasOptions configures SyncReadyReplicas behavior. +type SyncReadyReplicasOptions struct { + // SkipAiproxySync avoids scheduling LLMAiproxySyncTask (e.g. create task uses a child sync instead). + SkipAiproxySync bool +} + // SyncReadyReplicas recomputes ReadyReplicas from SLLM instances, persists it // to the deployment row, and transitions the deployment status based on replica // health: @@ -391,7 +417,11 @@ func (model *SLLMDeployment) RealDelete(ctx context.Context, userCred mcclient.T // are not overridden once set. // // Call after create/scale tasks finish and on every instance status change. -func (model *SLLMDeployment) SyncReadyReplicas(ctx context.Context, userCred mcclient.TokenCredential) error { +func (model *SLLMDeployment) SyncReadyReplicas(ctx context.Context, userCred mcclient.TokenCredential, opts ...SyncReadyReplicasOptions) error { + skipAiproxySync := false + if len(opts) > 0 { + skipAiproxySync = opts[0].SkipAiproxySync + } var rows []deploymentReplicaStatusRow err := GetLLMManager().Query("status"). Equals("llm_deployment_id", model.Id). @@ -416,10 +446,28 @@ func (model *SLLMDeployment) SyncReadyReplicas(ctx context.Context, userCred mcc if !canUpdateReplicaHealthStatus(model.Status) { return nil } + oldStatus := model.Status + if err := model.transitionReplicaHealthStatus(ctx, userCred, desired, summary.Running); err != nil { + return err + } + if !skipAiproxySync && model.AutoRegisterAiproxy && (desired == api.STATUS_READY || desired == api.LLM_DEPLOYMENT_STATUS_PARTIAL) { + if oldStatus != desired || desired == api.STATUS_READY { + if err := model.StartAiproxySyncTask(ctx, userCred, "", ""); err != nil { + log.Warningf("SyncReadyReplicas: start aiproxy sync for %s: %v", model.Name, err) + } + } + } + return nil +} + +func (model *SLLMDeployment) transitionReplicaHealthStatus(ctx context.Context, userCred mcclient.TokenCredential, desired string, running int) error { + if desired == "" { + return nil + } if model.Status == desired { return nil } - return model.SetStatus(ctx, userCred, desired, fmt.Sprintf("ready_replicas=%d/%d", summary.Running, model.Replicas)) + return model.SetStatus(ctx, userCred, desired, fmt.Sprintf("ready_replicas=%d/%d", running, model.Replicas)) } type deploymentReplicaStatusRow struct { @@ -482,7 +530,8 @@ func canUpdateReplicaHealthStatus(current string) bool { api.LLM_DEPLOYMENT_STATUS_CREATE_SKU_FAILED, api.LLM_STATUS_CREATE_FAIL, api.LLM_STATUS_DELETING, - api.LLM_STATUS_DELETE_FAILED: + api.LLM_STATUS_DELETE_FAILED, + api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNCING: return false } return true @@ -540,6 +589,12 @@ func (model *SLLMDeployment) PostCreate(ctx context.Context, userCred mcclient.T if input.HostPaths != nil && !input.HostPaths.IsZero() { model.HostPaths = input.HostPaths } + if input.AutoRegisterAiproxy != nil { + model.AutoRegisterAiproxy = *input.AutoRegisterAiproxy + } + if input.AiproxyModelPrefix != "" { + model.AiproxyModelPrefix = strings.TrimSpace(input.AiproxyModelPrefix) + } return nil }); err != nil { log.Errorf("SLLMDeployment.PostCreate persist instance template: %s", err) @@ -586,12 +641,52 @@ func (model *SLLMDeployment) PostUpdate(ctx context.Context, userCred mcclient.T log.Errorf("SLLMDeployment.PostUpdate start sync replicas task failed: %s", err) } } + if model.AutoRegisterAiproxy && (data.Contains("auto_register_aiproxy") || data.Contains("aiproxy_model_prefix")) { + if err := model.StartAiproxySyncTask(ctx, userCred, "", ""); err != nil { + log.Errorf("SLLMDeployment.PostUpdate start aiproxy sync task failed: %s", err) + } + } } // CustomizeDelete starts a cascade delete task that: // 1. Deletes all child SLLM instances (each via its own LLMDeleteTask, with this task as parent) // 2. After all instance delete tasks complete, deletes the deployment record itself +func (model *SLLMDeployment) PerformRegisterAiproxy( + ctx context.Context, + userCred mcclient.TokenCredential, + query jsonutils.JSONObject, + input jsonutils.JSONObject, +) (jsonutils.JSONObject, error) { + if _, err := db.Update(model, func() error { + model.AutoRegisterAiproxy = true + return nil + }); err != nil { + return nil, errors.Wrap(err, "enable auto_register_aiproxy") + } + if err := model.StartAiproxySyncTask(ctx, userCred, "", ""); err != nil { + return nil, err + } + return nil, nil +} + +func (model *SLLMDeployment) PerformUnregisterAiproxy( + ctx context.Context, + userCred mcclient.TokenCredential, + query jsonutils.JSONObject, + input jsonutils.JSONObject, +) (jsonutils.JSONObject, error) { + params := jsonutils.NewDict() + params.Set("unregister", jsonutils.JSONTrue) + if err := model.startAiproxySyncTaskWithParams(ctx, userCred, params, ""); err != nil { + return nil, err + } + return nil, nil +} + func (model *SLLMDeployment) CustomizeDelete(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) error { + if err := DeleteDeploymentAiproxyResources(ctx, model.Id); err != nil { + log.Warningf("CustomizeDelete: delete aiproxy resources for %s: %v", model.Name, err) + } // Set replicas to 0 to disable self-healing reconcile during teardown if _, err := db.Update(model, func() error { model.Replicas = 0 diff --git a/pkg/llm/tasks/llm/llm_aiproxy_sync_task.go b/pkg/llm/tasks/llm/llm_aiproxy_sync_task.go new file mode 100644 index 0000000000..a88c6a78cb --- /dev/null +++ b/pkg/llm/tasks/llm/llm_aiproxy_sync_task.go @@ -0,0 +1,69 @@ +package llm + +import ( + "context" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" + "yunion.io/x/onecloud/pkg/llm/models" +) + +type LLMAiproxySyncTask struct { + taskman.STask +} + +func init() { + taskman.RegisterTask(LLMAiproxySyncTask{}) +} + +func (task *LLMAiproxySyncTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { + dep := obj.(*models.SLLMDeployment) + if !dep.AutoRegisterAiproxy { + task.SetStageComplete(ctx, nil) + return + } + + llmId, _ := task.GetParams().GetString("llm_id") + unregister, _ := task.GetParams().Bool("unregister") + + var err error + if unregister { + err = models.UnregisterDeploymentAiproxy(ctx, task.UserCred, dep) + } else if llmId != "" { + llmObj, fetchErr := models.GetLLMManager().FetchById(llmId) + if fetchErr != nil { + err = fetchErr + } else { + llm := llmObj.(*models.SLLM) + if _, syncErr := models.SyncLlmInstance(ctx, task.UserCred, dep, llm); syncErr != nil { + err = syncErr + } else { + err = models.ReconcileDeploymentAiproxy(ctx, task.UserCred, dep) + } + } + } else { + err = models.ReconcileDeploymentAiproxy(ctx, task.UserCred, dep) + } + + if err != nil { + log.Errorf("LLMAiproxySyncTask deployment=%s: %v", dep.Name, err) + if !isDeploymentHealthyAfterAiproxySync(dep.Status) { + _ = dep.SetStatus(ctx, task.UserCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, err.Error()) + } + task.SetStageFailed(ctx, jsonutils.NewString(err.Error())) + return + } + task.SetStageComplete(ctx, nil) +} + +func isDeploymentHealthyAfterAiproxySync(status string) bool { + switch status { + case api.STATUS_READY, api.LLM_DEPLOYMENT_STATUS_PARTIAL: + return true + } + return false +} diff --git a/pkg/llm/tasks/llm/llm_delete_task.go b/pkg/llm/tasks/llm/llm_delete_task.go index 2a6cdcaeb3..972f3d8c2e 100644 --- a/pkg/llm/tasks/llm/llm_delete_task.go +++ b/pkg/llm/tasks/llm/llm_delete_task.go @@ -5,6 +5,7 @@ import ( "database/sql" "yunion.io/x/jsonutils" + "yunion.io/x/log" "yunion.io/x/pkg/errors" api "yunion.io/x/onecloud/pkg/apis/llm" @@ -115,6 +116,16 @@ func (task *LLMDeleteTask) OnLLMContainerDeleteComplete(ctx context.Context, llm // Capture LLMDeploymentId before deletion for self-healing reconcile llmDeploymentId := llm.LLMDeploymentId + if len(llmDeploymentId) > 0 { + if depObj, err := models.GetLLMDeploymentManager().FetchById(llmDeploymentId); err == nil { + dep := depObj.(*models.SLLMDeployment) + if dep.AutoRegisterAiproxy { + if err := models.UnsyncLlmInstance(ctx, task.UserCred, dep, llm.Id); err != nil { + log.Warningf("LLMDeleteTask: unsync aiproxy for llm %s: %v", llm.Id, err) + } + } + } + } err = llm.RealDelete(ctx, task.UserCred) if err != nil { diff --git a/pkg/llm/tasks/llm/llm_deployment_create_task.go b/pkg/llm/tasks/llm/llm_deployment_create_task.go index 9d3457e646..0e5a3e0a32 100644 --- a/pkg/llm/tasks/llm/llm_deployment_create_task.go +++ b/pkg/llm/tasks/llm/llm_deployment_create_task.go @@ -122,9 +122,28 @@ func (task *LLMDeploymentCreateTask) reconcileAndComplete(ctx context.Context, m // are already running. As each SLLM reaches running, its SetStatus override // re-runs SyncReadyReplicas, which will eventually flip us to ready. model.SetStatus(ctx, task.UserCred, api.LLM_DEPLOYMENT_STATUS_DEPLOYING, "instances created, waiting for running") - if err := model.SyncReadyReplicas(ctx, task.UserCred); err != nil { + if err := model.SyncReadyReplicas(ctx, task.UserCred, models.SyncReadyReplicasOptions{SkipAiproxySync: model.AutoRegisterAiproxy}); err != nil { log.Warningf("LLMDeploymentCreateTask: SyncReadyReplicas for %s: %s", model.Name, err) } + if model.AutoRegisterAiproxy { + task.SetStage("OnAiproxySyncComplete", nil) + if err := model.StartAiproxySyncTask(ctx, task.UserCred, "", task.GetTaskId()); err != nil { + log.Warningf("LLMDeploymentCreateTask: start aiproxy sync for %s: %v", model.Name, err) + task.SetStageComplete(ctx, nil) + } + return + } + task.SetStageComplete(ctx, nil) +} + +// OnAiproxySyncComplete is called after the child LLMAiproxySyncTask completes. +func (task *LLMDeploymentCreateTask) OnAiproxySyncComplete(ctx context.Context, model *models.SLLMDeployment, body jsonutils.JSONObject) { + task.SetStageComplete(ctx, nil) +} + +// OnAiproxySyncCompleteFailed is called if the child LLMAiproxySyncTask fails. +func (task *LLMDeploymentCreateTask) OnAiproxySyncCompleteFailed(ctx context.Context, model *models.SLLMDeployment, body jsonutils.JSONObject) { + log.Warningf("LLMDeploymentCreateTask: aiproxy sync failed for %s: %s", model.Name, body) task.SetStageComplete(ctx, nil) } diff --git a/pkg/llm/tasks/llm/llm_deployment_delete_task.go b/pkg/llm/tasks/llm/llm_deployment_delete_task.go index 1a3598126e..2dd54a68c6 100644 --- a/pkg/llm/tasks/llm/llm_deployment_delete_task.go +++ b/pkg/llm/tasks/llm/llm_deployment_delete_task.go @@ -100,6 +100,11 @@ func (task *LLMDeploymentDeleteTask) deleteDeployment(ctx context.Context, model // only SKUs automatically created for this deployment. skuId := deploymentManagedSkuIdForCascade(model) + // Safety net: ensure aiproxy catalog/routing rows are removed before the deployment row goes away. + if err := models.DeleteDeploymentAiproxyResources(ctx, model.Id); err != nil { + log.Warningf("LLMDeploymentDeleteTask: delete aiproxy resources for %s: %v", model.Name, err) + } + if err := model.RealDelete(ctx, task.UserCred); err != nil { log.Errorf("LLMDeploymentDeleteTask: RealDelete deployment %s: %s", model.Id, err) task.taskFailed(ctx, model, err) diff --git a/pkg/mcclient/options/aiproxy/resources.go b/pkg/mcclient/options/aiproxy/resources.go index a28b2f0203..c3099fc024 100644 --- a/pkg/mcclient/options/aiproxy/resources.go +++ b/pkg/mcclient/options/aiproxy/resources.go @@ -39,7 +39,9 @@ func mergeJSONStringField(params *jsonutils.JSONDict, key, raw string) error { type AiProviderListOptions struct { options.BaseListOptions - ProviderKey string `help:"filter by provider_key"` + ProviderKey string `help:"filter by provider_key"` + LlmDeploymentId string `help:"filter by llm_deployment_id" json:"llm_deployment_id"` + LlmId string `help:"filter by llm_id" json:"llm_id"` } func (o *AiProviderListOptions) Params() (jsonutils.JSONObject, error) { @@ -53,9 +55,11 @@ type AiProviderShowOptions struct { type AiProviderCreateOptions struct { options.BaseCreateOptions - ProviderKey string `help:"provider key (catalog identifier)" json:"provider_key"` - Config string `help:"provider config as JSON object string" json:"-"` - Enabled *bool `json:"enabled,omitempty"` + ProviderKey string `help:"provider key (catalog identifier)" json:"provider_key"` + Config string `help:"provider config as JSON object string" json:"-"` + LlmDeploymentId string `help:"source llm_deployment id" json:"llm_deployment_id"` + LlmId string `help:"source llm instance id" json:"llm_id"` + Enabled *bool `json:"enabled,omitempty"` } func (o *AiProviderCreateOptions) Params() (jsonutils.JSONObject, error) { @@ -68,12 +72,14 @@ func (o *AiProviderCreateOptions) Params() (jsonutils.JSONObject, error) { } type AiProviderUpdateOptions struct { - ID string `help:"ID or name" json:"-"` - Name string `json:"name,omitempty"` - Desc string `json:"description,omitempty"` - ProviderKey string `json:"provider_key,omitempty"` - Config string `help:"provider config JSON" json:"-"` - Enabled *bool `json:"enabled,omitempty"` + ID string `help:"ID or name" json:"-"` + Name string `json:"name,omitempty"` + Desc string `json:"description,omitempty"` + ProviderKey string `json:"provider_key,omitempty"` + Config string `help:"provider config JSON" json:"-"` + LlmDeploymentId string `json:"llm_deployment_id,omitempty"` + LlmId string `json:"llm_id,omitempty"` + Enabled *bool `json:"enabled,omitempty"` } func (o *AiProviderUpdateOptions) GetId() string { @@ -433,8 +439,8 @@ type AiRoutingModelDeleteOptions struct { type AiProxyNodeListOptions struct { options.BaseListOptions - Address string `help:"filter by address" json:"address"` - Domain string `help:"filter by domain" json:"domain"` + Address string `help:"filter by address" json:"address"` + AccessAddress string `help:"filter by access address" json:"access_address"` } func (o *AiProxyNodeListOptions) Params() (jsonutils.JSONObject, error) { @@ -448,10 +454,10 @@ type AiProxyNodeShowOptions struct { type AiProxyNodeCreateOptions struct { options.BaseCreateOptions - Address string `help:"reachable base URL (https://host:port or host:port)" json:"address"` - Domain string `help:"optional hostname without scheme or port" json:"domain,omitempty"` - HbTimeout int `help:"heartbeat timeout in seconds (default 120)" json:"hb_timeout,omitzero"` - Enabled *bool `help:"turn on enabled flag" json:"enabled,omitempty"` + Address string `help:"reachable base URL (https://host:port or host:port)" json:"address"` + AccessAddress string `help:"optional public access URL (https://host:port or host:port)" json:"access_address,omitempty"` + HbTimeout int `help:"heartbeat timeout in seconds (default 120)" json:"hb_timeout,omitzero"` + Enabled *bool `help:"turn on enabled flag" json:"enabled,omitempty"` } func (o *AiProxyNodeCreateOptions) Params() (jsonutils.JSONObject, error) { @@ -459,13 +465,13 @@ func (o *AiProxyNodeCreateOptions) Params() (jsonutils.JSONObject, error) { } type AiProxyNodeUpdateOptions struct { - ID string `help:"ID or name" json:"-"` - Name string `json:"name,omitempty"` - Desc string `json:"description,omitempty"` - Address string `help:"reachable base URL" json:"address,omitempty"` - Domain string `help:"hostname without scheme or port; empty string clears" json:"domain,omitempty"` - HbTimeout int `help:"heartbeat timeout in seconds" json:"hb_timeout,omitzero"` - Enabled *bool `json:"enabled,omitempty"` + ID string `help:"ID or name" json:"-"` + Name string `json:"name,omitempty"` + Desc string `json:"description,omitempty"` + Address string `help:"reachable base URL" json:"address,omitempty"` + AccessAddress string `help:"public access URL; empty string clears" json:"access_address,omitempty"` + HbTimeout int `help:"heartbeat timeout in seconds" json:"hb_timeout,omitzero"` + Enabled *bool `json:"enabled,omitempty"` } func (o *AiProxyNodeUpdateOptions) GetId() string { diff --git a/pkg/mcclient/options/llm/llm_deployment.go b/pkg/mcclient/options/llm/llm_deployment.go index e894557a09..e454905253 100644 --- a/pkg/mcclient/options/llm/llm_deployment.go +++ b/pkg/mcclient/options/llm/llm_deployment.go @@ -86,6 +86,8 @@ type LLMDeploymentCreateOptions struct { AutoGpuMemoryUtilization *bool `token:"auto-gpu-memory-utilization" help:"calculate GPU memory utilization from model VRAM and GPU memory" json:"auto_gpu_memory_utilization"` RestartOnError *bool `help:"restart on error" json:"restart_on_error"` AccessPolicy string `help:"access policy" choices:"public|authed|allowed_users" json:"access_policy"` + AutoRegisterAiproxy *bool `help:"auto register running replicas with aiproxy (default true; use --auto-register-aiproxy=false to disable)" json:"auto_register_aiproxy"` + AiproxyModelPrefix string `help:"deprecated; no longer affects aiproxy client model alias" json:"aiproxy_model_prefix"` } func (o *LLMDeploymentCreateOptions) Params() (jsonutils.JSONObject, error) { @@ -233,6 +235,8 @@ type LLMDeploymentUpdateOptions struct { GpuUtilization *float64 `token:"gpu-utilization" help:"Alias of --gpu-memory-utilization" json:"-"` AutoGpuMemoryUtilization *bool `token:"auto-gpu-memory-utilization" help:"calculate GPU memory utilization from model VRAM and GPU memory" json:"auto_gpu_memory_utilization"` AccessPolicy string `help:"access policy" json:"access_policy"` + AutoRegisterAiproxy *bool `help:"auto register running replicas with aiproxy (default true; use --auto-register-aiproxy=false to disable)" json:"auto_register_aiproxy"` + AiproxyModelPrefix *string `help:"deprecated; no longer affects aiproxy client model alias" json:"aiproxy_model_prefix"` } func (o *LLMDeploymentUpdateOptions) GetId() string { @@ -262,6 +266,30 @@ func (o *LLMDeploymentDeleteOptions) Params() (jsonutils.JSONObject, error) { return options.StructToParams(o) } +type LLMDeploymentRegisterAiproxyOptions struct { + options.BaseIdOptions +} + +func (o *LLMDeploymentRegisterAiproxyOptions) GetId() string { + return o.ID +} + +func (o *LLMDeploymentRegisterAiproxyOptions) Params() (jsonutils.JSONObject, error) { + return jsonutils.NewDict(), nil +} + +type LLMDeploymentUnregisterAiproxyOptions struct { + options.BaseIdOptions +} + +func (o *LLMDeploymentUnregisterAiproxyOptions) GetId() string { + return o.ID +} + +func (o *LLMDeploymentUnregisterAiproxyOptions) Params() (jsonutils.JSONObject, error) { + return jsonutils.NewDict(), nil +} + func applyGpuUtilizationAlias(params *jsonutils.JSONDict, gpuMemoryUtilization, gpuUtilization *float64) error { if gpuMemoryUtilization != nil && gpuUtilization != nil { return fmt.Errorf("--gpu-memory-utilization and --gpu-utilization are aliases; specify only one")