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

#!/usr/bin/env python3

import asyncio
import json
import sqlite3
import urllib.request
from collections import Counter
from pathlib import Path

DB = "data/telephony.sqlite3"
OUT = Path("data/historical_multi_intent.json")
OLLAMA = "http://127.0.0.1:11434/api/generate"
MODEL = "qwen3:8b"

INTENTS = [
    "VERSANDSTATUS",
    "LIEFERVERZUG",
    "ADRESSÄNDERUNG",
    "REKLAMATION",
    "TRANSPORTSCHADEN",
    "FEHLLIEFERUNG",
    "RECHNUNG",
    "ZAHLUNG",
    "WIDERRUF",
    "RETOURE",
    "PFLANZENBERATUNG",
    "SORTENBERATUNG",
    "PFLEGEFRAGE",
    "BESTANDSANFRAGE",
    "VORBESTELLUNG",
    "BESTELLUNG",
    "BESTELLÄNDERUNG",
    "STORNIERUNG",
    "B2B",
    "GROSSHANDEL",
    "SONSTIGES",
]

ACTIONS = [
    "KEINE_AKTION",
    "KUNDENKONTEXT_PRÜFEN",
    "BESTELLUNG_PRÜFEN",
    "BESTELLUNG_ÄNDERN",
    "BESTELLUNG_STORNIEREN",
    "BESTELLUNGEN_ZUSAMMENFÜHREN",
    "VERSANDSTATUS_PRÜFEN",
    "LIEFERTERMIN_PRÜFEN",
    "ADRESSE_ÄNDERN",
    "REKLAMATION_ERFASSEN",
    "TRANSPORTSCHADEN_ERFASSEN",
    "FEHLLIEFERUNG_PRÜFEN",
    "RECHNUNG_PRÜFEN",
    "RECHNUNG_KORRIGIEREN",
    "ZAHLUNG_PRÜFEN",
    "GUTSCHRIFT_ERSTELLEN",
    "RETOURE_ERFASSEN",
    "PRODUKT_IDENTIFIZIEREN",
    "BESTAND_PRÜFEN",
    "SORTEN_ALTERNATIVE_PRÜFEN",
    "PFLEGEHINWEIS_GEBEN",
    "BILD_ANFORDERN",
    "MENSCHLICHE_BEARBEITUNG_ERFORDERLICH",
    "RÜCKRUF",
    "E-MAIL_SENDEN",
]

def load_json(value):
    try:
        return json.loads(value or "{}")
    except Exception:
        return {}

def ask_qwen(prompt):
    payload = {
        "model": MODEL,
        "prompt": prompt,
        "stream": False,
        "format": "json",
        "options": {"temperature": 0},
    }

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

    with urllib.request.urlopen(req, timeout=300) as r:
        return json.loads(r.read().decode()).get("response", "")

def parse_json(text):
    text = text.strip()
    if text.startswith("```"):
        text = text.replace("```json", "", 1)
        text = text.replace("```", "").strip()
    return json.loads(text)

