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
55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
"""特征门（T-G5）：单轮/长度/意图黑名单/仓库级信号（纯函数）。
|
||
|
||
D-G1：t1_hard_ok=False 时禁止判 T1（硬门）；repo_signals=True 时倾向 T3。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from gateway.sense.config import SenseConfig
|
||
|
||
_REPO_SIGNALS = ("多文件", "仓库", "项目结构", "跨模块", "架构", "脚手架",
|
||
"migration", "refactor", "scaffold", "repository")
|
||
|
||
|
||
@dataclass
|
||
class Features:
|
||
turns: int = 1 # 非 system 消息数
|
||
est_tokens: int = 0 # 字符/3 估算（中文友好近似）
|
||
single_turn: bool = True
|
||
intent_blocked: bool = False # 命中意图黑名单
|
||
repo_signals: bool = False # 仓库级/多文件信号
|
||
over_length: bool = False # 超出 t1 长度门
|
||
t1_hard_ok: bool = True # 任一 T1 硬门不过即 False
|
||
|
||
|
||
def _to_text_and_turns(text_or_messages) -> tuple[str, int]:
|
||
if isinstance(text_or_messages, str):
|
||
return text_or_messages, 1
|
||
msgs = [m for m in (text_or_messages or []) if isinstance(m, dict)]
|
||
non_system = [m for m in msgs if str(m.get("role")) != "system"]
|
||
text = "\n".join(str(m.get("content") or "") for m in non_system)
|
||
return text, max(1, len(non_system))
|
||
|
||
|
||
def gate(text_or_messages, consumer: str, cfg: SenseConfig) -> Features:
|
||
"""特征门（§7 签名）：返回门特征 + t1_hard_ok。"""
|
||
text, turns = _to_text_and_turns(text_or_messages)
|
||
est_tokens = len(text) // 3
|
||
single_turn = turns <= 1
|
||
|
||
blocked = any(word in text for word in cfg.intent_blacklist)
|
||
repo = any(word in text for word in _REPO_SIGNALS)
|
||
over_length = est_tokens > cfg.t1_max_tokens
|
||
multi_turn = turns > cfg.t1_max_turns
|
||
|
||
t1_hard_ok = single_turn and not over_length and not blocked and not repo \
|
||
and not multi_turn
|
||
|
||
return Features(
|
||
turns=turns, est_tokens=est_tokens, single_turn=single_turn,
|
||
intent_blocked=blocked, repo_signals=repo,
|
||
over_length=over_length, t1_hard_ok=t1_hard_ok)
|