#!/usr/bin/env python3 """Small, deterministic CRM benchmark for models exposed through Ollama.""" from __future__ import annotations import argparse import json import re import time import urllib.request CASES = [ { "name": "contact_signature", "prompt": """Extract contact data from this email as JSON with exactly these keys: name, title, company, mobile, email. Do not mistake greetings such as 'Kind regards' for a job title. Use null when a value is absent. Kind regards Ida Gundersen Technical Advisor & Co-owner Mobile: +45 42 25 59 08 DK: +45 55 86 05 00 Email: ida@createx-onstage.com Web: createx-onstage.com Createx Storegade 4C | 4780 Stege, DK""", "checks": [ ("name", lambda value: "ida gundersen" in value.lower()), ("title", lambda value: "technical advisor" in value.lower() and "kind regards" not in value.lower()), ("company", lambda value: "createx" in value.lower()), ("mobile", lambda value: "42255908" in re.sub(r"\D", "", value)), ("email", lambda value: value.lower() == "ida@createx-onstage.com"), ], }, { "name": "internet_invoice", "prompt": """Extract this supplier invoice line as JSON with exactly these keys: reference, address, postal_code, city, ip_ranges, purchase_price_dkk. ip_ranges must be an array and purchase_price_dkk a number. GlobalConnect faktura 3018657 NKA-027964 | Firskovvej 36 | 2800 Kongens Lyngby | Internet 1 Gbit IP: 217.74.219.56/30 og 152.115.61.32/27 | Månedlig kostpris: 2.495,00 kr.""", "checks": [ ("reference", lambda value: "027964" in value), ("address", lambda value: "firskovvej 36" in value.lower()), ("postal_code", lambda value: str(value) == "2800"), ("city", lambda value: "lyngby" in value.lower()), ("ip_ranges", lambda value: set(value) == {"217.74.219.56/30", "152.115.61.32/27"}), ("purchase_price_dkk", lambda value: abs(float(value) - 2495.0) < 0.01), ], }, { "name": "support_solution", "prompt": """Analyze this support case and return JSON with exactly these keys: category, probable_cause, diagnostic_steps, customer_reply. diagnostic_steps must be an array of short actions. Do not claim the problem has been fixed. Customer: Internet works by IP address, but websites do not open by name on all PCs. The router is reachable and 8.8.8.8 responds to ping. The issue started after the customer changed DNS settings this morning.""", "checks": [ ("cause", lambda value: "dns" in json.dumps(value).lower()), ("steps", lambda value: isinstance(value, list) and len(value) >= 2), ("uncertainty", lambda value: not any(word in value.lower() for word in ("løst", "fixed", "resolved"))), ], }, { "name": "danish_rewrite", "prompt": """Rewrite this as a concise, professional Danish customer email. Preserve every fact and do not invent a resolution. Return JSON with exactly one key: text. hej vi kan se jeres forbindelse NKA-027964 på Firskovvej 36 har været nede siden kl 08:15. vi undersøger det hos globalconnect og vender tilbage senest kl 10:00.""", "checks": [ ("reference", lambda value: "NKA-027964" in value), ("address", lambda value: "Firskovvej 36" in value), ("times", lambda value: "08:15" in value and "10:00" in value), ("supplier", lambda value: "globalconnect" in value.lower()), ], }, ] def ask(endpoint: str, model: str, prompt: str, timeout: int) -> tuple[dict, float]: payload = { "model": model, "stream": False, "think": False, "format": "json", "messages": [ {"role": "system", "content": "Return only valid JSON. Follow the requested schema exactly."}, {"role": "user", "content": prompt}, ], "options": {"temperature": 0, "num_ctx": 8192}, } started = time.perf_counter() request = urllib.request.Request( f"{endpoint.rstrip('/')}/api/chat", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, ) with urllib.request.urlopen(request, timeout=timeout) as response: envelope = json.load(response) elapsed = time.perf_counter() - started content = envelope.get("message", {}).get("content", "") return json.loads(content), elapsed def score(case: dict, result: dict) -> tuple[int, list[str]]: passed = [] if case["name"] == "support_solution": values = { "cause": result, "steps": result.get("diagnostic_steps"), "uncertainty": str(result.get("customer_reply", "")), } elif case["name"] == "danish_rewrite": values = {label: str(result.get("text", "")) for label, _ in case["checks"]} else: values = result for label, check in case["checks"]: try: if check(values.get(label) if isinstance(values, dict) else values): passed.append(label) except (AttributeError, TypeError, ValueError): pass return len(passed), passed def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--endpoint", default="http://172.16.31.195:11434") parser.add_argument("--models", nargs="+", default=["qwen2.5:7b", "qwen3.5:9b"]) parser.add_argument("--cases", nargs="+", help="Run only the named benchmark cases") parser.add_argument("--timeout", type=int, default=180) args = parser.parse_args() report = [] for model in args.models: model_score = 0 model_max = 0 model_seconds = 0.0 cases = [] selected_cases = [case for case in CASES if not args.cases or case["name"] in args.cases] for case in selected_cases: maximum = len(case["checks"]) model_max += maximum try: result, seconds = ask(args.endpoint, model, case["prompt"], args.timeout) points, passed = score(case, result) error = None except Exception as exc: # benchmark must report failures and continue result, seconds, points, passed, error = None, 0.0, 0, [], str(exc) model_score += points model_seconds += seconds cases.append({ "case": case["name"], "score": points, "max": maximum, "seconds": round(seconds, 2), "passed": passed, "error": error, "result": result, }) report.append({ "model": model, "score": model_score, "max": model_max, "seconds": round(model_seconds, 2), "cases": cases, }) print(json.dumps(report, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()