async def main():
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row

    rows = con.execute("""
        SELECT
            a.id AS analysis_id,
            a.analysis_json,
            t.transcript_json,
            c.source_caller_id,
            c.destination_caller_id,
            c.start_time
        FROM analyses a
        JOIN transcripts t ON t.id = a.transcript_id
        JOIN cdr_calls c ON c.id = a.cdr_row_id
        WHERE a.cdr_row_id IS NOT NULL
        ORDER BY a.id
    """).fetchall()

    con.close()

    print(f"Historische Gespräche: {len(rows)}")

    results = []

    for n, row in enumerate(rows, 1):
        existing = load_json(row["analysis_json"])
        transcript = row["transcript_json"] or ""

        print(
            f"[{n}/{len(rows)}] "
            f"analysis={row['analysis_id']}"
        )

        prompt = f"""
Du analysierst ein echtes deutsches Kundentelefonat eines
Gartenpflanzen-Webshops für eine Telefon-KI.

WICHTIG:
Ein Gespräch kann MEHRERE Anliegen enthalten.

ZUERST muss der Gesprächszustand bestimmt werden:

- HUMAN_CONVERSATION:
  tatsächliches Gespräch mit einem Kunden/Mitarbeiter und verwertbarem Inhalt
- NO_CONTENT:
  leeres/fehlerhaftes Transkript, keine verwertbaren Gesprächsinhalte
- VOICEMAIL:
  automatische Mailbox/Ansage ohne echtes Kundengespräch
- UNKNOWN:
  Inhalt vorhanden, aber nicht zuverlässig klassifizierbar

Nur bei HUMAN_CONVERSATION werden Intents klassifiziert.

Ermittle:
1. call_state
2. primary_intent = wichtigstes tatsächliches Anliegen
3. secondary_intents = ALLE weiteren tatsächlich behandelten Anliegen
4. entities = konkret genannte Bestellungen und Produkte
5. actions = tatsächlich notwendige oder vereinbarte Aktionen

SEHR WICHTIG:
SONSTIGES ist der LETZTE FALLBACK.

Wenn das Gespräch beispielsweise:
- eine Pflegefrage enthält → PFLEGEFRAGE
- eine Rechnungskorrektur enthält → RECHNUNG
- eine Beanstandung enthält → REKLAMATION
- eine beschädigte Lieferung enthält → TRANSPORTSCHADEN
- eine Bestellung verändert → BESTELLÄNDERUNG
- mehrere Aufträge/Bestellungen behandelt → BESTELLUNG bzw. BESTELLÄNDERUNG
- Verfügbarkeit eines Artikels behandelt → BESTANDSANFRAGE

darf NICHT SONSTIGES gewählt werden.

Wenn mehrere dieser Themen vorkommen, müssen sie als
primary_intent + secondary_intents abgebildet werden.

Beispiel:
Bestelländerung + Versandstatus + Lieferproblem

→ primary_intent = BESTELLÄNDERUNG
→ secondary_intents = [VERSANDSTATUS, LIEFERVERZUG]

Beispiel:
Rechnung falsch + Gutschrift

→ primary_intent = RECHNUNG
→ secondary_intents = [REKLAMATION]

Beispiel:
Beschädigte Pflanzen + Qualitätsbeanstandung

→ primary_intent = REKLAMATION
→ secondary_intents = [TRANSPORTSCHADEN]

SONSTIGES darf nur verwendet werden, wenn nach Prüfung
aller definierten Kategorien wirklich keine passt.

Niemals mehrere Anliegen in einen einzigen Intent quetschen.

INTENTS:
{json.dumps(INTENTS, ensure_ascii=False)}

AKTIONEN:
{json.dumps(ACTIONS, ensure_ascii=False)}

Regeln:
- Nur tatsächlich aus dem Gespräch ableiten.
- Nicht aus Vermutungen ergänzen.
- BESTELLUNG verwenden, wenn eine Bestellung aufgegeben,
  ergänzt oder konkret bearbeitet wird.
- BESTELLÄNDERUNG verwenden, wenn eine bestehende Bestellung
  verändert wird.
- STORNIERUNG verwenden, wenn eine Bestellung aufgehoben werden soll.
- VERSANDSTATUS und LIEFERVERZUG unterscheiden.
- REKLAMATION ist eine Beanstandung.
- TRANSPORTSCHADEN ist eine durch Transport verursachte Beschädigung.
- PFLANZENBERATUNG = allgemeine Beratung.
- SORTENBERATUNG = Auswahl zwischen konkreten Sorten.
- PFLEGEFRAGE = Pflege/Gesundheit/Schnitt/Düngung etc.
- BESTANDSANFRAGE = Verfügbarkeit.
- SONSTIGES nur, wenn keine andere Kategorie passt.

Ein bereits vorhandenes Qwen-Ergebnis ist nur Hilfskontext.
Das Transkript ist maßgeblich.

BESTEHENDE ANALYSE:
{json.dumps(existing, ensure_ascii=False, indent=2)}

TRANSKRIPT:
{transcript}

Antworte ausschließlich als JSON:

{{
  "call_state": "HUMAN_CONVERSATION",
  "primary_intent": "INTENT",
  "secondary_intents": [],
  "customer_goal": "",
  "entities": {{
    "orders": [],
    "products": []
  }},
  "actions": [],
  "requires_customer_context": true,
  "requires_order_context": false,
  "requires_product_context": false,
  "requires_previous_contact_context": false,
  "requires_human_action": false,
  "confidence": 0.0,
  "reason": ""
}}
"""

        try:
            raw = await asyncio.to_thread(ask_qwen, prompt)
            result = parse_json(raw)
        except Exception as exc:
            print(f"  FEHLER: {exc}")
            result = {
                "primary_intent": "SONSTIGES",
                "secondary_intents": [],
                "customer_goal": "",
                "entities": {"orders": [], "products": []},
                "actions": ["MENSCHLICHE_BEARBEITUNG_ERFORDERLICH"],
                "requires_customer_context": True,
                "requires_order_context": False,
                "requires_product_context": False,
                "requires_previous_contact_context": False,
                "requires_human_action": True,
                "confidence": 0,
                "reason": str(exc),
            }

        if result.get("primary_intent") not in INTENTS:
            result["primary_intent"] = "SONSTIGES"

        result["secondary_intents"] = [
            x for x in result.get("secondary_intents", [])
            if x in INTENTS and x != result["primary_intent"]
        ]

        result["actions"] = [
            x for x in result.get("actions", [])
            if x in ACTIONS
        ]

        result["_analysis_id"] = row["analysis_id"]
        result["_cdr_row_id"] = row["analysis_id"]
        result["_start_time"] = row["start_time"]

        results.append(result)

    primary = Counter()
    secondary = Counter()
    actions = Counter()

    for r in results:
        primary[r["primary_intent"]] += 1
        secondary.update(r["secondary_intents"])
        actions.update(r["actions"])

    output = {
        "model": MODEL,
        "count": len(results),
        "primary_intents": dict(primary.most_common()),
        "secondary_intents": dict(secondary.most_common()),
        "actions": dict(actions.most_common()),
        "calls": results,
    }

    OUT.write_text(
        json.dumps(output, ensure_ascii=False, indent=2)
    )

    print("\n" + "=" * 72)
    print("MULTI-INTENT ANALYSE FERTIG")
    print("=" * 72)

    print("\nPRIMARY INTENTS")
    for k, v in primary.most_common():
        print(f"{v:3}  {k}")

    print("\nSECONDARY INTENTS")
    for k, v in secondary.most_common():
        print(f"{v:3}  {k}")

    print("\nAKTIONEN")
    for k, v in actions.most_common():
        print(f"{v:3}  {k}")

    print(f"\nGespeichert: {OUT}")

if __name__ == "__main__":
    asyncio.run(main())