fanhuadesenlinnn/cloudpods
1
0
Fork 0
mirror of https://hubproxy.babadafafafafa.cn/https://github.com/yunionio/cloudpods.git synced 2026-09-20 08:03:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(llm): sync llm deployment with aiproxy catalog and routing (#25057)

Browse Source 
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.
This commit is contained in:
Zexi Li
2026-06-22 20:14:01 +08:00
committed by GitHub
parent 4d6f4452c0
commit 4a4b3f8676
28 changed files with 1959 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/climc/shell/llm/llm_deployment.go
View File

@@ -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))
}

9
docs/aiproxy/functional-test-climc.md
View File

@@ -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 <node-id>
```
@@ -124,6 +124,8 @@ climc ai-proxy-node-disable <node-id>
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}]'
```

95
docs/llm/llm-deployment-aiproxy.md Normal file
View File

@@ -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 <sku> \
--net <net> \
--replicas 2
```
关闭自动关联：
```bash
climc llm-deployment-create my-qwen \
--auto-register-aiproxy=false \
--llm-sku-id <sku> \
--net <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 <deployment_id>
climc ai-provider-list --llm-id <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":"<deployment_name>-<upstream_model_key>","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` 会同步更新上述名称。

28
pkg/aiproxy/models/ai_providers.go
View File

@@ -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
}

80
pkg/aiproxy/models/ai_proxy_nodes.go
View File

@@ -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
}

13
pkg/aiproxy/models/ai_routing_models.go
View File

@@ -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
}

34
pkg/aiproxy/models/ai_routings.go
View File

@@ -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
}

19
pkg/aiproxy/models/aiproxy_catalog_validate.go
View File

@@ -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
}

38
pkg/aiproxy/models/chat_upstream.go
View File

@@ -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")
}

54
pkg/aiproxy/models/list_models.go
View File

@@ -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)
}

62
pkg/aiproxy/models/list_models_test.go
View File

@@ -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")
}
}

7
pkg/aiproxy/models/proxy_node_local.go
View File

@@ -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 == "" {

24
pkg/apis/aiproxy/ai_provider.go
View File

@@ -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"`
}

28
pkg/apis/aiproxy/ai_proxy_node.go
View File

@@ -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.

4
pkg/apis/aiproxy/ai_routing.go
View File

@@ -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"`

25
pkg/apis/llm/aiproxy_bindings.go Normal file
View File

@@ -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
}

38
pkg/apis/llm/llm_deployment.go
View File

@@ -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.

118
pkg/llm/models/instantmodel.go
View File

@@ -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 {

14
pkg/llm/models/llm.go
View File

@@ -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
}

935
pkg/llm/models/llm_aiproxy_sync.go Normal file
View File

@@ -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)
}

166
pkg/llm/models/llm_aiproxy_sync_test.go Normal file
View File

@@ -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)
}
}
}

101
pkg/llm/models/llm_deployment.go
View File

@@ -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

69
pkg/llm/tasks/llm/llm_aiproxy_sync_task.go Normal file
View File

@@ -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
}

11
pkg/llm/tasks/llm/llm_delete_task.go
View File

@@ -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 {

21
pkg/llm/tasks/llm/llm_deployment_create_task.go
View File

@@ -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)
}

5
pkg/llm/tasks/llm/llm_deployment_delete_task.go
View File

@@ -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)

52
pkg/mcclient/options/aiproxy/resources.go
View File

@@ -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 {

28
pkg/mcclient/options/llm/llm_deployment.go
View File

@@ -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")