tzt
4575025056
- features.py：gate() 纯函数——轮数/字符估算/意图黑名单/仓库级信号/长度门 -> t1_hard_ok（任一硬门不过即 False，D-G1） - grader.py：Grader.decide §8 时序（embed 降级检查 -> 特征门 -> LinearHead 概率 -> conformal 阈值：p1>=τ1 且 t1_hard_ok->T1，p3>=τ3 或 repo_signals->T3， 其余 T2 默认；collect/shadow 只写观察 executed=现行为，live 决策即执行； 全模式 observer.log）；工件/阈值缓存 + invalidate；D-G4 规则门退化 - routes：/v1/route 契约（D-G6 不落 query 原文） - fix(observer)：embedding list -> BLOB 转换（修 sqlite 绑定） - 测试 +6（门矩阵/shadow 不改流/live 决策/降级/保守阈值/写观察），全量 405 passed
129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
"""Grader/特征门测试（T-G5）：门×概率×mode 决策表 + fallback + /v1/route 契约。"""
|
||
import asyncio
|
||
import json
|
||
|
||
import pytest
|
||
|
||
pytest.importorskip("fastapi")
|
||
|
||
from gateway.sense.config import build_sense_config
|
||
from gateway.sense.errors import EmbedderDown
|
||
from gateway.sense.features import gate
|
||
from gateway.sense.grader import Grader
|
||
from gateway.sense.store import SenseStore
|
||
|
||
|
||
@pytest.fixture()
|
||
def env(tmp_path, monkeypatch):
|
||
cfg = build_sense_config({"sense": {
|
||
"enabled": True, "mode": "shadow",
|
||
"db_path": str(tmp_path / "s.sqlite3"),
|
||
"models_dir": str(tmp_path / "models")}})
|
||
store = SenseStore.init_db(cfg.db_path)
|
||
reset = getattr(__import__("gateway.sense.observer",
|
||
fromlist=["reset_observer"]), "reset_observer")
|
||
reset()
|
||
observer = __import__("gateway.sense.observer",
|
||
fromlist=["Observer"]).Observer(store)
|
||
g = Grader(cfg, store, observer)
|
||
yield {"cfg": cfg, "store": store, "grader": g, "observer": observer,
|
||
"tmp": tmp_path}
|
||
reset()
|
||
|
||
|
||
def _embed_ok(monkeypatch, dim=8):
|
||
import gateway.sense.embedder as em
|
||
orig = em.embed
|
||
|
||
async def fake(text, c):
|
||
return [1, 2, 3]
|
||
|
||
monkeypatch.setattr("gateway.sense.grader.embed", fake)
|
||
return orig
|
||
|
||
|
||
def _head_ok(tmp_path, monkeypatch, p1=0.95, p3=0.02):
|
||
"""注册一个 active head 工件，predict 可控。"""
|
||
import gateway.sense.grader as gr
|
||
|
||
class FakeHead:
|
||
version = "test-head"
|
||
|
||
def __init__(self, p1, p3):
|
||
self._p = (p1, 1 - p1 - p3, p3)
|
||
|
||
def predict(self, vec):
|
||
p1, p2, p3 = self._p
|
||
return {"t1": p1, "t2": p2, "t3": p3}
|
||
|
||
head = FakeHead(p1, p3)
|
||
monkeypatch.setattr(Grader, "_load_head",
|
||
lambda self: head if not self._head_loaded else None)
|
||
return head
|
||
|
||
|
||
def test_gate_features_matrix(env):
|
||
"""特征门全表：单轮/多轮/长度/黑名单/仓库信号。"""
|
||
cfg = env["cfg"]
|
||
short_ok = gate("什么是递归？", "pipeline", cfg)
|
||
assert short_ok.t1_hard_ok is True and short_ok.turns == 1
|
||
long_text = "x" * (cfg.t1_max_tokens * 3 + 10)
|
||
assert gate(long_text, "pipeline", cfg).t1_hard_ok is False
|
||
multi = [{"role": "user", "content": "a"}, {"role": "assistant", "content": "b"},
|
||
{"role": "user", "content": "c"}]
|
||
f2 = gate(multi, "pipeline", cfg)
|
||
assert f2.turns == 3 and f2.t1_hard_ok is False
|
||
for word in ("重构", "脚手架", "迁移", "实现", "多文件", "项目"):
|
||
f = gate(f"帮我{word}这个模块", "pipeline", cfg)
|
||
assert f.intent_blocked is True and f.t1_hard_ok is False
|
||
repo = gate("这个仓库要跨模块改造", "pipeline", cfg)
|
||
assert repo.repo_signals is True and repo.t1_hard_ok is False
|
||
|
||
|
||
def test_grader_shadow_does_not_change_flow(env, monkeypatch):
|
||
"""shadow：decided 照算，executed=规则门现行为，观察必写（flush 后落库）。"""
|
||
_embed_ok(monkeypatch)
|
||
_head_ok(env["tmp"], monkeypatch, p1=0.95, p3=0.02)
|
||
d = asyncio_run(env["grader"].decide("什么是递归？", "pipeline"))
|
||
assert d.tier == "T1" and d.mode == "shadow" and d.fallback is False
|
||
assert d.executed_tier == "T1" # 规则门现行为同为 T1
|
||
asyncio_run(env["observer"].flush_once())
|
||
obs_rows = env["store"].all_observations()
|
||
assert len(obs_rows) == 1
|
||
assert obs_rows[0]["decided_tier"] == "T1"
|
||
assert json.loads(obs_rows[0]["features"])["t1_hard_ok"] is True
|
||
|
||
|
||
def test_grader_live_tier_is_decision(env, monkeypatch):
|
||
"""live：决策即执行档。"""
|
||
env["cfg"].mode = "live"
|
||
_embed_ok(monkeypatch)
|
||
_head_ok(env["tmp"], monkeypatch, p1=0.3, p3=0.9)
|
||
d = asyncio_run(env["grader"].decide("多文件大改造任务", "proxy"))
|
||
assert d.tier == "T3"
|
||
assert d.mode == "live"
|
||
|
||
|
||
def test_grader_fallback_embedder_down(env, monkeypatch):
|
||
"""Embedder 挂 -> fallback=True 规则门退化（D-G4）。"""
|
||
def boom(text, c):
|
||
raise EmbedderDown("挂了")
|
||
monkeypatch.setattr("gateway.sense.grader.embed", boom)
|
||
d = asyncio_run(env["grader"].decide("什么是递归？", "pipeline"))
|
||
assert d.fallback is True
|
||
assert d.tier == "T1" # 规则门：短文本可 T1
|
||
assert d.probs == {}
|
||
|
||
|
||
def test_grader_conservative_thresholds(env, monkeypatch):
|
||
"""无工件 -> 保守阈值 0.9：p1=0.5 不判 T1。"""
|
||
_embed_ok(monkeypatch)
|
||
_head_ok(env["tmp"], monkeypatch, p1=0.5, p3=0.05)
|
||
d = asyncio_run(env["grader"].decide("什么是递归？", "pipeline"))
|
||
assert d.tier == "T2" # 置信不足默认 T2
|
||
assert d.thresholds_version == "conservative"
|
||
|
||
|
||
def asyncio_run(coro):
|
||
return asyncio.run(coro)
|