tests/test_jev_shadow_changes_nothing.py
Rama main · réplica de lectura del repositorio original.
"""jev 影子模式：只觀測，不改變任何召回結果。
接一個新的判斷模型有兩種搞砸的方式，這份測試各擋一種：
**它偷偷改變了行為。** 影子模式的全部價值在於「先量再決定」。只要它動到
kept/dropped，之後拿來校準門檻的那份資料就是它自己造成的，而不是餘弦真正的
決定。這個專案在「驗證了有動作，不是動得對」上有紀錄在案。
**「沒跑成」被寫成「判斷為不相關」。** jev 回 None 代表沒跑（憑證沒設、端點
打不通、回應形狀不對），不是 0。寫成同一件事的話，jev 一掛掉，分析資料會長得
像「它覺得每一筆都不相關」，而如果之後讓它決定，知識庫會靜默關閉——那正是這個
專案吃過最大虧的那種失敗（recall 壞了 12 週沒人發現）。
還有一條是效能：jev 實測 1.85 秒，而 questions 是平行評估的，所以 N 個候選必須
一次呼叫。逐一呼叫的話 10 個候選要 18 秒，比召回本身還慢。
"""
from __future__ import annotations
import json
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
import xkb_jev # noqa: E402
from xkb_memory_service import Store # noqa: E402
class _Inline:
"""把背景執行緒改成同步執行，好在測試裡斷言它做了什麼。"""
def __init__(self, target=None, daemon=None, name=None):
self._target = target
def start(self):
self._target()
class TheClientTellsSilenceFromRejection(unittest.TestCase):
def setUp(self):
self.env = {"LLM_API_URL": "https://api.example/v1", "LLM_API_KEY": "k"}
def test_no_credentials_means_did_not_run_not_irrelevant(self):
with mock.patch.object(xkb_jev, "runtime_env", return_value={}):
self.assertFalse(xkb_jev.available())
self.assertIsNone(xkb_jev.judge("s", {"q": {"type": "noul"}}))
self.assertIsNone(xkb_jev.relevance("問題", [("a", "內容")]))
def test_a_transport_failure_returns_none(self):
with mock.patch.object(xkb_jev, "runtime_env", return_value=self.env), \
mock.patch.object(xkb_jev.urllib.request, "urlopen",
side_effect=OSError("連不上")):
self.assertIsNone(xkb_jev.relevance("問題", [("a", "內容")]))
def test_a_response_of_the_wrong_shape_returns_none(self):
"""回了 200 但形狀不對，一樣是「沒跑成」。
對不上任何候選卻回一個空 dict 的話，呼叫端會把它當成「全部不相關」。
"""
with mock.patch.object(xkb_jev, "runtime_env", return_value=self.env), \
mock.patch.object(xkb_jev, "judge", return_value={"別的": {"noul": 1}}):
self.assertIsNone(xkb_jev.relevance("問題", [("a", "內容")]))
def test_a_real_zero_is_a_real_answer(self):
with mock.patch.object(xkb_jev, "runtime_env", return_value=self.env), \
mock.patch.object(xkb_jev, "judge",
return_value={"q0": {"type": "noul", "noul": 0.0}}):
self.assertEqual(xkb_jev.relevance("問題", [("a", "內容")]), {"a": 0.0})
class OneCallForEveryCandidate(unittest.TestCase):
"""questions 是平行評估的。逐一呼叫會比召回本身還慢。"""
def test_ten_candidates_are_one_request(self):
seen = []
def spy(state, questions, **kwargs):
seen.append(questions)
return {name: {"type": "noul", "noul": 0.5} for name in questions}
candidates = [(f"cards/{i}.md", f"內容 {i}") for i in range(10)]
with mock.patch.object(xkb_jev, "runtime_env",
return_value={"LLM_API_URL": "u", "LLM_API_KEY": "k"}), \
mock.patch.object(xkb_jev, "judge", spy):
out = xkb_jev.relevance("問題", candidates)
self.assertEqual(len(seen), 1, "十個候選要一次呼叫，不是十次")
self.assertEqual(len(seen[0]), 10)
self.assertEqual(len(out), 10)
def test_question_names_do_not_leak_file_paths(self):
"""問題名稱會變成回應 JSON 的欄位名，而候選的 key 是含斜線的路徑。"""
seen = {}
def spy(state, questions, **kwargs):
seen.update(questions)
return {name: {"type": "noul", "noul": 0.5} for name in questions}
with mock.patch.object(xkb_jev, "runtime_env",
return_value={"LLM_API_URL": "u", "LLM_API_KEY": "k"}), \
mock.patch.object(xkb_jev, "judge", spy):
out = xkb_jev.relevance("問題", [("01-topic/2040096164452958475.md", "內容")])
for name in seen:
self.assertNotIn("/", name)
self.assertIn("01-topic/2040096164452958475.md", out,
"序號要對得回原本的 key")
def test_a_long_card_is_trimmed(self):
seen = {}
def spy(state, questions, **kwargs):
seen.update(questions)
return {name: {"type": "noul", "noul": 0.5} for name in questions}
with mock.patch.object(xkb_jev, "runtime_env",
return_value={"LLM_API_URL": "u", "LLM_API_KEY": "k"}), \
mock.patch.object(xkb_jev, "judge", spy):
xkb_jev.relevance("問題", [("a", "很長" * 5000)])
body = next(iter(seen.values()))["instructions"]
self.assertLess(len(body), 1200, "instructions 是判斷準則，不是整篇文件")
class ShadowModeDecidesNothing(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.store = Store(Path(self.tmp.name) / "memory.sqlite")
self.records = [
{"id": "cards/good.md", "title": "好的", "summary": "正面回答"},
{"id": "cards/weak.md", "title": "弱的", "summary": "沾到邊"},
]
def _rows(self) -> list[dict]:
with self.store.connect() as db:
db.row_factory = __import__("sqlite3").Row
return [dict(r) for r in db.execute(
"SELECT * FROM relevance_shadow ORDER BY id")]
def _drop(self, *, jev_result):
"""跑一次過濾，jev 回 jev_result。回傳 (kept, dropped, 寫進去的列)。"""
catalog = self.store.catalog
# 影子模式只在 jev 還沒接管決定時存在。接管之後它就是第二次呼叫同一個
# 模型，所以預設關掉——要測影子模式就得明確說「現在不是 jev 在決定」。
with mock.patch.dict("os.environ", {"XKB_JEV_DECIDE": "0"}), \
mock.patch.object(xkb_jev, "available", return_value=True), \
mock.patch.object(xkb_jev, "relevance", return_value=jev_result), \
mock.patch("xkb_memory_service.threading.Thread", _Inline), \
mock.patch("xkb_memory_service.xkb_relevance.filter_irrelevant",
return_value=([self.records[0]], 1,
{"cards/good.md": 0.72, "cards/weak.md": 0.41})), \
mock.patch("xkb_memory_service.xkb_relevance.vector_key",
side_effect=lambda rid: rid), \
mock.patch("xkb_memory_service.xkb_relevance.min_similarity",
return_value=0.55):
kept, dropped = catalog._drop_irrelevant("問題", list(self.records))
return kept, dropped, self._rows()
def test_the_kept_set_is_the_cosine_decision(self):
"""jev 說全部都相關，也不准改變誰被留下來。"""
kept, dropped, rows = self._drop(
jev_result={"cards/good.md": 0.9, "cards/weak.md": 0.9})
self.assertEqual([r["id"] for r in kept], ["cards/good.md"])
self.assertEqual(dropped, 1)
self.assertEqual({r["record_id"]: r["kept"] for r in rows},
{"cards/good.md": 1, "cards/weak.md": 0})
def test_both_scores_are_recorded_side_by_side(self):
"""要回答的問題是「兩邊在哪裡不一致」，所以兩個分數都要留。"""
_kept, _dropped, rows = self._drop(
jev_result={"cards/good.md": 0.30, "cards/weak.md": 0.02})
by_id = {r["record_id"]: r for r in rows}
self.assertAlmostEqual(by_id["cards/good.md"]["cosine"], 0.72)
self.assertAlmostEqual(by_id["cards/good.md"]["jev"], 0.30)
self.assertAlmostEqual(by_id["cards/weak.md"]["cosine"], 0.41)
self.assertAlmostEqual(by_id["cards/weak.md"]["jev"], 0.02)
self.assertEqual({r["floor"] for r in rows}, {0.55})
self.assertEqual({r["query"] for r in rows}, {"問題"})
def test_jev_not_running_writes_nothing_rather_than_a_row_of_nulls(self):
"""一整批 jev=NULL 的列，在分析時長得像「它覺得每一筆都不相關」。"""
kept, dropped, rows = self._drop(jev_result=None)
self.assertEqual([r["id"] for r in kept], ["cards/good.md"])
self.assertEqual(dropped, 1)
self.assertEqual(rows, [], "沒跑成就不要留下會被誤讀的紀錄")
def test_a_crash_inside_the_shadow_never_breaks_recall(self):
with mock.patch.dict("os.environ", {"XKB_JEV_DECIDE": "0"}), \
mock.patch.object(xkb_jev, "available", return_value=True), \
mock.patch.object(xkb_jev, "relevance",
side_effect=RuntimeError("炸了")), \
mock.patch("xkb_memory_service.threading.Thread", _Inline), \
mock.patch("xkb_memory_service.xkb_relevance.filter_irrelevant",
return_value=([self.records[0]], 1, {})), \
mock.patch("xkb_memory_service.xkb_relevance.vector_key",
side_effect=lambda rid: rid), \
mock.patch("xkb_memory_service.xkb_relevance.min_similarity",
return_value=0.55):
kept, dropped = self.store.catalog._drop_irrelevant(
"問題", list(self.records))
self.assertEqual([r["id"] for r in kept], ["cards/good.md"])
self.assertEqual(dropped, 1)
def test_it_can_be_turned_off(self):
with mock.patch.dict("os.environ", {"XKB_JEV_SHADOW": "0",
"XKB_JEV_DECIDE": "0"}), \
mock.patch.object(xkb_jev, "available", return_value=True), \
mock.patch.object(xkb_jev, "relevance") as spy, \
mock.patch("xkb_memory_service.threading.Thread", _Inline), \
mock.patch("xkb_memory_service.xkb_relevance.filter_irrelevant",
return_value=([self.records[0]], 1, {})), \
mock.patch("xkb_memory_service.xkb_relevance.vector_key",
side_effect=lambda rid: rid), \
mock.patch("xkb_memory_service.xkb_relevance.min_similarity",
return_value=0.55):
self.store.catalog._drop_irrelevant("問題", list(self.records))
spy.assert_not_called()
self.assertEqual(self._rows(), [])
class ItDoesNotBlockRecall(unittest.TestCase):
def test_the_comparison_runs_off_the_recall_path(self):
"""jev 是 1.85 秒。影子模式下沒有東西依賴它的答案，不該讓任何人多等。"""
import inspect
from xkb_memory_service import KnowledgeCatalog
source = inspect.getsource(KnowledgeCatalog._shadow_compare)
self.assertIn("threading.Thread", source)
self.assertIn("daemon=True", source)
class TheJudgeModelIsNotTheChatModel(unittest.TestCase):
"""把 llm.json 的 model 改成 jev 會讓所有生成腳本一起壞掉。"""
def test_the_two_models_are_separate_settings(self):
config = json.loads((ROOT / "config" / "llm.json").read_text(encoding="utf-8"))
self.assertIn("judge_model", config)
self.assertNotEqual(config["model"], config["judge_model"])
def test_the_endpoint_is_not_chat_completions(self):
self.assertEqual(xkb_jev.JUDGE_PATH, "/systemone")
self.assertNotIn("chat", xkb_jev.JUDGE_PATH)
if __name__ == "__main__":
unittest.main()