guanlan-src/app_ontology/app_ontology_guanlan/ontology/llm_gate.py

# -*- coding: utf-8 -*-
"""离线模型闸 (2026-09-07, 用户令 "离线模型也要把关, 要稳定和可靠"): 所有对本机 Ollama 的调用经此一层.
保护: ① 确定性参数注入 (temperature=0, seed 固定; 同问同证据同答) ② 单飞 (同一时刻只一个生成请求, 防显存打架)
③ 瞬断重试一次 (连接失败/超时/5xx; 4xx 不重试, 原样抛给调用方处理) ④ 熔断 (连续 3 次失败 → 60 s 内直接报"本机模型未就绪", 不再压请求; 探针恢复)
⑤ 逐次审计留痕 logs/llm_audit.jsonl (模型/摘要/提示 sha/参数/耗时/输出 sha/错误) ⑥ 模型摘要钉死 (configs/models.json ↔ ollama /api/show digest)
不做: 不改提示词, 不改校闸 (那些在 fast_agent), 不做任何联网."""
from __future__ import annotations

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]
import hashlib, json, os, threading, time, urllib.error, urllib.request
from pathlib import Path
ROOT = _install_root(__file__)
import sys as _sys; _sys.path.insert(0, str(ROOT))
from src import paths as P          # 配置/日志路径唯一取用口
CFG = P.MODELS_JSON                      # 配置唯一取用口 (src/paths.py)
AUDIT = P.LOGS / "audit" / "llm_audit.jsonl"   # 审计流水统一落点 logs/audit/ (命名/格式约定见 docs §11)
ENDPOINT = os.environ.get("GUANLAN_OLLAMA", None)
DETERMINISTIC = {"temperature": 0, "seed": 7}
BREAK_N, BREAK_S = 3, 60
_lock = threading.Semaphore(1); _state = {"fail": 0, "open_until": 0.0, "last_err": None, "n_calls": 0, "n_retry": 0, "n_break": 0}; _digest = {}


def endpoint() -> str:
    if ENDPOINT: return ENDPOINT.rstrip("/")
    try: return json.loads(CFG.read_text(encoding="utf-8")).get("endpoint", "http://127.0.0.1:11434").rstrip("/")
    except Exception: return "http://127.0.0.1:11434"


def _sha(x) -> str: return hashlib.sha256(json.dumps(x, ensure_ascii=False, sort_keys=True, default=str).encode()).hexdigest()[:16]


def digest(model: str) -> str | None:
    """模型摘要 (缓存): /api/show 的 digest; 拿不到回 None (审计里如实记)."""
    if model in _digest: return _digest[model]
    try:   # /api/tags 每条带 digest (sha256:…); /api/show 不带 (09-07 实逮拿成 file_type "15")
        with urllib.request.urlopen(endpoint() + "/api/tags", timeout=10) as r:
            for m in json.load(r).get("models", []):
                if m.get("name") in (model, model + ":latest") or m.get("model") in (model, model + ":latest"): _digest[model] = str(m.get("digest"))[:19]; break
            else: _digest[model] = None
    except Exception: _digest[model] = None
    return _digest[model]


def _audit(rec: dict):
    try:
        AUDIT.parent.mkdir(exist_ok=True)
        with AUDIT.open("a", encoding="utf-8") as f: f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")
    except Exception: pass


class ModelDown(RuntimeError):
    """熔断中或连续失败: 本机模型未就绪 (调用方应报 503/明示, 不得静默回落)."""


def state() -> dict:
    now = time.time(); return dict(open=now < _state["open_until"], open_for_s=max(0, round(_state["open_until"] - now)), fails=_state["fail"], last_err=_state["last_err"], n_calls=_state["n_calls"], n_retry=_state["n_retry"], n_break=_state["n_break"], endpoint=endpoint(), audit=str(AUDIT))


def post(body: dict, timeout: int = 420, purpose: str = "chat") -> dict:
    """向 Ollama /api/chat (或 body 含 'prompt' 时 /api/generate) 发一次请求. 返回解析后的 JSON. 4xx 原样抛 HTTPError."""
    now = time.time()
    if now < _state["open_until"]: raise ModelDown(f"本机模型熔断中 (连续 {BREAK_N} 次失败, {round(_state['open_until'] - now)} s 后重试): {_state['last_err']}")
    body = dict(body); body["options"] = {**DETERMINISTIC, **(body.get("options") or {})}
    path = "/api/generate" if "prompt" in body else "/api/chat"; model = body.get("model")
    rec = dict(ts=time.strftime("%Y-%m-%dT%H:%M:%S"), purpose=purpose, model=model, digest=digest(model), prompt_sha16=_sha(body.get("messages") or body.get("prompt")), options=body["options"], think=body.get("think"), tools=bool(body.get("tools")))
    with _lock:
        _state["n_calls"] += 1; last = None
        for attempt in (1, 2):
            t0 = time.time()
            try:
                req = urllib.request.Request(endpoint() + path, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"})
                with urllib.request.urlopen(req, timeout=timeout) as r: d = json.load(r)
                out = (d.get("message") or {}).get("content") if "message" in d else d.get("response")
                rec.update(attempt=attempt, latency_s=round(time.time() - t0, 2), status="ok", output_sha16=_sha(out), out_chars=len(out or "")); _audit(rec)
                _state["fail"] = 0; return d
            except urllib.error.HTTPError as e:
                detail = e.read().decode("utf-8", "replace")[:300]; last = f"HTTP {e.code}: {detail[:120]}"
                if e.code < 500:   # 4xx = 调用方问题 (如模型不支持 tools), 不重试, 原样抛 (保留 fast_agent 的退化逻辑)
                    rec.update(attempt=attempt, latency_s=round(time.time() - t0, 2), status=f"http{e.code}", error=detail[:200]); _audit(rec)
                    e2 = urllib.error.HTTPError(e.url, e.code, e.msg, e.hdrs, None); e2.read = lambda: detail.encode(); raise e2
            except (urllib.error.URLError, TimeoutError, ConnectionError, OSError, json.JSONDecodeError) as e:
                last = f"{e.__class__.__name__}: {str(e)[:120]}"
            rec.update(attempt=attempt, latency_s=round(time.time() - t0, 2), status="fail", error=last); _audit(dict(rec))
            if attempt == 1: _state["n_retry"] += 1; time.sleep(1.0)
        _state["fail"] += 1; _state["last_err"] = last
        if _state["fail"] >= BREAK_N: _state["open_until"] = time.time() + BREAK_S; _state["n_break"] += 1; _state["fail"] = 0
        raise ModelDown(f"本机模型调用失败 (已重试 1 次): {last}")


def probe(model: str = "qwen3:8b", timeout: int = 60) -> dict:
    """小探针 (用 generate 非 tags: 大模型加载期 tags 仍 200). 成功即清熔断."""
    t0 = time.time()
    try:
        post({"model": model, "prompt": "1", "stream": False, "options": {"num_predict": 1}}, timeout=timeout, purpose="probe")
        _state["open_until"] = 0.0; return dict(ok=True, latency_s=round(time.time() - t0, 2), digest=digest(model))
    except Exception as e: return dict(ok=False, err=str(e)[:160], latency_s=round(time.time() - t0, 2))


def reset(): _state.update(fail=0, open_until=0.0, last_err=None); _digest.clear()