""" Offline evaluation harness for ModuleDesk's AI "Write reply" draft feature. Measures how close the app's auto-draft is to the reply the seller ACTUALLY sent, on already-resolved tickets from tenant_internal. Also measures € cost per draft. Usage: venv/bin/python scripts/eval_draft_quality.py [--n 8] [--schema tenant_internal] Cost assumptions (noted as approximations): gpt-4o: $2.50/1M prompt tokens, $10.00/1M completion tokens gpt-4o-mini: $0.15/1M prompt tokens, $0.60/1M completion tokens text-embedding-3-small: $0.02/1M tokens EUR/USD rate: 0.92 These are published OpenAI list prices as of mid-2025. """ from __future__ import annotations import argparse import json import os import sys import time from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional # Make project root importable sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # --------------------------------------------------------------------------- # Pricing constants (assumptions — see module docstring) # --------------------------------------------------------------------------- USD_EUR = 0.92 PRICE_USD_PER_1M = { "gpt-4o": {"in": 2.50, "out": 10.00}, "gpt-4o-mini": {"in": 0.15, "out": 0.60}, "text-embedding-3-small": {"in": 0.02, "out": 0.00}, } def usd_to_eur(usd: float) -> float: return usd * USD_EUR def tokens_to_eur(model: str, prompt_tokens: int, completion_tokens: int) -> float: p = PRICE_USD_PER_1M.get(model, {"in": 0.0, "out": 0.0}) usd = (prompt_tokens * p["in"] + completion_tokens * p["out"]) / 1_000_000 return usd_to_eur(usd) # --------------------------------------------------------------------------- # Token tracking: wrap openai client calls and intercept usage # --------------------------------------------------------------------------- @dataclass class CallRecord: call_type: str # classify | draft | embed | judge model: str prompt_tokens: int completion_tokens: int cost_eur: float @dataclass class RunAccumulator: calls: List[CallRecord] = field(default_factory=list) def record(self, call_type: str, model: str, prompt_tokens: int, completion_tokens: int) -> float: cost = tokens_to_eur(model, prompt_tokens, completion_tokens) self.calls.append(CallRecord(call_type, model, prompt_tokens, completion_tokens, cost)) return cost def total_cost(self) -> float: return sum(c.cost_eur for c in self.calls) # --------------------------------------------------------------------------- # Patched OpenAI client that forwards usage to an accumulator # --------------------------------------------------------------------------- import openai import numpy as np class TrackedOpenAI: """Thin wrapper around openai.OpenAI that records token usage to an accumulator.""" def __init__(self, real_client: openai.OpenAI, acc: RunAccumulator): self._client = real_client self._acc = acc self.chat = _ChatProxy(real_client, acc) self.embeddings = _EmbedProxy(real_client, acc) def __getattr__(self, name: str): return getattr(self._client, name) class _ChatProxy: def __init__(self, client, acc): self._client = client self._acc = acc class _CompletionsProxy: def __init__(self, client, acc): self._client = client self._acc = acc def create(self, **kwargs): resp = self._client.chat.completions.create(**kwargs) usage = resp.usage if usage: self._acc.record( "chat", kwargs.get("model", "unknown"), usage.prompt_tokens, usage.completion_tokens, ) return resp def __init__(self, client, acc): self._client = client self._acc = acc self.completions = _ChatProxy._CompletionsProxy(client, acc) class _EmbedProxy: def __init__(self, client, acc): self._client = client self._acc = acc def create(self, **kwargs): resp = self._client.embeddings.create(**kwargs) usage = resp.usage if usage: # Embeddings return total_tokens in prompt_tokens; no completion self._acc.record( "embed", kwargs.get("model", "unknown"), usage.prompt_tokens, 0, ) return resp # --------------------------------------------------------------------------- # Ticket selection # --------------------------------------------------------------------------- def select_eval_tickets(session, n: int = 8) -> List[Dict[str, Any]]: """ Select n closed tickets with: - At least one inbound message (the question) - A substantive outbound reply (> 150 chars, not screenshot-only) - Variety across products and languages Returns list of dicts with IDs only (no ORM objects — avoids session detachment). ORM objects are loaded per-ticket inside the main loop with a fresh session. """ from sqlalchemy import text # Pull candidates: one per product for variety, ordered by recency rows = session.execute(text(""" SELECT DISTINCT ON (at_.id_product) t.id AS ticket_id, at_.id AS thread_id, at_.subject, at_.product_name, at_.id_product, at_.customer_language, (SELECT id FROM addons_messages WHERE thread_id = at_.id AND direction = 'inbound' ORDER BY created_at ASC LIMIT 1) AS last_inbound_id, (SELECT id FROM addons_messages WHERE thread_id = at_.id AND direction = 'outbound' AND length(body_text) > 150 ORDER BY created_at ASC LIMIT 1) AS first_outbound_id, (SELECT body_text FROM addons_messages WHERE thread_id = at_.id AND direction = 'outbound' AND length(body_text) > 150 ORDER BY created_at ASC LIMIT 1) AS real_reply FROM tickets t JOIN addons_threads at_ ON at_.id = t.thread_id -- Ticket must have at least one inbound JOIN addons_messages mi ON mi.thread_id = at_.id AND mi.direction = 'inbound' -- Must have a substantive outbound reply JOIN addons_messages mo ON mo.thread_id = at_.id AND mo.direction = 'outbound' AND length(mo.body_text) > 150 WHERE t.status = 'closed' AND at_.id_product IS NOT NULL -- Exclude purely positive/screenshot replies in outbound AND NOT (lower(mo.body_text) LIKE '%screenshot%' AND length(mo.body_text) < 250) ORDER BY at_.id_product, t.id DESC LIMIT :limit """), {"limit": n * 3}).fetchall() # fetch extra for fallback # If we have fewer unique products than n, fill the remainder if len(rows) < n: extra = session.execute(text(""" SELECT t.id AS ticket_id, at_.id AS thread_id, at_.subject, at_.product_name, at_.id_product, at_.customer_language, (SELECT id FROM addons_messages WHERE thread_id = at_.id AND direction = 'inbound' ORDER BY created_at ASC LIMIT 1) AS last_inbound_id, (SELECT id FROM addons_messages WHERE thread_id = at_.id AND direction = 'outbound' AND length(body_text) > 150 ORDER BY created_at ASC LIMIT 1) AS first_outbound_id, (SELECT body_text FROM addons_messages WHERE thread_id = at_.id AND direction = 'outbound' AND length(body_text) > 150 ORDER BY created_at ASC LIMIT 1) AS real_reply FROM tickets t JOIN addons_threads at_ ON at_.id = t.thread_id JOIN addons_messages mi ON mi.thread_id = at_.id AND mi.direction = 'inbound' JOIN addons_messages mo ON mo.thread_id = at_.id AND mo.direction = 'outbound' AND length(mo.body_text) > 150 WHERE t.status = 'closed' ORDER BY t.id DESC LIMIT :limit """), {"limit": n * 2}).fetchall() seen_ids = {r.ticket_id for r in rows} for r in extra: if r.ticket_id not in seen_ids: rows.append(r) seen_ids.add(r.ticket_id) selected = rows[:n] # Return plain dicts (no ORM objects) to avoid session detachment issues result = [] for row in selected: if not row.last_inbound_id or not row.first_outbound_id or not row.real_reply: continue result.append({ "ticket_id": row.ticket_id, "thread_id": row.thread_id, "subject": row.subject, "product_name": row.product_name, "customer_language": row.customer_language, "inbound_msg_id": row.last_inbound_id, "real_reply": row.real_reply, }) return result # --------------------------------------------------------------------------- # Cosine similarity between two texts via embedding API # --------------------------------------------------------------------------- def cosine_similarity(vec_a: List[float], vec_b: List[float]) -> float: a = np.array(vec_a, dtype=np.float32) b = np.array(vec_b, dtype=np.float32) norm_a = np.linalg.norm(a) norm_b = np.linalg.norm(b) if norm_a < 1e-10 or norm_b < 1e-10: return 0.0 return float(np.dot(a, b) / (norm_a * norm_b)) # --------------------------------------------------------------------------- # LLM judge # --------------------------------------------------------------------------- HONEST_JUDGE_SYSTEM = """\ You are evaluating AI-generated customer support draft replies for a PrestaShop module seller. You will receive: 1. The customer's original inbound message 2. The AI-generated draft reply Rate whether this draft is something a competent, professional PrestaShop module seller could send RIGHT NOW — judging only from what the customer wrote. A draft is USABLE if it either: (a) Correctly answers or resolves the issue using safe, accurate general knowledge about PrestaShop and modules (no hallucinated specific steps), OR (b) Professionally asks for the specific information, access, or details needed to proceed (BO access, error logs, screenshots, PS/module version, etc.) — this is the CORRECT response when the issue cannot be diagnosed without more info. A draft is NOT USABLE if: - It hallucinates a specific fix or setting that cannot be known from the question alone - It is generic/unhelpful boilerplate that doesn't engage with the customer's actual issue - It is in the wrong language relative to the customer's message - The tone is significantly unprofessional or off-brand for a technical support reply - It refuses to help or adds unjustified disclaimers Be strict about hallucinated specifics (claiming a particular setting, file path, or code fix without knowing the actual issue = NOT usable). But DO credit ask-for-info replies — asking for BO access/logs when the issue is unclear is professional and correct. Answer ONLY with JSON: {"usable": true/false, "reason": "one sentence", "would_send_as_is_or_minor_edits": true/false} """ JUDGE_SYSTEM = """\ You are an expert evaluator for a customer support system. You will be given: 1. The AI-generated draft reply 2. The actual reply the seller sent Your task: determine whether the seller could send the AI draft with no more than minor copy-editing (fixing a word, adjusting a phrase). Answer strictly with JSON. Criteria for usable=true: - Covers the same core answer / resolution path as the real reply - Tone is professional and appropriate - No significant missing steps, wrong information, or hallucinations - Minor wording differences are fine Criteria for usable=false: - Wrong diagnosis or solution - Key information present in real reply is absent from draft - Tone is significantly off (too formal, too casual, wrong language) - Draft is a generic non-answer when the real reply is specific Output format (JSON only, no other text): { "usable": true or false, "reason": "one sentence reason", "tone_match_0_5": 0-5 integer (0=completely wrong tone, 5=perfect match) } """ def llm_judge( draft: str, real_reply: str, judge_client: openai.OpenAI, acc: RunAccumulator, ) -> Dict[str, Any]: """Call gpt-4o-mini to judge whether the draft is usable vs the real reply.""" user_content = ( f"AI-generated draft:\n---\n{draft[:2000]}\n---\n\n" f"Actual seller reply:\n---\n{real_reply[:2000]}\n---\n\n" "Respond with JSON only." ) try: resp = judge_client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": JUDGE_SYSTEM}, {"role": "user", "content": user_content}, ], temperature=0.0, max_tokens=200, response_format={"type": "json_object"}, ) usage = resp.usage if usage: acc.record("judge", "gpt-4o-mini", usage.prompt_tokens, usage.completion_tokens) content = resp.choices[0].message.content result = json.loads(content) return { "usable": bool(result.get("usable", False)), "reason": str(result.get("reason", "")), "tone_match_0_5": int(result.get("tone_match_0_5", 0)), } except Exception as exc: return {"usable": False, "reason": f"Judge call failed: {exc}", "tone_match_0_5": 0} def honest_usability_judge( inbound_text: str, draft: str, judge_client: openai.OpenAI, acc: RunAccumulator, ) -> Dict[str, Any]: """Call gpt-4o-mini to judge whether the draft is usable vs the customer's inbound message only.""" user_content = ( f"Customer's inbound message:\n---\n{inbound_text[:2000]}\n---\n\n" f"AI-generated draft reply:\n---\n{draft[:2000]}\n---\n\n" "Respond with JSON only." ) try: resp = judge_client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": HONEST_JUDGE_SYSTEM}, {"role": "user", "content": user_content}, ], temperature=0.0, max_tokens=200, response_format={"type": "json_object"}, ) usage = resp.usage if usage: acc.record("honest_judge", "gpt-4o-mini", usage.prompt_tokens, usage.completion_tokens) content = resp.choices[0].message.content result = json.loads(content) return { "usable": bool(result.get("usable", False)), "reason": str(result.get("reason", "")), "would_send_as_is_or_minor_edits": bool(result.get("would_send_as_is_or_minor_edits", False)), } except Exception as exc: return { "usable": False, "reason": f"Judge call failed: {exc}", "would_send_as_is_or_minor_edits": False, } # --------------------------------------------------------------------------- # Ask-for-info detection (Phase 4) # --------------------------------------------------------------------------- _ASK_INFO_PHRASES = [ # Access requests "back office", "back-office", "bo access", "ftp access", "admin access", "temporary access", "accès temporaire", "acceso temporal", "acceso al bo", # Log / error requests "error message", "error log", "console log", "browser console", "screenshot", "capture d'écran", "captura de pantalla", # Version requests "prestashop version", "module version", "versión de prestashop", "version du module", "versión del módulo", # Generic info-request signals "could you share", "could you provide", "please provide", "please share", "can you send", "can you share", "pouvez-vous", "puede enviarme", "would you be able", "podría", "pourriez-vous", ] def is_ask_for_info_reply(draft: str) -> bool: """ Return True if the draft is primarily asking the customer for information (diagnostic access, error details, version info, etc.) rather than providing a direct answer. Heuristic: the draft contains at least one ask-for-info phrase AND does NOT contain explicit solution/step phrases that indicate a direct answer was given. """ lower = draft.lower() ask_count = sum(1 for phrase in _ASK_INFO_PHRASES if phrase in lower) # Strong signal: at least 2 ask phrases or presence of access-specific ones has_access_ask = any(p in lower for p in ("back office", "ftp access", "bo access", "temporary access", "accès temporaire", "acceso temporal")) return ask_count >= 2 or has_access_ask # --------------------------------------------------------------------------- # Main evaluation loop # --------------------------------------------------------------------------- def run_eval(n: int = 20, schema: str = "tenant_internal") -> None: print(f"\n{'='*60}") print(f"ModuleDesk AI Draft Quality Evaluation") print(f"Schema: {schema} | N tickets: {n}") print(f"{'='*60}\n") # Bootstrap app config (loads .env) from supporthub.app.config import config from supporthub.app.db import session_scope, set_worker_tenant_schema # Validate OpenAI key before spending anything if not config.AI_API_KEY: print("ERROR: AI_API_KEY is not set in .env. Cannot proceed.") sys.exit(1) # Set worker schema so session_scope() without explicit schema argument # also uses the correct tenant set_worker_tenant_schema(schema) # Build a base OpenAI client and a per-run accumulator base_client = openai.OpenAI(api_key=config.AI_API_KEY, base_url=config.AI_BASE_URL) acc = RunAccumulator() # Wrap with tracking proxy tracked_client = TrackedOpenAI(base_client, acc) # Load embedding index (needed for RAG in generate_suggestion) from supporthub.app.services.embedding_service import embedding_index, EmbeddingService print("Loading embedding index...") embedding_index.load_from_db(schema=schema) embedding_index.load_doc_pages_from_db(schema=schema) print(f" Loaded {embedding_index.count} thread embeddings\n") # Build AIService — use a lightweight mock org with plan='internal' to bypass # all credit gating and plan checks (internal plan = unlimited, no BYOK). # This is the safest way to call AIService in a script context. class _InternalOrg: """Minimal org stub that satisfies plan_service checks without DB access.""" plan = "internal" trial_ends_at = None max_bonus_ends_at = None openai_api_key_encrypted = None ai_credits_used = 0 ai_credits_extra = 0 id = 0 org_id = 0 from supporthub.app.services.ai_service import AIService ai_service = AIService(org=_InternalOrg()) # Inject our tracked client so we capture all calls ai_service.client = tracked_client # Select eval tickets print(f"Selecting {n} eval tickets from {schema}...") with session_scope(schema=schema) as session: tickets = select_eval_tickets(session, n=n) # Detach objects from session before closing (load needed attrs eagerly) # We need to keep the session open during inference, so process inline session.expunge_all() print(f" Selected {len(tickets)} tickets\n") if not tickets: print("ERROR: No eligible tickets found. Cannot proceed.") sys.exit(1) # Process each ticket embed_svc = EmbeddingService() # Inject tracked client into embed service too embed_svc.client = tracked_client results = [] for i, item in enumerate(tickets, 1): print(f"[{i}/{len(tickets)}] ticket={item['ticket_id']} | {item['product_name']}") ticket_acc_start = len(acc.calls) cost_before = acc.total_cost() # Build detached proxy objects from raw DB values so there are no # live-session dependencies. SessionLocal is a scoped_session (thread-local), # so every session_scope() in this thread returns the same session object. # Any inner session_scope() that calls session.close() would detach our ORM # objects. By pre-loading into plain Python objects we sidestep this entirely. with session_scope(schema=schema) as session: from sqlalchemy import text from supporthub.app.models import ( Tickets, AddonsMessages, AddonsThreads, AISuggestions ) # Load raw ticket data trow = session.execute(text(""" SELECT t.id, t.thread_id, t.status, t.priority, t.is_read, t.is_starred, t.first_seen_at, t.last_activity_at, t.first_response_at, t.draft_html, t.created_at, t.updated_at FROM tickets t WHERE t.id = :tid """), {"tid": item["ticket_id"]}).one_or_none() # Load thread data throw = session.execute(text(""" SELECT at_.id, at_.provider_thread_id, at_.subject, at_.customer_display_name, at_.id_product, at_.product_name, at_.id_order, at_.ps_version, at_.customer_website, at_.is_support_expired, at_.customer_language, at_.created_at, at_.last_activity_at FROM addons_threads at_ WHERE at_.id = :tid """), {"tid": item["thread_id"]}).one_or_none() # Load all messages for the thread mrows = session.execute(text(""" SELECT id, thread_id, provider_message_id, direction, author_display, body_text, created_at, inserted_at FROM addons_messages WHERE thread_id = :tid ORDER BY created_at ASC """), {"tid": item["thread_id"]}).fetchall() # Find the inbound message inbound_row = next( (m for m in mrows if m.id == item["inbound_msg_id"]), None ) if not trow or not throw or not mrows or not inbound_row: print(f" SKIP: missing data for ticket {item['ticket_id']}") continue # Build lightweight proxy objects that AIService can traverse without a session import datetime as _dt class _MsgProxy: def __init__(self, r, thread_proxy=None): self.id = r.id self.thread_id = r.thread_id self.provider_message_id = r.provider_message_id self.direction = r.direction self.author_display = r.author_display self.body_text = r.body_text self.created_at = r.created_at self.inserted_at = r.inserted_at self._thread = thread_proxy # set after thread is built @property def thread(self): return self._thread class _ThreadProxy: def __init__(self, r): self.id = r.id self.provider_thread_id = r.provider_thread_id self.subject = r.subject self.customer_display_name = r.customer_display_name self.id_product = r.id_product self.product_name = r.product_name self.id_order = r.id_order self.ps_version = r.ps_version self.customer_website = r.customer_website self.is_support_expired = r.is_support_expired self.customer_language = r.customer_language self.created_at = r.created_at self.last_activity_at = r.last_activity_at self.messages = [] # filled below class _TicketProxy: def __init__(self, r, thread_proxy): self.id = r.id self.thread_id = r.thread_id self.status = r.status self.priority = r.priority self.is_read = r.is_read self.is_starred = r.is_starred self.first_seen_at = r.first_seen_at self.last_activity_at = r.last_activity_at self.first_response_at = r.first_response_at self.draft_html = r.draft_html self.created_at = r.created_at self.updated_at = r.updated_at self.thread = thread_proxy self.tags = [] self.ai_suggestions = [] thread_proxy = _ThreadProxy(throw) msg_proxies = [_MsgProxy(m, thread_proxy) for m in mrows] thread_proxy.messages = msg_proxies ticket_proxy = _TicketProxy(trow, thread_proxy) # Find the inbound message proxy inbound_proxy = next( (m for m in msg_proxies if m.id == item["inbound_msg_id"]), None ) if not inbound_proxy: print(f" SKIP: could not find inbound message proxy") continue # Generate draft by calling AIService internal methods directly. # We bypass generate_suggestion() to avoid the AISuggestions persistence # step (which opens a session_scope, commits, then returns a detached object # — accessing .draft_reply on a detached object triggers a lazy load error). # The internal calls _run_classification and _run_draft are stateless wrt # sessions and work fine with our proxy objects. confidence = None try: summary, classification_obj, classification_json_str, intent = \ ai_service._run_classification(ticket_proxy, inbound_proxy, session=None) if intent == "thanks_solved": # Mimick what generate_suggestion does for thanks_solved draft = ( "Thank you for your feedback! I'm glad the issue has been resolved. " "If you need any further assistance in the future, don't hesitate " "to reach out. Have a great day!" ) # thanks_solved: no RAG → lowest confidence confidence = ai_service._compute_confidence(0.0, 0, False, True, False) else: # Phase 3: _run_draft now returns (draft, confidence) draft, confidence = ai_service._run_draft( ticket_proxy, inbound_proxy, summary, classification_obj, session=None, instruction=None, ) except Exception as exc: print(f" ERROR generating draft: {exc}") results.append({ "ticket_id": item["ticket_id"], "error": str(exc), }) continue conf_band = confidence["band"] if confidence else "Unknown" conf_score = confidence["score"] if confidence else 0 draft_is_ask = is_ask_for_info_reply(draft) draft_type = "ask-for-info" if draft_is_ask else "answer" print(f" Draft length: {len(draft)} chars | Confidence: {conf_band} ({conf_score}/100) | Type: {draft_type}") # --- Score: semantic similarity via embeddings --- try: draft_vec_resp = base_client.embeddings.create( input=draft[:8000], model=config.EMBEDDING_MODEL, dimensions=config.EMBEDDING_DIMENSIONS, ) draft_vec = draft_vec_resp.data[0].embedding acc.record("embed", config.EMBEDDING_MODEL, draft_vec_resp.usage.prompt_tokens, 0) real_vec_resp = base_client.embeddings.create( input=item["real_reply"][:8000], model=config.EMBEDDING_MODEL, dimensions=config.EMBEDDING_DIMENSIONS, ) real_vec = real_vec_resp.data[0].embedding acc.record("embed", config.EMBEDDING_MODEL, real_vec_resp.usage.prompt_tokens, 0) similarity = cosine_similarity(draft_vec, real_vec) except Exception as exc: print(f" ERROR computing similarity: {exc}") similarity = None # --- Score: LLM judge (vs final reply) --- judge_result = llm_judge(draft, item["real_reply"], base_client, acc) # --- Score: Honest usability judge (inbound-only) --- honest_result = honest_usability_judge( inbound_proxy.body_text or "", draft, base_client, acc, ) # --- Cost for this ticket --- cost_this = acc.total_cost() - cost_before ticket_calls = acc.calls[ticket_acc_start:] call_summary = {} for c in ticket_calls: key = f"{c.call_type}:{c.model}" if key not in call_summary: call_summary[key] = {"prompt": 0, "completion": 0, "cost_eur": 0.0} call_summary[key]["prompt"] += c.prompt_tokens call_summary[key]["completion"] += c.completion_tokens call_summary[key]["cost_eur"] += c.cost_eur # Count draft tokens (from the draft call record) draft_tokens = sum( c.prompt_tokens + c.completion_tokens for c in ticket_calls if c.call_type == "chat" and "gpt-4o" in c.model and "mini" not in c.model ) print(f" Similarity: {similarity:.3f}" if similarity is not None else " Similarity: N/A") print(f" Judge (old): usable={judge_result['usable']} | tone={judge_result['tone_match_0_5']}/5") print(f" Reason (old): {judge_result['reason']}") print(f" Judge (honest): usable={honest_result['usable']} | would_send={honest_result['would_send_as_is_or_minor_edits']}") print(f" Reason (honest): {honest_result['reason']}") print(f" Cost: €{cost_this:.5f}") results.append({ "ticket_id": item["ticket_id"], "thread_id": item["thread_id"], "subject": item["subject"], "product_name": item["product_name"], "customer_language": item["customer_language"], "draft_preview": draft[:200], "real_reply_preview": item["real_reply"][:200], "similarity": round(similarity, 4) if similarity is not None else None, "usable": judge_result["usable"], "tone_match": judge_result["tone_match_0_5"], "judge_reason": judge_result["reason"], "draft_tokens": draft_tokens, "cost_eur": round(cost_this, 6), "call_breakdown": call_summary, # Phase 3: confidence data "confidence": confidence, # Phase 4: draft type "draft_type": draft_type, # Honest judge (inbound-only) "honest_usable": honest_result["usable"], "honest_reason": honest_result["reason"], "honest_would_send": honest_result["would_send_as_is_or_minor_edits"], }) # --------------------------------------------------------------------------- # Aggregate statistics # --------------------------------------------------------------------------- valid = [r for r in results if "error" not in r and r.get("similarity") is not None] usable_count = sum(1 for r in valid if r["usable"]) honest_usable_count = sum(1 for r in valid if r.get("honest_usable")) similarities = [r["similarity"] for r in valid] tones = [r["tone_match"] for r in valid] costs = [r["cost_eur"] for r in valid] median_sim = float(np.median(similarities)) if similarities else 0.0 mean_sim = float(np.mean(similarities)) if similarities else 0.0 pct_usable = (usable_count / len(valid) * 100) if valid else 0.0 honest_pct_usable = (honest_usable_count / len(valid) * 100) if valid else 0.0 avg_cost = float(np.mean(costs)) if costs else 0.0 avg_tone = float(np.mean(tones)) if tones else 0.0 proj_50 = avg_cost * 50 proj_200 = avg_cost * 200 total_run_cost = acc.total_cost() # Aggregate call costs agg_calls: Dict[str, Dict] = {} for c in acc.calls: key = f"{c.call_type}:{c.model}" if key not in agg_calls: agg_calls[key] = {"count": 0, "prompt_tokens": 0, "completion_tokens": 0, "cost_eur": 0.0} agg_calls[key]["count"] += 1 agg_calls[key]["prompt_tokens"] += c.prompt_tokens agg_calls[key]["completion_tokens"] += c.completion_tokens agg_calls[key]["cost_eur"] += c.cost_eur # Divergence analysis: collect judge reasons for unusable drafts divergence_reasons = [ {"ticket_id": r["ticket_id"], "product": r["product_name"], "reason": r["judge_reason"], "tone": r["tone_match"], "similarity": r["similarity"]} for r in valid if not r["usable"] ] # --------------------------------------------------------------------------- # Print summary # --------------------------------------------------------------------------- print(f"\n{'='*60}") print("BASELINE RESULTS") print(f"{'='*60}") print(f"Tickets evaluated: {len(valid)} / {len(tickets)} requested") print(f"Median semantic similarity: {median_sim:.3f}") print(f"Mean semantic similarity: {mean_sim:.3f}") print(f"Usable — old judge (vs final reply): {usable_count}/{len(valid)} = {pct_usable:.0f}%") print(f"Usable — honest judge (inbound-only): {honest_usable_count}/{len(valid)} = {honest_pct_usable:.0f}%") print(f"Avg tone match: {avg_tone:.1f}/5") print(f"\nCost breakdown:") for key, v in sorted(agg_calls.items()): print(f" {key}: {v['count']} calls, " f"{v['prompt_tokens']}+{v['completion_tokens']} tokens, " f"€{v['cost_eur']:.5f}") print(f"\nAvg cost/draft: €{avg_cost:.5f}") print(f"Total run cost: €{total_run_cost:.5f}") print(f"Projected cost for 50 tickets: €{proj_50:.4f}") print(f"Projected cost for 200 tickets: €{proj_200:.4f}") if divergence_reasons: print(f"\nTop divergence reasons ({len(divergence_reasons)} unusable drafts):") for d in divergence_reasons: print(f" ticket={d['ticket_id']} ({d['product']}): {d['reason']}") # Phase 3+4: Confidence calibration table with ask-for-info counts # Group valid results by confidence band and compute per-band stats conf_bands_order = ["High", "Medium", "Low", "Unknown"] conf_groups: Dict[str, List] = {b: [] for b in conf_bands_order} for r in valid: band = (r.get("confidence") or {}).get("band", "Unknown") if band not in conf_groups: band = "Unknown" conf_groups[band].append(r) ask_for_info_total = sum(1 for r in valid if r.get("draft_type") == "ask-for-info") low_ask_count = sum(1 for r in valid if (r.get("confidence") or {}).get("band") == "Low" and r.get("draft_type") == "ask-for-info") print(f"\nConfidence Calibration (Phase 4):") print(f" {'Band':<8} {'N':>3} {'Mean sim':>9} {'Old usbl':>9} {'Honest usbl':>12} {'Ask-info':>9}") print(f" {'-'*8} {'-'*3} {'-'*9} {'-'*9} {'-'*12} {'-'*9}") calibration_rows = [] for band in conf_bands_order: grp = conf_groups[band] if not grp: continue grp_sims = [r["similarity"] for r in grp if r["similarity"] is not None] grp_usable = sum(1 for r in grp if r["usable"]) grp_honest_usable = sum(1 for r in grp if r.get("honest_usable")) grp_ask = sum(1 for r in grp if r.get("draft_type") == "ask-for-info") grp_mean_sim = float(np.mean(grp_sims)) if grp_sims else 0.0 grp_pct_usable = (grp_usable / len(grp) * 100) if grp else 0.0 grp_pct_honest = (grp_honest_usable / len(grp) * 100) if grp else 0.0 print(f" {band:<8} {len(grp):>3} {grp_mean_sim:>9.3f} {grp_pct_usable:>8.0f}% {grp_pct_honest:>10.0f}% {grp_ask:>5}/{len(grp):<3}") calibration_rows.append({ "band": band, "count": len(grp), "mean_sim": round(grp_mean_sim, 4), "pct_usable": round(grp_pct_usable, 1), "honest_usable_count": grp_honest_usable, "ask_for_info_count": grp_ask, }) print(f"\n Ask-for-info drafts total: {ask_for_info_total}/{len(valid)}") print(f" Low-confidence → ask-for-info: {low_ask_count} (target: most Low drafts should ask)") # --------------------------------------------------------------------------- # Write JSON + markdown report # --------------------------------------------------------------------------- output = { "run_config": { "schema": schema, "n_requested": n, "n_evaluated": len(valid), "cost_assumptions_usd_per_1m": PRICE_USD_PER_1M, "eur_usd_rate": USD_EUR, }, "aggregates": { "median_similarity": round(median_sim, 4), "mean_similarity": round(mean_sim, 4), "pct_usable": round(pct_usable, 1), "honest_pct_usable": round(honest_pct_usable, 1), "avg_tone_match": round(avg_tone, 2), "avg_cost_eur_per_draft": round(avg_cost, 6), "total_run_cost_eur": round(total_run_cost, 6), "projected_50_eur": round(proj_50, 4), "projected_200_eur": round(proj_200, 4), }, "call_totals": agg_calls, "per_ticket": results, "divergence_reasons": divergence_reasons, "confidence_calibration": calibration_rows, "phase4": { "ask_for_info_total": ask_for_info_total, "low_band_ask_for_info": low_ask_count, }, } out_dir = Path(__file__).resolve().parents[1] / "tmp" out_dir.mkdir(exist_ok=True) json_path = out_dir / "eval-draft-baseline.json" md_path = out_dir / "eval-draft-baseline.md" with open(json_path, "w", encoding="utf-8") as f: json.dump(output, f, indent=2, ensure_ascii=False) # Phase 3+4: Build confidence calibration markdown section calib_md = "" if calibration_rows: calib_md = "\n## Confidence Calibration (Phase 3+4)\n\n" calib_md += "| Band | Count | Mean sim | Old usable | Honest usable | Ask-for-info |\n" calib_md += "|------|-------|----------|------------|---------------|--------------|\n" for row in calibration_rows: ask_str = f"{row.get('ask_for_info_count', 0)}/{row['count']}" honest_str = f"{row.get('honest_usable_count', 0)}/{row['count']}" calib_md += f"| {row['band']} | {row['count']} | {row['mean_sim']:.3f} | {row['pct_usable']:.0f}% | {honest_str} | {ask_str} |\n" calib_md += f"\n_Ask-for-info total: {ask_for_info_total}/{len(valid)}_\n" calib_md += f"_Low-confidence → ask-for-info: {low_ask_count} (Phase 4 target: Low drafts should ask, not invent)_\n" # Build markdown table md_rows = [] for r in results: if "error" in r: md_rows.append( f"| {r['ticket_id']} | ERROR | — | — | — | — | — | — | {r['error'][:60]} |" ) else: sim_str = f"{r['similarity']:.3f}" if r['similarity'] is not None else "N/A" usable_str = "YES" if r["usable"] else "NO" conf = r.get("confidence") or {} conf_str = f"{conf.get('band','?')} ({conf.get('score', 0)})" if conf else "—" md_rows.append( f"| {r['ticket_id']} " f"| {(r['product_name'] or '')[:28]} " f"| {r.get('customer_language') or 'en'} " f"| {sim_str} " f"| {usable_str} " f"| {r['tone_match']}/5 " f"| {conf_str} " f"| {r['draft_tokens']} " f"| €{r['cost_eur']:.5f} |" ) md_table = "\n".join(md_rows) divergence_md = "" if divergence_reasons: divergence_md = "\n## Top Divergence Reasons\n\n" for d in divergence_reasons: divergence_md += f"- **ticket {d['ticket_id']}** ({d['product']}): {d['reason']}\n" md_content = f"""\ # ModuleDesk AI Draft Quality — Baseline Evaluation **Date:** {time.strftime('%Y-%m-%d %H:%M UTC', time.gmtime())} **Schema:** `{schema}` | **N:** {len(valid)} tickets evaluated ## Cost Assumptions | Model | In ($/1M) | Out ($/1M) | |-------|-----------|------------| | gpt-4o | $2.50 | $10.00 | | gpt-4o-mini | $0.15 | $0.60 | | text-embedding-3-small | $0.02 | — | EUR/USD rate: {1/USD_EUR:.2f} (1 EUR = ${1/USD_EUR:.2f}) ## Aggregate Results | Metric | Value | |--------|-------| | Tickets evaluated | {len(valid)} | | Median semantic similarity | {median_sim:.3f} | | Mean semantic similarity | {mean_sim:.3f} | | % Usable — old judge (vs final reply) | {pct_usable:.0f}% | | **% Usable — honest judge (inbound-only)** | **{honest_pct_usable:.0f}%** | | Avg tone match | {avg_tone:.1f}/5 | | **Avg cost/draft** | **€{avg_cost:.5f}** | | Total run cost | €{total_run_cost:.5f} | | Projected cost (50 tickets) | €{proj_50:.4f} | | Projected cost (200 tickets) | €{proj_200:.4f} | ## Call-Level Token Breakdown | Call | Count | Prompt tok | Completion tok | Cost (€) | |------|-------|-----------|----------------|----------| """ + "\n".join( f"| {k} | {v['count']} | {v['prompt_tokens']} | {v['completion_tokens']} | €{v['cost_eur']:.5f} |" for k, v in sorted(agg_calls.items()) ) + f""" ## Per-Ticket Results | Ticket | Product | Lang | Similarity | Usable | Tone | Confidence | Draft tok | Cost | |--------|---------|------|-----------|--------|------|------------|-----------|------| {md_table} """ + divergence_md + calib_md + f""" ## Notes - Semantic similarity is cosine distance between `text-embedding-3-small` embeddings of the draft and the actual sent reply (range 0–1, higher is better; ≥0.85 is strong, 0.70–0.85 is usable). - Judge prompt asks gpt-4o-mini whether the seller could send the draft with **minor edits only**. - Cost does NOT include the evaluation embeddings or judge calls in the "per-draft" figure (those are eval overhead). The per-draft cost covers classify + draft + RAG embedding. - The classify+draft token counts come from the actual AIService calls; the `embed` rows reflect the query embedding computed during RAG context retrieval. """ with open(md_path, "w", encoding="utf-8") as f: f.write(md_content) print(f"\nReports written:") print(f" JSON: {json_path}") print(f" MD: {md_path}") # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": parser = argparse.ArgumentParser(description="Eval ModuleDesk AI draft quality") parser.add_argument("--n", type=int, default=20, help="Number of tickets to evaluate (default: 20)") parser.add_argument("--schema", default="tenant_internal", help="Tenant schema to use") args = parser.parse_args() run_eval(n=args.n, schema=args.schema)