"""Entity matching with Jev (TypeSafe) — phases 3-5 of the final plan. Mechanics (final plan v5 §2): - برون‌گروهی: ONE Jev request per FIELD that has mentions (requests run in parallel; latency ≈ one round-trip regardless of field count). - درون‌گروهی: inside a field request, ONE `choice` question per mention; every question rides with the SAME full candidate list (+ a `_none` option). - درون‌واریانسی: all STT spellings of one entity travel in ONE question, and the instructions ask the model to treat them as one name. Routing (thresholds from env; calibrated in phase 6): conf >= JEV_CONF_EXACT → exact (bind id, no ambiguity sub-message) JEV_CONF_FLOOR..JEV_CONF_EXACT → ambiguity (candidate buttons sub-message) below floor / _none → new (existing auto-create path) Fail-soft: a failed field simply yields no results for its mentions; callers fall back to the difflib path (`safe_resolve`). With JEV_ENABLED=false nothing in this module runs (zero regression). """ import asyncio import hashlib import json import logging from app import observability as obs from app.config import settings from app.jev import ask as jev_ask from app.nlu import normalize_text logger = logging.getLogger("bot.matching") FIELD_ORDER = ["project", "supplier", "material", "brand", "spec", "unit"] FIELD_FA = { "project": "پروژه", "supplier": "تأمین‌کننده", "material": "متریال", "brand": "برند", "spec": "مشخصه", "unit": "واحد", } # --------------------------------------------------------------------------- # candidate lists (full lists — no prefilter, per owner decision) # --------------------------------------------------------------------------- async def _candidates(resolver, field: str) -> list: if field == "project": return await resolver.projects() if field == "supplier": return await resolver.all_suppliers() if field == "material": return await resolver.all_materials() if field == "brand": return await resolver.brands() if field == "spec": return await resolver.specs() if field == "unit": return await resolver.units() return [] def _hint(field: str, cand) -> str: """Short, token-cheap description shown next to a candidate name.""" if field == "supplier": extra = cand.extra or {} bits = [b for b in (extra.get("category_name"), extra.get("location")) if b] return "، ".join(str(b) for b in bits) if bits else "تأمین‌کننده" return "" def _instructions(field: str, variants: list[str]) -> str: fa = FIELD_FA.get(field, field) vlist = "»، «".join(v for v in variants if v) if len(variants) > 1: return ( f"کاربر در پیام (صوتی) خود این {fa} را ذکر کرده و مدل‌های مختلف STT آن را به شکل‌های " f"«{vlist}» پیاده کرده‌اند؛ این‌ها چند نگارش از یک نام واحدند. با توجه به بافت جمله و " f"تشابه آوایی میان این واریانت‌ها، مشخص کن به کدام گزینه از فهرست اشاره دارد. " f"اگر هیچ‌کدام منطبق نیست یا تردید جدی داری «_none» را انتخاب کن." ) return ( f"کاربر «{vlist}» را به‌عنوان {fa} ذکر کرده است. این عبارت به کدام گزینه از فهرست اشاره دارد؟ " f"اگر هیچ‌کدام منطبق نیست یا تردید جدی داری «_none» را انتخاب کن." ) # --------------------------------------------------------------------------- from app.fallback_matcher import MatchingFailureError, match_with_deepseek # --------------------------------------------------------------------------- # the per-field fan-out # --------------------------------------------------------------------------- async def resolve_mentions( ctx, patch: dict, mentions: list[dict], trace_id: str | None = None, raw_input: object = None, ) -> dict: """Run Jev per field; on failure, trigger DeepSeek Crisis Fallback. Return a map keyed by normalized variant. Each value: {"field", "status": exact|ambiguity|new", "variants", "id"?, "name"?, "extra"?, "confidence"?, "candidates": [...]} """ by_field: dict[str, list[dict]] = {} for m in mentions or []: field = str((m or {}).get("field") or "").strip() variants = [str(v).strip() for v in ((m or {}).get("variants") or []) if str(v or "").strip()] if field in FIELD_ORDER and variants: by_field.setdefault(field, []).append({"field": field, "variants": variants[:6]}) results: dict[str, dict] = {} if not by_field: return results unhandled_mentions: list[dict] = [] unhandled_fields: set[str] = set() async def one_field(field: str, ms: list[dict]) -> None: try: cands = await _candidates(ctx.resolver, field) except Exception: logger.exception("matching: candidate fetch failed for %s", field) unhandled_mentions.extend(ms) unhandled_fields.add(field) return if not cands: unhandled_mentions.extend(ms) unhandled_fields.add(field) return by_name: dict[str, object] = {} criteria: dict[str, str] = {} for c in cands: if c.name in criteria: continue criteria[c.name] = _hint(field, c) by_name[c.name] = c criteria["_none"] = "هیچ‌کدام از موارد موجود نیست" capped = ms[: settings.JEV_MAX_QUESTIONS_PER_FIELD] questions = { f"mention_{i}": { "type": "choice", "instructions": _instructions(field, m["variants"]), "criteria": criteria, } for i, m in enumerate(capped) } state = { "field": field, "patch_context": json.dumps(patch, ensure_ascii=False, default=str)[:1500], } await obs.event(trace_id, "jev.request", field=field, payload={ "n_mentions": len(capped), "n_candidates": len(cands), }) resp = await jev_ask(state, questions) await obs.event(trace_id, "jev.result", field=field, ok=resp is not None, model=(resp or {}).get("model"), latency_ms=(resp or {}).get("latency_ms"), payload={"answers": (resp or {}).get("answers"), "usage": (resp or {}).get("usage")}) if resp is None: # Observability: log jev.failure event await obs.event(trace_id, "jev.failure", field=field, payload={ "error": "jev_unavailable_or_error", "reason": "HTTP error or timeout or no response", "model": settings.JEV_MODEL, "n_mentions": len(capped), }) unhandled_mentions.extend(capped) unhandled_fields.add(field) return answers = resp.get("answers") or {} decisions = [] for i, m in enumerate(capped): ans = answers.get(f"mention_{i}") or {} if not ans: unhandled_mentions.append(m) unhandled_fields.add(field) continue choice = ans.get("choice") conf = float(ans.get("confidence") or 0.0) probs = ans.get("probabilities") or {} cand = by_name.get(choice) if choice else None ranked = sorted( ({"id": by_name[n].id, "name": n, "prob": float(p or 0.0)} for n, p in probs.items() if n != "_none" and n in by_name), key=lambda x: -x["prob"], )[:6] base = {"field": field, "variants": m["variants"], "confidence": conf, "choice": (cand.name if cand is not None else None), "candidates": ranked} if cand is not None and conf >= settings.JEV_CONF_EXACT: res = {**base, "status": "exact", "id": cand.id, "name": cand.name, "extra": cand.extra or {}} elif cand is not None and conf >= settings.JEV_CONF_FLOOR: res = {**base, "status": "ambiguity", "id": cand.id, "name": cand.name, "extra": cand.extra or {}, "candidates": ranked or [{"id": cand.id, "name": cand.name, "prob": conf}]} elif cand is None: res = {**base, "status": "new"} else: # Below-floor pick: route to fallback matcher unhandled_mentions.append(m) unhandled_fields.add(field) continue for v in m["variants"]: results[normalize_text(v)] = res decisions.append({"variants": m["variants"], "status": res["status"], "choice": res.get("name"), "confidence": conf, "probs_top": [{k: round(v, 3)} for k, v in sorted(probs.items(), key=lambda kv: -kv[1])[:3]]}) await obs.event(trace_id, "routing.decision", field=field, payload={"source": "jev", "mentions": decisions}) if settings.JEV_ENABLED: await asyncio.gather(*(one_field(f, ms) for f, ms in by_field.items())) else: # Jev disabled -> all mentions route to DeepSeek fallback for f, ms in by_field.items(): unhandled_mentions.extend(ms) unhandled_fields.add(f) # If any mentions were unhandled / failed in Jev, run DeepSeek Crisis Fallback if unhandled_mentions: cands_by_field = {} for f in unhandled_fields: try: cands_by_field[f] = await _candidates(ctx.resolver, f) except Exception: cands_by_field[f] = [] fallback_res = await match_with_deepseek( raw_input=raw_input, mentions=unhandled_mentions, candidates_by_field=cands_by_field, patch_context=patch, trace_id=trace_id, ) results.update(fallback_res) return results def lookup(jev_map: dict | None, name: str | None) -> dict | None: """Find the Jev result for a raw name (by normalized variant).""" if not jev_map or not name: return None return jev_map.get(normalize_text(name)) # --------------------------------------------------------------------------- # writing results into drafts # --------------------------------------------------------------------------- def apply_choice(payload: dict, field: str, cand: dict, variants: list[str]) -> None: """Bind one chosen candidate (exact match or a user pick) into the payload.""" name, cid = cand.get("name"), cand.get("id") extra = cand.get("extra") or {} variants_norm = {normalize_text(v) for v in (variants or []) if v} if field == "supplier": payload["supplier_id"] = cid payload["supplier_name"] = name for k in ("location", "phone", "contact_person", "category_name"): if extra.get(k): payload[f"supplier_{k}"] = extra[k] elif field == "project": resolved = payload.setdefault("resolved_projects", {}) for v in (variants or []): resolved[v] = cid resolved[name] = cid if cid not in payload.get("project_ids", []): payload.setdefault("project_ids", []).append(cid) if not payload.get("project_id"): payload["project_id"] = cid elif field == "material": for item in payload.get("items") or []: if normalize_text(item.get("material_name") or "") in variants_norm: item["material_name"] = name item["material_id"] = cid item["material_row"] = {"id": cid, "canonical_unit_id": extra.get("canonical_unit_id")} elif field == "unit": for item in payload.get("items") or []: if normalize_text(item.get("unit_name") or "") in variants_norm: item["unit_id"] = cid item["unit_explicit"] = True item["unit_name"] = name if not payload.get("items"): payload["canonical_unit_id"] = cid payload["unit_name"] = name elif field == "brand": for item in payload.get("items") or []: if normalize_text(item.get("brand_name") or "") in variants_norm: item["brand_name"] = name elif field == "spec": for item in payload.get("items") or []: specs = item.get("specs") or [] for i, s in enumerate(specs): if normalize_text(s) in variants_norm: specs[i] = name def apply_exact(payload: dict, res: dict) -> None: """Bind one exact Jev match into a draft payload.""" apply_choice(payload, res.get("field"), {"id": res.get("id"), "name": res.get("name"), "extra": res.get("extra") or {}}, res.get("variants") or []) def normalize_brand_spec(payload: dict, jev_map: dict | None) -> None: """Canonicalize brand/spec strings via exact matches (spec effect per owner: spelling normalization only).""" if not jev_map: return for item in payload.get("items") or []: brand = item.get("brand_name") if brand: hit = lookup(jev_map, brand) if hit and hit.get("status") == "exact": item["brand_name"] = hit["name"] specs = item.get("specs") or [] for i, s in enumerate(specs): hit = lookup(jev_map, s) if hit and hit.get("status") == "exact": specs[i] = hit["name"] # --------------------------------------------------------------------------- # phase 4 — ambiguity sub-messages with candidate buttons # --------------------------------------------------------------------------- def _amb_id(field: str, variants: list[str]) -> str: digest = hashlib.sha1("|".join(sorted(variants)).encode()).hexdigest()[:4] return f"em_{field[:4]}_{digest}" def _fa_num(n: int) -> str: return str(n).translate(str.maketrans("0123456789", "۰۱۲۳۴۵۶۷۸۹")) def create_ambiguities(payload: dict, jev_map: dict | None) -> None: """One `entity_match` ambiguity entry per ambiguous mention (deduped).""" if not jev_map: return ambs = payload.setdefault("ambiguities", []) seen = {a.get("id") for a in ambs} unique: dict[int, dict] = {} for res in jev_map.values(): unique[id(res)] = res # dedupe identical result objects for res in unique.values(): if res.get("status") != "ambiguity" or not res.get("candidates"): continue amb_id = _amb_id(res["field"], res.get("variants") or []) if amb_id in seen: continue cands = res["candidates"][:6] lines = [f"• {c['name']} — {_fa_num(int(round(c['prob'] * 100)))}٪" for c in cands] ambs.append({ "id": amb_id, "kind": "entity_match", "field": res["field"], "variants": res.get("variants") or [], "candidates": [{"id": c["id"], "name": c["name"]} for c in cands], "title": f"تطبیق {FIELD_FA.get(res['field'], res['field'])}: «{' / '.join(res.get('variants') or [])}»", "details": "\n".join(lines), "status": "pending", }) seen.add(amb_id)