""" Acoustic-text fusion (v0.4.0) -- combines the audio sentiment model's per-sentence output with the text LLM's rubric scores. Design: DETERMINISTIC POST-HOC FUSION, not prompt injection. The LLM judges what text can prove; the acoustic model contributes what text cannot see (tone, emotion, escalation intensity); a documented formula combines them. This keeps the LLM's variance out of the acoustic signal and makes every fused score reproducible and explainable. Math ---- Valence map: Positive=+1, Neutral/Mixed=0, Negative=-1. Quality dimension fusion (empathy, professionalism <- AGENT channel; customer_satisfaction <- CUSTOMER channel, final third weighted 2x): acoustic_score a = 3 + 2 * weighted_mean(valence) # maps [-1,1] -> [1,5] w_a = W_ACOUSTIC_CAP * coverage # coverage = processed/total fused = clamp(round((1 - w_a) * text + w_a * a), 1, 5) Text stays the primary signal (cap 0.4) because the rubric's behaviours are text-defined; acoustics modulate. Zero acoustic data -> w_a = 0 (text only). Escalation fusion (CUSTOMER channel escalation_score trajectory): late_mean = mean(score over final third), peak = max(score) acoustic tier (v0.4.1, fitted on the 22-call batch -- see threshold comments): escalate if late_mean >= 0.50, review if late_mean >= 0.30, none otherwise. peak is recorded as context but does NOT set a tier. Merge (v0.4.1): tiers that agree merge deterministically ("agreement"); tiers that disagree are left to the arbitrate LLM node ("disputed" -> resolved as "llm_arbitration"), because disagreement is where a fixed formula has no information to prefer one channel over the other. Data layout: data/sentiment/{domain}/{call_id}.json (model output) data/sentiment/{domain}/{call_id}_segments.json (the EXACT sentence segmentation the model ran against -- seq_ids are only meaningful against their own segmentation) """ import os import json import paths SENTIMENT_ROOT = str(paths.SENTIMENT_ROOT) W_ACOUSTIC_CAP = 0.4 # max acoustic weight at 100% coverage LATE_FRACTION = 1 / 3 # "end of call" window for trajectory metrics LATE_WEIGHT = 2.0 # extra weight on final-third customer valence # Acoustic escalation tiers -- fitted on the 22-call batch (2026-07, # batch_summary.json, n=21 calls with paired acoustic data): # late_mean: min 0.166, p25 0.219, median 0.227, p75 0.257, p90 0.281, # max 0.406 (single outlier; next highest 0.281). # -> review at 0.30 sits above the calm cluster's p90 with margin; only the # outlier call crosses it. escalate at 0.50 is deliberately outside the # observed range: nothing in this batch warrants an acoustic-only # escalate, and the bar should stay high until data demands lowering it. # peak: median 0.558, p90 0.696 -- even calm calls routinely spike past the # old 0.60 trigger, so peak has no discriminating power as a threshold. # It is kept as recorded context (arbitrator/investigator input) only. ESC_THRESHOLDS = { "escalate": {"late_mean": 0.50}, "review": {"late_mean": 0.30}, } VALENCE = {"Positive": 1.0, "Negative": -1.0, "Neutral": 0.0, "Mixed": 0.0} RISK_ORDER = {"none": 0, "review": 1, "escalate": 2} # which quality dimension is informed by which channel FUSED_DIMS = { "professionalism": {"channel": "AGENT", "late_weighted": False}, "empathy": {"channel": "AGENT", "late_weighted": False}, "customer_satisfaction": {"channel": "CUSTOMER", "late_weighted": True}, } def load_acoustic(call_id, domain): """ Join the sentiment model output with the segmentation it ran against. Returns list of rows {seq_id, pos, speaker, valence, emotion, escalation_score, ok} or None if no acoustic data exists for this call. """ d = os.path.join(SENTIMENT_ROOT, domain.lower()) sent_path = os.path.join(d, f"{call_id}.json") seg_path = os.path.join(d, f"{call_id}_segments.json") if not (os.path.exists(sent_path) and os.path.exists(seg_path)): return None with open(sent_path, encoding="utf-8") as f: sentiment = json.load(f) with open(seg_path, encoding="utf-8") as f: segments = json.load(f) speakers = {s["seq_id"]: s["speaker"] for s in segments["sentences"]} n = max(speakers) or 1 rows = [] for s in sentiment["segments"]: sid = s["seq_id"] ok = s.get("processing_status") == "success" and s.get("sentiment") rows.append({ "seq_id": sid, "pos": sid / n, # 0..1 position in call "speaker": speakers.get(sid), "valence": VALENCE.get(s.get("sentiment"), 0.0) if ok else None, "emotion": s.get("dominant_emotion"), "escalation_score": s.get("escalation_score"), "ok": bool(ok), }) return rows def _channel_valence(rows, channel, late_weighted): """(weighted mean valence, coverage) for one speaker channel.""" chan = [r for r in rows if r["speaker"] == channel] if not chan: return None, 0.0 ok = [r for r in chan if r["ok"]] coverage = len(ok) / len(chan) if not ok: return None, 0.0 wsum = vsum = 0.0 for r in ok: w = LATE_WEIGHT if (late_weighted and r["pos"] >= 1 - LATE_FRACTION) else 1.0 wsum += w vsum += w * r["valence"] return vsum / wsum, coverage def _acoustic_escalation(rows): """Acoustic risk tier from the customer escalation_score trajectory.""" cust = [r for r in rows if r["speaker"] == "CUSTOMER" and r["ok"] and r["escalation_score"] is not None] if not cust: return None, None, None late = [r["escalation_score"] for r in cust if r["pos"] >= 1 - LATE_FRACTION] late_mean = sum(late) / len(late) if late else 0.0 peak = max(r["escalation_score"] for r in cust) for tier in ("escalate", "review"): if late_mean >= ESC_THRESHOLDS[tier]["late_mean"]: return tier, round(late_mean, 3), round(peak, 3) return "none", round(late_mean, 3), round(peak, 3) def trajectory(rows, buckets=48): """ Downsample the per-sentence acoustic signal into fixed position buckets for the frontend timeline: [{p, esc, cust_val, agent_val}, ...]. p = bucket midpoint (0..1); esc = mean customer escalation_score; *_val = mean valence per channel. Fields are null where a bucket has no processed sentences for that channel. """ out = [] for b in range(buckets): lo, hi = b / buckets, (b + 1) / buckets cell = [r for r in rows if r["ok"] and lo <= r["pos"] < hi] esc = [r["escalation_score"] for r in cell if r["speaker"] == "CUSTOMER" and r["escalation_score"] is not None] cv = [r["valence"] for r in cell if r["speaker"] == "CUSTOMER" and r["valence"] is not None] av = [r["valence"] for r in cell if r["speaker"] == "AGENT" and r["valence"] is not None] out.append({ "p": round((lo + hi) / 2, 4), "esc": round(sum(esc) / len(esc), 3) if esc else None, "cust_val": round(sum(cv) / len(cv), 3) if cv else None, "agent_val": round(sum(av) / len(av), 3) if av else None, }) return out def fuse_evaluation(ev, rows): """ Mutate an evaluation dict in place: fuse acoustic signal into the three audio-informed quality dimensions and the escalation risk level. Every dimension gets a `hybrid` provenance block (method "text_only" when acoustics did not contribute) so the UI can explain each score. Returns a short summary dict for node_meta. """ fused_dims = [] for name, dim in (ev.get("quality") or {}).items(): if not isinstance(dim, dict) or "score" not in dim: continue text_score = dim["score"] cfg = FUSED_DIMS.get(name) vbar, coverage = (_channel_valence(rows, cfg["channel"], cfg["late_weighted"]) if (cfg and rows) else (None, 0.0)) if cfg and vbar is not None: a = 3 + 2 * vbar w_a = W_ACOUSTIC_CAP * coverage fused = max(1, min(5, round((1 - w_a) * text_score + w_a * a))) dim["score"] = fused dim["hybrid"] = { "text_score": text_score, "acoustic_score": round(a, 2), "text_weight": round(1 - w_a, 2), "acoustic_weight": round(w_a, 2), "coverage": round(coverage, 2), "channel": cfg["channel"], "method": "weighted_mean", } fused_dims.append(name) else: dim["hybrid"] = { "text_score": text_score, "acoustic_score": None, "text_weight": 1.0, "acoustic_weight": 0.0, "coverage": None, "channel": cfg["channel"] if cfg else None, "method": "text_only", } # Escalation merge (v0.4.1): agreement is trivially deterministic; # disagreement carries no information about which channel to trust, so it # is NOT resolved here -- the graph routes disputed calls to the arbitrate # LLM node, which weighs both signals against the transcript. Until (or # unless) arbitration runs, the text tier stands. disputed = False esc = ev.get("escalation") or {} text_risk = esc.get("risk_level") a_risk, late_mean, peak = _acoustic_escalation(rows) if rows else (None, None, None) if text_risk and a_risk is not None: disputed = a_risk != text_risk esc["hybrid"] = { "text_risk": text_risk, "acoustic_risk": a_risk, "late_mean_escalation": late_mean, "peak_escalation": peak, "method": "agreement" if not disputed else "text_only", "arbitration_rationale": None, } elif text_risk: esc["hybrid"] = { "text_risk": text_risk, "acoustic_risk": None, "late_mean_escalation": None, "peak_escalation": None, "method": "text_only", "arbitration_rationale": None, } # downsampled acoustic timeline for the frontend Agent-findings lanes if rows: ev["_acoustic"] = {"trajectory": trajectory(rows), "n_sentences": len(rows)} return { "acoustic_available": bool(rows), "fused_dims": fused_dims, "acoustic_risk": a_risk, "disputed": disputed, }