120 lines
5.7 KiB
Python
120 lines
5.7 KiB
Python
"""测量：读取器的 score_margin 能否区分「同一属性的更新」与「另一属性的事实」。
|
||
|
||
背景：为了让自动层能版本化 Agent 经接口写入的记录，曾尝试用读取器判断"这一轮讲的是哪件事"再继承冲突键；
|
||
结果误伤了不相关事实（存了 默认语言=日语 与 项目代号=蓝鲸-47 后，说「我的项目代号是 蓝鲸-47」
|
||
却把默认语言那条取代掉）。因此动更新语义之前，先量 margin 分布。
|
||
|
||
做法（走正在运行的控制面，不额外占卡）：
|
||
正类 = 先存「我的<A>是 v1」，再问「我的<A>改成 v2」 —— 期望：读取器明确指向那条记录
|
||
负类 = 先存「我的<A>是 v1」，再问「我的<B>是 v2」(B≠A) —— 期望：不该指向那条记录
|
||
每类各跑一轮，输出 need_memory / stop_reason / margin / 命中是否是那条记录，并给两类区间。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import statistics
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
BASE = "http://127.0.0.1:8766/v1"
|
||
|
||
# 取自 data/realistic_v2/eval.manifest.json 的 eval 属性族（未出现在训练里）
|
||
FAMILIES = [
|
||
"主库地址", "仓库库位", "代码仓库", "供应商", "保险到期", "值班电话",
|
||
"值班表", "发布窗口", "合同编号", "告警阈值", "周会时间", "培训周期",
|
||
]
|
||
VALUES_A = ["A-7719", "v3.14.2", "64GB", "B3-204", "91.5%", "C1-105", "D-9021", "v1.9.7"]
|
||
VALUES_B = ["v4.0.0-rc1", "128GB", "E-4402", "73%", "F-1180", "v2.0.1", "G-6613", "88%"]
|
||
|
||
|
||
def call(path: str, payload: dict | None = None, method: str = "POST") -> dict:
|
||
url = f"{BASE}{path}"
|
||
data = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else None
|
||
request = urllib.request.Request(
|
||
url, data=data, headers={"Content-Type": "application/json; charset=utf-8"}, method=method
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=120) as response:
|
||
return json.loads(response.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as error:
|
||
return {"__error__": f"HTTP {error.code}: {error.read().decode('utf-8', 'replace')[:200]}"}
|
||
|
||
|
||
def trial(store_attribute: str, store_value: str, query: str) -> dict:
|
||
call("/nm2/reset", {})
|
||
written = call("/nm2/write", {
|
||
"text": f"我的{store_attribute}是 {store_value}。",
|
||
"entity": "user", "attribute": store_attribute, "value": store_value,
|
||
})
|
||
record_id = (written.get("record") or {}).get("record_id", "")
|
||
probed = call("/nm2/probe", {"query": query})
|
||
if "__error__" in probed:
|
||
return {"error": probed["__error__"]}
|
||
hits = probed.get("records") or []
|
||
return {
|
||
"need_memory": bool(probed.get("need_memory")),
|
||
"stop_reason": str(probed.get("stop_reason")),
|
||
"margin": probed.get("score_margin"),
|
||
"top_score": probed.get("top_score"),
|
||
"hit_is_stored": bool(hits) and str(hits[0].get("record_id")) == record_id,
|
||
"hits": len(hits),
|
||
}
|
||
|
||
|
||
def summarise(name: str, rows: list[dict]) -> dict:
|
||
margins = [float(r["margin"]) for r in rows if r.get("margin") is not None]
|
||
usable = [r for r in rows if "error" not in r]
|
||
pointed = sum(1 for r in usable if r.get("hit_is_stored"))
|
||
print(f"\n=== {name}（{len(usable)} 次）===")
|
||
print(f" need_memory 为真: {sum(1 for r in usable if r['need_memory'])}/{len(usable)}")
|
||
print(f" 命中就是那条已存记录: {pointed}/{len(usable)}")
|
||
if margins:
|
||
print(f" margin: 最小 {min(margins):.4f} 中位 {statistics.median(margins):.4f} 最大 {max(margins):.4f}")
|
||
print(f" margin 取值: {', '.join(f'{m:.4f}' for m in sorted(margins))}")
|
||
reasons: dict[str, int] = {}
|
||
for r in usable:
|
||
reasons[r["stop_reason"]] = reasons.get(r["stop_reason"], 0) + 1
|
||
print(f" stop_reason: {reasons}")
|
||
return {"margins": margins, "pointed": pointed, "total": len(usable)}
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--rounds", type=int, default=1, help="每个属性跑几轮")
|
||
args = parser.parse_args()
|
||
|
||
positive: list[dict] = []
|
||
negative: list[dict] = []
|
||
for round_index in range(args.rounds):
|
||
for index, attribute in enumerate(FAMILIES):
|
||
value = VALUES_A[(index + round_index) % len(VALUES_A)]
|
||
other = FAMILIES[(index + 1) % len(FAMILIES)]
|
||
other_value = VALUES_B[(index + round_index) % len(VALUES_B)]
|
||
|
||
# 正类：同一属性的更新（"改成"）
|
||
positive.append(trial(attribute, value, f"我的{attribute}改成 {VALUES_B[(index + round_index) % len(VALUES_B)]}。"))
|
||
# 负类：另一个属性的事实
|
||
negative.append(trial(attribute, value, f"我的{other}是 {other_value}。"))
|
||
|
||
pos = summarise("正类：同一属性的更新", positive)
|
||
neg = summarise("负类：另一属性的事实", negative)
|
||
|
||
print("\n=== 判别力检查 ===")
|
||
if pos["margins"] and neg["margins"]:
|
||
print(f" 正类 margin 区间: [{min(pos['margins']):.4f}, {max(pos['margins']):.4f}]")
|
||
print(f" 负类 margin 区间: [{min(neg['margins']):.4f}, {max(neg['margins']):.4f}]")
|
||
overlap_lo = max(min(pos["margins"]), min(neg["margins"]))
|
||
overlap_hi = min(max(pos["margins"]), max(neg["margins"]))
|
||
if overlap_hi > overlap_lo:
|
||
print(f" → 两类区间重叠于 [{overlap_lo:.4f}, {overlap_hi:.4f}]：**单靠 margin 无法干净分开**")
|
||
else:
|
||
print(" → 两类区间不重叠：存在可用门限")
|
||
print(f" 命中率：正类 {pos['pointed']}/{pos['total']}，负类 {neg['pointed']}/{neg['total']}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|