#!/usr/bin/env python3 """jev：只做判斷的模型。XKB 用它回答「這筆知識有沒有回答這個問題」。 jev 不是聊天模型，所以 `_llm.call` 打不到它——它沒有 `/v1/chat/completions`： POST {LLM_API_URL}/systemone {"model": "jev-1.13", "state": "<要判斷的內容>", "questions": {"<名稱>": {"type": "noul|choice|score", "instructions": ...}}} 回傳的是有型別的答案，不是文字。`noul` 回 0~1 的機率，`choice` 回選項， `score` 回等級。計費只算輸入。 **為什麼值得接。** 2026-09-24 在正式資料上量過：一次呼叫裡塞四個判斷，總共 1.85 秒——跟單一判斷一樣，因為 questions 是平行評估的。所以 N 個候選只要一次 往返，不是 N 次。而它的分離度遠勝餘弦： 同一批候選 jev 餘弦 正面回答 0.31 / 0.54 0.480 完全不相關 0.02 / 0.01 0.446 / 0.435 餘弦把相關與不相關壓在 0.435~0.480 這 7% 的區間裡，而固定門檻 0.55 就壓在 那一段的上方——正式資料上 980 次召回丟掉了 3,886 筆候選，而抽樣檢查發現 「食品展的客戶通常怎麼找」這種問題的正確答案就落在 0.480，被丟掉了。餘弦分不 出「講同一個主題」跟「回答了這個問題」，jev 的 noul 正好就是後者。 **這支不做決定。** 它只回答。現行 Knowledge Service 預設在合併點採用判斷； `XKB_JEV_DECIDE=0` 才停用，影子比較另由 `XKB_JEV_SHADOW` 控制。 既有相關性政策由呼叫端管理，本 adapter 不另設門檻。 **失敗一律開放通過。** 判斷不出來時回 None，而不是 0。這兩件事差很多：0 是 「jev 說不相關」，None 是「jev 沒跑」。把它們寫成同一件事，就會變成 jev 一掛掉 知識庫就靜默關閉——那正是這個專案吃過最大虧的那種失敗。 """ from __future__ import annotations import json import sys import time from concurrent.futures import ThreadPoolExecutor import urllib.error import urllib.request from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools")) import xkb_failures from runtime_config import runtime_env, llm_service_env _SKILL_DIR = Path(__file__).resolve().parent.parent _CONFIG_FILE = _SKILL_DIR / "config" / "llm.json" # Judge has its own XKB_JUDGE_API_URL/API_KEY route, independent of generation. # Legacy LLM_* is supported only when no scoped judge endpoint/key is set. DEFAULT_JUDGE_MODEL = "jev-1.13" JUDGE_PATH = "/systemone" # 判斷要嘛很快要嘛沒用。實測 1.85 秒，留三倍餘裕；超過就當它沒跑。 # 召回路徑上多等十秒比少一筆弱候選糟得多。 TIMEOUT_SECONDS = 6 def judge_model() -> str: env_model = runtime_env().get("XKB_JUDGE_MODEL", "") if env_model: return env_model try: cfg = json.loads(_CONFIG_FILE.read_text(encoding="utf-8")) return cfg.get("judge_model") or DEFAULT_JUDGE_MODEL except Exception as err: # noqa: BLE001 if _CONFIG_FILE.exists(): xkb_failures.note("jev config", err, detail=str(_CONFIG_FILE)) return DEFAULT_JUDGE_MODEL def available() -> bool: """有沒有可用的憑證。沒有的話呼叫端要當作「沒跑」，不是「判斷為否」。""" settings = llm_service_env("judge") return bool(settings.get("LLM_API_URL") and settings.get("LLM_API_KEY")) MAX_BODY_BYTES = 32 * 1024 def _body(state: str, questions: dict) -> bytes: return json.dumps({"model": judge_model(), "state": state, "questions": questions}, ensure_ascii=False, separators=(",", ":")).encode("utf-8") def judge(state: str, questions: dict[str, dict], *, timeout: float = TIMEOUT_SECONDS) -> dict | None: """問 jev 一組問題。回 answers；判斷不出來回 None。 None 與「答案是 0」是兩件不同的事，呼叫端必須分開處理——見模組說明。 """ if not state or not questions or not available(): return None settings = llm_service_env("judge") url = settings["LLM_API_URL"].rstrip("/") + JUDGE_PATH body = _body(state, questions) if len(body) > MAX_BODY_BYTES: xkb_failures.note("jev budget", ValueError("request exceeds 32 KiB")) return None request = urllib.request.Request( url, data=body, method="POST", headers={"Content-Type": "application/json", "Authorization": f"Bearer {settings['LLM_API_KEY']}"}) try: with urllib.request.urlopen(request, timeout=timeout) as response: payload = json.loads(response.read().decode("utf-8")) except (urllib.error.URLError, OSError, json.JSONDecodeError, ValueError) as err: xkb_failures.note("jev judge", err, detail=url) return None if not isinstance(payload, dict): return None answers = payload.get("answers") if not isinstance(answers, dict): data = payload.get("data") answers = data.get("answers") if isinstance(data, dict) else None return answers if isinstance(answers, dict) else None def relevance(query: str, candidates: list[tuple[str, str]], *, timeout: float = TIMEOUT_SECONDS) -> dict[str, float] | None: """每個候選有沒有回答這個問題。回 {key: 0~1}；判斷不出來回 None。 以最終序列化 UTF-8 bytes 分批，每批含 model/state 不超過 32 KiB。 小批維持一次呼叫；大批最多四路並行，共用 timeout deadline。 缺失或失敗批次不產生否定答案，呼叫端保留並警告未判斷項目。 候選文字會被截斷：instructions 是判斷準則，不是整篇文件，而卡片可以很長。 標題加摘要足以判斷「能否推進當前需求」——真要整篇才判斷得出來的情況， 那筆本來就不該直接注入。 """ if not query or not candidates: return None questions = {} slots: dict[str, list[str]] = {} text_slots: dict[str, str] = {} for index, (key, text) in enumerate(candidates): if not key or not text: continue # All candidates share the same current-turn state. Identical effective # questions need one verdict, even when multiple retrieval legs found # them. Keep every key for fan-out; do not collapse retrieval support. effective_text = _trim(text) if effective_text in text_slots: slots[text_slots[effective_text]].append(key) continue # 問題名稱用序號，不用候選的 key：key 是檔案路徑，裡面有斜線與中文， # 而它會變成回應 JSON 的欄位名。序號對回去就好。 slot = f"q{index}" text_slots[effective_text] = slot slots[slot] = [key] questions[slot] = { "type": "noul", "instructions": ("這段知識能否提供推進當前需求的做法、限制或經驗；" "只匹配過去話題而無助於當前需求不算相關。" f"知識內容：{effective_text}"), } if not questions: return None state = ("判斷證據能否推進使用者當前需求，陳述計畫、困難或限制也可能需要知識。" "前文只協助理解指代；當前發言改變方向時，以當前需求為準。\n" "使用者已暫停或放棄的話題不算當前需求。已在助手前文完整說過的做法，" "除非使用者要求重述或出現新的適用條件，否則沒有新增幫助；" "未回應不代表接受或拒絕。\n" f"對話情境：{_trim(query, 4096)}") batches, batch = [], {} for slot, question in questions.items(): proposed = {**batch, slot: question} if len(_body(state, proposed)) > MAX_BODY_BYTES: if batch: batches.append(batch) batch = {slot: question} else: batch = proposed if batch: batches.append(batch) # A shared deadline bounds queued work; small requests stay one call. deadline = time.monotonic() + timeout def run(batch): remaining = deadline - time.monotonic() return judge(state, batch, timeout=remaining) if remaining > 0 else None if len(batches) == 1: results = [run(batches[0])] else: with ThreadPoolExecutor(max_workers=min(4, len(batches))) as pool: results = list(pool.map(run, batches)) answers = {} for result in results: if result is not None: answers.update(result) if not answers: return None out: dict[str, float] = {} for slot, keys in slots.items(): answer = answers.get(slot) if not isinstance(answer, dict): continue value = answer.get("noul") if isinstance(value, (int, float)) and not isinstance(value, bool) and 0 <= value <= 1: for key in keys: out[key] = float(value) # 一個都對不上就是回應的形狀跟預期不同——那是「沒跑成」，不是「全部不相關」。 return out or None def _trim(text: str, limit: int = 900) -> str: text = " ".join(str(text or "").split()) return text if len(text) <= limit else text[:limit] + "…" import xkb_usage # noqa: E402 — 量測誰在跑，見 scripts/xkb_usage.py if __name__ == "__main__": xkb_usage.record(__file__) demo_query = "食品展的客戶通常怎麼找" demo = [ ("cards/a.md", "公開企業名錄通常不直接提供完整聯絡資訊；食品展官方" "資料可能只有公司名稱、攤位、官網、產品與簡介，沒有電話。"), ("cards/b.md", "AI 影片生成的鏡頭運動提示詞寫法，包含推軌、環繞與特寫。"), ] scores = relevance(demo_query, demo) if scores is None: print("jev 沒跑成——憑證沒設，或端點打不通。這不代表候選不相關。") else: print(f"問題：{demo_query}") for key, value in sorted(scores.items(), key=lambda kv: -kv[1]): print(f" {value:.2f} {key}")