3cx-telefonie-middleware/tools/benchmark_zammad_taxonomy.py

#!/usr/bin/env python3

import json
import sqlite3
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed

DB = "data/zammad_analysis.sqlite3"
URL = "http://127.0.0.1:11434/api/chat"
MODEL = "qwen3:8b"

WORKERS = 2
TEST_LIMIT = 10

TAXONOMY = """
PRIMÄRINTENTS

VERSANDSTATUS
LIEFERVERZUG
ADRESSÄNDERUNG
REKLAMATION
TRANSPORTSCHADEN
FEHLLIEFERUNG
RECHNUNG
ZAHLUNG
WIDERRUF
RETOURE
PFLANZENBERATUNG
SORTENBERATUNG
PFLEGEFRAGE
BESTANDSANFRAGE
VORBESTELLUNG
B2B
GROSSHANDEL
DÜNGER_FEHLT
SONSTIGES


TAGS / MERKMALE

QUALITAETSPROBLEM
VERWECHSLUNG
SCHLECHTVERPACKT
FALSCHESORTEGELIEFERT
PROBLEMMITLIEFERDIENST
LIEFERDIENST_DHL
LIEFERDIENST_GLS
LIEFERDIENST_UNBEKANNT
TRANSPORTSCHADEN
PFLANZENKRANKHEIT
PFLANZENSCHADEN
WACHSTUMSPROBLEM
"""

SYSTEM = f"""
Du analysierst historische Kundenservice-Tickets von Schmid Gartenpflanzen.

Bestimme das tatsächliche Kundenanliegen anhand des KUNDENBEITRAGS.
Die Antwort des Kundenservice dient nur als Kontext.

{TAXONOMY}

WICHTIGE KLASSIFIKATIONSREGELN:

1. BESTANDSANFRAGE

Jede Frage nach aktueller oder zukünftiger Verfügbarkeit,
Lieferbarkeit oder Bestellbarkeit eines Produkts.

Beispiele:
- "Wann ist die Rose wieder verfügbar?"
- "Ab wann wieder bestellbar?"
- "Ist die Sorte im Herbst wieder lieferbar?"
- "Können Sie mich informieren, sobald sie wieder verfügbar ist?"
- "Kann ich die Rose im Juli bekommen?"

Alle diese Fälle sind BESTANDSANFRAGE.

2. VORBESTELLUNG

VORBESTELLUNG nur verwenden, wenn der Kunde tatsächlich eine
Vormerkung, Reservierung oder verbindliche Vorabbestellung verlangt.

Eine reine Frage nach der zukünftigen Verfügbarkeit ist KEINE
VORBESTELLUNG.

3. WIDERRUF

Stornierung oder Rückabwicklung einer Bestellung auf Kundenwunsch,
einschließlich Stornierung vor Versand.

4. REKLAMATION

Beanstandung eines gelieferten Produkts oder einer erbrachten
Leistung wegen Qualität, Zustand oder Beschädigung.

5. TRANSPORTSCHADEN

Wenn ein Schaden offensichtlich durch Transport oder Versand
entstanden ist. Kann zusätzlich als Tag TRANSPORTSCHADEN erscheinen.

6. FEHLLIEFERUNG

Falscher Artikel, falsche Sorte oder falsches Produkt geliefert.

7. DÜNGER_FEHLT

Bestellter Dünger fehlt in der Lieferung.

8. QUALITÄT

QUALITAETSPROBLEM ist kein eigener Primärintent.
Es ist ein Tag und kann zusammen mit REKLAMATION verwendet werden.

PFLANZENKRANKHEIT, PFLANZENSCHADEN und WACHSTUMSPROBLEM sind
ebenfalls Merkmale/Tags.

9. LIEFERDIENST

PROBLEMMITLIEFERDIENST ist ein Problem-/Sekundärmerkmal.

Wenn DHL ausdrücklich genannt wird:
LIEFERDIENST_DHL

Wenn GLS ausdrücklich genannt wird:
LIEFERDIENST_GLS

Wenn ein Paketdienstproblem vorliegt, aber der Dienst unbekannt ist:
LIEFERDIENST_UNBEKANNT

10. MULTI-INTENT

Mehrere gleichzeitig vorhandene Kundenanliegen müssen erfasst werden.

primary_intent enthält das wichtigste Anliegen.
secondary_intents enthält weitere relevante Primärintents.
tags enthält konkrete Problem- und Kontextmerkmale.

11. TAXONOMIE-LÜCKEN

Keine neuen Primärintents eigenmächtig erfinden.

Wenn kein Primärintent sinnvoll passt:
primary_intent = SONSTIGES
taxonomy_fit = POOR oder PARTIAL
taxonomy_candidate = möglicher fehlender Intent.

Antworte ausschließlich als gültiges JSON.
"""

