tzt
3250e4e099
- normalizer.py：resolve_bucket（X-Campus-Bucket 头 > 映射 > default，D-P2）； normalize_messages（role\u0001content\u0002 串接，剔易变字段与时间戳行， 多模态 content 取 text）；is_cacheable（system 外 >1 条 = 多轮不缓存，D-P5）； canonical_hash（bucket|doc_version|sha256(norm)——模板/资料不参与哈希， 资料更新 = 版本+1 旧键失效）；shape（[canonical_system(无才注入)] -> [课程资料前缀(文件读取,不入git)] -> [原 messages]，易变顶层字段剔除） - 测试 +8：同义同哈希/易变剔除/模板不入哈希/doc_version 失效/多轮判定/ 整形顺序+资料注入/桶解析回落/序列化形态，全量 370 passed
128 lines
5.2 KiB
Python
128 lines
5.2 KiB
Python
"""请求规范化与桶解析（T-P5）。
|
||
|
||
规范化规则（§6，决定缓存命中率的代码）：
|
||
1. messages 序列化：role 与 content 交替拼接为 `role\\u0001content\\u0002`；
|
||
2. 剔除易变字段：temperature/frequency_penalty/seed/request_id/时间戳类内容行；
|
||
3. 系统模板与资料前缀不参与 L1 哈希（桶+doc_version 已表达），只参与上游整形；
|
||
4. 整形后消息顺序固定：[canonical_system] -> [doc_prefix] -> [原 messages]；
|
||
5. 多轮判定：len(messages) > 2（system 之外 >1 条）-> cacheable=False（D-P5）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from gateway.proxy.ledgerutil import _today # noqa: F401 （保持导入面稳定）
|
||
|
||
_SEP_R = "\u0001"
|
||
_SEP_M = "\u0002"
|
||
|
||
# 易变字段（规范化时从 body 剔除，不参与哈希）
|
||
_VOLATILE_FIELDS = ("temperature", "frequency_penalty", "presence_penalty",
|
||
"seed", "request_id", "user", "logprobs", "top_logprobs")
|
||
# 时间戳类内容行（消息内容若整体匹配则该行剔除后再序列化）
|
||
_TIMESTAMP_LINE = re.compile(
|
||
r"^\s*(当前时间|时间|now|time|date|日期)\s*[:：].*$", re.IGNORECASE)
|
||
|
||
|
||
@dataclass
|
||
class BucketResolution:
|
||
"""桶解析结果。"""
|
||
name: str
|
||
cfg: Any # BucketCfg
|
||
cacheable: bool # D-P5：多轮 False
|
||
norm_text: str # 规范化后的 messages 文本（L2 语义比对用）
|
||
norm_hash: str # bucket|doc_version|sha256(norm)（L1 键）
|
||
|
||
|
||
def resolve_bucket(body: dict, headers: Dict[str, str], cfg) -> Any:
|
||
"""桶解析（D-P2）：X-Campus-Bucket 头 > model->bucket 映射 > default。"""
|
||
name = (headers.get("x-campus-bucket")
|
||
or headers.get("X-Campus-Bucket") or "").strip()
|
||
if not name:
|
||
# model->bucket 映射（配置预留：bucket 名即映射；MVP 仅 default）
|
||
name = "default"
|
||
return cfg.bucket(name)
|
||
|
||
|
||
def _norm_content(content: Any) -> str:
|
||
if isinstance(content, str):
|
||
lines = [ln for ln in content.splitlines()
|
||
if ln.strip() and not _TIMESTAMP_LINE.match(ln)]
|
||
return "\n".join(lines).strip()
|
||
if isinstance(content, list):
|
||
# 多模态 content parts：取 text 项拼接
|
||
parts = []
|
||
for p in content:
|
||
if isinstance(p, dict) and isinstance(p.get("text"), str):
|
||
parts.append(_norm_content(p["text"]))
|
||
return "\n".join(parts)
|
||
return ""
|
||
|
||
|
||
def normalize_messages(body: dict) -> str:
|
||
"""规则 1+2：剔易变行后按 role\\u0001content\\u0002 串接（system 亦参与——
|
||
客户端自带 system 属请求语义；桶模板/资料前缀不在这里，见 shape）。"""
|
||
out: List[str] = []
|
||
for m in body.get("messages") or []:
|
||
if not isinstance(m, dict):
|
||
continue
|
||
role = str(m.get("role") or "")
|
||
text = _norm_content(m.get("content"))
|
||
out.append(role + _SEP_R + text + _SEP_M)
|
||
return "".join(out)
|
||
|
||
|
||
def is_cacheable(body: dict) -> bool:
|
||
"""规则 5：system 之外消息数 > 1 -> 多轮 -> 不缓存（D-P5）。"""
|
||
msgs = [m for m in (body.get("messages") or []) if isinstance(m, dict)]
|
||
non_system = [m for m in msgs if str(m.get("role")) != "system"]
|
||
return len(non_system) <= 1
|
||
|
||
|
||
def canonical_hash(bucket: str, doc_version: int, body: dict) -> str:
|
||
"""规则 3：缓存键 = bucket + '|' + doc_version + '|' + sha256(norm)。
|
||
|
||
桶模板/资料前缀不参与哈希（由 bucket+doc_version 表达，资料更新 = 版本+1）。
|
||
"""
|
||
norm = normalize_messages(body)
|
||
digest = hashlib.sha256(norm.encode("utf-8")).hexdigest()
|
||
return f"{bucket}|{doc_version}|{digest}"
|
||
|
||
|
||
def _load_doc_prefix(bucket_cfg) -> str:
|
||
"""读桶资料前缀文件（§3：不入库不入 git；缺失返回空串）。"""
|
||
path = getattr(bucket_cfg, "doc_prefix_file", None)
|
||
if not path:
|
||
return ""
|
||
try:
|
||
from pathlib import Path
|
||
return Path(path).read_text(encoding="utf-8")
|
||
except OSError:
|
||
return ""
|
||
|
||
|
||
def shape(body: dict, bucket_cfg) -> dict:
|
||
"""规则 4：整形上游请求体 —— [canonical_system] -> [doc_prefix] -> [原 messages]。
|
||
|
||
原请求自带的 system 保留在原位（其属请求语义）；桶模板仅在没有
|
||
system 时注入，资料前缀始终插在 system 之后、其余消息之前。
|
||
剔除易变顶层字段后的浅拷贝返回。
|
||
"""
|
||
shaped = {k: v for k, v in body.items() if k not in _VOLATILE_FIELDS}
|
||
msgs: List[Dict[str, Any]] = [dict(m) for m in (body.get("messages") or [])
|
||
if isinstance(m, dict)]
|
||
has_system = any(str(m.get("role")) == "system" for m in msgs)
|
||
prefix_block: List[Dict[str, Any]] = []
|
||
if not has_system and bucket_cfg.system_template:
|
||
prefix_block.append({"role": "system", "content": bucket_cfg.system_template})
|
||
doc = _load_doc_prefix(bucket_cfg)
|
||
if doc:
|
||
prefix_block.append({"role": "system",
|
||
"content": f"[课程资料 v{bucket_cfg.doc_version}]\n{doc}"})
|
||
shaped["messages"] = prefix_block + msgs
|
||
return shaped
|