From 787824a19e2c59104d0a4e253414b37a42c9ee91 Mon Sep 17 00:00:00 2001 From: "reza.g" Date: Fri, 16 Oct 2026 03:13:21 +0330 Subject: [PATCH] feat(matching): implement two-stage matching with DeepSeek arbiter, fail-stop guard, and zero-probability filter --- bot/app/fallback_matcher.py | 282 +++++++++++++++++++++++++++- bot/app/handlers/__init__.py | 24 ++- bot/app/matching.py | 80 ++++---- bot/app/nlu.py | 17 +- bot/tools/test_fallback_matching.py | 100 +++++++++- bot/tools/test_matching.py | 20 ++ 6 files changed, 475 insertions(+), 48 deletions(-) diff --git a/bot/app/fallback_matcher.py b/bot/app/fallback_matcher.py index 86916cd..e7d0408 100644 --- a/bot/app/fallback_matcher.py +++ b/bot/app/fallback_matcher.py @@ -48,6 +48,66 @@ class MatchingFailureError(Exception): pass +ARBITRATION_SYSTEM_PROMPT = """تو قاضی و داور نهایی هوشمند تطبیق موجودیت‌های تدارکاتی و ساختمانی در یک سیستم ERP هستی. +یک مدل جستجوی برداری اولیه (به نام Jev) برای هر منشن کاربر، کاتالوگ دیتابیس را بررسی کرده و گزینه‌ای را پیشنهاد داده است. +وظیفه خطیر تو: +تحلیل هم‌زمان ۱) پیام اولیه کاربر، ۲) کاندیداهای کاتالوگ دیتابیس، و ۳) پیشنهاد مدل Jev، و صدور «حکم قطعی نهایی». + +یک شیء JSON و فقط JSON برگردان با ساختار زیر: +{ + "decisions": [ + { + "field": "supplier|material|project|unit|brand|spec", + "variant": "نگارش منشن", + "status": "exact|ambiguity|new", + "id": "شناسه رکورد کاتالوگ یا null", + "name": "نام رسمی رکورد کاتالوگ یا null", + "candidates": [ + {"id": "شناسه", "name": "نام", "prob": 0.0..1.0} + ], + "inferred_category_slug": "یکی از ۸ اسلاگ مجاز فقط برای تامین‌کننده new یا null", + "adjudication_action": "endorsed_jev|vetoed_to_new|demoted_to_ambiguity", + "reasoning": "توضیح کوتاه فنی دلیل قضاوت" + } + ] +} + +قوانین حیاتی داوری: +۱. عدم اعتماد کورکورانه به Jev: + - مدل Jev صرفاً یک مدل آماری/برداری است و تفاوت سایزها، رده‌ها، ابعاد فنی و استانداردهای صنعتی (مانند لوله ۶ اینچ در برابر لوله ۴ اینچ، میلگرد ۱۶ در برابر ۲۰، یا تفاوت دو برند متفاوت) را تشخیص نمی‌دهد. + - تو باید کلام کاربر را با پیشنهاد Jev مقایسه کنی. + +۲. حالات صدور حکم: + الف) تأیید انتخاب Jev (status: "exact"): + اگر پیشنهاد Jev دقیقاً و با احتساب نام تجاری و مشخصات فنی با خواسته کاربر در پیام مطابقت دارد. + - فیلد id و name را با مقادیر دقیق رکورد کاتالوگ پر کن. candidates خالی باشد ([]). + + ب) وتوی Jev و اعلام قلم جدید (status: "new"): + اگر کالا، تأمین‌کننده یا پروژه مورد نظر کاربر در کاتالوگ ارسالی وجود ندارد (حتی اگر Jev به اشتباه به دلیل شباهت حروفی گزینه‌ای را با اطمینان بالا پیشنهاد داده باشد). + - پیشنهاد Jev را بلافاصله باطل و وتو کن! + - وضعیت را "new" بگذار و id حتماً null باشد. هرگز قلم ناموجود را به زور به کاتالوگ نچسبان. + - برای تأمین‌کننده جدید (supplier)، مناسب‌ترین اسلاگ را از بین ۸ دسته مجاز در inferred_category_slug قرار بده. + + ج) تنزل به ابهام (status: "ambiguity"): + اگر کاربر عبارت کلی گفته (مثلاً «میلگرد») و در کاتالوگ چند گزینه محتمل وجود دارد، یا بین انتخاب Jev و سایر کاندیداها تردید فنی وجود دارد. + - وضعیت را "ambiguity" بگذار و گزینه‌های منتخب را در candidates بگذار. + +۳. تحلیل ورودی‌های صوتی (Voice Mode): + اگر ورودی صوتی است (input_mode: "voice")، آرایه کامل ترانویسی موتورهای STT (مانند Whisper، Vosk و غیره) به همراه consensus_text در اختیارت است. + شواهد صوتی، تلفظ‌ها، خطاهای شنیداری و لهجه‌های کارگاهی را ملاک قضاوت قرار بده. + +۴. اسلاگ‌های مجاز دسته‌بندی برای تأمین‌کننده جدید: + * metals-steel-sections + * mechanical-installations-pipes + * electrical-equipment-supplies + * construction-materials + * construction-chemicals + * doors-windows-glass + * finishing-interior-decoration + * construction-tools-equipment +""" + + FALLBACK_SYSTEM_PROMPT = """تو موتور تطبیق و پالایش هوشمند موجودیت‌های تدارکاتی یک سیستم ساختمانی و صنعتی ERP در ایران هستی. سرویس جانبی تطبیق کاتالوگ در دسترس نیست و اکنون تو وظیفه داری با درک مهندسی، تجاری و معنایی خود، عبارات ذکرشده توسط کاربر را مستقیماً با رکوردهای کاتالوگ دیتابیس تطبیق دهی. @@ -115,6 +175,7 @@ async def match_with_deepseek( candidates_by_field: dict[str, list], patch_context: dict | None, trace_id: str | None, + consensus_text: str | None = None, ) -> dict[str, dict]: """Execute DeepSeek crisis matching for unmapped mentions. @@ -141,8 +202,20 @@ async def match_with_deepseek( field_cands.append(item) catalog_summary[fld] = field_cands[:80] # token-safe ceiling per field + if isinstance(raw_input, list): + reference_input = { + "input_mode": "voice", + "stt_multi_transcripts": raw_input, + "consensus_text": consensus_text or "", + } + else: + reference_input = { + "input_mode": "text", + "raw_text": str(raw_input or ""), + } + payload_data = { - "raw_user_input": raw_input, + "reference_user_input": reference_input, "entity_mentions": mentions, "catalog_candidates": catalog_summary, "patch_context": patch_context or {}, @@ -270,3 +343,210 @@ async def match_with_deepseek( }) return results_map + + +async def adjudicate_with_deepseek( + raw_input: object, + consensus_text: str | None, + items: list[dict], + patch_context: dict | None, + trace_id: str | None, +) -> dict[str, dict]: + """Execute DeepSeek final arbitration over Jev recommendations. + + Takes Jev-assessed items and makes final binding decisions. + Raises MatchingFailureError if DeepSeek fails or returns unparseable output. + """ + if not items: + return {} + + if not settings.AI_BASE_URL or not settings.AI_API_KEY: + logger.error("fallback_matcher: AI provider not configured for arbitration") + raise MatchingFailureError("AI provider not configured") + + if isinstance(raw_input, list): + reference_input = { + "input_mode": "voice", + "stt_multi_transcripts": raw_input, + "consensus_text": consensus_text or "", + } + else: + reference_input = { + "input_mode": "text", + "raw_text": str(raw_input or ""), + } + + arbitration_items_payload = [] + # Candidate lookup by (field, id) and (field, name) for metadata + cand_lookup: dict[tuple[str, str], object] = {} + + for it in items: + fld = it["field"] + c_list = it.get("candidates") or [] + serialized_cands = [] + for c in c_list: + cid = str(getattr(c, "id", "")) + cname = str(getattr(c, "name", "")) + cand_lookup[(fld, cid)] = c + cand_lookup[(fld, normalize_text(cname))] = c + item_entry = {"id": cid, "name": cname} + extra = getattr(c, "extra", {}) or {} + for k in ("category_name", "location", "canonical_unit_id", "code"): + if extra.get(k): + item_entry[k] = extra[k] + serialized_cands.append(item_entry) + + raw_probs = it.get("jev_probabilities") or {} + non_zero_probs = {k: v for k, v in raw_probs.items() if float(v or 0.0) > 0.0} + + arbitration_items_payload.append({ + "field": fld, + "variants": it.get("variants") or [], + "catalog_candidates": serialized_cands[:60], + "jev_assessment": { + "recommended_choice": it.get("jev_choice"), + "recommended_id": it.get("jev_choice_id"), + "confidence": it.get("jev_confidence"), + "probabilities": non_zero_probs, + "top_ranked": it.get("jev_ranked") or [], + }, + }) + + payload_data = { + "reference_user_input": reference_input, + "items_to_adjudicate": arbitration_items_payload, + "patch_context": patch_context or {}, + } + user_content = json.dumps(payload_data, ensure_ascii=False) + + t0 = time.perf_counter() + try: + async with httpx.AsyncClient(timeout=settings.LLM_TIMEOUT_SECONDS) as client: + resp = await client.post( + f"{settings.AI_BASE_URL.rstrip('/')}/chat/completions", + headers={"Authorization": f"Bearer {settings.AI_API_KEY}"}, + json={ + "model": settings.LLM_MODEL, + "messages": [ + {"role": "system", "content": ARBITRATION_SYSTEM_PROMPT}, + {"role": "user", "content": user_content}, + ], + "temperature": 0.1, + "response_format": {"type": "json_object"}, + }, + ) + latency_ms = int((time.perf_counter() - t0) * 1000) + + if resp.status_code != 200: + logger.error("adjudicate_with_deepseek: HTTP %s from DeepSeek: %s", resp.status_code, resp.text[:300]) + await obs.event(trace_id, "matching.arbitration.failure", ok=False, payload={ + "status_code": resp.status_code, + "error": "http_error", + "body": resp.text[:300], + }) + raise MatchingFailureError(f"DeepSeek returned HTTP {resp.status_code}") + + content = ((resp.json() or {}).get("choices") or [{}])[0].get("message", {}).get("content", "") + usage = (resp.json() or {}).get("usage") or {} + parsed = _extract_json(content) + if not parsed or not isinstance(parsed.get("decisions"), list): + logger.error("adjudicate_with_deepseek: unparseable JSON from DeepSeek: %r", content[:300]) + await obs.event(trace_id, "matching.arbitration.failure", ok=False, payload={ + "error": "unparseable_json", + "raw": content[:500], + }) + raise MatchingFailureError("DeepSeek returned unparseable arbitration JSON") + + except httpx.HTTPError as exc: + latency_ms = int((time.perf_counter() - t0) * 1000) + logger.exception("adjudicate_with_deepseek: HTTP network error calling DeepSeek") + await obs.event(trace_id, "matching.arbitration.failure", ok=False, payload={ + "error": "network_exception", + "details": str(exc), + "latency_ms": latency_ms, + }) + raise MatchingFailureError("Network exception calling DeepSeek") from exc + except Exception as exc: + if isinstance(exc, MatchingFailureError): + raise + latency_ms = int((time.perf_counter() - t0) * 1000) + logger.exception("adjudicate_with_deepseek: Unexpected error") + await obs.event(trace_id, "matching.arbitration.failure", ok=False, payload={ + "error": "unexpected_exception", + "details": str(exc), + "latency_ms": latency_ms, + }) + raise MatchingFailureError("Unexpected error calling DeepSeek") from exc + + results_map: dict[str, dict] = {} + actions_count = {"endorsed_jev": 0, "vetoed_to_new": 0, "demoted_to_ambiguity": 0, "other": 0} + statuses_count = {"exact": 0, "ambiguity": 0, "new": 0} + + # Map decisions back to results_map + for dec in parsed.get("decisions", []): + if not isinstance(dec, dict): + continue + field = str(dec.get("field") or "").strip() + variant = str(dec.get("variant") or "").strip() + status = str(dec.get("status") or "").strip().lower() + if status not in ("exact", "ambiguity", "new"): + status = "new" + + action = str(dec.get("adjudication_action") or "other") + actions_count[action] = actions_count.get(action, 0) + 1 + statuses_count[status] = statuses_count.get(status, 0) + 1 + + cand_id = str(dec.get("id")) if dec.get("id") else None + cand_name = str(dec.get("name")) if dec.get("name") else None + cand_obj = None + if cand_id: + cand_obj = cand_lookup.get((field, cand_id)) + if not cand_obj and cand_name: + cand_obj = cand_lookup.get((field, normalize_text(cand_name))) + extra = getattr(cand_obj, "extra", {}) or {} + + amb_candidates = [] + for c in dec.get("candidates") or []: + if isinstance(c, dict) and c.get("id") and c.get("name"): + amb_candidates.append({ + "id": str(c["id"]), + "name": str(c["name"]), + "prob": float(c.get("prob") or 0.5), + }) + + # Find matching item from items for variants + matched_item = next( + (it for it in items if it["field"] == field and any(normalize_text(v) == normalize_text(variant) for v in it.get("variants", []))), + None + ) + variants = matched_item.get("variants") if matched_item else [variant] + + entry = { + "field": field, + "status": status, + "variants": variants, + "id": cand_id, + "name": cand_name, + "extra": extra, + "candidates": amb_candidates, + "source": "deepseek-arbiter", + "adjudication_action": action, + "reasoning": dec.get("reasoning"), + } + if dec.get("inferred_category_slug"): + entry["inferred_category_slug"] = str(dec["inferred_category_slug"]) + + for v in variants: + results_map[normalize_text(v)] = entry + + # Observability: matching.arbitration.result + await obs.event(trace_id, "matching.arbitration.result", ok=True, payload={ + "latency_ms": latency_ms, + "usage": usage, + "actions": actions_count, + "statuses": statuses_count, + "n_items": len(items), + "n_resolved": len(results_map), + }) + + return results_map diff --git a/bot/app/handlers/__init__.py b/bot/app/handlers/__init__.py index c1a4793..296d15a 100644 --- a/bot/app/handlers/__init__.py +++ b/bot/app/handlers/__init__.py @@ -13,7 +13,8 @@ import logging from app import drafts as draft_store from app import observability as obs from app.config import settings -from app.nlu import understand +from app.fallback_matcher import MatchingFailureError +from app.nlu import NLUFailureError, understand from app.runtime import BotContext, LinkedUser from app.storage import save_proforma_file from app.stt import transcribe, transcribe_multi @@ -222,23 +223,34 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None: draft_summary = _draft_summary(target_draft) if target_draft else None input_payload = voice_transcripts if voice_transcripts else text - result = await understand(settings, input_payload, draft_summary, trace_id=trace_id) + try: + result = await understand(settings, input_payload, draft_summary, trace_id=trace_id) + except (MatchingFailureError, NLUFailureError): + logger.error("NLU processing failed completely (Universal Fail-Stop Guard)") + await ctx.send( + chat_id, + "⚠️ سامانه در پردازش هوشمند موجودیت‌ها با اختلال مواجه شد. لطفاً دوباره پیام را ارسال کنید.", + reply_to=message_id, + ) + return + intent, patch = result.get("intent"), result.get("patch") or {} - consensus_text = result.get("consensus_text") or text + consensus_text = result.get("consensus_text") or (text if isinstance(text, str) else "") mentions = result.get("entity_mentions") or [] await obs.event(trace_id, "entities.extracted", payload={"patch": patch, "intent": intent, "entity_mentions": mentions}) - # Entity matching (Jev primary, DeepSeek crisis fallback) + # Entity matching (Two-Stage: Jev scorer + DeepSeek final arbiter; DeepSeek crisis fallback if Jev down) jev_map: dict = {} if mentions and intent in ("register_quote", "update_draft", "add_material"): try: from app import matching jev_map = await matching.resolve_mentions( - ctx, patch, mentions, trace_id=trace_id, raw_input=input_payload + ctx, patch, mentions, trace_id=trace_id, raw_input=input_payload, + consensus_text=consensus_text, ) except matching.MatchingFailureError: - logger.error("entity matching failed completely (crisis fallback failed)") + logger.error("entity matching failed completely (Universal Fail-Stop Guard)") await ctx.send( chat_id, "⚠️ سامانه در پردازش هوشمند موجودیت‌ها با اختلال مواجه شد. لطفاً دوباره پیام را ارسال کنید.", diff --git a/bot/app/matching.py b/bot/app/matching.py index cc124bd..5b9bc67 100644 --- a/bot/app/matching.py +++ b/bot/app/matching.py @@ -87,7 +87,7 @@ def _instructions(field: str, variants: list[str]) -> str: # --------------------------------------------------------------------------- -from app.fallback_matcher import MatchingFailureError, match_with_deepseek +from app.fallback_matcher import MatchingFailureError, match_with_deepseek, adjudicate_with_deepseek # --------------------------------------------------------------------------- # the per-field fan-out @@ -99,8 +99,9 @@ async def resolve_mentions( mentions: list[dict], trace_id: str | None = None, raw_input: object = None, + consensus_text: str | None = None, ) -> dict: - """Run Jev per field; on failure, trigger DeepSeek Crisis Fallback. + """Run Two-Stage Matching: Jev scorer + DeepSeek final arbiter; DeepSeek crisis fallback if Jev down. Return a map keyed by normalized variant. Each value: {"field", "status": exact|ambiguity|new", "variants", @@ -119,6 +120,8 @@ async def resolve_mentions( unhandled_mentions: list[dict] = [] unhandled_fields: set[str] = set() + jev_arbitration_items: list[dict] = [] + cands_by_field: dict[str, list] = {} async def one_field(field: str, ms: list[dict]) -> None: try: @@ -133,6 +136,8 @@ async def resolve_mentions( unhandled_fields.add(field) return + cands_by_field[field] = cands + by_name: dict[str, object] = {} criteria: dict[str, str] = {} for c in cands: @@ -187,39 +192,36 @@ async def resolve_mentions( choice = ans.get("choice") conf = float(ans.get("confidence") or 0.0) probs = ans.get("probabilities") or {} + non_zero_probs = {k: v for k, v in probs.items() if float(v or 0.0) > 0.0} 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), + for n, p in non_zero_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} + # Collect for DeepSeek final arbitration (never write directly to results) + jev_arbitration_items.append({ + "field": field, + "variants": m["variants"], + "candidates": cands, + "jev_choice": cand.name if cand else ("_none" if choice == "_none" else None), + "jev_choice_id": cand.id if cand else None, + "jev_confidence": conf, + "jev_probabilities": non_zero_probs, + "jev_ranked": ranked, + }) + + decisions.append({ + "variants": m["variants"], + "jev_choice": cand.name if cand else choice, + "confidence": conf, + "probs_top": [{k: round(v, 3)} for k, v in + sorted(probs.items(), key=lambda kv: -kv[1])[:3]], + }) - 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}) + payload={"source": "jev_scored", "mentions": decisions}) if settings.JEV_ENABLED: await asyncio.gather(*(one_field(f, ms) for f, ms in by_field.items())) @@ -229,20 +231,32 @@ async def resolve_mentions( unhandled_mentions.extend(ms) unhandled_fields.add(f) - # If any mentions were unhandled / failed in Jev, run DeepSeek Crisis Fallback + # 1. Final Arbitration: Pass Jev recommendations + raw user input + DB catalog to DeepSeek + if jev_arbitration_items: + adjudicated_res = await adjudicate_with_deepseek( + raw_input=raw_input, + consensus_text=consensus_text, + items=jev_arbitration_items, + patch_context=patch, + trace_id=trace_id, + ) + results.update(adjudicated_res) + + # 2. Crisis Fallback: 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] = [] + if f not in cands_by_field: + 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, + consensus_text=consensus_text, ) results.update(fallback_res) diff --git a/bot/app/nlu.py b/bot/app/nlu.py index d5413ad..5c5827b 100644 --- a/bot/app/nlu.py +++ b/bot/app/nlu.py @@ -16,6 +16,11 @@ from app.config import Settings logger = logging.getLogger("bot.nlu") + +class NLUFailureError(Exception): + """Raised when NLU understanding fails due to LLM / network failure.""" + pass + PERSIAN_DIGITS = str.maketrans("۰۱۲۳۴۵۶۷۸۹٠١٢٣٤٥٦٧٨٩", "01234567890123456789") @@ -233,14 +238,14 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary: logger.error("LLM HTTP %s: %s", resp.status_code, resp.text[:300]) await obs.event(trace_id, "nlu.result", ok=False, model=settings.LLM_MODEL, payload={"error": f"HTTP {resp.status_code}", "body": resp.text[:300]}) - return fallback + raise NLUFailureError(f"HTTP {resp.status_code}") content = ((resp.json() or {}).get("choices") or [{}])[0].get("message", {}).get("content", "") parsed = _extract_json(content) if not parsed or parsed.get("intent") not in INTENTS: logger.warning("unparseable LLM output: %r", content[:200]) await obs.event(trace_id, "nlu.result", ok=False, model=settings.LLM_MODEL, payload={"error": "unparseable", "raw": content[:500]}) - return fallback + raise NLUFailureError("unparseable LLM output") parsed.setdefault("patch", {}) parsed["entity_mentions"] = sanitize_mentions(parsed.get("entity_mentions")) if not parsed.get("consensus_text"): @@ -253,8 +258,10 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary: "entity_mentions": parsed.get("entity_mentions"), }) return parsed - except (httpx.HTTPError, json.JSONDecodeError, KeyError, TypeError): + except Exception as exc: + if isinstance(exc, NLUFailureError): + raise logger.exception("NLU failure") await obs.event(trace_id, "nlu.result", ok=False, model=settings.LLM_MODEL, - payload={"error": "exception"}) - return fallback + payload={"error": "exception", "details": str(exc)}) + raise NLUFailureError(f"NLU exception: {exc}") from exc diff --git a/bot/tools/test_fallback_matching.py b/bot/tools/test_fallback_matching.py index 7cad91f..1c41c91 100644 --- a/bot/tools/test_fallback_matching.py +++ b/bot/tools/test_fallback_matching.py @@ -116,7 +116,7 @@ async def test_jev_failure_triggers_fallback() -> None: matching.jev_ask = lambda state, qs: asyncio.sleep(0.001, result=None) # Mock DeepSeek fallback matcher - async def fake_fallback(raw_input, mentions, candidates_by_field, patch_context, trace_id): + async def fake_fallback(*args, **kwargs): return { normalize_text("میلگرد ۱۴"): { "field": "material", @@ -181,12 +181,84 @@ async def test_jev_failure_triggers_fallback() -> None: matching.match_with_deepseek = orig_deepseek +async def test_two_stage_adjudication_flow() -> None: + """2. When Jev succeeds, outputs are NOT directly committed; they are sent to DeepSeek Arbiter.""" + ctx = SimpleNamespace(resolver=FakeResolver()) + saved_key = settings.TYPESAFE_API_KEY + settings.TYPESAFE_API_KEY = "test-typesafe-key" + + orig_jev_ask = matching.jev_ask + matching.jev_ask = lambda state, qs: asyncio.sleep(0.001, result={ + "answers": { + "mention_0": { + "choice": "میلگرد ۱۴ آجدار", + "confidence": 0.95, + "probabilities": {"میلگرد ۱۴ آجدار": 0.95, "تسمه فلزی": 0.0, "_none": 0.05}, + }, + "mention_1": { + "choice": "لوله پی‌وی‌سی ۴ اینچ", + "confidence": 0.82, + "probabilities": {"لوله پی‌وی‌سی ۴ اینچ": 0.82, "لوله پلی‌اتیلن": 0.0, "_none": 0.18}, + }, + } + }) + + arbiter_called_with = [] + + async def fake_arbiter(*args, **kwargs): + arbiter_called_with.append((args, kwargs)) + items = kwargs.get("items") or (args[2] if len(args) > 2 else []) + res = {} + for it in items: + v0 = it["variants"][0] + if v0 == "میلگرد ۱۴": + # Endorsed + res[normalize_text(v0)] = { + "field": it["field"], "status": "exact", "id": "m1", "name": "میلگرد ۱۴ آجدار", + "variants": it["variants"], "source": "deepseek-arbiter", "adjudication_action": "endorsed_jev", + } + elif v0 == "لوله ۶ اینچ": + # Vetoed Jev (Jev wrongly picked 4-inch pipe) + res[normalize_text(v0)] = { + "field": it["field"], "status": "new", "id": None, "name": None, + "variants": it["variants"], "source": "deepseek-arbiter", "adjudication_action": "vetoed_to_new", + } + return res + + orig_arbiter = matching.adjudicate_with_deepseek + matching.adjudicate_with_deepseek = fake_arbiter + + try: + mentions = [ + {"field": "material", "variants": ["میلگرد ۱۴"]}, + {"field": "material", "variants": ["لوله ۶ اینچ"]}, + ] + res = await matching.resolve_mentions( + ctx, {"items": []}, mentions, trace_id="t_arb", + raw_input="استعلام میلگرد ۱۴ و لوله ۶ اینچ", + ) + + check("Two-stage arbitration called when Jev succeeds", len(arbiter_called_with) == 1) + passed_items = arbiter_called_with[0][1].get("items") or arbiter_called_with[0][0][2] + m0_probs = passed_items[0]["jev_probabilities"] + check("Zero probability items filtered out", "تسمه فلزی" not in m0_probs) + check("Non-zero probability preserved exactly", m0_probs.get("میلگرد ۱۴ آجدار") == 0.95) + check("_none with non-zero probability preserved", m0_probs.get("_none") == 0.05) + check("Catalog candidates list complete and untouched", len(passed_items[0]["candidates"]) > 0) + check("Arbiter endorsed valid Jev choice as exact", res[normalize_text("میلگرد ۱۴")]["status"] == "exact") + check("Arbiter vetoed wrong Jev choice to new", res[normalize_text("لوله ۶ اینچ")]["status"] == "new" and res[normalize_text("لوله ۶ اینچ")]["id"] is None) + finally: + settings.TYPESAFE_API_KEY = saved_key + matching.jev_ask = orig_jev_ask + matching.adjudicate_with_deepseek = orig_arbiter + + async def test_defensive_failure_guard() -> None: - """2. If DeepSeek fallback fails or returns unparseable JSON, MatchingFailureError is raised.""" + """3. Universal Fail-Stop Guard: If DeepSeek (fallback or arbiter or NLU) fails, MatchingFailureError is raised.""" ctx = SimpleNamespace(resolver=FakeResolver()) matching.jev_ask = lambda state, qs: asyncio.sleep(0.001, result=None) - async def boom_fallback(raw_input, mentions, candidates_by_field, patch_context, trace_id): + async def boom_fallback(*args, **kwargs): raise MatchingFailureError("DeepSeek API 500 error") orig_deepseek = matching.match_with_deepseek @@ -203,6 +275,27 @@ async def test_defensive_failure_guard() -> None: finally: matching.match_with_deepseek = orig_deepseek + # Also test Arbiter failure raises MatchingFailureError + matching.jev_ask = lambda state, qs: asyncio.sleep(0.001, result={"answers": {"mention_0": {"choice": "m1", "confidence": 0.9}}}) + async def boom_arbiter(*args, **kwargs): + raise MatchingFailureError("Arbiter DeepSeek timeout") + + orig_arbiter = matching.adjudicate_with_deepseek + matching.adjudicate_with_deepseek = boom_arbiter + try: + failed_arb = False + try: + await matching.resolve_mentions(ctx, {"items": []}, mentions, trace_id="t2_arb") + except MatchingFailureError: + failed_arb = True + check("Arbiter failure raises MatchingFailureError (stops cleanly)", failed_arb) + finally: + matching.adjudicate_with_deepseek = orig_arbiter + + # Test NLUFailureError is subclass of Exception + from app.nlu import NLUFailureError + check("NLUFailureError is subclass of Exception", issubclass(NLUFailureError, Exception)) + async def test_quote_resolve_payload_zero_difflib() -> None: """3. resolve_payload in quote.py respects Jev/fallback map and never runs difflib.""" @@ -292,6 +385,7 @@ async def main() -> None: fallback_matcher.obs.event = lambda *a, **k: asyncio.sleep(0.0001) try: await test_jev_failure_triggers_fallback() + await test_two_stage_adjudication_flow() await test_defensive_failure_guard() await test_quote_resolve_payload_zero_difflib() await test_quote_commit_defensive_get_or_create() diff --git a/bot/tools/test_matching.py b/bot/tools/test_matching.py index efde916..26f5c2d 100644 --- a/bot/tools/test_matching.py +++ b/bot/tools/test_matching.py @@ -81,6 +81,25 @@ async def mock_tests() -> None: for m in ms for v in m.get("variants", [])} matching.match_with_deepseek = mock_fallback_below_floor + orig_arbiter = matching.adjudicate_with_deepseek + async def mock_arbiter(*args, **kwargs): + items = kwargs.get("items") or (args[2] if len(args) > 2 else []) + res = {} + for it in items: + choice = it.get("jev_choice") + conf = it.get("jev_confidence", 0.0) + if choice == "منشانه" and conf >= 0.8: + entry = {"status": "exact", "id": "p2", "name": "منشانه", "field": it["field"], "variants": it["variants"]} + elif choice == "دفتر" and conf >= 0.6: + entry = {"status": "ambiguity", "id": "p1", "name": "دفتر", "field": it["field"], "variants": it["variants"], + "candidates": [{"id": "p1", "name": "دفتر", "prob": conf}]} + else: + entry = {"status": "new", "id": None, "name": None, "field": it["field"], "variants": it["variants"], "candidates": []} + for v in it["variants"]: + res[normalize_text(v)] = entry + return res + matching.adjudicate_with_deepseek = mock_arbiter + mentions = [ {"field": "project", "variants": ["منشیانه"]}, {"field": "project", "variants": ["دفتر", "دفر"]}, @@ -91,6 +110,7 @@ async def mock_tests() -> None: res = await matching.resolve_mentions(ctx, {"items": []}, mentions, None) finally: matching.match_with_deepseek = orig_fallback + matching.adjudicate_with_deepseek = orig_arbiter r0 = matching.lookup(res, "منشیانه") r1 = matching.lookup(res, "دفر") # variant key resolves too