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