bmc_hub/scripts/benchmark_ollama_crm.py

#!/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()