OUTPUT_SCHEMA = """
{
  "primary_intent": "...",
  "secondary_intents": [],
  "tags": [],
  "complaint_type": null,
  "actions": [],
  "summary": "...",
  "taxonomy_fit": "GOOD|PARTIAL|POOR",
  "taxonomy_candidate": null,
  "confidence": 0.0
}
"""

SYSTEM += "\nErwartetes JSON-Format:\n" + OUTPUT_SCHEMA


def classify(row):

    prompt = f"""
Ticket #{row["ticket_number"]}

Betreff:
{row["title"] or ""}

KUNDENBEITRAG:
{row["customer_text"] or ""}

KUNDENSERVICE-KONTEXT:
{row["agent_text"] or ""}
"""

    payload = {
        "model": MODEL,
        "messages": [
            {
                "role": "system",
                "content": SYSTEM,
            },
            {
                "role": "user",
                "content": prompt,
            },
        ],
        "stream": False,
        "format": "json",
        "options": {
            "temperature": 0,
        },
    }

    request = urllib.request.Request(
        URL,
        data=json.dumps(payload).encode(),
        headers={
            "Content-Type": "application/json",
        },
    )

    started = time.monotonic()

    with urllib.request.urlopen(
        request,
        timeout=300,
    ) as response:
        result = json.loads(response.read())

    elapsed = time.monotonic() - started

    content = result["message"]["content"]
    parsed = json.loads(content)

    return row, elapsed, parsed


def main():

    db = sqlite3.connect(DB)
    db.row_factory = sqlite3.Row

    rows = db.execute("""
        SELECT
            ticket_id,
            ticket_number,
            title,
            customer_text,
            agent_text
        FROM cases
        WHERE analysis_status = 'ANALYZABLE'
        ORDER BY ticket_id
        LIMIT ?
    """, (TEST_LIMIT,)).fetchall()

    db.close()

    print("=" * 80)
    print("ZAMMAD TAXONOMIE V2 – PARALLEL BENCHMARK")
    print("=" * 80)
    print(f"Modell:   {MODEL}")
    print(f"Tickets:  {len(rows)}")
    print(f"Worker:   {WORKERS}")
    print()

    started = time.monotonic()
    results = []

    with ThreadPoolExecutor(max_workers=WORKERS) as executor:

        futures = [
            executor.submit(classify, row)
            for row in rows
        ]

        for i, future in enumerate(
            as_completed(futures),
            1,
        ):

            try:
                row, elapsed, result = future.result()

                results.append(
                    (
                        row,
                        elapsed,
                        result,
                    )
                )

                print(
                    f"[{i:02}/{len(rows)}] "
                    f"#{row['ticket_number']} "
                    f"{elapsed:6.1f}s → "
                    f"{result.get('primary_intent')} / "
                    f"{result.get('taxonomy_fit')} / "
                    f"{result.get('confidence')}"
                )

                if result.get("secondary_intents"):
                    print(
                        f"    secondary: "
                        f"{result['secondary_intents']}"
                    )

                if result.get("tags"):
                    print(
                        f"    tags: "
                        f"{result['tags']}"
                    )

                if result.get("taxonomy_candidate"):
                    print(
                        f"    candidate: "
                        f"{result['taxonomy_candidate']}"
                    )

            except Exception as exc:
                print(
                    f"[{i:02}/{len(rows)}] ERROR: {exc}"
                )

    total = time.monotonic() - started

    print()
    print("=" * 80)
    print("BENCHMARK")
    print("=" * 80)

    if results:

        print(f"Erfolgreich:       {len(results)}")
        print(f"Gesamtzeit:        {total:.1f}s")
        print(
            f"Ø Einzelrequest:   "
            f"{sum(x[1] for x in results) / len(results):.1f}s"
        )
        print(
            f"Durchsatz:         "
            f"{len(results) / total * 3600:.1f} Tickets/Stunde"
        )

        throughput = len(results) / total * 3600

        print(
            f"Hochrechnung 4404: "
            f"{4404 / throughput:.1f} Stunden"
        )

        print(
            f"Hochrechnung Tage: "
            f"{4404 / throughput / 24:.2f}"
        )


if __name__ == "__main__":
    main()