3cx-telefonie-middleware/tools/classify_historical_intents.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_intent_classification.json")

OLLAMA = "http://127.0.0.1:11434/api/generate"
MODEL = "qwen3:8b"

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


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


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

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

    with urllib.request.urlopen(request, timeout=300) as response:
        data = json.loads(response.read().decode("utf-8"))

    return data.get("response", "")


def parse_model_json(text):
    text = text.strip()

    if text.startswith("```"):
        text = text.replace("```json", "", 1)
        text = text.replace("```", "")
        text = text.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 index, row in enumerate(rows, 1):

        analysis = load_json(row["analysis_json"])

        transcript = row["transcript_json"] or ""

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

        prompt = f"""
Du klassifizierst ein reales deutsches Kundentelefonat
für einen Gartenpflanzen-Webshop.

Wir entwickeln daraus eine Telefon-KI.
Die Klassifikation muss deshalb konservativ und praxisnah sein.

Wähle genau EINEN primary_intent aus dieser Taxonomie:

{json.dumps(TAXONOMY, ensure_ascii=False)}

Regeln:

VERSANDSTATUS:
Der Kunde fragt nach dem aktuellen Versand-/Paketstatus.

LIEFERVERZUG:
Eine erwartete Lieferung ist überfällig oder verspätet.

ADRESSÄNDERUNG:
Änderung einer Liefer-, Rechnungs- oder Kontaktadresse.

REKLAMATION:
Beanstandung eines gelieferten Produkts oder einer Leistung.

TRANSPORTSCHADEN:
Produkt wurde beim Transport beschädigt.

FEHLLIEFERUNG:
Es wurde ein falscher Artikel geliefert.

RECHNUNG:
Frage zur Rechnung, Rechnungsstellung oder Rechnungsdaten.

ZAHLUNG:
Zahlungsstatus, Zahlungsart, Zahlungseingang oder Zahlungsproblem.

WIDERRUF:
Widerruf eines Kaufs.

RETOURE:
Rückgabe/Rücksendung eines Artikels.

PFLANZENBERATUNG:
Allgemeine Beratung zu Pflanzen, Standort, Pflanzung oder Auswahl.

SORTENBERATUNG:
Konkrete Beratung zur Auswahl oder Eignung einer Sorte.

PFLEGEFRAGE:
Pflege, Schnitt, Düngung, Krankheiten oder Überwinterung.

BESTANDSANFRAGE:
Frage, ob eine konkrete Pflanze/ein Produkt verfügbar ist.

VORBESTELLUNG:
Vorbestellung bzw. Reservierung eines noch nicht verfügbaren Artikels.

B2B:
Geschäftskunde mit individuellem gewerblichen Anliegen.

GROSSHANDEL:
Gewerblicher Großhandel bzw. größere Mengen / Wiederverkauf.

SONSTIGES:
Wenn keine Kategorie ausreichend passt.

WICHTIG:

- Nicht aus Kundennummern, Telefonnummern oder Produktnamen raten.
- Das tatsächliche Gespräch ist maßgeblich.
- Bei mehreren Themen ist das Hauptanliegen des Anrufers primary_intent.
- Weitere eindeutig vorhandene Anliegen gehören in secondary_intents.
- Keine Kategorie nur deshalb wählen, weil sie in der bestehenden Qwen-Analyse steht.
- Die bestehende Analyse ist nur Hilfskontext.
- Der vollständige Transkripttext ist maßgeblich.

Bestehende Qwen-Analyse:
{json.dumps(analysis, ensure_ascii=False, indent=2)}

Transkript:
{transcript}

Antworte ausschließlich als JSON:

{{
  "primary_intent": "EIN_WERT_AUS_DER_TAXONOMIE",
  "secondary_intents": [],
  "customer_goal": "kurze präzise Beschreibung",
  "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": "kurze Begründung"
}}
"""

        try:
            response = await asyncio.to_thread(
                ollama_generate,
                prompt,
            )

            classification = parse_model_json(response)

        except Exception as exc:
            print(f"  FEHLER: {exc}")
            classification = {
                "primary_intent": "SONSTIGES",
                "secondary_intents": [],
                "customer_goal": None,
                "requires_customer_context": True,
                "requires_order_context": False,
                "requires_product_context": False,
                "requires_previous_contact_context": False,
                "requires_human_action": True,
                "confidence": 0,
                "reason": f"LLM error: {exc}",
            }

        if classification.get("primary_intent") not in TAXONOMY:
            classification["primary_intent"] = "SONSTIGES"

        classification["_analysis_id"] = row["analysis_id"]
        classification["_cdr_row_id"] = row["analysis_id"]
        classification["_rec_id"] = None
        classification["_start_time"] = row["start_time"]

        results.append(classification)

    intent_counts = Counter(
        r["primary_intent"]
        for r in results
    )

    context_counts = Counter()

    for r in results:
        for field in (
            "requires_customer_context",
            "requires_order_context",
            "requires_product_context",
            "requires_previous_contact_context",
            "requires_human_action",
        ):
            if r.get(field):
                context_counts[field] += 1

    output = {
        "model": MODEL,
        "count": len(results),
        "taxonomy": TAXONOMY,
        "intent_counts": dict(intent_counts.most_common()),
        "context_requirements": dict(context_counts),
        "calls": results,
    }

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

    print()
    print("=" * 70)
    print("HISTORISCHE INTENT-KLASSIFIKATION FERTIG")
    print("=" * 70)

    print("\nINTENTS:")
    for intent, count in intent_counts.most_common():
        print(f"{count:3}  {intent}")

    print("\nKONTEXTBEDARF:")
    for key, count in context_counts.most_common():
        print(f"{count:3}  {key}")

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


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