ml-services/evaluation/rubric.py

"""
Formal, versioned translation of the domain expert's QA rubric into a
machine-readable schema (Pydantic v2). This is the OUTPUT CONTRACT the LLM
evaluation layer must satisfy.

Design principle -- MODALITY SPLIT
----------------------------------
The expert's rubric mixes two kinds of judgment:
  * TEXT-derivable   : can be decided from the transcript alone.
  * AUDIO/paralinguistic: tone, pace, energy, "sounds satisfied" -- these
                          require the acoustic model (WP3) and must NOT be
                          guessed from text. They are represented here but
                          flagged `requires_audio=True` and left null by the
                          text LLM, to be filled by audio-text fusion (WBS 3.4).

Each leaf judgment carries EVIDENCE (a verbatim quote + speaker + timestamp) so
every score is provable -- no ungrounded compliance passes. This directly
supports the "Verbatim Evidence Extractor" track later.

RUBRIC_VERSION is bumped whenever criteria change so labelled data stays traceable.
"""
from __future__ import annotations
from typing import Optional, List, Literal
from pydantic import BaseModel, Field

RUBRIC_VERSION = "0.1.0"          # legacy monolithic path (extract.py)
RUBRIC_VERSION_GRAPH = "0.4.1"    # graph path: 0.2.0 added customer_satisfaction,
                                  # 0.3.0 added evidence re-anchor loop +
                                  # escalation investigation node,
                                  # 0.4.0 added acoustic-text fusion (hybrid
                                  # scores) + expected-workflow track,
                                  # 0.4.1 refit fusion thresholds on the
                                  # 22-call batch + arbitrate node on
                                  # text/acoustic disagreement

Speaker = Literal["AGENT", "CUSTOMER"]


# ─────────────────────────────────────────────────────────────────────────────
# Shared evidence object -- makes every judgment provable
# ─────────────────────────────────────────────────────────────────────────────
class Evidence(BaseModel):
    quote: str = Field(description="Verbatim text copied from the transcript (no paraphrasing).")
    speaker: Speaker
    timestamp: Optional[str] = Field(default=None, description="mm:ss of the turn, if available.")


# ─────────────────────────────────────────────────────────────────────────────
# 1. COMPLIANCE  (all TEXT-derivable; binary pass/fail with N/A)
#    passed = True | False | None(=not applicable, e.g. recording disclosure
#    "when needed", or transfer steps when there was no transfer)
# ─────────────────────────────────────────────────────────────────────────────
class ComplianceItem(BaseModel):
    passed: Optional[bool] = Field(description="True=met, False=violated, None=not applicable to this call.")
    evidence: Optional[Evidence] = Field(default=None, description="Required when passed is True or False.")
    note: Optional[str] = None


class ComplianceChecklist(BaseModel):
    # "Announce their name"
    name_announced: ComplianceItem
    # "Announce company name"
    company_announced: ComplianceItem
    # "State 'this call may be recorded' (when needed)"  -> None if not needed
    recording_disclosure: ComplianceItem
    # "Confirm customer identity before discussing account" (last 4 / acct # / customer ID / DOB)
    identity_verified: ComplianceItem
    identity_method: Optional[str] = Field(
        default=None, description="e.g. 'last 4 digits', 'account number', 'date of birth', 'customer ID'.")
    # "Provide resolution / next steps before closing"
    resolution_provided: ComplianceItem
    # "If call transferred, provide information of next steps" -> None if no transfer
    transfer_next_steps: ComplianceItem


# ─────────────────────────────────────────────────────────────────────────────
# 2. QUALITY  (1-5 each). Each dimension lists the concrete signals the rubric
#    names, split into text vs audio. The text LLM fills text signals + a score;
#    `requires_audio` flags dimensions whose FULL judgment also needs WP3.
# ─────────────────────────────────────────────────────────────────────────────
class HybridScore(BaseModel):
    """Provenance for a fused score (v0.4.0). method='text_only' means the
    acoustic model contributed nothing (no data / dimension is text-defined)."""
    text_score: int = Field(ge=1, le=5, description="LLM score from text signals alone.")
    acoustic_score: Optional[float] = Field(default=None, description="Acoustic subscore mapped to 1-5.")
    text_weight: float
    acoustic_weight: float
    coverage: Optional[float] = Field(
        default=None, description="Fraction of channel sentences the audio model processed.")
    channel: Optional[str] = Field(default=None, description="Speaker channel the acoustic signal came from.")
    method: Literal["weighted_mean", "text_only"] = "text_only"


