"""Phase-6 calibration: A/B difflib vs Jev + confidence-threshold sweep. docker compose exec bot python tools/calibrate_matching.py - Runs a fixed scenario set (seeded catalog + realistic STT-ish variants and negatives) through BOTH matchers: * difflib baseline: resolve.safe_resolve (the pre-Jev path) * Jev: matching.resolve_mentions (one request per field — that is how the pipeline actually runs آن) - Every run is recorded as real observability data (traces with source="calibration") plus a markdown report under data/reports/. - Prints a summary with recommended JEV_CONF_EXACT / JEV_CONF_FLOOR values. Nothing is sent to Bale and no drafts are created. """ import asyncio import json import sys from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app import matching, observability as obs from app.api.client import ErpClient from app.config import settings from app.resolve import Resolver, safe_resolve # (field, variants, truth) — truth=None means "must NOT match anything". SCENARIOS = [ ("project", ["دفتر"], "دفتر"), ("project", ["منشیانه"], "منشانه"), ("project", ["منشانه"], "منشانه"), ("project", ["باق اسپارت"], "باغ اسپارت"), ("project", ["پردیس"], "پردیس کاشان"), ("project", ["بابک"], "بابک"), ("project", ["آسمون"], "آسمان"), ("project", ["تست یک"], "تست 1"), ("project", ["انبار قدیم"], None), ("supplier", ["مصطفوی"], "بازرگانی مصطفوی"), ("supplier", ["مستفدی"], "بازرگانی مصطفوی"), ("supplier", ["مصطفوی", "مستفدی"], "بازرگانی مصطفوی"), ("supplier", ["فولاد مبارکه"], "نمایندگی فولاد مبارکه"), ("supplier", ["پلیکا البرز"], "پلیکا فروش البرز"), ("supplier", ["راجین"], "راجین"), ("supplier", ["آهن فروشی آزاد"], None), ("material", ["گچ و خاک"], "گچ و خاک"), ("material", ["سیمان"], "سیمان پرتلند"), ("material", ["میلگرد"], "میلگرد آجدار"), ("material", ["تیر آهن"], "تیرآهن"), ("material", ["لوله پلیکا"], "لوله پلیکا"), ("material", ["کاغذ A4"], None), ("unit", ["کیلو"], "کیلوگرم"), ("unit", ["تن"], "تن"), ("unit", ["کیسه"], "کیسه"), ("unit", ["شاخه"], "شاخه"), ("unit", ["جعبه"], None), ("brand", ["ذوب آهن"], "ذوب آهن اصفهان"), ("brand", ["مبارکه"], "فولاد مبارکه"), ("brand", ["سیمان تهران"], "سیمان تهران"), ("spec", ["سایز 14"], "سایز ۱۴"), ("spec", ["50 کیلویی"], "۵۰ کیلویی"), ("spec", ["6 اینچ"], "۶ اینچ"), ("spec", ["40×40"], "۴۰×۴۰"), ("spec", ["۲ میلی"], "۲ میل"), ] SWEEP = [round(0.55 + 0.05 * i, 2) for i in range(9)] # 0.55 … 0.95 async def run() -> None: api = ErpClient(settings.API_BASE_URL, settings.API_TIMEOUT_SECONDS, settings.SERVICE_TOKEN) resolver = Resolver(api) ctx = SimpleNamespace(resolver=resolver) fields: dict[str, list[dict]] = {} for field, variants, truth in SCENARIOS: fields.setdefault(field, []).append({"field": field, "variants": variants, "truth": truth, "difflib": None, "jev": None}) for field, scs in fields.items(): trace = await obs.trace_start(0, None, "text", source="calibration") # --- difflib baseline ------------------------------------------------- for sc in scs: res = await safe_resolve(resolver, field, sc["variants"][0]) cands = [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]] sc["difflib"] = {"status": res.status, "top": cands[0]["name"] if cands else None, "cands": [c["name"] for c in cands]} await obs.log_resolution(trace, field, sc["variants"][0], res.status, cands) # --- Jev (one batch for the whole field) ------------------------------ # Collect RAW picks: temporarily widen the bands so every confidence is # returned; classification below uses the CONFIGURED thresholds. saved_exact, saved_floor = settings.JEV_CONF_EXACT, settings.JEV_CONF_FLOOR settings.JEV_CONF_EXACT, settings.JEV_CONF_FLOOR = 1.01, 0.0 try: jev = await matching.resolve_mentions( ctx, {}, [{"field": s["field"], "variants": s["variants"]} for s in scs], trace) finally: settings.JEV_CONF_EXACT, settings.JEV_CONF_FLOOR = saved_exact, saved_floor for sc in scs: hit = matching.lookup(jev, sc["variants"][0]) raw_conf = None if hit is None else hit.get("confidence") raw_choice = None if hit is None else hit.get("choice") if raw_choice is None: sc["jev"] = {"status": "new", "choice": None, "conf": raw_conf} elif raw_conf >= saved_exact: sc["jev"] = {"status": "exact", "choice": raw_choice, "conf": raw_conf} elif raw_conf >= saved_floor: sc["jev"] = {"status": "ambiguity", "choice": raw_choice, "conf": raw_conf} else: sc["jev"] = {"status": "lowtrust", "choice": raw_choice, "conf": raw_conf} await obs.event(trace, "calibration.truth", payload={ "field": field, "rows": [{"variants": s["variants"], "truth": s["truth"], "difflib": s["difflib"], "jev": s["jev"]} for s in scs], }) # ------------------------- metrics --------------------------------------- pos = [s for s in sum(fields.values(), []) if s["truth"]] neg = [s for s in sum(fields.values(), []) if not s["truth"]] d_auto = d_amb = d_miss = 0 for s in pos: d = s["difflib"] if d["top"] == s["truth"]: d_auto += 1 elif s["truth"] in d["cands"]: d_amb += 1 else: d_miss += 1 d_neg_ok = sum(1 for s in neg if s["difflib"]["status"] == "none") j_exact_ok = sum(1 for s in pos if s["jev"]["status"] == "exact" and s["jev"]["choice"] == s["truth"]) j_exact_bad = sum(1 for s in pos if s["jev"]["status"] == "exact" and s["jev"]["choice"] != s["truth"]) j_amb = sum(1 for s in pos if s["jev"]["status"] == "ambiguity") j_low = sum(1 for s in pos if s["jev"]["status"] == "lowtrust") j_new = sum(1 for s in pos if s["jev"]["status"] == "new") # negatives: rejected cleanly / asked the user (safe but noisy) / wrongly bound j_neg_ok = sum(1 for s in neg if s["jev"]["status"] in ("new", "lowtrust")) j_neg_ask = sum(1 for s in neg if s["jev"]["status"] == "ambiguity") j_neg_bad = sum(1 for s in neg if s["jev"]["status"] == "exact") # --------------------- threshold sweep (EXACT) --------------------------- # Binds = positives above the threshold PLUS any pick on a negative scenario # (a wrong pick above EXACT would auto-bind — it must drag precision down). sweep_rows = [] for th in SWEEP: pos_binds = [s for s in pos if s["jev"]["choice"] and (s["jev"]["conf"] or 0) >= th] neg_binds = [s for s in neg if s["jev"]["choice"] and (s["jev"]["conf"] or 0) >= th] ok = sum(1 for s in pos_binds if s["jev"]["choice"] == s["truth"]) binds = len(pos_binds) + len(neg_binds) prec = (ok / binds) if binds else 0.0 cov = len(pos_binds) / len(pos) sweep_rows.append({"th": th, "binds": binds, "correct": ok, "precision": round(prec, 3), "coverage": round(cov, 3)}) # Conservative recommendation (risk-asymmetric: a wrong auto-bind costs more # than an extra confirmation tap; the sample here is still small): # - EXACT: keep the configured value while its precision is >= 0.95; raise it # to the first threshold with precision >= 0.95 otherwise. # - FLOOR: raise just above the highest confidence of a WRONG pick (a wrong # pick below the floor never again becomes a candidate button). binds_now = [r for r in sweep_rows if r["th"] == settings.JEV_CONF_EXACT] prec_now = binds_now[0]["precision"] if binds_now else 0.0 rec_exact = settings.JEV_CONF_EXACT if prec_now < 0.95: for row in sweep_rows: if row["precision"] >= 0.95 and row["coverage"] >= 0.30: rec_exact = row["th"] break correct_confs = sorted(s["jev"]["conf"] for s in pos if s["jev"]["choice"] == s["truth"] and s["jev"]["conf"]) # a wrong pick on a NEGATIVE scenario (truth=None) matters too: the floor # must sit above it so it never becomes a candidate button offer. wrong_confs = sorted(s["jev"]["conf"] for s in pos + neg if s["jev"]["choice"] and s["jev"]["choice"] != s["truth"] and s["jev"]["conf"]) rec_floor = settings.JEV_CONF_FLOOR if wrong_confs: import math # The same scenario's confidence jitters ±0.05 across identical runs # (جعبه observed 0.61/0.60/0.58), so sit one step above the highest wrong # pick + margin — routing uses `conf >= FLOOR`, so equality leaks through. target = max(wrong_confs) + 0.05 rec_floor = min(rec_exact, round(math.floor(target * 20 + 0.5) / 20.0, 2)) if rec_floor <= max(wrong_confs): rec_floor = min(rec_exact, round((math.floor(max(wrong_confs) * 20) + 1) / 20.0, 2)) # how much coverage a more aggressive EXACT would buy (info only) lower_info = next((r for r in reversed(sweep_rows) if r["th"] < settings.JEV_CONF_EXACT and r["precision"] >= 0.98), None) # ------------------------------ report ----------------------------------- ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M") out_dir = Path(settings.OBS_VOICE_DIR).parent / "reports" out_dir.mkdir(parents=True, exist_ok=True) report = out_dir / f"calibration_{ts}.md" lines = [ "# گزارش کالیبراسیون تطبیق موجودیت‌ها (فاز ۶)", f"\nتاریخ: {ts} UTC · مدل Jev پین‌شده: `{settings.JEV_MODEL}` · " f"آستانه‌های فعلی: EXACT={settings.JEV_CONF_EXACT} · FLOOR={settings.JEV_CONF_FLOOR}", f"\nسناریوها: {len(pos)} مثبت + {len(neg)} منفی (بدون تطبیق) = {len(pos) + len(neg)}", "\n> روش: انتخاب‌های خام Jev با باند بازِ موقت جمع‌آوری می‌شوند و «وضعیت» هر " "سناریو با آستانه‌های پیکربندی‌شده بالا محاسبه می‌شود. status `lowtrust` یعنی " "انتخاب زیر FLOOR بود و در اجرای واقعی رد می‌شود → مسیر difflib ادامه پیدا می‌کند.", "\n## خلاصه", f"- **difflib (پایه):** اتصال درست {d_auto}/{len(pos)} · ابهام {d_amb} · از دست رفته {d_miss} " f"· رد درست منفی‌ها {d_neg_ok}/{len(neg)}", f"- **Jev:** exact-درست {j_exact_ok} · exact-غلط {j_exact_bad} · ابهام(سؤال از کاربر) {j_amb} · " f"کم‌اعتماد→difflib {j_low} · new {j_new}", f"- **منفی‌ها:** ردِ تمیز {j_neg_ok}/{len(neg)} · سؤال از کاربر {j_neg_ask} · پذیرش غلط {j_neg_bad}", f"- **پیشنهاد آستانه:** `JEV_CONF_EXACT={rec_exact}` · `JEV_CONF_FLOOR={rec_floor}`" + (f" _(داده فعلی از کاهش EXACT تا {lower_info['th']} با دقت {lower_info['precision']} " f"و پوشش {lower_info['coverage']} حمایت می‌کند — با حجم داده واقعی بازبینی شود)_" if lower_info else ""), "\n## جدول sweep آستانه EXACT", "| آستانه | اتصال خودکار (مثبت+منفی) | درست | دقت | پوشش مثبت‌ها |", "|---|---|---|---|---|", ] for r in sweep_rows: lines.append(f"| {r['th']} | {r['binds']} | {r['correct']} | {r['precision']} | {r['coverage']} |") lines += [ "\n## توزیع confidence", f"- انتخاب‌های درست: {correct_confs or '—'}", f"- انتخاب‌های غلط: {wrong_confs or '—'}", ] for field, scs in fields.items(): lines += [f"\n## فیلد {field}", "| variants | حقیقت | difflib | Jev (choice/conf) |", "|---|---|---|---|"] for s in scs: j = s["jev"] jtxt = "—" if not j else f"{j['status']}: {j['choice'] or '_none'} @ {j['conf']}" if j and j["status"] == "lowtrust": jtxt += " (→ مسیر difflib)" lines.append(f"| {' / '.join(s['variants'])} | {s['truth'] or '—'} | " f"{s['difflib']['status']}: {s['difflib']['top'] or '—'} | {jtxt} |") report.write_text("\n".join(lines), encoding="utf-8") print("\n".join(lines[:14])) print(f"\nگزارش کامل: {report}") print(f"\nپیشنهاد: JEV_CONF_EXACT={rec_exact} JEV_CONF_FLOOR={rec_floor}") if __name__ == "__main__": asyncio.run(run())