""" IPM Limits accuracy test: compare Marco's stated limits against real ChromaDB data. For each scenario: 1. Pull the real IPM chunk directly from ChromaDB 2. Ask Marco the same question 3. Extract numbers from both 4. Compare — flag mismatches """ import httpx import time import sys import io import re import os from pathlib import Path sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") # Add project root to path so we can import app modules sys.path.insert(0, str(Path(__file__).parent.parent)) os.environ.setdefault("PYTHONPATH", str(Path(__file__).parent.parent)) BASE_URL = "http://localhost:8000" TIMEOUT = 60.0 SEP = "-" * 80 # ── Real data lookup via IPMRetriever ───────────────────────────────────────── def get_real_ipm_data(crop_code: str, disease_query: str, farming_type: str = "conventional") -> dict: """ Pull the real IPM chunk from ChromaDB for this crop+disease. Returns dict with raw chunk text and key extracted fields. """ from app.services.ipm_retriever import ipm_retriever results = ipm_retriever.search_excel( query=disease_query, crop_code=crop_code, farming_type=farming_type if farming_type == "organic" else None, n_results=3, ) if not results: return {"found": False, "chunks": []} chunks = [] for r in results: text = r.get("text", "") meta = r.get("metadata", {}) chunks.append({ "crop": meta.get("crop_full", ""), "pest": meta.get("pest_name", ""), "text": text, "raw_limits": _extract_limits_from_chunk(text, farming_type), }) return {"found": True, "chunks": chunks} def _extract_limits_from_chunk(text: str, farming_type: str) -> list: """ Extract treatment limits from a chunk text. Looks for patterns like: (max N/anno), max N trattamenti, [gruppo max N], "Al massimo N interventi", "Massimo N interventi" """ limits = [] # Pattern: (max N/anno) or (max N/stagione) for m in re.finditer(r"max\s+(\d+)\s*/\s*anno", text, re.IGNORECASE): limits.append({"type": "per_anno", "value": int(m.group(1)), "context": m.group(0)}) # Pattern: [gruppo max N] for m in re.finditer(r"gruppo\s+max\s+(\d+)", text, re.IGNORECASE): limits.append({"type": "gruppo", "value": int(m.group(1)), "context": m.group(0)}) # Pattern: numero massimo trattamenti / max N trattamenti / Al massimo N interventi for m in re.finditer(r"(?:al\s+)?(?:massimo|max)(?:\s+di)?\s+(\d+)\s+(?:trattament|interventi|applicazion)", text, re.IGNORECASE): limits.append({"type": "trattamenti", "value": int(m.group(1)), "context": m.group(0)}) # Pattern: "massimo N" standalone (e.g. "Massimo 10 interventi tra...") for m in re.finditer(r"[Mm]assimo\s+(\d+)\s+interventi", text): limits.append({"type": "trattamenti", "value": int(m.group(1)), "context": m.group(0)}) return limits def _extract_numbers_from_response(response: str) -> list: """Extract treatment-related numbers from Marco's response.""" numbers = [] patterns = [ # Italian: "massimo di 2 trattamenti", "al massimo 6", "max 3/anno" r"(?:max|massimo|al massimo)(?:\s+di)?\s+(\d+)\s*(?:trattament|applicazion|volte|per anno|/anno|stagione|interventi)", # Italian: "limitato a 2 per anno", "limitato a 2 trattamenti" r"limitato\s+a\s+(\d+)\s*(?:per\s+(?:anno|stagione)|trattament|applicazion|volte)", # Italian: "Fino a X volte per stagione", "fino a X trattamenti per stagione" r"fino\s+a\s+(\d+)\s+(?:volte|trattament|applicazion)\s+(?:per|alla?)\s+stagione", # Italian: "fino a X volte" (bare) r"fino\s+a\s+(\d+)\s+(?:volte|trattament|applicazion)", # Italian: "puoi applicarlo/effettuare fino a X" r"puoi\s+(?:\w+\s+){0,4}fino\s+a\s+(\d+)", # Italian: "X volte per stagione" r"(\d+)\s+volte\s+per\s+stagione", # Italian: "Applicazioni: Fino a X" / "Applicazioni: X" r"[Aa]pplicazioni\s*:\s*(?:[Ff]ino\s+a\s+)?(\d+)", # Italian: "2 trattamenti per anno/stagione" r"(\d+)\s+(?:trattament|applicazion|interventi)\s+(?:per|all')\s*(?:anno|stagione)", r"(\d+)\s+(?:trattament|applicazion|interventi)\s+massim", # Italian: "X–Y volte/trattamenti" — take first number of range r"(\d+)(?:–|-)\d+\s+(?:volte|trattament|applicazion)\s+per\s+stagione", r"da\s+(\d+)\s+a\s+\d+\s+(?:trattament|applicazion|volte)", # Label app format: "Grapevine: 2–6" or "2-6 per season" r"(?:Grapevine|Vite|Fruit trees|Tomato|Pomodoro)\s*:\s*(\d+)", r"label apps.*?(\d+)", # English r"maximum\s+(?:of\s+)?(\d+)\s+(?:application|treatment|time)", r"(\d+)\s+(?:application|treatment)s?\s+per\s+(?:season|year)", r"up\s+to\s+(\d+)\s+(?:application|treatment)", r"(\d+)(?:–|-)\d+\s+(?:application|treatment|time)s?\s+per", ] for pat in patterns: for m in re.finditer(pat, response, re.IGNORECASE): try: numbers.append(int(m.group(1))) except (ValueError, IndexError): pass return list(set(numbers)) def ask_marco(client: httpx.Client, session_id: str, message: str, crop_type: str = None, farming_type: str = "conventional", bbch_stage: str = None) -> str: payload = { "session_id": session_id, "message": message, "farming_type": farming_type, "weather_enabled": False, } if crop_type: payload["crop_type"] = crop_type if bbch_stage: payload["bbch_stage"] = bbch_stage resp = client.post(f"{BASE_URL}/api/chat", json=payload, timeout=TIMEOUT) if resp.status_code != 200: return f"[ERROR {resp.status_code}]" return resp.json().get("response", "") # ── Test scenarios ───────────────────────────────────────────────────────────── SCENARIOS = [ { "id": 1, "label": "Vite / oidio / conventional", "crop_code": "vite", "farming_type": "conventional", "bbch_stage": "57", "ipm_query": "vite oidio oidium sostanze trattamento massimo", "question": "ho l'oidio sulle mie viti BBCH 57, cosa uso e quanti trattamenti posso fare?", }, { "id": 2, "label": "Vite / peronospora / conventional", "crop_code": "vite", "farming_type": "conventional", "bbch_stage": "71", "ipm_query": "vite peronospora Plasmopara viticola sostanze trattamento massimo", "question": "peronospora sulle viti BBCH 71, quali prodotti e quanti trattamenti?", }, { "id": 3, "label": "Vite / botrytis / organic", "crop_code": "vite", "farming_type": "organic", "bbch_stage": "79", "ipm_query": "vite botrytis muffa grigia biologico sostanze ammesse massimo", "question": "ho botrytis sulle viti BBCH 79, sono in biologico, cosa uso e quante volte?", }, { "id": 4, "label": "Pomodoro / peronospora / conventional", "crop_code": "pomodoro", "farming_type": "conventional", "bbch_stage": "71", "ipm_query": "pomodoro peronospora sostanze trattamento massimo numero", "question": "peronospora sul pomodoro BBCH 71, prodotto AllTech e limite trattamenti IPM?", }, { "id": 5, "label": "Pomodoro / alternaria / conventional", "crop_code": "pomodoro", "farming_type": "conventional", "bbch_stage": "71", "ipm_query": "pomodoro alternaria early blight sostanze trattamento massimo", "question": "alternaria sul pomodoro BBCH 71, cosa uso e quanti trattamenti max?", }, { "id": 6, "label": "Pomodoro / botrytis / organic", "crop_code": "pomodoro", "farming_type": "organic", "bbch_stage": "79", "ipm_query": "pomodoro botrytis biologico sostanze ammesse massimo", "question": "botrytis sul pomodoro biologico BBCH 79, prodotto e limite trattamenti?", }, { "id": 7, "label": "Melo / ticchiolatura / conventional", "crop_code": "melo", "farming_type": "conventional", "bbch_stage": "65", "ipm_query": "melo ticchiolatura Venturia inaequalis sostanze massimo trattamenti", "question": "ticchiolatura sul melo BBCH 65, dimmi prodotto AllTech e max trattamenti IPM", }, { "id": 8, "label": "Melo / oidio / organic", "crop_code": "melo", "farming_type": "organic", "bbch_stage": "65", "ipm_query": "melo oidio biologico sostanze ammesse massimo", "question": "oidio sul melo biologico BBCH 65, cosa posso usare e quante volte?", }, { "id": 9, "label": "Pero / ticchiolatura / conventional", "crop_code": "pero", "farming_type": "conventional", "bbch_stage": "65", "ipm_query": "pero ticchiolatura Venturia pirina sostanze massimo trattamenti", "question": "ticchiolatura sul pero BBCH 65, prodotto AllTech e limite trattamenti IPM?", }, { "id": 10, "label": "Fragola / botrytis / organic", "crop_code": "fragola", "farming_type": "organic", "bbch_stage": "71", "ipm_query": "fragola botrytis biologico sostanze ammesse massimo trattamenti", "question": "botrytis sulle fragole biologico BBCH 71, prodotto e quanti trattamenti?", }, { "id": 11, "label": "Olivo / occhio di pavone / conventional", "crop_code": "olivo", "farming_type": "conventional", "bbch_stage": "60", "ipm_query": "olivo occhio di pavone Spilocaea oleagina sostanze massimo", "question": "occhio di pavone sull'olivo BBCH 60, cosa uso e quanti trattamenti max?", }, { "id": 12, "label": "Vite / peronospora / organic", "crop_code": "vite", "farming_type": "organic", "bbch_stage": "57", "ipm_query": "vite peronospora biologico sostanze ammesse massimo trattamenti", "question": "peronospora sulle viti biologico BBCH 57, prodotto e limite trattamenti IPM?", }, { "id": 13, "label": "Pomodoro / oidio / conventional", "crop_code": "pomodoro", "farming_type": "conventional", "bbch_stage": "65", "ipm_query": "pomodoro oidio sostanze massimo trattamenti", "question": "oidio sul pomodoro BBCH 65, prodotto AllTech e max trattamenti secondo IPM?", }, { "id": 14, "label": "Melo / peronospora / conventional", "crop_code": "melo", "farming_type": "conventional", "bbch_stage": "71", "ipm_query": "melo peronospora Phytophthora cactorum sostanze massimo", "question": "peronospora sul melo BBCH 71, prodotto e limite trattamenti IPM?", }, { "id": 15, "label": "Vite / oidio / organic", "crop_code": "vite", "farming_type": "organic", "bbch_stage": "71", "ipm_query": "vite oidio biologico sostanze ammesse massimo", "question": "oidio sulle viti biologico BBCH 71, cosa posso usare e quante volte?", }, ] def main(): print("\n" + "=" * 80) print(" IPM LIMITS vs REAL DATA — 15 scenarios") print("=" * 80 + "\n") results = [] with httpx.Client() as client: for sc in SCENARIOS: session_id = f"ipm_vs_data_{sc['id']}_{int(time.time())}" print(SEP) print(f" [{sc['id']:02d}] {sc['label']}") print(SEP) # ── Step 1: Get real data from ChromaDB ────────────────────────── real = get_real_ipm_data(sc["crop_code"], sc["ipm_query"], sc["farming_type"]) print(f" REAL DATA ({sc['crop_code']} / {sc['farming_type']}):") if not real["found"]: print(" [NOT FOUND in ChromaDB]") real_limits = [] real_substances_text = "" else: real_limits = [] real_substances_text = "" for i, chunk in enumerate(real["chunks"][:2]): print(f" Chunk {i+1}: {chunk['pest']} on {chunk['crop']}") # Print relevant lines from chunk for line in chunk["text"].split("\n"): line = line.strip() if any(kw in line.lower() for kw in [ "sostanz", "massimo", "max", "limite", "limitaz", "gruppo", "trattament", "biologico", "convenzional" ]): print(f" > {line[:120]}") real_substances_text += line + " " real_limits.extend(chunk["raw_limits"]) if real_limits: print(f" EXTRACTED LIMITS: {real_limits}") else: print(f" EXTRACTED LIMITS: none found (may be in free text)") # ── Step 2: Ask Marco ──────────────────────────────────────────── print(f"\n USER: {sc['question']}") t0 = time.time() response = ask_marco( client, session_id, sc["question"], crop_type=sc["crop_code"], farming_type=sc["farming_type"], bbch_stage=sc.get("bbch_stage"), ) elapsed = time.time() - t0 print(f" MARCO ({elapsed:.1f}s):") print(f" {response[:600]}{'...' if len(response) > 600 else ''}") # ── Step 3: Extract numbers from Marco's response ──────────────── marco_numbers = _extract_numbers_from_response(response) real_numbers = [l["value"] for l in real_limits] print(f"\n COMPARISON:") print(f" Real limits (from data) : {real_numbers if real_numbers else 'not extracted (check raw text)'}") print(f" Marco stated numbers : {marco_numbers if marco_numbers else 'none detected'}") # ── Step 4: Verdict ────────────────────────────────────────────── # If real_numbers found, check if Marco's numbers match or include them # If no real_numbers extracted, check at minimum that Marco stated SOME limit if real_numbers and marco_numbers: # Check if any of Marco's numbers match any real limit matches = [n for n in marco_numbers if n in real_numbers] correct = len(matches) > 0 verdict = "MATCH" if correct else "MISMATCH" detail = f"real={real_numbers}, marco={marco_numbers}" elif real_numbers and not marco_numbers: correct = False verdict = "MISSING" detail = f"real data has limits {real_numbers} but Marco stated no number" elif not real_numbers and marco_numbers: correct = True # Can't verify, but at least stated something verdict = "UNVERIFIABLE (stated)" detail = f"no limit extracted from raw data, Marco stated {marco_numbers}" else: correct = None verdict = "UNVERIFIABLE" detail = "neither data nor Marco had extractable numbers" icon = "[OK] " if correct else ("[?] " if correct is None else "[FAIL]") print(f" {icon} {verdict}: {detail}") results.append({ "id": sc["id"], "label": sc["label"], "correct": correct, "verdict": verdict, "real_numbers": real_numbers, "marco_numbers": marco_numbers, "real_text_snippet": real_substances_text[:200], }) print() # ── Summary ──────────────────────────────────────────────────────────────── print("\n" + "=" * 80) print(" SUMMARY — Real Data vs Marco") print("=" * 80) print(f" {'ID':<4} {'Verdict':<20} {'Real':<15} {'Marco':<15} Label") print(f" {'-'*4} {'-'*20} {'-'*15} {'-'*15} {'-'*30}") for r in results: icon = "[OK]" if r["correct"] else ("[?] " if r["correct"] is None else "[!]") print(f" {r['id']:<4} {icon} {r['verdict']:<16} {str(r['real_numbers']):<15} {str(r['marco_numbers']):<15} {r['label']}") verifiable = [r for r in results if r["correct"] is not None] correct_v = [r for r in verifiable if r["correct"]] unverif = [r for r in results if r["correct"] is None] matches = [r for r in verifiable if r["verdict"] == "MATCH"] mismatches = [r for r in verifiable if r["verdict"] == "MISMATCH"] missing = [r for r in verifiable if r["verdict"] == "MISSING"] print(f"\n Verifiable scenarios : {len(verifiable)}/15") print(f" MATCH (correct) : {len(matches)}") print(f" MISMATCH (wrong num) : {len(mismatches)}") print(f" MISSING (no number) : {len(missing)}") print(f" Unverifiable : {len(unverif)}") if verifiable: acc = len(correct_v) / len(verifiable) * 100 print(f" Accuracy (verifiable): {acc:.1f}%") print("=" * 80 + "\n") if __name__ == "__main__": main()