Pine
865f937f41
- compute_client.list_models(status): 拉取 compute 引擎 admin 模型 - 新增 compute_catalog.py(替代静态 pineagents_catalog): Model行→桌面端形状(id/name/group/价格/能力/actual_model), 仅 status=1 - rbac_opc /opc/compute/models|prices 改读 admin 模型; /v1/models 与之一致 - 删 pineagents_catalog.py(静态, 不再引用) - tests: compute_catalog._map 2 通过
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""算力中心模型目录：把 compute 引擎的 admin 模型（models 表，status=1）映射为桌面端可读形状。
|
||
|
||
替代原静态 pineagents_catalog：OPC 桌面端展示/使用的模型 = admin 端新增的模型（算力中心）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from . import compute_client
|
||
|
||
|
||
def _map(row: dict) -> dict:
|
||
"""Model 行 → 桌面端模型形状。"""
|
||
name = row.get("name") or row.get("model_name") or ""
|
||
return {
|
||
"id": row.get("model_name") or row.get("name") or "",
|
||
"model_name": row.get("model_name", ""),
|
||
"name": name,
|
||
"actual_model": row.get("actual_model") or row.get("model_name", ""),
|
||
"company": "",
|
||
"group": row.get("group", "default"),
|
||
"billing_unit": row.get("billing_unit", "/百万tokens"),
|
||
"input_price": row.get("input_price", 0) or 0,
|
||
"output_price": row.get("output_price", 0) or 0,
|
||
"cache_hit_price": row.get("cache_hit_price", 0) or 0,
|
||
"vision_support": bool(row.get("vision_support", False)),
|
||
"image_support": bool(row.get("image_support", False)),
|
||
"audio_support": bool(row.get("audio_support", False)),
|
||
"video_support": bool(row.get("video_support", False)),
|
||
"tags": (row.get("tags") or "").split(",") if row.get("tags") else [],
|
||
"vendor_id": row.get("vendor_id", 0),
|
||
}
|
||
|
||
|
||
async def catalog() -> list[dict]:
|
||
"""从 compute 引擎拉取 admin 模型（status=1）并映射。空/不可达 → 返回空列表。"""
|
||
try:
|
||
items = await compute_client.list_models(status=1)
|
||
except compute_client.ComputeError:
|
||
return []
|
||
return [_map(row) for row in items if (row.get("status", 1) == 1)]
|
||
|
||
|
||
async def models() -> list[dict]:
|
||
return await catalog()
|
||
|
||
|
||
async def prices() -> list[dict]:
|
||
return [
|
||
{"id": m["id"], "name": m["name"], "input": m["input_price"],
|
||
"output": m["output_price"], "cache_hit": m["cache_hit_price"],
|
||
"unit": m["billing_unit"]}
|
||
for m in await catalog()
|
||
]
|
||
|
||
|
||
async def model_ids() -> list[str]:
|
||
"""供 /v1/models 使用：admin 模型的对外 id。"""
|
||
return [m["id"] for m in await catalog() if m["id"]]
|