From 78a410b773ce8a0af3b0918f9a02f8af9f54490a Mon Sep 17 00:00:00 2001 From: tzt <14718231+flying-travel@user.noreply.gitee.com> Date: Sun, 30 Aug 2026 21:00:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20T2=20=E8=BF=90=E7=BB=B4=E5=B1=82=20?= =?UTF-8?q?hw=5Fprofile=20+=20llama=5Fserver=20=E8=BF=9B=E7=A8=8B=E7=AE=A1?= =?UTF-8?q?=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 6 +- runtime/__init__.py | 6 + runtime/hw_profile.py | 140 +++++++++++++++ runtime/llama_server.py | 264 ++++++++++++++++++++++++++++ tests/_ports.py | 8 + tests/fixtures/fake_llama_server.py | 54 ++++++ tests/test_hw_profile.py | 114 ++++++++++++ tests/test_llama_server.py | 152 ++++++++++++++++ 任务拆解与执行计划.md | 108 ++++++++++++ 9 files changed, 851 insertions(+), 1 deletion(-) create mode 100644 runtime/__init__.py create mode 100644 runtime/hw_profile.py create mode 100644 runtime/llama_server.py create mode 100644 tests/_ports.py create mode 100644 tests/fixtures/fake_llama_server.py create mode 100644 tests/test_hw_profile.py create mode 100644 tests/test_llama_server.py create mode 100644 任务拆解与执行计划.md diff --git a/.gitignore b/.gitignore index 8fbe5bf..928aa65 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/runtime/__init__.py b/runtime/__init__.py new file mode 100644 index 0000000..fe9383b --- /dev/null +++ b/runtime/__init__.py @@ -0,0 +1,6 @@ +"""运维层（runtime）：本地 llama.cpp 运行时与硬件档位管理。 + +与 router_system 核心解耦：本包允许使用 httpx/fastapi 等第三方依赖， +用于真实本地模型（llama-server）的进程生命周期管理与硬件适配。 +核心协议（交流文本）仍在 router_system 内保持零第三方依赖。 +""" diff --git a/runtime/hw_profile.py b/runtime/hw_profile.py new file mode 100644 index 0000000..62a2aa3 --- /dev/null +++ b/runtime/hw_profile.py @@ -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', '')}]" + ) diff --git a/runtime/llama_server.py b/runtime/llama_server.py new file mode 100644 index 0000000..bcb023c --- /dev/null +++ b/runtime/llama_server.py @@ -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"), + ) diff --git a/tests/_ports.py b/tests/_ports.py new file mode 100644 index 0000000..8a4ebb0 --- /dev/null +++ b/tests/_ports.py @@ -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] diff --git a/tests/fixtures/fake_llama_server.py b/tests/fixtures/fake_llama_server.py new file mode 100644 index 0000000..edc3bed --- /dev/null +++ b/tests/fixtures/fake_llama_server.py @@ -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()) diff --git a/tests/test_hw_profile.py b/tests/test_hw_profile.py new file mode 100644 index 0000000..fde6f2e --- /dev/null +++ b/tests/test_hw_profile.py @@ -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" diff --git a/tests/test_llama_server.py b/tests/test_llama_server.py new file mode 100644 index 0000000..b59eb6b --- /dev/null +++ b/tests/test_llama_server.py @@ -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 diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md new file mode 100644 index 0000000..81dd85c --- /dev/null +++ b/任务拆解与执行计划.md @@ -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 改写） | ⬜ | | +