Prompt、模型与工具版本追踪

mkdir l7-versions
cd l7-versions

"""保存 Prompt、模型与工具的联合清单，用固定退款文本逐项切换版本。"""

from copy import deepcopy
import hashlib
import json
import re

PROMPTS = {
    "p1": {"instruction": "判断退货资格", "include_reason": False},
    "p2": {"instruction": "判断退货资格并说明政策依据", "include_reason": True},
}
MODELS = {
    "m1": {"name": "local-parser-1", "chinese_twenty": False},
    "m2": {"name": "local-parser-2", "chinese_twenty": True},
}
TOOLS = {
    "t1": {"schema": "return-policy-v1", "max_days": 7, "policy_id": "book-7"},
    "t2": {"schema": "return-policy-v2", "max_days": 30, "policy_id": "book-30"},
}


def canonical(value):
    # 键排序后同一份内容得到相同字节，摘要不受键插入顺序影响。
    return json.dumps(value, ensure_ascii=False, sort_keys=True,
                      separators=(",", ":")).encode("utf-8")


def snapshot(prompt, model, tool):
    # 拷贝具体内容，后续别名或目录改变也不会改写这次运行的记录。
    manifest = {"refs": {"prompt": prompt, "model": model, "tool": tool},
                "prompt": deepcopy(PROMPTS[prompt]),
                "model": deepcopy(MODELS[model]),
                "tool": deepcopy(TOOLS[tool]),
                "config": {"temperature": 0, "parser_revision": "code-1"}}
    manifest["fingerprint"] = hashlib.sha256(canonical(manifest)).hexdigest()
    return manifest


def verify(manifest):
    saved = deepcopy(manifest)
    fingerprint = saved.pop("fingerprint")
    if hashlib.sha256(canonical(saved)).hexdigest() != fingerprint:
        raise ValueError("manifest content changed")


def run(text, manifest):
    verify(manifest)
    # 两个本地解析器有确定的能力差异，每个版本真的参与计算。
    normalized = text
    if manifest["model"]["chinese_twenty"]:
        normalized = normalized.replace("二十", "20")
    match = re.search(r"(\d+)天", normalized)
    days = int(match.group(1)) if match else None
    policy = manifest["tool"]
    decision = ("needs_review" if days is None else
                "eligible" if days <= policy["max_days"] else "ineligible")
    result = {"decision": decision}
    if manifest["prompt"]["include_reason"]:
        result["reason"] = ("days_unknown" if days is None else
                            f"{days}d vs {policy['max_days']}d")
        result["policy_id"] = policy["policy_id"]
    return result


def main():
    text = "收货二十天想退货"
    for refs in (("p1", "m1", "t1"), ("p2", "m1", "t1"),
                 ("p2", "m2", "t1"), ("p2", "m2", "t2")):
        manifest = snapshot(*refs)
        print(f"{'/'.join(refs)} fingerprint={manifest['fingerprint'][:12]} "
              f"result={json.dumps(run(text, manifest), ensure_ascii=False, sort_keys=True)}")


if __name__ == "__main__":
    main()

ls
python -X utf8 version_lab.py

p1/m1/t1 fingerprint=5a06f68646cc result={"decision": "needs_review"}
p2/m1/t1 fingerprint=469f5dc5e4f9 result={"decision": "needs_review", "policy_id": "book-7", "reason": "days_unknown"}
p2/m2/t1 fingerprint=e084e5490b87 result={"decision": "ineligible", "policy_id": "book-7", "reason": "20d vs 7d"}
p2/m2/t2 fingerprint=9b7819e36992 result={"decision": "eligible", "policy_id": "book-30", "reason": "20d vs 30d"}

ls
python -X utf8 -m unittest -v test_version_lab.py