scripts/xkb_jev.py
Rama main · réplica de lectura del repositorio original.
#!/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}")