class QualityDimension(BaseModel):
    score: int = Field(ge=1, le=5, description="1=poor, 5=excellent. Text LLM score; "
                       "overwritten with the fused score when acoustic data exists (v0.4.0).")
    signals_present: List[str] = Field(default_factory=list, description="Rubric behaviours observed.")
    signals_absent: List[str] = Field(default_factory=list, description="Expected behaviours not observed.")
    evidence: List[Evidence] = Field(default_factory=list)
    requires_audio: bool = Field(
        default=False, description="True if a complete judgment also needs acoustic signals (WP3).")
    hybrid: Optional[HybridScore] = Field(
        default=None, description="How this score was computed (v0.4.0+). Null in older evaluations.")


class QualityDimensions(BaseModel):
    # a. Efficiency: FCR, appropriate duration, unnecessary holds/transfers (text-derivable;
    #    duration is metadata)
    efficiency: QualityDimension
    # c. Problem Resolution: options A/B/C, pros/cons, explains WHY, respects choice, issue solved
    problem_resolution: QualityDimension
    # b. Clarity: open-ended questions, "does that make sense?", confirms understanding, recap
    clarity: QualityDimension
    # d. Professionalism: grammar/complete sentences/no slang (TEXT) + pace/confident voice (AUDIO)
    professionalism: QualityDimension   # requires_audio likely True
    # e. Empathy: validation phrases + name use (TEXT) + flat/robotic tone, matches emotion (AUDIO)
    empathy: QualityDimension           # requires_audio likely True
    # f. Customer Satisfaction: problem addressed, clear resolution, followed through (TEXT)
    #    + "sounds satisfied", not frustrated at end (AUDIO). Added in v0.2.0.
    customer_satisfaction: Optional[QualityDimension] = Field(
        default=None,
        description="Customer outcome signals. Added in v0.2.0; null in v0.1.0 evaluations."
    )


# ─────────────────────────────────────────────────────────────────────────────
# 3. ESCALATION RISK
#    Red flags are mostly TEXT (phrases). Customer emotion is text-inferred here
#    but should be CONFIRMED by the audio model before it drives an action.
# ─────────────────────────────────────────────────────────────────────────────
RedFlag = Literal[
    "manager_requested",        # "I want to speak to a manager"
    "competitor_switch",        # "I'm switching to [competitor]"
    "repeat_attempts",          # "I've called a couple of times"
    "issue_too_complex",        # agent struggling / repeated holds
    "explicit_dissatisfaction", # other strong negative statements
]


class EscalationFusion(BaseModel):
    """Provenance for the fused risk level.
    v0.4.0 fused via ordinal max (more severe tier wins, "ordinal_max").
    v0.4.1: tiers that AGREE merge deterministically ("agreement"); tiers
    that DISAGREE are resolved by an arbitration LLM node that weighs both
    signals in context ("llm_arbitration") -- disagreement is exactly where
    mechanical merging has nothing to stand on (batch evidence: of 10 calls
    where either channel flagged risk, both flagged on only 1)."""
    text_risk: Literal["none", "review", "escalate"]
    acoustic_risk: Optional[Literal["none", "review", "escalate"]] = None
    late_mean_escalation: Optional[float] = Field(
        default=None, description="Mean customer escalation_score over the final third of the call.")
    peak_escalation: Optional[float] = Field(
        default=None, description="Max customer escalation_score anywhere in the call. Context only "
                                  "since v0.4.1 -- not a tier trigger (batch median 0.56 gave it no "
                                  "discriminating power).")
    method: Literal["agreement", "llm_arbitration", "ordinal_max", "text_only"] = "text_only"
    arbitration_rationale: Optional[str] = Field(
        default=None, description="Arbitrator's reasoning when method='llm_arbitration'.")


class EscalationRisk(BaseModel):
    red_flags: List[RedFlag] = Field(default_factory=list)
    customer_emotion_text: Literal["calm", "mild_frustration", "frustrated", "angry", "distressed"] = Field(
        description="Emotion inferred from TEXT only; audio model confirms before action.")
    risk_level: Literal["none", "review", "escalate"] = Field(
        description="none / review (manager may review) / escalate (immediate). "
                    "Fused with the acoustic tier when audio data exists (v0.4.0).")
    evidence: List[Evidence] = Field(default_factory=list)
    requires_audio: bool = Field(
        default=True, description="Emotion intensity needs acoustic confirmation.")
    hybrid: Optional[EscalationFusion] = Field(
        default=None, description="How risk_level was computed (v0.4.0+). Null in older evaluations.")


