20 KiB
20 KiB
实施方案：语义分析器与三级任务分级（T-G0…T-G8 执行版）
编写日期：2026-09-05 状态：执行版（设计依据《方案_校园AI代理层.md》§7；冲突以本文为准） 前置依赖：代理层 T-P0…T-P8（观察埋点依赖其账本/网关基建；T-G0/G1 可与之并行）。 预期读者：负责实现的 AI Agent。必读：§1 锁定决策、§2 三级规格、§11 测试纪律。 总工期约 11 个工作日；里程碑 M-G1（采集）→ M-G2（shadow）→ M-G3（live 分流）。
0. 范围与非目标
范围：语义分析器（共享 Embedding 底座 + 轻量分级头）→ 三级难度档位 → 三类消费方 （v2 编码管线 / 校园代理 / 端侧客户端）的执行逻辑分流；观察-标签-校准-重训数据闭环； shadow→live 灰度；审计抽样接 ReviewQueue。
非目标（超纲不做）：LoRA/vLLM 分类服务（预留接口，T-G8 可选实验）；多语言分类头； 实时在线学习（夜间批处理为准）；情感/内容安全分类。
核心改动约束：router_system/ 仅允许一处受控改动——pipeline.py 注入可选
tier_fn 钩子（默认 None = 现行为逐字节不变，见 D-G5）；其余全部落在 gateway/sense/。
1. 锁定决策（实现中不得擅改）
- D-G1 分级是建议，阶梯是权威：TierDecision 只决定入口档位；任何档位失败都沿 T1→T2→T3 升级阶梯爬升（复用现有验证器/升级回路）。分析器错误的上界 = 一次浪费的尝试。
- D-G2 标签只来自结果观测：禁止人工标注难度作为训练集主体（人工仅审计抽样纠偏）。 true_tier 由夜间任务从 outcome 推导（§4 规则）。
- D-G3 一切决策带版本：每条 TierDecision 记录 policy_version（= 模型工件版本 + 阈值版本）；阈值由 split-conformal 夜间重算，数据不足（<500 条标签）时沿用 last-good。
- D-G4 降级阶梯：Embedder 挂 → 决策退化为规则门（§2 特征门）+ 默认 T2； 分类器工件缺失 → 同上。任何智能组件故障不得影响代理可用性。
- D-G5 零依赖与钩子：serving 路径零新依赖（线性头推理 = 纯 Python 点积，768 维
int8 ≈ 0.1ms）；numpy 仅离线训练用（requirements-ml.txt，可选安装）。
pipeline.py只加tier_fn: Optional[Callable]构造参数与一处 if 分支。 - D-G6 隐私与留存：embedding 存 int8 BLOB、按课程桶分区；观察记录留存 180 天
（与账本归档对齐，夜间任务顺带清理）；
/v1/route不落原始 query 全文（只存哈希+特征）。 - D-G7 灰度开关：
sense.enabled（总开关，默认 false）+sense.mode：collect（只记不决策）|shadow（决策只写日志比对）|live（真分流）。 非 live 模式下所有消费方行为与现状一致。 - D-P 系全部继承（毫元整数/凭据/单进程等），本文不重复。
2. 三级任务分级规格（罗列）
| 档 | 典型任务 | 特征门（硬性，任一不过即降档） | 执行逻辑 | 验证方式 | 失败升级 | 目标占比 |
|---|---|---|---|---|---|---|
| T1 简单 | 概念解释、改写/翻译/摘要、格式转换、简单计算、单函数补全、报错一句话解释 | 单轮；输入 ≤512 tok；无仓库/多文件上下文；意图不在黑名单（实现/重构/脚手架/迁移/多文件）；code 域限解释/补全/单函数 | 本地小模型直答（客户端本机 / 校端 local-small 池条目） | code→接地验证（跑测试/facts）；QA→结构检查+facts 抽检 | →T2 | 40–55% |
| T2 中等 | 单文件函数实现、修 bug、带约束生成、模板代码、数据解析 | 单文件可完成；输入输出明确；验证可自动化；分析器置信中段 | 单次云端直答（budget 档，deepseek-chat）或本地大模型直答（校端 9B+ 空闲时优先本地） | code→跑测试；QA→LLM rubric 抽检 | →T3 | 30–45% |
| T3 复杂 | 多文件/跨模块重构、项目脚手架、多约束需求、架构决策、长周期 agent 任务 | 仓库级上下文；需多步计划；多验收点；历史升级信号（曾从 T1/T2 升级） | 四阶段管线：云 brief → 本地大模型构建+检测 → 云 decide 修复 → 云 final_review | 管线内置（Worker 接地验证 + final_review + 审计抽样） | 终审打回→管线内循环（rounds_cap） | 10–20% |
消费方 × 档位 = 执行逻辑（同一 TierDecision，三类后果）：
| consumer | T1 | T2 | T3 |
|---|---|---|---|
pipeline（v2 编码智能体） |
现有 fast path（本地直答） | 新增档：单次云端直答（跳过 brief/review） | 完整四阶段管线 |
proxy（API 透传） |
model_pool local-small 端点 |
budget 云端条目 |
premium 云端条目（响应头建议走管线） |
client（端侧） |
本机模型直答 | 经代理云端直答 | 完整管线（本机=Worker，云=Architect） |
3. 文件清单
新建：
gateway/sense/
├── __init__.py # build_sense_router(settings) -> APIRouter；唯一组装点
├── config.py # SenseConfig：buckets/consumer 映射/阈值路径/mode
├── errors.py # EmbedderDown/ArtifactMissing
├── embedder.py # llama-server /v1/embeddings 客户端 + int8 量化 + 降级
├── features.py # 特征门：token 估算/轮数/意图黑名单/多文件检测（纯函数）
├── classifier.py # TierHead 接口：LinearHead(npz 纯 Python 推理) | LoraRemote(vLLM，预留)
├── calibrate.py # split-conformal 阈值计算 + last-good 回退 + 阈值工件读写
├── labeler.py # 夜间 true_tier 推导 + 180d 清理
├── observer.py # 观察写入（to_thread + 100ms 批量）；消费方埋点 SDK
├── grader.py # 决策组合：probs + 特征门 + conformal 阈值 + mode
├── store.py # sense.sqlite3 DDL + 查询
└── routes.py # /v1/route、/v1/embeddings、/sense/admin/*
scripts/train_tier_head.py # 离线训练（numpy，requirements-ml.txt）
scripts/sense_report.py # shadow 一致率/分布/覆盖率报告（E-G1 数据管道）
webapp/src/views/… # 不新增页面；Settings 加 sense 区块、Metrics 加 tier 卡（T-G7）
tests/test_sense_{embedder,features,classifier,calibrate,labeler,grader,routes}.py
修改：gateway/api.py（include_router，sense.enabled 门控，≤5 行）；
router_system/pipeline.py（tier_fn 钩子，D-G5 唯一受控改动）；
gateway/proxy/routes.py（T1/T2/T3 → model_pool 档位映射，sense.mode=live 时启用）；
config/settings.json（sense 段）；任务拆解与执行计划.md、AGENTS.md（登记）。
4. 数据模型（data/sense.sqlite3，WAL；全部经 to_thread）
CREATE TABLE tier_observations(
id INTEGER PRIMARY KEY, ts INTEGER NOT NULL,
request_id TEXT NOT NULL, consumer TEXT NOT NULL, -- pipeline|proxy|client
bucket TEXT NOT NULL DEFAULT 'default', domain TEXT DEFAULT '',
decided_tier TEXT NOT NULL, executed_tier TEXT NOT NULL, -- shadow 期二者可不同
probs TEXT NOT NULL, -- JSON {t1,t2,t3}
policy_version TEXT NOT NULL, features TEXT NOT NULL, -- 门特征 JSON（不含 query 原文）
embedding BLOB, -- int8 量化向量（query 哈希可关联）
outcome TEXT DEFAULT '', -- ok|verified|escalated|failed|user_retry|timeout
true_tier TEXT DEFAULT '', human_override TEXT DEFAULT '');
CREATE INDEX idx_obs_ts ON tier_observations(ts);
CREATE INDEX idx_obs_policy ON tier_observations(policy_version);
CREATE TABLE sense_artifacts( -- 模型/阈值工件登记
version TEXT PRIMARY KEY, kind TEXT NOT NULL, -- head|thresholds
path TEXT NOT NULL, metrics TEXT NOT NULL, created_ts INTEGER NOT NULL, active INTEGER DEFAULT 0);
true_tier 夜间推导规则（labeler.py，确定性）：
- executed=T1 且 outcome∈{ok,verified} 且 24h 内无 user_retry → true=T1；
- outcome∈{escalated,failed,user_retry,timeout} → true = executed 的下一档（T1→T2→T3，T3 保持 T3）；
- consumer=pipeline 且管线发出 brief（route 含 plan:multi）→ true=T3；
- human_override 非空 → 以人工为准（审计抽样回写）。
5. 配置 schema（settings.json 的 sense 段）
"sense": {
"enabled": false,
"mode": "collect", // collect | shadow | live
"embedder": {"base_url": "http://127.0.0.1:8902/v1",
"model": "bge-m3-Q4_K_M", "timeout_s": 5, "dim": 1024},
"features": {"t1_max_tokens": 512, "t1_max_turns": 2,
"intent_blacklist": ["重构","脚手架","迁移","实现","多文件","项目"],
"code_t1_kinds": ["解释","补全"]},
"policy": {"alpha": 0.05, // conformal：P(true>T1|判T1) ≤ α
"min_labels": 500, "t2_prefer_local_when_idle": true},
"consumers": {"proxy": {"t1": "local-small", "t2": "budget", "t3": "premium"}}
}
model_pool 条目需相应增加 tier_hint 字段（local-small/budget/premium），缺失时按
现有 local/budget/premium 映射兜底。
6. API 契约
| 端点 | 说明 |
|---|---|
POST /v1/route |
{query 或 messages, consumer, domain?} → {tier, probs, confidence, thresholds_version, head_version, mode, fallback:bool}；不落 query 原文 |
POST /v1/embeddings |
OpenAI 兼容透传 embedder（客户端/代理共用） |
GET /sense/admin/agreement?since= |
shadow 一致率、档位分布、逐域混淆矩阵（E-G1） |
POST /sense/admin/calibrate |
手动触发 conformal 重算（夜间任务自动跑） |
POST /sense/admin/train |
手动触发线性头训练（新工件 version，默认不 active，人工切换） |
POST /sense/admin/promote |
工件/阈值切 active（带 last-good 回滚参数） |
live 晋升门（shadow → live，三条同时满足）：一致率 ≥85%；conformal 覆盖实测 P(true>T1|判T1) ≤ α+2%；审计抽样（ReviewQueue）无"判 T1 实为 T3"的严重误判。
7. 模块接口签名
# embedder.py
async def embed(text: str) -> list[int] # int8 量化向量；失败 raise EmbedderDown
# features.py
def gate(text_or_messages, consumer, cfg) -> Features # 门特征 + t1_hard_ok: bool
# classifier.py
class LinearHead: load(path); predict(vec) -> dict[str,float] # 纯 Python 点积 softmax
class LoraRemote: predict(vec) -> dict[str,float] # vLLM /v1/classify，T-G8 预留
# calibrate.py
def compute_thresholds(labels, probs, alpha) -> Thresholds # split-conformal
def load_active(store) -> Thresholds # 无合格工件→last-good→内置保守值
# grader.py
async def decide(query_or_messages, consumer, domain, store, cfg, now) -> TierDecision
# TierDecision: tier, probs, hard_gates, thresholds_version, head_version, fallback
# 决策顺序：embedder 降级检查 → 特征门（t1_hard_ok）→ conformal 阈值比较 → mode 裁剪
# collect/shadow：decided_tier 照算、executed_tier=现行为，只写观察不改流
# observer.py
async def log(obs: Observation) -> None # 批量缓冲 100ms 刷盘（to_thread）
# labeler.py
def derive_true_tiers(store, now) -> int # 返回回填条数；顺带 180d 清理
8. 分级决策流程（grader.py 主时序）
输入 → embedder.embed（挂→fallback=T2+规则门，标记 fallback:true）
→ features.gate（t1_hard_ok）
→ classifier.predict(vec) → {p1,p2,p3}
→ conformal 阈值 τ1（P(true>T1|判T1)≤α）：
p1 ≥ τ1 且 t1_hard_ok → T1
p3 ≥ τ3 或 特征含仓库级/多验收点 → T3
其余 → T2（含置信不足默认）
→ mode 裁剪：collect/shadow → 只写 tier_observations（decided≠executed）
live → 返回决策，消费方执行
→ observer.log（全模式必写）
升级阶梯（消费方内实现，非 grader 职责）：T1 失败（验证不过/超时）→ T2； T2 失败（测试仍不过/用户重试）→ T3；阶梯每爬一级写 observer（outcome=escalated）。
9. 与现有系统的集成点
- v2 管线（
pipeline.py，D-G5 钩子）：build_v2_pipeline(..., tier_fn=...)；tier_fn=None现行为不变。live 时：T1=fast path；T2=新档（单次云端直答， 复用 ArchitectClient 单次调用路径，跳过 brief/review）；T3=完整管线。 现有测试零回归（tier_fn 缺省路径逐字节一致）。 - 代理层（
proxy/routes.py）：sense.mode=live时，转发前调 grader， 按 §2 消费方表映射 model_pool 档位；响应头X-Campus-Tier透出（可解释性）。 - 客户端：
/v1/route直调；端侧 UI 显示"端侧/云端"来源与一键升级 （复用 v3 前端基建，T-G7）。 - ReviewQueue：T1 决策按
review.sample_rate抽样入队（人工核"判 T1 实为更高档"， human_override 回写观察表）。
10. 工件与训练（scripts/train_tier_head.py）
- 输入：tier_observations 中 true_tier 非空且 policy_version 对应 embedding 可解析的行； 按时间切分 train/val/calib = 70/15/15。
- 模型：线性有序三分类（softmax 回归，纯 numpy，类别权重均衡）；输出
data/sense_models/{version}/head.npz + metrics.json（macro-F1、逐档 P/R、AUC-ordinal）。 - 阈值：calibrate.compute_thresholds(calib 集, α) →
thresholds.json（τ1/τ3 + 实测覆盖率）。 - 登记 sense_artifacts（active=0），人工 /sense/admin/promote 切换；回滚 = promote 旧版本。
11. 测试计划（隔离：tmp sqlite / 假时钟 / 假 embedder=确定性向量 / MockTransport）
| 文件 | 必测用例 |
|---|---|
| test_sense_embedder | 正常向量；int8 量化可逆性（cosine 误差 <1e-2）；超时/挂→EmbedderDown |
| test_sense_features | 特征门全表：单轮/多轮、长度边界、意图黑名单逐词、code 域白名单 |
| test_sense_classifier | 纯 Python 点积与 numpy 参考实现一致（黄金向量）；工件缺失→ArtifactMissing |
| test_sense_calibrate | 合成分布：α=0.05 时实测覆盖率 ≤α+2%；<500 标签→last-good；工件版本化读写 |
| test_sense_labeler | §4 四条推导规则逐条；T3 保持 T3；180d 清理；human_override 优先 |
| test_sense_grader | 决策表全组合（门×概率×mode）：shadow 不改流、live 才分流；fallback 路径 |
| test_sense_routes | /v1/route 契约；/v1/embeddings 透传；admin 三端点鉴权；不落 query 原文（DB 断言） |
| test_pipeline_tier_fn | tier_fn=None 现测试全绿；注入后三档分流各达预期路径；T2 档跳过 brief/review |
12. WBS：T-G0…T-G8（每任务一 commit feat(sense): T-Gn 描述）
| # | 任务 | 产出 | 验收 | 估时 | 依赖 |
|---|---|---|---|---|---|
| T-G0 | 骨架 | gateway/sense/ 包 + DDL + sense.enabled 门控挂路由 |
全量测试绿；开关关闭 /sense/* 404 |
0.5d | T-P0 |
| T-G1 | Embedder | embedder.py + int8 + 降级 + /v1/embeddings |
test_sense_embedder 绿；llama-server 真机冒烟 | 1d | T-G0 |
| T-G2 | 观察埋点 | observer.py + 三消费方埋点（pipeline 经 tier_fn=None 时的 executed 记录 + proxy/client） | test_sense_labeler 的观察写入部分绿；真机跑 10 请求观察表有行 | 1d | T-G0 |
| T-G3 | 标签+校准 | labeler.py 夜间任务 + calibrate.py + 工件表 | test_sense_labeler/calibrate 绿；合成数据覆盖率达标 | 1.5d | T-G2 |
| T-G4 | 线性头 | train_tier_head.py + classifier.LinearHead + 工件登记 | 黄金向量一致性；合成集 macro-F1 报告 | 1.5d | T-G3 |
| T-G5 | Grader | grader.py + features.py + /v1/route + mode 三态 |
test_sense_grader/routes 绿 | 1.5d | T-G4 |
| T-G6 | live 分流 | pipeline tier_fn 三档 + proxy 档位映射 + 升级阶梯插 T2 档 | test_pipeline_tier_fn 绿（含零回归断言）；真机三档各走通 | 1.5d | T-G5 |
| T-G7 | 审计+前端 | ReviewQueue 抽样接线 + Metrics tier 卡 + 客户端来源显示/一键升级 | 抽样入队可见；npm run build 产物更新 |
1d | T-G6 |
| T-G8 | 实验（+可选 LoRA） | sense_report.py：E-G1 一致率/分布/覆盖率、E-G3 成本延迟 delta；（可选）LoraRemote+E-G2 | shadow≥2 周报告入 research/v2_experiments/；晋升门全过 |
1.5d | T-G7 |
依赖链：T-G0 → T-G1 → T-G2 → T-G3 → T-G4 → T-G5 → T-G6 → T-G7 → T-G8。 总验收 = E-G1 报告：一致率 ≥85%、conformal 覆盖 ≤α+2%、T1 档实测占比落在 40–55% 区间、 引入 T2 档后管线云端调用次数下降可测。
13. 风险与对策
| 风险 | 对策 |
|---|---|
| 冷启动无标签 | collect 模式先用规则门+现行为跑 2–4 周；夜间标签自动积累；min_labels 门前不 live |
| 误判 T1 伤教育质量 | conformal α 保守 + 特征门硬拦 + 升级阶梯兜底 + 审计抽样；严重误判（判 T1 实为 T3）一票否决晋升 |
| Embedder 单点 | 降级阶梯（D-G4）；校端双实例；客户端缓存常用 embedding 不做（隐私），仅服务端 |
| 观察表膨胀 | 180d 清理 + int8 BLOB + 批量写；按月分表预留 |
| 阈值漂移（上游/学期更替） | 夜间 conformal 重算 + policy_version 全链路可追溯 + last-good 回滚 |
| pipeline.py 改动引入回归 | D-G5 钩子缺省逐字节一致 + test_pipeline_tier_fn 零回归断言 |
14. 架构研讨定稿（2026-09-05）：变体 B（向量检索式）与混合路线
两种架构的本质：A（线性头/LoRA）= 记忆的压缩蒸馏（参数化判别边界）； B（向量检索式，无分类器）= 未压缩记忆（embedding + true_tier 标签库上检索定档）。 B 三变体：B1 kNN 加权投票 / B2 最近原型（每档中心向量）/ B3 conformal-kNN （kNN 距离作 nonconformity 得分，与 D-G3 的 conformal 校准同构）。
对比结论：B 的决定性优势 = Day 0 冷启动（每档 10–50 条精选原型即可用，A 需 ≥500 标签 ≈ 2–4 周流量）+ 即时漂移适应（新桶加样即刻生效，A 滞后 ≤24h）+ 天然可解释 （"与这 3 条历史升级任务相似"）。A 的优势 = 恒定 0.1ms 推理、全局边界外推、KB 级工件、 平滑概率直接套 conformal。共同盲区：T3 检测依赖结构特征（多文件/仓库级），语义信号弱—— 特征门（§7 features.gate）权重不可降。
定稿：混合阶梯，主线不变、B 提前插入：
- 新增 T-G3b（T-G3 后、T-G4 前，+1.5d）：KnnHead 实现（TierHead 接口新实现， 复用 semcache M3 预留的 hnswlib 按桶分区设计；B1 投票起步，B3 conformal-kNN 为阈值来源； 每桶每档样本封顶 + 去重 + 老化压缩）。冷启动期 shadow 由此提前产出一致率报告。
- T-G4 线性头照常，上线即 hybrid fusion：p_final = λ·p_head + (1−λ)·p_knn， λ 按头验证 F1 退火（0 → 0.7）；新桶可按桶独立用 B（λ=0）。
- 分歧 = 漂移检测器：head 与 kNN 分档不一致率按桶按天统计，超阈值 → 告警 + 自动触发重训 + 临时收紧 τ1（保守）。单用 A 时不可观测的"模型变老"从此有信号。
- E-G2 扩为四方案对比：B1-kNN / A1-线性 / hybrid /（可选）A2-LoRA； 新增指标"漂移适应速度"（注入新课桶后误判率恢复天数，预期 B≫A——论文中 "为什么必须混合"的实验证据）。文献锚点：ICL-Router（调研已录，冷启动友好）、 Conformal 2607.25018（B3 依据）、S-LoRA（A2 服务形态）。
- 工程增量 ≈ +1.5 工日（KnnHead ~150 行 + 索引复用）；classifier.py 的 TierHead 接口当初即为多形态头预留，grader 加一行 fusion 插值。
验收追加（并入 T-G8）：hybrid 上线后漂移告警通路演示（注入合成新桶 → 告警 → 收紧 → 重训 → 恢复全链路走通）；四方案对比表入 E-G2 报告。