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

#!/usr/bin/env python3

import json
import sqlite3
import time
import urllib.request

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

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

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

Klassifiziere das tatsächliche Kundenanliegen.

Bestehende Primärtaxonomie:
{TAXONOMY}

Wichtige Regeln:
- PRODUKTQUALITÄT unter REKLAMATION oder PFLANZENBERATUNG.
- SCHLECHTVERPACKT unter REKLAMATION bzw. TRANSPORTSCHADEN.
- FALSCHESORTEGELIEFERT unter FEHLLIEFERUNG.
- DÜNGER_FEHLT ist ein eigener Intent.
- Mehrere Anliegen gleichzeitig müssen als Multi-Intent erfasst werden.
- Keine neuen Primärintents erfinden.

Antworte ausschließlich als JSON:
{{
  "primary_intent": "...",
  "secondary_intents": [],
  "complaint_type": null,
  "actions": [],
  "summary": "...",
  "taxonomy_fit": "GOOD|PARTIAL|POOR",
  "taxonomy_candidate": null,
  "confidence": 0.0
}}
"""


def ask(case):

    prompt = f"""
Ticket #{case['ticket_number']}
Betreff:
{case['title'] or ''}

KUNDEN:
{case['customer_text'] or ''}

KUNDENSERVICE:
{case['agent_text'] or ''}

SONSTIGE FACHLICHE ANTWORTEN:
{case['unknown_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.time()

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

    elapsed = time.time() - started

    content = result["message"]["content"]

    parsed = json.loads(content)

    return elapsed, parsed


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

rows = db.execute("""
    SELECT
        ticket_id,
        ticket_number,
        title,
        customer_text,
        agent_text,
        unknown_text
    FROM cases
    WHERE classification_status = 'NEW'
    ORDER BY ticket_id
    LIMIT 10
""").fetchall()

print("=" * 72)
print("OLLAMA API TEST")
print("=" * 72)
print(f"Tickets: {len(rows)}")
print(f"Modell:  {MODEL}")
print()

for row in rows:

    elapsed, result = ask(row)

    print(
        f"#{row['ticket_number']} "
        f"{elapsed:.1f}s "
        f"→ {result.get('primary_intent')} "
        f"secondary={result.get('secondary_intents')} "
        f"confidence={result.get('confidence')}"
    )

    print(
        f"  {result.get('summary', '')[:250]}"
    )

db.close()