diff --git a/core/inference-backend/cann_probe.py b/core/inference-backend/cann_probe.py new file mode 100644 index 0000000..905fe61 --- /dev/null +++ b/core/inference-backend/cann_probe.py @@ -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 可执行但未解析到昇腾设备", + } diff --git a/core/inference-backend/factory.py b/core/inference-backend/factory.py index f626d93..8e55449 100644 --- a/core/inference-backend/factory.py +++ b/core/inference-backend/factory.py @@ -40,14 +40,18 @@ def build_backend(config: dict) -> InferenceBackend: ) cls = BACKEND_REGISTRY[name] inf = config.get("inference", {}) or {} - return cls( + kwargs = dict( endpoint=inf.get("endpoint", ""), model=inf.get("model", "iaop-default"), 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: diff --git a/core/inference-backend/npu_backend.py b/core/inference-backend/npu_backend.py index 9aea2c2..5ca8544 100644 --- a/core/inference-backend/npu_backend.py +++ b/core/inference-backend/npu_backend.py @@ -1,12 +1,27 @@ # -*- coding: utf-8 -*- """华为昇腾 NPU 推理后端实现（ACL/CANN）。 -对应 PRD 5.6：昇腾实现（ACL/CANN）——通过昇腾推理服务 -（MindIE / onnxruntime-ascend）的 OpenAI 兼容接口对外提供推理， -与 NVIDIA GPU 后端实现同一 `InferenceBackend` 接口： -**切换后端仅改适配层配置，业务代码零改动**。 +对应 PRD 5.6 与 issue #59「昇腾 NPU 后端适配（CANN 对接）」： +昇腾实现（ACL/CANN）通过昇腾推理服务（MindIE / onnxruntime-ascend）的 +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.cann_probe import probe_cann_environment class AscendNpuBackend(InferenceBackend): @@ -23,24 +38,68 @@ class AscendNpuBackend(InferenceBackend): self.device = device # ascend-310p | ascend-910b self.cann_version = cann_version # CANN 工具链版本 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: - """加载模型到 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 - if not self.endpoint: + cann = self.cann_info() + if self.endpoint: + body = self._post_json( + "/acl/models/load", + {"model": model_name, "device": self.device, + "cann_version": self.cann_version}) + self._loaded = body.get("status") in ("ok", "loaded", "ready") + return body + if cann["available"]: + self._acl_load() self._loaded = True return {"status": "ok", "backend": self.backend_name, "model": model_name, "device": self.device, - "reason": "dry-run（未配置 endpoint）"} - body = self._post_json("/acl/models/load", - {"model": model_name, "device": self.device, - "cann_version": self.cann_version}) - self._loaded = body.get("status") in ("ok", "loaded", "ready") - return body + "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: """昇腾推理（MindIE OpenAI 兼容 /v1/chat/completions）。""" - import time started = time.monotonic() payload = request.to_payload() payload["model"] = payload["model"] or self.model @@ -56,11 +115,11 @@ class AscendNpuBackend(InferenceBackend): latency_ms=latency_ms, meta={"runtime": self.runtime, "device": self.device, "cann_version": self.cann_version, "model": self.model, - "raw": body}, + "cann": self.cann_info(), "raw": body}, ) def health(self) -> dict: - """健康巡检：探测 /health，返回后端/设备/CANN 版本信息。""" + """健康巡检：探测 /health，返回后端/设备/CANN 环境信息。""" base = self._healthz() base.update({ "backend": self.backend_name, @@ -69,13 +128,30 @@ class AscendNpuBackend(InferenceBackend): "cann_version": self.cann_version, "model": self.model, "loaded": self._loaded, + "cann": self.cann_info(), }) return base 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 - 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": "dry-run（未配置 endpoint）"} - return self._post_json("/acl/models/unload", {"model": self.model}) + "reason": "直连模式：ACL 资源已释放"} + return {"status": "ok", "backend": self.backend_name, + "reason": "dry-run（未配置 endpoint）"} diff --git a/core/inference-backend/scripts/verify_npu_backend.py b/core/inference-backend/scripts/verify_npu_backend.py new file mode 100644 index 0000000..ab4866c --- /dev/null +++ b/core/inference-backend/scripts/verify_npu_backend.py @@ -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())