# ─────────────────────────────────────────────────────────────────────────────
# 4. INVESTIGATION (v0.3.0) -- produced ONLY when risk_level != "none".
#    A conditional deep-dive that turns escalation red flags into a
#    manager-ready incident report.
# ─────────────────────────────────────────────────────────────────────────────
class InvestigationReport(BaseModel):
    summary: str = Field(description="2-3 sentence incident summary for a manager.")
    contributing_factors: List[str] = Field(
        default_factory=list, description="What led to the escalation risk (agent + customer side).")
    recommended_action: str = Field(description="Concrete next step for the reviewing manager.")
    priority: Literal["low", "medium", "high"] = Field(
        description="Review urgency. escalate risk -> high, review risk -> low/medium.")
    evidence: List[Evidence] = Field(default_factory=list)


# ─────────────────────────────────────────────────────────────────────────────
# 5. EXPECTED WORKFLOW (v0.4.0) -- three isolated LLM phases:
#    a. subject: one sentence stating what the customer contacted about
#       (the REQUEST only, never the outcome).
#    b. expected steps: generated from domain + subject ONLY -- the model has
#       not seen the transcript, so expectations cannot be contaminated by
#       what actually happened on the call.
#    c. check: audits the transcript against that independent checklist.
# ─────────────────────────────────────────────────────────────────────────────
class WorkflowStep(BaseModel):
    step: str = Field(description="Short imperative phrase, e.g. 'Verify customer identity'.")
    rationale: Optional[str] = Field(default=None, description="Why this step is expected for this call type.")
    met: Optional[bool] = Field(
        default=None, description="True=performed, False=missed, None=not determinable/applicable. "
                                  "Filled by the check phase; null before checking.")
    evidence: Optional[Evidence] = None


class CallWorkflow(BaseModel):
    subject: str = Field(description="One-sentence gist of what the customer wanted (request, not outcome).")
    expected_steps: List[WorkflowStep] = Field(
        description="4-8 boxes a competent agent should check for a call with this subject.")


# ─────────────────────────────────────────────────────────────────────────────
# Top-level evaluation payload
# ─────────────────────────────────────────────────────────────────────────────
class CallMetadata(BaseModel):
    call_id: str
    domain: str                       # injected as "Call Type" lens
    accent: Optional[str] = None
    duration_seconds: Optional[float] = None
    transcript_model: Optional[str] = None


class CallEvaluation(BaseModel):
    rubric_version: str = RUBRIC_VERSION
    metadata: CallMetadata
    compliance: ComplianceChecklist
    quality: QualityDimensions
    escalation: EscalationRisk
    investigation: Optional[InvestigationReport] = Field(
        default=None,
        description="Deep-dive incident report. Only populated when risk_level != 'none' (v0.3.0+).")
    workflow: Optional[CallWorkflow] = Field(
        default=None,
        description="Subject-derived expected workflow + audit results (v0.4.0+).")
    overall_summary: Optional[str] = Field(
        default=None, description="2-3 sentence plain-language summary for the dashboard.")


# Documentation map: which rubric areas are text vs audio vs metadata.
# Used by the prompt builder (Phase 1) and the report.
MODALITY_MAP = {
    "compliance.name_announced":        "text",
    "compliance.company_announced":     "text",
    "compliance.recording_disclosure":  "text",
    "compliance.identity_verified":     "text",
    "compliance.resolution_provided":   "text",
    "compliance.transfer_next_steps":   "text",
    "quality.efficiency":               "text+metadata",  # duration from metadata
    "quality.problem_resolution":       "text",
    "quality.clarity":                  "text",
    "quality.professionalism":          "text+audio",     # grammar=text, tone/pace=audio
    "quality.empathy":                  "text+audio",     # phrases=text, warmth/tone=audio
    "escalation.red_flags":             "text",
    "escalation.customer_emotion":      "text+audio",      # words=text, intensity=audio
    "quality.customer_satisfaction":    "text+audio",      # outcome=text, "sounds satisfied"=audio
}


if __name__ == "__main__":
    # emit the JSON schema so we can eyeball the contract
    import json
    print(f"RUBRIC_VERSION = {RUBRIC_VERSION}\n")
    print(json.dumps(CallEvaluation.model_json_schema(), indent=2)[:1500], "...")