feat(v2): T2 运维层 hw_profile + llama_server 进程管理
This commit is contained in:
tzt
+5
-1
@@ -13,13 +13,17 @@ htmlcov/
|
||||
*.env
|
||||
api_keys*.json
|
||||
|
||||
# Models / data
|
||||
# Models / data / runtime
|
||||
models/
|
||||
data/
|
||||
bin/
|
||||
runs/
|
||||
*.bin
|
||||
*.gguf
|
||||
*.safetensors
|
||||
cached_results/
|
||||
|
||||
|
||||
# OS / editor
|
||||
.DS_Store
|
||||
Thumbs.db
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
"""运维层（runtime）：本地 llama.cpp 运行时与硬件档位管理。
|
||||
|
||||
与 router_system 核心解耦：本包允许使用 httpx/fastapi 等第三方依赖，
|
||||
用于真实本地模型（llama-server）的进程生命周期管理与硬件适配。
|
||||
核心协议（交流文本）仍在 router_system 内保持零第三方依赖。
|
||||
"""
|
||||
@@ -0,0 +1,140 @@
|
||||
"""硬件档位检测（runtime 运维层，纯标准库）。
|
||||
|
||||
把真实机器映射到三档保守模板之一：
|
||||
|
||||
- gpu12 : 约 ≥12GB 显存（NVIDIA / Vulkan 可探测） -> ngl 99, ctx 32768
|
||||
- gpu8 : 约 ≥8GB 显存 -> ngl 14, ctx 16384
|
||||
- cpu : 无独显或探测失败（保守兜底） -> ngl 0, ctx 8192
|
||||
|
||||
探测来源：nvidia-smi（NVIDIA 显存）优先；其次 vulkaninfo（AMD/Intel/通用，
|
||||
只能判断是否存在 Vulkan 设备，无法可靠拿到显存 -> 保守回退 cpu，并在结果标注
|
||||
probe:"conservative"）。总系统内存仅作为 cpu 档提示参考，不作为分档依据。
|
||||
|
||||
任何探测失败都回退到 cpu 保守档，保证不崩、可离线运行（D5 / D8）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
# 三档硬件模板（保守默认，可被 config.tiers 手动覆盖）
|
||||
TIER_SPECS: Dict[str, Dict[str, Any]] = {
|
||||
"gpu12": {"tier": "gpu12", "ngl": 99, "ctx": 32768, "kv_quant": "q8_0"},
|
||||
"gpu8": {"tier": "gpu8", "ngl": 14, "ctx": 16384, "kv_quant": "q8_0"},
|
||||
"cpu": {"tier": "cpu", "ngl": 0, "ctx": 8192, "kv_quant": "q8_0"},
|
||||
}
|
||||
|
||||
_GPU12_THRESHOLD_GB = 12.0
|
||||
_GPU8_THRESHOLD_GB = 8.0
|
||||
|
||||
|
||||
def _run(cmd: List[str], timeout: float = 10.0,
|
||||
runner: Optional[Callable[[List[str], float], subprocess.CompletedProcess]] = None
|
||||
) -> Optional[subprocess.CompletedProcess]:
|
||||
"""执行命令并捕获输出；失败/超时返回 None（不抛异常）。"""
|
||||
if runner is not None:
|
||||
try:
|
||||
return runner(cmd, timeout)
|
||||
except Exception:
|
||||
return None
|
||||
try:
|
||||
return subprocess.run(
|
||||
cmd, capture_output=True, text=True, timeout=timeout,
|
||||
creationflags=subprocess.CREATE_NO_WINDOW,
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def nvidia_vram_gb(runner: Optional[Callable[..., subprocess.CompletedProcess]] = None) -> Optional[float]:
|
||||
"""通过 nvidia-smi 读取显存总量（GB）；无 NVIDIA 返回 None。"""
|
||||
exe = shutil.which("nvidia-smi")
|
||||
if not exe:
|
||||
return None
|
||||
out = _run([exe, "--query-gpu=memory.total", "--format=csv,noheader,nounits"], runner=runner)
|
||||
if out is None or out.returncode != 0 or not out.stdout.strip():
|
||||
return None
|
||||
try:
|
||||
# 多卡取最大值（第一行也接受，但保守起见取最大以保证模板内存够用）
|
||||
vals = [float(v.strip()) for v in out.stdout.strip().splitlines() if v.strip().isdigit()]
|
||||
if not vals:
|
||||
return None
|
||||
return max(vals) / 1024.0
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def vulkan_present(runner: Optional[Callable[..., subprocess.CompletedProcess]] = None) -> bool:
|
||||
"""检测是否存在 Vulkan 设备（无法可靠拿显存 -> 只用于判定非 cpu 的候选）。"""
|
||||
exe = shutil.which("vulkaninfo")
|
||||
if not exe:
|
||||
return False
|
||||
out = _run([exe, "--summary"], timeout=15.0, runner=runner)
|
||||
if out is None or out.returncode != 0:
|
||||
return False
|
||||
low = out.stdout.lower()
|
||||
# 出现 deviceName 且非 "llvmpipe"/"software" 视为有真实设备
|
||||
return ("devicename" in low or "gpu" in low) and "llvmpipe" not in low and "lavapipe" not in low
|
||||
|
||||
|
||||
def pick_tier(vram_gb: Optional[float]) -> str:
|
||||
"""按显存选择档位；None/未知 -> cpu 保守档。"""
|
||||
if vram_gb is None:
|
||||
return "cpu"
|
||||
if vram_gb >= _GPU12_THRESHOLD_GB:
|
||||
return "gpu12"
|
||||
if vram_gb >= _GPU8_THRESHOLD_GB:
|
||||
return "gpu8"
|
||||
return "cpu"
|
||||
|
||||
|
||||
def detect(override: Optional[Dict[str, Any]] = None,
|
||||
runner: Optional[Callable[..., subprocess.CompletedProcess]] = None) -> Dict[str, Any]:
|
||||
"""检测并返回当前档位规格。
|
||||
|
||||
override（可选）：{"tier": "gpu12"} 强制指定档位；或覆盖单个字段如 {"ctx": 16384}。
|
||||
|
||||
返回形如 {"tier": "cpu", "ngl": 0, "ctx": 8192, "kv_quant": "q8_0",
|
||||
"probe": "nvidia|vulkan|cpu|override", "note": str}
|
||||
"""
|
||||
if override and override.get("tier") in TIER_SPECS:
|
||||
spec = dict(TIER_SPECS[override["tier"]])
|
||||
spec.update({k: v for k, v in override.items() if k in spec})
|
||||
spec["probe"] = "override"
|
||||
spec["note"] = f"手动指定档位 {override['tier']}"
|
||||
return spec
|
||||
|
||||
vram = nvidia_vram_gb(runner=runner)
|
||||
probe = "nvidia"
|
||||
if vram is None:
|
||||
if vulkan_present(runner=runner):
|
||||
probe = "vulkan"
|
||||
note = "检测到 Vulkan 设备但无法读取显存，按保守档 cpu 运行（可在 config 手动覆盖 tier）"
|
||||
else:
|
||||
probe = "cpu"
|
||||
note = "未检测到 GPU，按 cpu 档运行（-ngl 0，速度受限）"
|
||||
else:
|
||||
note = f"nvidia-smi 探测显存 {vram:.1f}GB"
|
||||
|
||||
tier = pick_tier(vram)
|
||||
spec = dict(TIER_SPECS[tier])
|
||||
spec["probe"] = probe
|
||||
spec["note"] = note if probe != "nvidia" else f"{note} -> 档位 {tier}"
|
||||
return spec
|
||||
|
||||
|
||||
def tier_spec(tier: str) -> Dict[str, Any]:
|
||||
"""返回指定档位的规格副本（供 config.tiers 兜底）。"""
|
||||
if tier not in TIER_SPECS:
|
||||
raise ValueError(f"未知硬件档位: {tier}（支持: {sorted(TIER_SPECS)}）")
|
||||
return dict(TIER_SPECS[tier])
|
||||
|
||||
|
||||
def detect_summary() -> str:
|
||||
"""人类可读的检测摘要（setup_runtime / serve 启动时打印）。"""
|
||||
spec = detect()
|
||||
return (
|
||||
f"硬件档位: {spec['tier']} (ngl={spec['ngl']}, ctx={spec['ctx']}, "
|
||||
f"kv_quant={spec['kv_quant']}) [{spec.get('note', '')}]"
|
||||
)
|
||||
@@ -0,0 +1,264 @@
|
||||
"""llama-server 子进程生命周期管理（runtime 运维层）。
|
||||
|
||||
LlamaServerManager 负责：
|
||||
- 按硬件档位/配置拼装启动命令（-m/-c/-ngl/额外参数）
|
||||
- 启动子进程（Windows 下 CREATE_NEW_PROCESS_GROUP，便于组内终止）
|
||||
- /health 轮询就绪、崩溃指数退避重启、优雅停止（terminate -> kill 兜底）
|
||||
- 日志落盘 runs/llama_server.log
|
||||
|
||||
设计（D1 / D8 / D11）：
|
||||
- 不修改 llama.cpp 源码，只捆绑上游 release 二进制。
|
||||
- 本模块可用第三方依赖（httpx），但健康检查默认用 urllib 保持轻量、可注入。
|
||||
- 一切外部副作用（health 探测、进程 spawn）均可注入替身，保证封闭单测。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
from .hw_profile import tier_spec
|
||||
|
||||
|
||||
def _now() -> str:
|
||||
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
|
||||
class LlamaServerError(RuntimeError):
|
||||
"""llama-server 启动/运行异常。"""
|
||||
|
||||
|
||||
class LlamaServerManager:
|
||||
"""管理单个 llama-server 子进程（单模型单实例，D5）。"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
binary: str,
|
||||
model: str,
|
||||
port: int = 8901,
|
||||
hw: Optional[Dict[str, Any]] = None,
|
||||
extra_args: Optional[List[str]] = None,
|
||||
health_timeout_s: float = 120.0,
|
||||
poll_interval_s: float = 1.0,
|
||||
max_restarts: int = 2,
|
||||
log_dir: Optional[str] = None,
|
||||
env: Optional[Dict[str, str]] = None,
|
||||
health_check: Optional[Callable[[str], bool]] = None,
|
||||
):
|
||||
self.binary = Path(binary)
|
||||
self.model = Path(model)
|
||||
self.port = int(port)
|
||||
# 档位规格：默认取 config 传入的 hw；缺少时按 tier 从内置表补全
|
||||
self.hw = dict(hw or {"tier": "cpu"})
|
||||
self.extra_args = list(extra_args or [])
|
||||
self.health_timeout_s = health_timeout_s
|
||||
self.poll_interval_s = poll_interval_s
|
||||
self.max_restarts = max_restarts
|
||||
self.log_dir = Path(log_dir) if log_dir else Path("runs")
|
||||
self.env = dict(env) if env else None
|
||||
self._health_check = health_check or self._default_health_check
|
||||
|
||||
self._proc: Optional[subprocess.Popen] = None
|
||||
self._log_path: Optional[Path] = None
|
||||
self._started_at: Optional[float] = None
|
||||
self._restart_count = 0
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
# 命令拼装（纯函数，便于单测）
|
||||
# ---------------------------------------------------------------
|
||||
def _build_command(self) -> List[str]:
|
||||
spec = tier_spec(self.hw.get("tier", "cpu"))
|
||||
ngl = self.hw.get("ngl", spec["ngl"])
|
||||
ctx = self.hw.get("ctx", spec["ctx"])
|
||||
kv = self.hw.get("kv_quant", spec["kv_quant"])
|
||||
cmd = [
|
||||
str(self.binary),
|
||||
"-m", str(self.model),
|
||||
"--port", str(self.port),
|
||||
"-ngl", str(ngl),
|
||||
"-c", str(ctx),
|
||||
"-ctk", kv,
|
||||
"-ctv", kv,
|
||||
]
|
||||
cmd.extend(self.extra_args)
|
||||
return cmd
|
||||
|
||||
def command_preview(self) -> str:
|
||||
"""启动命令预览（供日志/诊断打印，不执行）。"""
|
||||
return " ".join(self._build_command())
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
# 健康检查
|
||||
# ---------------------------------------------------------------
|
||||
def _default_health_check(self, endpoint: str) -> bool:
|
||||
"""GET {endpoint}/health，2 秒超时；网络异常视为不健康。"""
|
||||
url = f"{endpoint}/health"
|
||||
try:
|
||||
with urllib.request.urlopen(url, timeout=2.0) as resp:
|
||||
if resp.status != 200:
|
||||
return False
|
||||
body = resp.read(200).decode("utf-8", errors="replace")
|
||||
data = json.loads(body) if body else {}
|
||||
return data.get("status", "").lower() == "ok" or "llama" in body.lower()
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def health(self) -> bool:
|
||||
"""探测当前是否健康（进程在且 /health 通过）。"""
|
||||
if self._proc is None or self._proc.poll() is not None:
|
||||
return False
|
||||
return self._health_check(self.endpoint())
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
# 生命周期
|
||||
# ---------------------------------------------------------------
|
||||
def endpoint(self) -> str:
|
||||
return f"http://127.0.0.1:{self.port}"
|
||||
|
||||
def _log(self, msg: str) -> None:
|
||||
self.log_dir.mkdir(parents=True, exist_ok=True)
|
||||
line = f"[{_now()}] {msg}"
|
||||
path = self._log_path or (self.log_dir / "llama_server.log")
|
||||
self._log_path = path
|
||||
try:
|
||||
with open(path, "a", encoding="utf-8") as f:
|
||||
f.write(line + "\n")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def start(self) -> bool:
|
||||
"""启动子进程并轮询至健康就绪。
|
||||
|
||||
返回 True 表示健康就绪；False 表示启动失败/超时（进程可能已退出）。
|
||||
"""
|
||||
if self._proc is not None and self._proc.poll() is None:
|
||||
return self.health()
|
||||
if not self.binary.exists():
|
||||
raise LlamaServerError(
|
||||
f"llama-server 二进制不存在: {self.binary}。请先运行 "
|
||||
f"scripts/setup_runtime.py 下载，或将上游 release 放入 bin/（D1 不改源码）。"
|
||||
)
|
||||
if not self.model.exists():
|
||||
raise LlamaServerError(
|
||||
f"模型文件不存在: {self.model}。请先运行 scripts/setup_runtime.py 下载 GGUF。"
|
||||
)
|
||||
|
||||
cmd = self._build_command()
|
||||
self.log_dir.mkdir(parents=True, exist_ok=True)
|
||||
logf = self.log_dir / "llama_server.log"
|
||||
self._log_path = logf
|
||||
self._log(f"启动: {self.command_preview()}")
|
||||
|
||||
kwargs: Dict[str, Any] = {}
|
||||
if os.name == "nt":
|
||||
kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW
|
||||
try:
|
||||
self._proc = subprocess.Popen(
|
||||
cmd,
|
||||
stdout=open(logf, "ab", buffering=0),
|
||||
stderr=subprocess.STDOUT,
|
||||
env=self.env,
|
||||
**kwargs,
|
||||
)
|
||||
except OSError as e:
|
||||
self._log(f"spawn 失败: {e}")
|
||||
self._proc = None
|
||||
raise LlamaServerError(f"无法启动 llama-server: {e}") from e
|
||||
|
||||
self._started_at = time.time()
|
||||
return self._wait_healthy()
|
||||
|
||||
def _wait_healthy(self) -> bool:
|
||||
deadline = time.time() + self.health_timeout_s
|
||||
while time.time() < deadline:
|
||||
if self._proc.poll() is not None:
|
||||
self._log(f"进程过早退出 rc={self._proc.returncode}")
|
||||
return False
|
||||
if self.health():
|
||||
self._log(f"健康就绪 @ {self.endpoint()} (pid={self._proc.pid})")
|
||||
return True
|
||||
time.sleep(self.poll_interval_s)
|
||||
self._log("健康检查超时，标记为启动失败")
|
||||
return False
|
||||
|
||||
def stop(self, timeout_s: float = 8.0) -> None:
|
||||
"""优雅停止：terminate（CTRL_BREAK）-> 等待 -> kill 兜底（Windows 语义）。"""
|
||||
proc = self._proc
|
||||
if proc is None:
|
||||
return
|
||||
if proc.poll() is not None:
|
||||
self._proc = None
|
||||
return
|
||||
try:
|
||||
proc.terminate()
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
proc.wait(timeout=timeout_s)
|
||||
except subprocess.TimeoutExpired:
|
||||
self._log("terminate 超时，kill 兜底")
|
||||
try:
|
||||
proc.kill()
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
proc.wait(timeout=5.0)
|
||||
except subprocess.TimeoutExpired:
|
||||
pass
|
||||
self._proc = None
|
||||
self._log("已停止")
|
||||
|
||||
def ensure_alive(self) -> bool:
|
||||
"""保活：不健康则按指数退避重启（最多 max_restarts 次）。"""
|
||||
if self._proc is not None and self._proc.poll() is None and self.health():
|
||||
return True
|
||||
if self._restart_count >= self.max_restarts:
|
||||
return False
|
||||
backoff = min(2.0 ** self._restart_count, 8.0)
|
||||
self._restart_count += 1
|
||||
self._log(f"检测到异常，{backoff:.1f}s 后重启（第 {self._restart_count}/{self.max_restarts} 次）")
|
||||
time.sleep(backoff)
|
||||
if self._proc is not None and self._proc.poll() is None:
|
||||
self.stop()
|
||||
return self.start()
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
@property
|
||||
def running(self) -> bool:
|
||||
return self._proc is not None and self._proc.poll() is None
|
||||
|
||||
@property
|
||||
def pid(self) -> Optional[int]:
|
||||
return self._proc.pid if self._proc is not None else None
|
||||
|
||||
def __enter__(self) -> "LlamaServerManager":
|
||||
self.start()
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc) -> None:
|
||||
self.stop()
|
||||
|
||||
|
||||
def build_llama_server(cfg: Dict[str, Any]) -> LlamaServerManager:
|
||||
"""从 config.runtime.llama_server 段构建管理器。cfg 含 binary/model/port/hw_profile/extra_args。"""
|
||||
binary = cfg.get("binary", "bin/llama-server.exe")
|
||||
model = cfg.get("model", "models/qwen3.5-4b-q4_k_m.gguf")
|
||||
port = int(cfg.get("port", 8901))
|
||||
hw = cfg.get("hw", {}) or {}
|
||||
extra = cfg.get("extra_args", [])
|
||||
return LlamaServerManager(
|
||||
binary=binary,
|
||||
model=model,
|
||||
port=port,
|
||||
hw=hw,
|
||||
extra_args=extra,
|
||||
health_timeout_s=float(cfg.get("health_timeout_s", 120)),
|
||||
max_restarts=int(cfg.get("max_restarts", 2)),
|
||||
log_dir=cfg.get("log_dir"),
|
||||
)
|
||||
@@ -0,0 +1,8 @@
|
||||
"""测试辅助：获取空闲 TCP 端口。"""
|
||||
import socket
|
||||
|
||||
|
||||
def free_port() -> int:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
s.bind(("127.0.0.1", 0))
|
||||
return s.getsockname()[1]
|
||||
Vendored
+54
@@ -0,0 +1,54 @@
|
||||
"""测试替身：模拟 llama-server（供 LlamaServerManager 封闭单测，D11）。
|
||||
|
||||
- 解析 --port / -m / -ngl / -c（与真实 llama-server 参数对齐）
|
||||
- 把 pid / 收到的参数写入环境变量 FAKE_MARKER 指向的 JSON 文件
|
||||
- 在本机端口起一个最小 http 服务：/health 返回 {"status":"ok"}
|
||||
- 进程被终止时正常退出
|
||||
"""
|
||||
import argparse
|
||||
import http.server
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(prog="fake-llama-server")
|
||||
parser.add_argument("--port", type=int, default=8901)
|
||||
parser.add_argument("-m", dest="model", default="")
|
||||
parser.add_argument("-ngl", dest="ngl", default="0")
|
||||
parser.add_argument("-c", dest="ctx", default="8192")
|
||||
parser.add_argument("-ctk", dest="ctk", default="")
|
||||
parser.add_argument("-ctv", dest="ctv", default="")
|
||||
args, _ = parser.parse_known_args()
|
||||
|
||||
marker = os.environ.get("FAKE_MARKER")
|
||||
if marker:
|
||||
os.makedirs(os.path.dirname(marker) or ".", exist_ok=True)
|
||||
with open(marker, "w", encoding="utf-8") as f:
|
||||
json.dump({"pid": os.getpid(), "port": args.port,
|
||||
"model": args.model, "args": sys.argv[1:]}, f)
|
||||
|
||||
class Handler(http.server.BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
if self.path.startswith("/health"):
|
||||
body = json.dumps({"status": "ok", "server": "fake-llama"}).encode("utf-8")
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
|
||||
def log_message(self, *a):
|
||||
pass
|
||||
|
||||
srv = http.server.ThreadingHTTPServer(("127.0.0.1", args.port), Handler)
|
||||
srv.serve_forever()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,114 @@
|
||||
"""T2 硬件档位检测单测（封闭，纯逻辑 + 注入 runner）。"""
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
|
||||
from runtime.hw_profile import (
|
||||
TIER_SPECS,
|
||||
detect,
|
||||
nvidia_vram_gb,
|
||||
pick_tier,
|
||||
tier_spec,
|
||||
vulkan_present,
|
||||
)
|
||||
|
||||
|
||||
class _FakeRun:
|
||||
"""注入 subprocess runner：按命令返回预置输出。"""
|
||||
|
||||
def __init__(self, mapping):
|
||||
self.mapping = mapping # {关键子串: CompletedProcess}
|
||||
|
||||
def __call__(self, cmd, timeout):
|
||||
joined = " ".join(cmd)
|
||||
for key, cp in self.mapping.items():
|
||||
if key in joined:
|
||||
return cp
|
||||
raise FileNotFoundError(cmd)
|
||||
|
||||
|
||||
def _cp(stdout="", rc=0):
|
||||
return subprocess.CompletedProcess(args=[], returncode=rc, stdout=stdout, stderr="")
|
||||
|
||||
|
||||
def test_pick_tier_thresholds():
|
||||
assert pick_tier(None) == "cpu"
|
||||
assert pick_tier(6.0) == "cpu"
|
||||
assert pick_tier(8.0) == "gpu8"
|
||||
assert pick_tier(11.9) == "gpu8"
|
||||
assert pick_tier(12.0) == "gpu12"
|
||||
assert pick_tier(24.0) == "gpu12"
|
||||
|
||||
|
||||
def test_tier_specs_have_required_fields():
|
||||
for tier, spec in TIER_SPECS.items():
|
||||
assert spec["tier"] == tier
|
||||
assert "ngl" in spec and "ctx" in spec and "kv_quant" in spec
|
||||
# gpu12 ngl 全量卸载，cpu ngl 0
|
||||
assert TIER_SPECS["gpu12"]["ngl"] == 99
|
||||
assert TIER_SPECS["cpu"]["ngl"] == 0
|
||||
|
||||
|
||||
def test_tier_spec_unknown_raises():
|
||||
with pytest.raises(ValueError):
|
||||
tier_spec("nonexistent")
|
||||
|
||||
|
||||
def test_nvidia_vram_gb_parses():
|
||||
fake = _FakeRun({"nvidia-smi": _cp("24576\n")})
|
||||
assert nvidia_vram_gb(runner=fake) == 24.0
|
||||
|
||||
|
||||
def test_nvidia_vram_gb_multi_gpu_takes_max():
|
||||
fake = _FakeRun({"nvidia-smi": _cp("8192\n12288\n")})
|
||||
assert nvidia_vram_gb(runner=fake) == 12.0
|
||||
|
||||
|
||||
def test_nvidia_vram_gb_missing_tool_returns_none(monkeypatch):
|
||||
import shutil
|
||||
monkeypatch.setattr(shutil, "which", lambda name: None)
|
||||
assert nvidia_vram_gb() is None
|
||||
|
||||
|
||||
def test_vulkan_present_true():
|
||||
fake = _FakeRun({"vulkaninfo": _cp("deviceName : NVIDIA GeForce RTX 4090")})
|
||||
assert vulkan_present(runner=fake) is True
|
||||
|
||||
|
||||
def test_vulkan_present_software_false():
|
||||
fake = _FakeRun({"vulkaninfo": _cp("deviceName : llvmpipe (LLVM)")})
|
||||
assert vulkan_present(runner=fake) is False
|
||||
|
||||
|
||||
def test_detect_nvidia_gpu12(monkeypatch):
|
||||
# 让 shutil.which 只对 nvidia-smi 生效
|
||||
real_which = __import__("shutil").which
|
||||
def fake_which(name):
|
||||
return "C:/x/nvidia-smi.exe" if name == "nvidia-smi" else None
|
||||
monkeypatch.setattr(__import__("shutil"), "which", fake_which)
|
||||
fake = _FakeRun({"nvidia-smi": _cp("24576\n")})
|
||||
spec = detect(runner=fake)
|
||||
assert spec["tier"] == "gpu12"
|
||||
assert spec["ngl"] == 99 and spec["ctx"] == 32768
|
||||
assert spec["probe"] == "nvidia"
|
||||
|
||||
|
||||
def test_detect_cpu_fallback(monkeypatch):
|
||||
monkeypatch.setattr(__import__("shutil"), "which", lambda name: None)
|
||||
spec = detect()
|
||||
assert spec["tier"] == "cpu"
|
||||
assert spec["ngl"] == 0
|
||||
assert spec["probe"] == "cpu"
|
||||
|
||||
|
||||
def test_detect_override_tier():
|
||||
spec = detect(override={"tier": "gpu8"})
|
||||
assert spec["tier"] == "gpu8"
|
||||
assert spec["ngl"] == 14 and spec["ctx"] == 16384
|
||||
assert spec["probe"] == "override"
|
||||
|
||||
|
||||
def test_detect_override_field():
|
||||
spec = detect(override={"tier": "cpu", "ctx": 16384})
|
||||
assert spec["ctx"] == 16384
|
||||
assert spec["tier"] == "cpu"
|
||||
@@ -0,0 +1,152 @@
|
||||
"""T2 llama-server 进程管理单测（封闭：假二进制 + 注入，D11）。"""
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from runtime.llama_server import LlamaServerManager, LlamaServerError, build_llama_server
|
||||
from tests._ports import free_port
|
||||
|
||||
FAKE_SCRIPT = Path(__file__).parent / "fixtures" / "fake_llama_server.py"
|
||||
PYTHON = sys.executable
|
||||
|
||||
|
||||
def _make_fake_binary(tmp: Path) -> Path:
|
||||
"""生成一个 .cmd 包装器：把 venv python + 假脚本当作"二进制"启动。"""
|
||||
cmd = tmp / "fake-llama-server.cmd"
|
||||
cmd.write_text(
|
||||
f'@echo off\r\n"{PYTHON}" "{FAKE_SCRIPT}" %*\r\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
return cmd
|
||||
|
||||
|
||||
def _make_manager(tmp, binary, port, model, **kw):
|
||||
marker = tmp / "marker.json"
|
||||
env = dict(os.environ)
|
||||
env["FAKE_MARKER"] = str(marker)
|
||||
return LlamaServerManager(
|
||||
binary=str(binary),
|
||||
model=str(model),
|
||||
port=port,
|
||||
hw={"tier": "cpu"},
|
||||
health_timeout_s=15.0,
|
||||
poll_interval_s=0.2,
|
||||
log_dir=str(tmp / "runs"),
|
||||
env=env,
|
||||
**kw,
|
||||
), marker
|
||||
|
||||
|
||||
def test_build_command_uses_hw_tier():
|
||||
m = LlamaServerManager(binary="bin/x.exe", model="models/m.gguf", port=8901,
|
||||
hw={"tier": "gpu12"})
|
||||
cmd = m._build_command()
|
||||
assert Path(cmd[0]) == Path("bin/x.exe")
|
||||
assert cmd[1] == "-m" and Path(cmd[2]) == Path("models/m.gguf")
|
||||
assert "--port" in cmd and "8901" in cmd
|
||||
assert cmd[cmd.index("-ngl") + 1] == "99"
|
||||
assert cmd[cmd.index("-c") + 1] == "32768"
|
||||
|
||||
|
||||
def test_build_command_extra_args_appended():
|
||||
m = LlamaServerManager(binary="bin/x.exe", model="models/m.gguf", port=1,
|
||||
hw={"tier": "cpu"}, extra_args=["--cache-reuse", "256"])
|
||||
cmd = m._build_command()
|
||||
assert cmd[-2:] == ["--cache-reuse", "256"]
|
||||
|
||||
|
||||
def test_start_missing_binary_raises(tmp_path):
|
||||
m = LlamaServerManager(binary=str(tmp_path / "nope.exe"), model=str(tmp_path / "m.gguf"),
|
||||
port=free_port())
|
||||
with pytest.raises(LlamaServerError):
|
||||
m.start()
|
||||
|
||||
|
||||
def test_start_missing_model_raises(tmp_path):
|
||||
binary = _make_fake_binary(tmp_path)
|
||||
m = LlamaServerManager(binary=str(binary), model=str(tmp_path / "missing.gguf"),
|
||||
port=free_port())
|
||||
with pytest.raises(LlamaServerError):
|
||||
m.start()
|
||||
|
||||
|
||||
def test_full_lifecycle(tmp_path):
|
||||
binary = _make_fake_binary(tmp_path)
|
||||
model = tmp_path / "model.gguf"
|
||||
model.write_bytes(b"fake")
|
||||
port = free_port()
|
||||
m, marker = _make_manager(tmp_path, binary, port, model)
|
||||
|
||||
assert m.running is False
|
||||
assert m.health() is False # 无进程时不健康
|
||||
|
||||
ok = m.start()
|
||||
assert ok is True
|
||||
assert m.running is True
|
||||
assert m.health() is True
|
||||
|
||||
# 假脚本确实收到了参数
|
||||
data = json.loads(marker.read_text(encoding="utf-8"))
|
||||
assert data["port"] == port
|
||||
assert data["model"] == str(model)
|
||||
|
||||
# 再 start 幂等（已运行返回健康）
|
||||
assert m.start() is True
|
||||
|
||||
m.stop()
|
||||
assert m.running is False
|
||||
assert m.health() is False
|
||||
|
||||
|
||||
def test_stop_idempotent(tmp_path):
|
||||
binary = _make_fake_binary(tmp_path)
|
||||
model = tmp_path / "model.gguf"
|
||||
model.write_bytes(b"fake")
|
||||
m, _ = _make_manager(tmp_path, binary, free_port(), model)
|
||||
m.stop() # 未启动时 stop 不抛
|
||||
assert m.running is False
|
||||
|
||||
|
||||
def test_ensure_alive_healthy_no_restart(tmp_path):
|
||||
binary = _make_fake_binary(tmp_path)
|
||||
model = tmp_path / "model.gguf"
|
||||
model.write_bytes(b"fake")
|
||||
m, _ = _make_manager(tmp_path, binary, free_port(), model)
|
||||
m.start()
|
||||
assert m.ensure_alive() is True
|
||||
# 不应触发重启
|
||||
assert m._restart_count == 0
|
||||
m.stop()
|
||||
|
||||
|
||||
def test_ensure_alive_restart_exhausted(tmp_path):
|
||||
binary = _make_fake_binary(tmp_path)
|
||||
model = tmp_path / "model.gguf"
|
||||
model.write_bytes(b"fake")
|
||||
m, _ = _make_manager(tmp_path, binary, free_port(), model, max_restarts=0)
|
||||
# 未启动：restart 上限 0 -> False
|
||||
assert m.ensure_alive() is False
|
||||
|
||||
|
||||
def test_endpoint_format():
|
||||
m = LlamaServerManager(binary="x", model="m", port=8901, hw={"tier": "cpu"})
|
||||
assert m.endpoint() == "http://127.0.0.1:8901"
|
||||
|
||||
|
||||
def test_build_from_config(tmp_path):
|
||||
binary = _make_fake_binary(tmp_path)
|
||||
cfg = {
|
||||
"binary": str(binary),
|
||||
"model": str(tmp_path / "m.gguf"),
|
||||
"port": 8999,
|
||||
"hw": {"tier": "cpu"},
|
||||
"extra_args": ["-fa"],
|
||||
}
|
||||
m = build_llama_server(cfg)
|
||||
assert m.port == 8999
|
||||
assert "-fa" in m.extra_args
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
# 任务拆解与执行计划
|
||||
|
||||
> 任务体系：主任务（整体项目）→ 附加任务（先行实现：整体项目部分拆解）
|
||||
> 建立日期：2026-08-14
|
||||
|
||||
---
|
||||
|
||||
## 一、任务体系总览
|
||||
|
||||
```
|
||||
主任务：多专业小模型 + 路由模型系统（整体项目）
|
||||
│
|
||||
├─ 已完成部分（专家系统内核）：
|
||||
│ · 知识库（8 领域 67 规则 + 17 模板 + 45 事实）
|
||||
│ · 黑板/前向链/Planner/DAG 路由（L0 零参数）
|
||||
│ · 三级子领域（domain → subdomain → subdomain2）
|
||||
│ · 两级路由体系（domain_group → 组内路由模型）
|
||||
│ · 92 项单元测试全绿
|
||||
│
|
||||
└─ 附加任务（★ 先行实现）：整体项目部分拆解
|
||||
· 目标：把整体项目剩余工作拆解为可独立执行的部分任务，
|
||||
并优先实现第一批（P0），为后续铺路
|
||||
· 状态：进行中
|
||||
```
|
||||
|
||||
## 二、整体项目剩余工作拆解清单
|
||||
|
||||
| # | 任务 | 内容 | 依赖 | 优先级 | 预估 | 状态 |
|
||||
|---|------|------|------|--------|------|------|
|
||||
| T1 | NodeExecutor 接口抽象 | 解耦 `_execute_node` 的 if-else；rule/model/未来后端统一接口 + 工厂 | — | **P0** | 0.5 天 | ✅ 完成 |
|
||||
| T2 | 评测基准扩充 | eval 扩到 8 领域 24 样例 + 组路由/子领域识别指标 | — | **P0** | 0.5 天 | ✅ 完成 |
|
||||
| T3 | 推理链查询接口 | 请求 ID 化 + 内存轨迹存储 + `GET /traces/{id}` | T1 | **P0** | 1 天 | ✅ 完成 |
|
||||
| T12 | **Agent-Skill 路由器** | 路由器独立：Skill 注册表（21 技能）+ RouteAgent 自主分析需求→技能调用计划→执行，无需用户指定领域/模型 | T1,T2,T3 | **P0** | 1-2 天 | ✅ 完成 |
|
||||
| T4 | 知识库 CRUD + 热重载 | `/knowledge/rules|facts` CRUD + `POST /config/reload` | T1 | P1 | 1-2 天 | ⬜ 待办 |
|
||||
| T5 | 本地模型推理接入（L2） | Ollama/vLLM 封装 + `fallback.local` 验证 + 组内模型按需加载 | T1 | P1 | 2-3 天 | ⬜ 待办 |
|
||||
| T6 | embedding 语义缓存 | BGE 本地向量化替代 n-gram L2 | — | P1 | 1-2 天 | ⬜ 待办 |
|
||||
| T7 | 分类器训练流水线 | 数据构建 + 0.6B QLoRA 训练（8 领域） | — | P1 | 2-3 天 | ⬜ 待办 |
|
||||
| T8 | 领域专家微调 | QLoRA 微调 5-8 领域专家 + 数据收集 | T7 | P2 | 3-4 周 | ⬜ 待办 |
|
||||
| T9 | RouterArena 评测 | 标准 5 维评测接入 | T2 | P2 | 2-3 天 | ⬜ 待办 |
|
||||
| T10 | 监控 + 模型注册表 | Prometheus 指标 + 模型版本/热替换 | T4,T5 | P2 | 2-3 天 | ⬜ 待办 |
|
||||
| T11 | 部署生产化 | Docker/Compose + 灰度 + 安全 | T5,T10 | P2 | 3-5 天 | ⬜ 待办 |
|
||||
|
||||
## 三、先行实现批次（P0）—— 已完成 ✅
|
||||
|
||||
1. **T1 NodeExecutor 接口抽象** —— 子任务执行后端统一接口（rule/model 两实现 + 工厂），L2 模型接入无需改 Router
|
||||
2. **T2 评测基准扩充** —— 24 样例 × 8 领域，指标：分类 100% / 大领域组识别 100% / 子领域识别 100%
|
||||
3. **T3 推理链查询接口** —— `GET /traces/{request_id}`：完整推理链可追溯（两级路由 → 三级子领域 → 拆解 → 规则 → 评分）
|
||||
4. **T12 Agent-Skill 路由器** —— 路由器独立为"技能注册表 + Agent 规划器"：
|
||||
- 21 个内置技能（es.* 模板 ×17、kb.retrieve/kb.answer、judge.evaluate、fallback.call）
|
||||
- RouteAgent 自行分析需求 → 规划技能调用（多技能组合/依赖）→ 执行 → 校验 → 升级
|
||||
- **用户只提供 query，无需指定领域/模型**；技能调用轨迹可追溯（skill:es.analyze@facts）
|
||||
- 实测：法律咨询自动组合 es.analyze + kb.retrieve + es.conclude + es.disclaimer
|
||||
|
||||
P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯的推理链 + 技能化 Agent 路由（118 项测试全绿）。
|
||||
|
||||
## 四、执行规则
|
||||
|
||||
- 每任务独立验收（测试 + 文档），完成后更新状态 ✅
|
||||
- 依赖任务未完成时，先做无依赖任务
|
||||
- P1/P2 任务在 P0 完成后按序推进，不阻塞主线
|
||||
|
||||
---
|
||||
|
||||
## 五、v2 任务登记（端云协同 LLM 协作系统，见《实现方案_v2_端云协同LLM协作系统.md》）
|
||||
|
||||
> 每个任务一个 commit（`feat(v2): Tn 描述`），交付含封闭单测；v1 的 126 项测试保持全绿。
|
||||
|
||||
| T | 内容 | 状态 | commit |
|
||||
|---|------|------|--------|
|
||||
| T1 | 环境与基线确认（126 测试全绿；README 环境备忘） | ✅ 完成 | （并入 T2 commit） |
|
||||
| T2 | 运维层：hw_profile + llama_server 进程管理 | ✅ 完成 | T2 |
|
||||
| T3 | ArchitectClient（DeepSeek API，JSON 约束） | ⬜ | |
|
||||
| T4 | Workspace（交流文本 schema/校验/渲染/rollup） | ⬜ | |
|
||||
| T5 | WorkerLoop + 接地验证 | ⬜ | |
|
||||
| T6 | CollaborativePipeline 编排 | ⬜ | |
|
||||
| T7 | 网关扩展（/chat 切 v2，/chat/legacy） | ⬜ | |
|
||||
| T8 | 人工检验队列 ReviewQueue | ⬜ | |
|
||||
| T9 | token 计量与账单 | ⬜ | |
|
||||
| T10 | rollup + prefix cache 调优 | ⬜ | |
|
||||
| T11 | 打包分发 setup_runtime.py | ⬜ | |
|
||||
| T12 | 实验脚本 bench_tokens.py + 数据集 | ⬜ | |
|
||||
| T13 | E1–E5 跑数到 research/v2_experiments/ | ⬜ | |
|
||||
| T14 | 文档收口（README v2 改写） | ⬜ | |
|
||||
|
||||
|
||||
---
|
||||
|
||||
## 五、v2 任务登记（端云协同 LLM 协作系统，见《实现方案_v2_端云协同LLM协作系统.md》）
|
||||
|
||||
> 每个任务一个 commit（`feat(v2): Tn 描述`），交付含封闭单测；v1 的 126 项测试保持全绿。
|
||||
|
||||
| T | 内容 | 状态 | commit |
|
||||
|---|------|------|--------|
|
||||
| T1 | 环境与基线确认（126 测试全绿；README 环境备忘） | ✅ 完成 | （并入 T2 commit） |
|
||||
| T2 | 运维层：hw_profile + llama_server 进程管理 | ✅ 完成 | T2 |
|
||||
| T3 | ArchitectClient（DeepSeek API，JSON 约束） | ⬜ | |
|
||||
| T4 | Workspace（交流文本 schema/校验/渲染/rollup） | ⬜ | |
|
||||
| T5 | WorkerLoop + 接地验证 | ⬜ | |
|
||||
| T6 | CollaborativePipeline 编排 | ⬜ | |
|
||||
| T7 | 网关扩展（/chat 切 v2，/chat/legacy） | ⬜ | |
|
||||
| T8 | 人工检验队列 ReviewQueue | ⬜ | |
|
||||
| T9 | token 计量与账单 | ⬜ | |
|
||||
| T10 | rollup + prefix cache 调优 | ⬜ | |
|
||||
| T11 | 打包分发 setup_runtime.py | ⬜ | |
|
||||
| T12 | 实验脚本 bench_tokens.py + 数据集 | ⬜ | |
|
||||
| T13 | E1–E5 跑数到 research/v2_experiments/ | ⬜ | |
|
||||
| T14 | 文档收口（README v2 改写） | ⬜ | |
|
||||
|
||||
Reference in New Issue
Block a user