feat: 完成 issue #59 ⑥ 昇腾 NPU 后端适配(CANN 对接)

This commit is contained in:
2026-08-05 04:13:23 +08:00
parent 5d9a76c3ff
commit 95dace5ef5
4 changed files with 584 additions and 23 deletions
+224
View File
@@ -0,0 +1,224 @@
# -*- coding: utf-8 -*-
"""昇腾 CANN 运行时环境探测(issue #59「昇腾 NPU 后端适配(CANN 对接)」)。
目标:
- 在**无 CANN 硬件的开发/CI 环境**下安全降级(available=False,绝不抛异常);
- 在**有 CANN 的生产环境**正确上报设备数量/名称与工具链版本,
供 :class:`~inference_backend.npu_backend.AscendNpuBackend` 的健康巡检
与直连模式(本地 ACL 生命周期管理)使用。
探测优先级(取第一个可用的运行时):
1. ``torch_npu`` —— PyTorch 昇腾插件(``torch.npu.*``);
2. ``acl`` —— CANN 基础 ACL Python API(``acl.init`` / ``acl.rt.*``);
3. ``npu-smi`` —— 昇腾命令行工具(``npu-smi info`` 输出解析)。
ACL API 各 CANN 版本存在差异(部分接口返回 ``(ret, value)`` 元组、
部分直接返回值),本模块统一用 ``_unwrap`` 兼容两种风格,并逐调用
try/except 兜底 —— 探测逻辑在任何版本/异常下都不会让调用方崩溃。
"""
from __future__ import annotations
import shutil
import subprocess
from typing import Optional
#: 探测失败兜底结果(永不抛异常)
_NO_CANN = {
"available": False,
"runtime": "none",
"cann_version": "",
"device_count": 0,
"device_names": [],
"detail": "未检测到 CANN 运行时(torch_npu / acl / npu-smi 均不可用)",
}
def probe_cann_environment() -> dict:
"""探测当前环境的昇腾 CANN 运行时,返回结构化结果。
Returns:
dict,固定键:``available`` / ``runtime`` / ``cann_version`` /
``device_count`` / ``device_names`` / ``detail``。
任一探测路径失败均不抛异常,仅将 ``available`` 置为 False。
"""
for probe in (_probe_torch_npu, _probe_acl, _probe_npu_smi):
info = probe()
if info is not None:
return info
return dict(_NO_CANN)
def _unwrap(ret_value):
"""兼容 ACL API 两种返回风格:``(ret, value)`` 元组或直接返回 value。"""
if isinstance(ret_value, tuple):
if len(ret_value) >= 2:
return ret_value[1]
return None
return ret_value
def _version_of(module, name: str) -> str:
"""尽力取模块版本号(torch_npu.__version__ / acl.__version__)。"""
try:
ver = getattr(module, "__version__", "")
return str(ver) if ver else ""
except Exception:
return ""
def _probe_torch_npu() -> Optional[dict]:
"""探测 torch_npu(PyTorch 昇腾插件)路径。"""
try:
import torch # noqa: F401
import torch_npu # noqa: F401
except Exception:
return None # 未安装该运行时,交给下一个探测路径
base = {
"runtime": "torch_npu",
"cann_version": _version_of(torch_npu, "torch_npu"),
}
try:
if not torch_npu.npu.is_available():
return {
**base,
"available": False,
"device_count": 0,
"device_names": [],
"detail": "torch_npu 可导入但 NPU 不可用(未检测到昇腾设备)",
}
count = int(torch.npu.device_count())
names = []
for i in range(count):
try:
names.append(str(torch.npu.get_device_name(i)))
except Exception:
names.append("ascend-device-%d" % i)
return {
**base,
"available": True,
"device_count": count,
"device_names": names,
"detail": "torch_npu 探测成功(%d 张昇腾设备)" % count,
}
except Exception as exc: # noqa: BLE001 —— 探测必须容错
return {
**base,
"available": False,
"device_count": 0,
"device_names": [],
"detail": "torch_npu 探测失败: %s" % exc,
}
def _probe_acl() -> Optional[dict]:
"""探测 CANN ACL Python API 路径。"""
try:
import acl
except Exception:
return None # 未安装该运行时,交给下一个探测路径
base = {
"runtime": "acl",
"cann_version": _version_of(acl, "acl"),
}
try:
ret = acl.init()
# 注意:不能用 `ret not in (0, None, True)` —— Python 中 1 == True,
# 会把错误码 1 误判为成功;ACL 约定 ret=0(ACL_SUCCESS)为成功。
if not (ret is True or ret is None or ret == 0):
return {
**base,
"available": False,
"device_count": 0,
"device_names": [],
"detail": "acl.init() 返回错误码 %s" % ret,
}
try:
count = int(_unwrap(acl.rt.get_device_count()))
except Exception:
count = 0
names = []
for i in range(count):
name = None
for method in ("get_device_name", "get_soc_name"):
fn = getattr(acl.rt, method, None)
if fn is None:
continue
try:
name = _unwrap(fn(i))
break
except Exception:
name = None
names.append(str(name) if name else "ascend-device-%d" % i)
# CANN 版本:优先 acl.rt.get_version(),其次模块 __version__
try:
ver = _unwrap(acl.rt.get_version()) or base["cann_version"]
except Exception:
ver = base["cann_version"]
return {
**base,
"cann_version": str(ver) if ver else "",
"available": True,
"device_count": count,
"device_names": names,
"detail": "ACL 探测成功(%d 张昇腾设备)" % count,
}
except Exception as exc: # noqa: BLE001
return {
**base,
"available": False,
"device_count": 0,
"device_names": [],
"detail": "ACL 探测失败: %s" % exc,
}
finally:
try:
acl.finalize() # 探测后立即释放,避免占用设备
except Exception:
pass
def _probe_npu_smi() -> Optional[dict]:
"""探测 npu-smi 命令行工具路径。"""
exe = shutil.which("npu-smi")
if not exe:
return None # 未安装该工具
try:
proc = subprocess.run(
[exe, "info", "-l"],
capture_output=True, text=True, timeout=10,
check=False,
)
text = proc.stdout + "\n" + proc.stderr
except Exception as exc:
return {
"available": False,
"runtime": "npu-smi",
"cann_version": "",
"device_count": 0,
"device_names": [],
"detail": "npu-smi 执行失败: %s" % exc,
}
# 解析 "Device Count : N" 与 "Name : xxx"(npu-smi info -l 常见输出)
device_count = 0
names = []
for line in text.splitlines():
low = line.lower()
if "device count" in low and ":" in line:
try:
device_count = int(line.split(":")[-1].strip())
except ValueError:
device_count = 0
elif "name" in low and ":" in line:
name = line.split(":", 1)[-1].strip()
if name:
names.append(name)
available = device_count > 0 or bool(names)
return {
"available": available,
"runtime": "npu-smi",
"cann_version": "",
"device_count": device_count,
"device_names": names[: max(device_count, len(names))],
"detail": "npu-smi 探测成功(%d 张昇腾设备)" % device_count
if available else "npu-smi 可执行但未解析到昇腾设备",
}
+8 -4
View File
@@ -40,14 +40,18 @@ def build_backend(config: dict) -> InferenceBackend:
) )
cls = BACKEND_REGISTRY[name] cls = BACKEND_REGISTRY[name]
inf = config.get("inference", {}) or {} inf = config.get("inference", {}) or {}
return cls( kwargs = dict(
endpoint=inf.get("endpoint", ""), endpoint=inf.get("endpoint", ""),
model=inf.get("model", "iaop-default"), model=inf.get("model", "iaop-default"),
timeout_seconds=float(inf.get("timeout_seconds", 10)), timeout_seconds=float(inf.get("timeout_seconds", 10)),
runtime=inf.get("runtime", ""),
device=inf.get("device", ""),
cann_version=inf.get("cann_version", ""),
) )
# 空值/缺省不覆盖后端类默认(如昇腾 npu 默认 mindie / ascend-910b / CANN 8.0),
# 保证「切换后端仅改 backend 字段」时硬件相关参数落到正确的默认值。
for key in ("runtime", "device", "cann_version"):
value = inf.get(key)
if value:
kwargs[key] = value
return cls(**kwargs)
def load_backend_config(path: str) -> dict: def load_backend_config(path: str) -> dict:
+93 -17
View File
@@ -1,12 +1,27 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""华为昇腾 NPU 推理后端实现(ACL/CANN)。 """华为昇腾 NPU 推理后端实现(ACL/CANN)。
对应 PRD 5.6:昇腾实现(ACL/CANN)——通过昇腾推理服务 对应 PRD 5.6 与 issue #59「昇腾 NPU 后端适配(CANN 对接)」:
(MindIE / onnxruntime-ascend)的 OpenAI 兼容接口对外提供推理, 昇腾实现(ACL/CANN)通过昇腾推理服务(MindIE / onnxruntime-ascend)的
与 NVIDIA GPU 后端实现同一 `InferenceBackend` 接口: OpenAI 兼容接口对外提供推理,与 NVIDIA GPU 后端实现同一
**切换后端仅改适配层配置,业务代码零改动**。 `InferenceBackend` 接口:**切换后端仅改适配层配置,业务代码零改动**。
两种运行模式(由 endpoint 是否配置决定):
1. **服务模式**(endpoint 非空):对接 MindIE / onnxruntime-ascend 的
OpenAI 兼容 HTTP 接口,走 ``/acl/models/load``、``/v1/chat/completions``、
``/acl/models/unload``、``/health``;
2. **直连模式**(endpoint 为空且本机可检测到 CANN):调用本地 ACL Python
API(``acl.init`` / ``acl.rt.set_device`` / ``acl.rt.reset_device`` /
``acl.finalize``)管理昇腾设备与模型加载;
endpoint 为空且无 CANN 环境时进入 dry-run(适配层就绪,便于离线验证)。
健康巡检会附带 :func:`~inference_backend.cann_probe.probe_cann_environment`
探测到的 CANN 设备/工具链信息(``cann`` 段),便于运维确认后端就绪状态。
""" """
import time
from inference_backend.base import InferenceBackend, InferRequest, InferResult from inference_backend.base import InferenceBackend, InferRequest, InferResult
from inference_backend.cann_probe import probe_cann_environment
class AscendNpuBackend(InferenceBackend): class AscendNpuBackend(InferenceBackend):
@@ -23,24 +38,68 @@ class AscendNpuBackend(InferenceBackend):
self.device = device # ascend-310p | ascend-910b self.device = device # ascend-310p | ascend-910b
self.cann_version = cann_version # CANN 工具链版本 self.cann_version = cann_version # CANN 工具链版本
self._loaded = False self._loaded = False
self._acl_initialized = False
self._cann_cache = None
# ---- CANN 环境探测(懒加载,探测结果缓存于实例) ----
def cann_info(self) -> dict:
"""返回本机 CANN 环境探测结果(见 cann_probe.probe_cann_environment)。"""
if self._cann_cache is None:
self._cann_cache = probe_cann_environment()
return self._cann_cache
# ---- 统一接口实现 ----
def load_model(self, model_name: str | None = None) -> dict: def load_model(self, model_name: str | None = None) -> dict:
"""加载模型到 NPU(ACL aclmdlLoadFromFile 语义)。""" """加载模型到 NPU。
服务模式:``POST /acl/models/load``(ACL aclmdlLoadFromFile 语义);
直连模式:本地 ``acl.init()`` + ``acl.rt.set_device(device_id)``;
无 endpoint 且无 CANN 环境:dry-run,仅上报适配层就绪。
"""
model_name = model_name or self.model model_name = model_name or self.model
if not self.endpoint: cann = self.cann_info()
self._loaded = True if self.endpoint:
return {"status": "ok", "backend": self.backend_name, body = self._post_json(
"model": model_name, "device": self.device, "/acl/models/load",
"reason": "dry-run(未配置 endpoint)"}
body = self._post_json("/acl/models/load",
{"model": model_name, "device": self.device, {"model": model_name, "device": self.device,
"cann_version": self.cann_version}) "cann_version": self.cann_version})
self._loaded = body.get("status") in ("ok", "loaded", "ready") self._loaded = body.get("status") in ("ok", "loaded", "ready")
return body return body
if cann["available"]:
self._acl_load()
self._loaded = True
return {"status": "ok", "backend": self.backend_name,
"model": model_name, "device": self.device,
"runtime": self.runtime, "cann": cann,
"reason": "直连模式:CANN ACL 已初始化并绑定设备"}
self._loaded = True
return {"status": "ok", "backend": self.backend_name,
"model": model_name, "device": self.device,
"cann": cann,
"reason": "dry-run(未配置 endpoint 且无 CANN 环境,适配层就绪)"}
def _acl_load(self) -> None:
"""直连模式初始化 CANN ACL 并把上下文绑定到首张昇腾设备。"""
try:
import acl
except Exception as exc: # 探测与实际导入之间环境可能变化,容错
raise RuntimeError(f"CANN ACL 不可用,无法直连加载: {exc}") from exc
ret = acl.init()
# 注意:不能用 `ret not in (0, None, True)` —— Python 中 1 == True,
# 会把错误码 1 误判为成功;ACL 约定 ret=0(ACL_SUCCESS)为成功。
if not (ret is True or ret is None or ret == 0):
raise RuntimeError(f"acl.init() 失败: ret={ret}")
device_id = 0 # 默认首卡;多卡资源调度由部署侧配置扩展
ret = acl.rt.set_device(device_id)
if not (ret is True or ret is None or ret == 0):
acl.finalize()
raise RuntimeError(f"acl.rt.set_device({device_id}) 失败: ret={ret}")
self._acl_initialized = True
def infer(self, request: InferRequest) -> InferResult: def infer(self, request: InferRequest) -> InferResult:
"""昇腾推理(MindIE OpenAI 兼容 /v1/chat/completions)。""" """昇腾推理(MindIE OpenAI 兼容 /v1/chat/completions)。"""
import time
started = time.monotonic() started = time.monotonic()
payload = request.to_payload() payload = request.to_payload()
payload["model"] = payload["model"] or self.model payload["model"] = payload["model"] or self.model
@@ -56,11 +115,11 @@ class AscendNpuBackend(InferenceBackend):
latency_ms=latency_ms, latency_ms=latency_ms,
meta={"runtime": self.runtime, "device": self.device, meta={"runtime": self.runtime, "device": self.device,
"cann_version": self.cann_version, "model": self.model, "cann_version": self.cann_version, "model": self.model,
"raw": body}, "cann": self.cann_info(), "raw": body},
) )
def health(self) -> dict: def health(self) -> dict:
"""健康巡检:探测 /health,返回后端/设备/CANN 版本信息。""" """健康巡检:探测 /health,返回后端/设备/CANN 环境信息。"""
base = self._healthz() base = self._healthz()
base.update({ base.update({
"backend": self.backend_name, "backend": self.backend_name,
@@ -69,13 +128,30 @@ class AscendNpuBackend(InferenceBackend):
"cann_version": self.cann_version, "cann_version": self.cann_version,
"model": self.model, "model": self.model,
"loaded": self._loaded, "loaded": self._loaded,
"cann": self.cann_info(),
}) })
return base return base
def unload(self) -> dict: def unload(self) -> dict:
"""卸载模型、释放 NPU 资源(ACL aclmdlUnload 语义)。""" """卸载模型、释放 NPU 资源。
服务模式:``POST /acl/models/unload``(ACL aclmdlUnload 语义);
直连模式:``acl.rt.reset_device`` + ``acl.finalize``;
无 endpoint 且未直连加载:dry-run。
"""
self._loaded = False self._loaded = False
if not self.endpoint: if self.endpoint:
return self._post_json("/acl/models/unload", {"model": self.model})
if self._acl_initialized:
try:
import acl
acl.rt.reset_device(0)
acl.finalize()
except Exception as exc: # noqa: BLE001
return {"status": "warn", "backend": self.backend_name,
"reason": f"ACL 资源释放失败: {exc}"}
self._acl_initialized = False
return {"status": "ok", "backend": self.backend_name,
"reason": "直连模式:ACL 资源已释放"}
return {"status": "ok", "backend": self.backend_name, return {"status": "ok", "backend": self.backend_name,
"reason": "dry-run(未配置 endpoint)"} "reason": "dry-run(未配置 endpoint)"}
return self._post_json("/acl/models/unload", {"model": self.model})
@@ -0,0 +1,257 @@
# -*- coding: utf-8 -*-
"""昇腾 NPU 后端适配(CANN 对接)验收脚本(issue #59,PRD 5.6)。
验证能力点:
1. **配置资产与注册表**:模板配置可解析且含 backend 选择键;npu 已注册;
2. **CANN 环境探测**:无 CANN 环境优雅降级(available=False 且不抛异常);
注入模拟 ACL 后探测解析正确(设备数/名称/工具链版本);
3. **统一接口生命周期**:dry-run(无 endpoint、无 CANN)下
load_model / health / unload 状态 ok,health 附带 cann 信息;
4. **切换后端仅改配置**:同一业务调用面在 gpu ↔ npu 下等价(backend_name 变化、
接口方法可调用、硬件参数落到各自默认值);
5. **直连模式 ACL 生命周期**:注入模拟 ACL 后 load_model 触发
acl.init + rt.set_device,unload 触发 rt.reset_device + acl.finalize;
6. **服务模式(MindIE)载荷**:infer 走 /v1/chat/completions 且携带 model;
负例:未注册后端报 ValueError。
用法(在仓库根目录或 core/inference-backend 目录下均可):
python core/inference-backend/scripts/verify_npu_backend.py
退出码:0 = 全部通过;1 = 存在未达标项。
"""
from __future__ import annotations
import os
import sys
import types
import unittest.mock as mock
# 目录名含连字符(inference-backend),以包名 inference_backend 挂载到 sys.modules
_BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, _BACKEND_DIR)
if "inference_backend" not in sys.modules:
_pkg = types.ModuleType("inference_backend")
_pkg.__path__ = [_BACKEND_DIR]
sys.modules["inference_backend"] = _pkg
from inference_backend.base import InferRequest # noqa: E402
from inference_backend.cann_probe import probe_cann_environment # noqa: E402
from inference_backend.factory import ( # noqa: E402
BACKEND_REGISTRY, build_backend, default_config_path, load_backend_config,
)
# ---------------------------------------------------------------------------
# 模拟 CANN ACL 模块(用于验证探测/直连解析逻辑,本机无真实 CANN)
# ---------------------------------------------------------------------------
def make_fake_acl(init_ret=0, device_count=2, version="CANN 8.0.RC1",
set_device_ret=0):
"""构造一个行为可控的 fake ``acl`` 模块(兼容 (ret, value) 元组风格)。"""
rt = types.SimpleNamespace(
get_device_count=lambda: (0, device_count),
get_device_name=lambda i: (0, "ascend-910b-%d" % i),
get_version=lambda: (0, version),
set_device=lambda d: set_device_ret,
reset_device=lambda d: 0,
)
calls = {"init": [], "set_device": [], "reset_device": [], "finalize": []}
acl = types.SimpleNamespace(
rt=rt,
init=lambda *a, **k: (calls["init"].append(a) or init_ret),
finalize=lambda *a, **k: calls["finalize"].append(a) or 0,
)
rt.set_device = lambda d: (calls["set_device"].append(d) or set_device_ret)
rt.reset_device = lambda d: calls["reset_device"].append(d) or 0
return acl, calls
@mock.patch.dict("sys.modules", {"acl": None, "torch_npu": None})
@mock.patch("inference_backend.cann_probe.shutil.which", return_value=None)
def _probe_without_cann(_which):
"""强制无 CANN 环境:sys.modules 置 None 使 import 失败,且无 npu-smi。"""
return probe_cann_environment()
# ---------------------------------------------------------------------------
# 场景
# ---------------------------------------------------------------------------
def scenario_01_config_and_registry() -> bool:
print("== 场景 1:配置资产与可插拔注册表 ==")
cfg = load_backend_config(default_config_path())
ok = "backend" in cfg and cfg["backend"].lower() in ("gpu", "npu")
print(f" [A] 模板配置可解析且含 backend 选择键 -> {'PASS' if ok else 'FAIL'}")
ok2 = sorted(BACKEND_REGISTRY) == ["gpu", "npu"]
print(f" [B] 注册表覆盖 gpu/npu -> {'PASS' if ok2 else 'FAIL'}")
print(f" -> 场景 1 {'PASS' if ok and ok2 else 'FAIL'}\n")
return ok and ok2
def scenario_02_cann_probe() -> bool:
print("== 场景 2:CANN 环境探测 ==")
ok = True
info = _probe_without_cann()
degrade_ok = (info["available"] is False and info["runtime"] == "none"
and info["device_count"] == 0)
print(f" [A] 无 CANN 环境优雅降级(available=False, runtime=none)"
f" -> {'PASS' if degrade_ok else 'FAIL'}")
ok = ok and degrade_ok
with mock.patch.dict("sys.modules", {"acl": make_fake_acl()[0]}):
info = probe_cann_environment()
parse_ok = (info["available"] is True and info["runtime"] == "acl"
and info["device_count"] == 2
and info["device_names"] == ["ascend-910b-0", "ascend-910b-1"]
and info["cann_version"] == "CANN 8.0.RC1")
print(f" [B] 模拟 ACL 探测解析(2 设备/名称/版本)"
f" -> {'PASS' if parse_ok else 'FAIL'}")
ok = ok and parse_ok
acl_fail, _ = make_fake_acl(init_ret=1)
with mock.patch.dict("sys.modules", {"acl": acl_fail}):
info = probe_cann_environment()
fail_ok = info["available"] is False and "错误码" in info["detail"]
print(f" [C] ACL init 失败降级(available=False 且带原因)"
f" -> {'PASS' if fail_ok else 'FAIL'}")
ok = ok and fail_ok
print(f" -> 场景 2 {'PASS' if ok else 'FAIL'}\n")
return ok
def scenario_03_dry_run_lifecycle() -> bool:
print("== 场景 3:统一接口生命周期(dry-run) ==")
cfg = load_backend_config(default_config_path())
cfg["backend"] = "npu"
cfg["inference"]["endpoint"] = ""
cfg["inference"].pop("runtime", None)
cfg["inference"].pop("device", None)
backend = build_backend(cfg)
ok = True
loaded = backend.load_model()
ok = ok and loaded["status"] == "ok"
health = backend.health()
ok = ok and health["status"] == "ok" and health["loaded"] is True
ok = ok and "cann" in health and "device_names" in health["cann"]
unloaded = backend.unload()
ok = ok and unloaded["status"] == "ok"
print(f" load/health/unload 状态 ok、health 附 CANN 信息"
f" -> {'PASS' if ok else 'FAIL'}")
print(f" -> 场景 3 {'PASS' if ok else 'FAIL'}\n")
return ok
def scenario_04_switch_backend() -> bool:
print("== 场景 4:切换后端仅改配置(gpu ↔ npu 同一调用面) ==")
cfg = load_backend_config(default_config_path())
ok = True
gpu = build_backend(cfg)
cfg["backend"] = "npu"
cfg["inference"].pop("runtime", None)
cfg["inference"].pop("device", None)
npu = build_backend(cfg)
methods = ("load_model", "infer", "health", "unload")
surface_ok = (gpu.backend_name == "gpu" and npu.backend_name == "npu"
and all(callable(getattr(gpu, m)) and callable(getattr(npu, m))
for m in methods))
print(f" [A] 同一接口调用面,backend_name 随配置切换 -> "
f"{'PASS' if surface_ok else 'FAIL'}")
ok = ok and surface_ok
# 空值不覆盖后端默认:npu 落到 mindie / ascend-910b / CANN 8.0
defaults_ok = (npu.runtime == "mindie" and npu.device == "ascend-910b"
and npu.cann_version == "8.0")
print(f" [B] 硬件参数落到昇腾默认值(mindie/ascend-910b/CANN 8.0) -> "
f"{'PASS' if defaults_ok else 'FAIL'}")
ok = ok and defaults_ok
try:
build_backend({"backend": "tpu"})
neg_ok = False
except ValueError:
neg_ok = True
print(f" [C] 负例:未注册后端报 ValueError -> {'PASS' if neg_ok else 'FAIL'}")
ok = ok and neg_ok
print(f" -> 场景 4 {'PASS' if ok else 'FAIL'}\n")
return ok
def scenario_05_direct_mode_acl() -> bool:
print("== 场景 5:直连模式 ACL 生命周期 ==")
fake, calls = make_fake_acl()
with mock.patch.dict("sys.modules", {"acl": fake}):
cfg = load_backend_config(default_config_path())
cfg["backend"] = "npu"
cfg["inference"]["endpoint"] = ""
backend = build_backend(cfg)
loaded = backend.load_model()
load_ok = (loaded["status"] == "ok"
and "直连模式" in loaded["reason"]
and calls["set_device"] == [0])
unloaded = backend.unload()
# 探测本身会各调用一次 init/finalize;此处断言直连触发的确定性信号
unload_ok = (unloaded["status"] == "ok"
and calls["reset_device"] == [0]
and len(calls["finalize"]) >= 2)
health = backend.health()
health_ok = health["cann"]["available"] is True
ok = load_ok and unload_ok and health_ok
print(f" load 触发 acl.init+set_device、unload 触发 reset+finalize -> "
f"{'PASS' if ok else 'FAIL'}")
print(f" -> 场景 5 {'PASS' if ok else 'FAIL'}\n")
return ok
def scenario_06_service_mode_infer() -> bool:
print("== 场景 6:服务模式(MindIE OpenAI 兼容)载荷 ==")
cfg = load_backend_config(default_config_path())
cfg["backend"] = "npu"
cfg["inference"]["endpoint"] = "http://iaop-npu-svc:8000/v1"
cfg["inference"]["device"] = "ascend-910b"
cfg["inference"]["runtime"] = "mindie"
backend = build_backend(cfg)
fake_body = {"choices": [{"message": {"content": "炉温偏高,建议降低加料比"}}]}
with mock.patch.object(backend, "_post_json",
return_value=fake_body) as post:
result = backend.infer(InferRequest(prompt="请解释炉温报警"))
_, payload = post.call_args[0]
ok = (post.call_args[0][0] == "/v1/chat/completions"
and payload["model"] == backend.model
and result.text == "炉温偏高,建议降低加料比"
and result.backend == "npu")
print(f" infer 走 /v1/chat/completions 且携带 model、解析结果正确 -> "
f"{'PASS' if ok else 'FAIL'}")
print(f" -> 场景 6 {'PASS' if ok else 'FAIL'}\n")
return ok
# ---------------------------------------------------------------------------
def main() -> int:
results = [
("配置资产与注册表", scenario_01_config_and_registry()),
("CANN 环境探测", scenario_02_cann_probe()),
("dry-run 生命周期", scenario_03_dry_run_lifecycle()),
("切换后端仅改配置", scenario_04_switch_backend()),
("直连模式 ACL 生命周期", scenario_05_direct_mode_acl()),
("服务模式 infer 载荷", scenario_06_service_mode_infer()),
]
print("=" * 48)
all_ok = True
for name, ok in results:
print(f" {name}: {'PASS' if ok else 'FAIL'}")
all_ok = all_ok and ok
print("=" * 48)
print("全部通过" if all_ok else "存在未达标项")
return 0 if all_ok else 1
if __name__ == "__main__":
sys.exit(main())