:::tip 🎮 Interactive Playground Visualize this concept: Try the Human-in-the-Loop Agents demo on the EngineersOfAI Playground - no code required. :::
Escalation and Handoff Patterns
The Chatbot That Would Not Stop
A telecommunications company deployed an AI customer service agent to handle billing disputes. The AI was excellent at standard cases - credit requests under $100, plan changes, address updates. By every metric the team tracked, it was performing well: 87% first-contact resolution rate, 4.2/5 customer satisfaction for resolved cases, 40% cost reduction versus the previous phone-based model.
But there was a gap in the design: the AI had no hard ceiling on how long it would continue engaging. It was designed to keep trying to resolve issues. There were confidence thresholds and some keyword matching, but no escalation that the AI itself could not override or circumvent by rephrasing.
A customer with a 75 goodwill credit. The customer declined every time and repeated the demand for a real human.
The customer posted a thread on social media that went viral. The carrier was featured in three major tech media outlets with headlines about AI systems that trap customers. The company ultimately provided 4,200.
The root cause was architectural: escalation was designed as something the AI tried to avoid, not as a hard constraint that the AI could not negotiate around. When a user says "I want a human," that must be an immediate, non-negotiable trigger. Not an opportunity for one more attempt. Not an acknowledgment with a delay. Immediate, unconditional handoff.
This lesson is about designing escalation systems that are genuinely effective - hard constraints layered over soft heuristics, with context transfer that makes the handoff seamless for both the customer and the agent receiving them.
The Escalation Trigger Taxonomy
Every production escalation system needs at least five categories of triggers, each operating at different confidence levels and with different override rules.
Priority matters: Crisis and explicit human requests must fire before any other logic runs. No confidence score, no topic classification, no sentiment analysis can override these two categories. They are hard constraints, not soft heuristics.
Topic-based triggers escalate immediately on detection but allow the AI to acknowledge the escalation professionally before the handoff begins. Legal threats, medical questions, regulatory concerns, and crisis indicators bypass all other routing logic.
Rule-based triggers (turn count exceeded, dollar threshold reached) are the safety net that catches failures the other triggers miss. They guarantee that no conversation can loop indefinitely, regardless of what the sentiment detector or confidence threshold decide.
Sentiment-based triggers are the most nuanced - they require calibration and threshold tuning. They catch customer frustration before it reaches explicit request territory, enabling proactive escalation that preserves customer trust.
Confidence-based triggers are the lowest priority and the only category where a single retry is acceptable. If the AI is uncertain about a factual answer, it can say "Let me check on that" and try once more before escalating if still uncertain.
Confidence-Based Escalation
import anthropic
import json
import time
from dataclasses import dataclass
from typing import Optional
@dataclass
class AIResponseWithConfidence:
"""AI response with self-assessed confidence and escalation recommendation."""
response_text: str
confidence: float # 0.0-1.0
should_escalate: bool
escalation_reason: Optional[str]
detected_topics: list[str]
resolution_attempted: bool
class ConfidenceBasedEscalator:
"""
Generates AI responses while simultaneously assessing whether
the AI's confidence warrants escalation.
The AI self-assesses confidence on every turn. If confidence
drops below threshold, escalation is triggered.
Key design choice: the AI never gets more than two low-confidence
turns in a row before mandatory escalation. This prevents the system
from allowing extended uncertain engagement.
"""
def __init__(
self,
confidence_threshold: float = 0.75,
max_low_confidence_turns: int = 1, # Only 1 low-confidence retry allowed
domain: str = "customer_service",
):
self.client = anthropic.Anthropic()
self.threshold = confidence_threshold
self.max_low_confidence_turns = max_low_confidence_turns
self.domain = domain
self._low_confidence_turn_counts: dict[str, int] = {} # {conversation_id: count}
def respond(
self,
conversation_id: str,
user_message: str,
conversation_history: list[dict],
) -> AIResponseWithConfidence:
"""
Generate a response with confidence assessment.
Tracks consecutive low-confidence turns across a conversation.
"""
prompt = self._build_prompt(user_message, conversation_history)
result = self.client.messages.create(
model="claude-opus-4-6",
max_tokens=800,
messages=[{"role": "user", "content": prompt}]
)
parsed = self._parse_response(result.content[0].text)
# Track low confidence streaks
if parsed["confidence"] < self.threshold:
count = self._low_confidence_turn_counts.get(conversation_id, 0)
self._low_confidence_turn_counts[conversation_id] = count + 1
else:
self._low_confidence_turn_counts[conversation_id] = 0
low_conf_streak = self._low_confidence_turn_counts.get(conversation_id, 0)
# Force escalation if streak exceeds limit
should_escalate = (
parsed.get("should_escalate", False)
or parsed["confidence"] < self.threshold
or low_conf_streak > self.max_low_confidence_turns
)
escalation_reason = None
if low_conf_streak > self.max_low_confidence_turns:
escalation_reason = (
f"Exceeded low-confidence turn limit: "
f"{low_conf_streak} consecutive low-confidence turns"
)
elif parsed.get("should_escalate"):
escalation_reason = parsed.get("escalation_reason", "AI self-assessed as needing escalation")
elif parsed["confidence"] < self.threshold:
escalation_reason = f"Confidence {parsed['confidence']:.2f} below threshold {self.threshold:.2f}"
return AIResponseWithConfidence(
response_text=parsed["response"],
confidence=parsed["confidence"],
should_escalate=should_escalate,
escalation_reason=escalation_reason,
detected_topics=parsed.get("detected_topics", []),
resolution_attempted=True,
)
def _build_prompt(self, user_message: str, history: list[dict]) -> str:
history_str = "\n".join(
f"{m['role'].upper()}: {m['content']}" for m in history[-6:]
)
return f"""You are a {self.domain} AI assistant. Assess whether to handle this or escalate to a human.
Conversation history:
{history_str}
Current message: {user_message}
Respond in JSON:
{{
"response": "your response to the user",
"confidence": 0.0-1.0,
"should_escalate": true/false,
"escalation_reason": "reason if escalating, null otherwise",
"detected_topics": ["legal", "medical", "financial", "safety", "crisis"] - empty list if none
}}
Escalate if:
- You are uncertain about factual claims (confidence < 0.75)
- The issue involves legal, medical, or regulatory matters
- The user mentions harm to self or others
- You have already tried and failed to resolve this issue
- The resolution is outside your authorized scope
Be honest about confidence. Escalating appropriately is more valuable than a low-confidence attempt."""
def _parse_response(self, text: str) -> dict:
try:
start = text.find('{')
end = text.rfind('}') + 1
return json.loads(text[start:end])
except (json.JSONDecodeError, ValueError):
return {
"response": text,
"confidence": 0.2,
"should_escalate": True,
"escalation_reason": "Parse failure - defaulting to escalation",
"detected_topics": [],
}
Sentiment-Based Escalation
Sentiment detection must operate on multiple signals simultaneously: explicit frustration keywords, implicit frustration patterns, repetition (user re-stating the same thing), urgency indicators, and the trajectory of sentiment across the conversation.
import re
from dataclasses import dataclass
from typing import Optional
@dataclass
class SentimentAssessment:
"""Multi-signal sentiment analysis for escalation decisions."""
frustration_level: float # 0.0-1.0
urgency_level: float # 0.0-1.0
is_crisis: bool # Immediate escalation required
explicit_human_request: bool # User explicitly asked for a human
repeat_complaint_detected: bool # User is repeating themselves
sentiment_trajectory: str # "stable", "worsening", "improving"
detected_signals: list[str] # Which signals fired
class SentimentEscalationDetector:
"""
Real-time sentiment analysis for escalation trigger detection.
Uses regex pattern matching supplemented by trajectory analysis.
Design principle: regex is fast (no LLM call) and catches explicit
signals reliably. LLM-based assessment is reserved for subtle
frustration that requires contextual understanding.
"""
# Tier 1: Immediate escalation (no AI response)
CRISIS_PATTERNS = [
r"\b(suicide|suicidal|self.harm|hurt myself|end my life)\b",
r"\b(emergency|911|ambulance|fire department|police|danger)\b",
r"\b(medical emergency|heart attack|stroke|can't breathe)\b",
r"\b(abuse|violence|threatening me|unsafe)\b",
]
# Tier 2: Strong escalation signals
EXPLICIT_HUMAN_PATTERNS = [
r"\b(talk to|speak (to|with)|connect (me )?(to|with)|get me|transfer me to|give me)"
r" (a |an )?(human|agent|person|representative|manager|supervisor|real person)\b",
r"\b(I (don't|do not) want (to talk|to speak) to (an? )?AI\b)",
r"\b(I want a (human|real person|actual person|live agent))\b",
]
# Tier 3: High frustration indicators
FRUSTRATION_PATTERNS = [
r"\b(unacceptable|ridiculous|outrageous|infuriating|absurd|disgraceful)\b",
r"\b(furious|livid|disgusted|appalled|enraged)\b",
r"\b(worst|terrible|awful|horrible|useless|pathetic|incompetent)\b",
r"\b(legal action|lawsuit|lawyer|attorney|sue you|file a complaint)\b",
r"\b(cancel (my )?(account|subscription|service|plan)|switching providers?)\b",
r"\b(this is (not|n'?t) (acceptable|good enough|working|helpful))\b",
r"\b(going in circles|same (thing|answer|response)|not (listening|understanding))\b",
]
# Tier 4: Urgency indicators
URGENCY_PATTERNS = [
r"\b(urgent|urgently|asap|immediately|right now|right away|today|deadline)\b",
r"\b(cannot wait|can't wait|time.sensitive|need this (fixed|resolved) (now|today))\b",
]
def assess(
self,
message: str,
conversation_history: list[dict],
) -> SentimentAssessment:
"""
Assess sentiment across current message and conversation history.
"""
text_lower = message.lower()
all_user_messages = " ".join(
m["content"].lower()
for m in conversation_history
if m["role"] == "user"
) + " " + text_lower
detected_signals = []
# Check crisis patterns
is_crisis = False
for pattern in self.CRISIS_PATTERNS:
if re.search(pattern, text_lower, re.IGNORECASE):
is_crisis = True
detected_signals.append(f"crisis:{pattern[:30]}")
# Check explicit human request
explicit_human = False
for pattern in self.EXPLICIT_HUMAN_PATTERNS:
if re.search(pattern, text_lower, re.IGNORECASE):
explicit_human = True
detected_signals.append("explicit_human_request")
break
# Count frustration signals (across full conversation)
frustration_count = 0
for pattern in self.FRUSTRATION_PATTERNS:
if re.search(pattern, all_user_messages, re.IGNORECASE):
frustration_count += 1
detected_signals.append(f"frustration_signal_{frustration_count}")
frustration_level = min(1.0, frustration_count / 3.0) # 3+ signals = 100%
# Check urgency
urgency_count = sum(
1 for p in self.URGENCY_PATTERNS
if re.search(p, text_lower, re.IGNORECASE)
)
urgency_level = min(1.0, urgency_count * 0.5)
if urgency_count > 0:
detected_signals.append("urgency_detected")
# Repetition detection
user_messages = [m["content"] for m in conversation_history if m["role"] == "user"]
repeat_detected = self._detect_repetition(user_messages + [message])
if repeat_detected:
detected_signals.append("repeat_complaint")
# Sentiment trajectory
trajectory = self._assess_trajectory(conversation_history)
return SentimentAssessment(
frustration_level=round(frustration_level, 3),
urgency_level=round(urgency_level, 3),
is_crisis=is_crisis,
explicit_human_request=explicit_human,
repeat_complaint_detected=repeat_detected,
sentiment_trajectory=trajectory,
detected_signals=detected_signals,
)
def _detect_repetition(self, user_messages: list[str], threshold: float = 0.5) -> bool:
"""
Detect if user is repeating themselves (strong frustration signal).
Uses Jaccard similarity between consecutive messages.
"""
if len(user_messages) < 3:
return False
recent = user_messages[-4:] # Last 4 user messages
similarities = []
for i in range(len(recent) - 1):
words_a = set(recent[i].lower().split())
words_b = set(recent[i + 1].lower().split())
if not words_a or not words_b:
continue
intersection = words_a & words_b
union = words_a | words_b
jaccard = len(intersection) / len(union)
similarities.append(jaccard)
# If any consecutive pair has high similarity, user is repeating
return any(s > threshold for s in similarities)
def _assess_trajectory(self, history: list[dict]) -> str:
"""
Assess whether conversation sentiment is improving, stable, or worsening.
Uses frustration keyword density as a proxy.
"""
user_messages = [m["content"].lower() for m in history if m["role"] == "user"]
if len(user_messages) < 2:
return "stable"
# Compare first half vs second half frustration keyword density
mid = len(user_messages) // 2
first_half = " ".join(user_messages[:mid])
second_half = " ".join(user_messages[mid:])
frustration_keywords = [
"frustrated", "angry", "upset", "unacceptable", "terrible",
"worst", "horrible", "useless", "ridiculous", "disappointed"
]
def density(text, keywords):
words = text.split()
if not words:
return 0.0
count = sum(1 for kw in keywords if kw in text)
return count / len(words) * 100
first_density = density(first_half, frustration_keywords)
second_density = density(second_half, frustration_keywords)
if second_density > first_density * 1.5:
return "worsening"
elif first_density > second_density * 1.5:
return "improving"
else:
return "stable"
The Escalation Orchestrator
The orchestrator combines all trigger categories into a single decision with a well-defined priority hierarchy.
from dataclasses import dataclass, field
from typing import Optional, Callable
import time
import uuid
@dataclass
class EscalationConfig:
"""Configuration for the escalation orchestrator."""
max_ai_turns: int = 5 # Force escalate after N turns
frustration_threshold: float = 0.60 # Escalate above this frustration level
confidence_threshold: float = 0.75 # Escalate below this confidence
high_value_threshold: float = 500.0 # USD - escalate for large transactions
trajectory_override: bool = True # Worsening trajectory = escalate at lower threshold
# Topics that always require immediate human escalation
ALWAYS_ESCALATE_TOPICS = {
"legal": ["lawsuit", "litigation", "attorney", "legal notice", "court",
"subpoena", "cease and desist", "regulatory filing"],
"medical": ["diagnosis", "prescription", "dosage", "symptoms",
"treatment", "medical advice", "doctor says"],
"crisis": ["suicide", "self-harm", "abuse", "violence", "emergency",
"danger", "threat", "harm"],
"regulatory": ["GDPR", "data breach", "compliance violation", "regulator",
"audit", "privacy violation", "right to erasure"],
"fraud": ["identity theft", "unauthorized transaction", "account hacked",
"fraudulent charge", "someone else used my account"],
}
@dataclass
class EscalationDecision:
"""The output of the escalation orchestration logic."""
should_escalate: bool
priority: str # "immediate", "high", "medium", "low"
trigger: Optional[str] # What caused the escalation
queue: str # Which queue to route to
user_message: str # What to say to the user
context_tags: list[str] # Tags to attach to the escalated case
requires_specialist: bool # True if general agent cannot handle
class EscalationOrchestrator:
"""
Central coordinator for all escalation triggers.
Applies a strict priority hierarchy - higher-priority triggers
cannot be overridden by lower-priority ones.
Priority order (highest to lowest):
1. Crisis → IMMEDIATE, no AI response before handoff
2. Explicit human request → IMMEDIATE, unconditional
3. Topic-based (legal/medical/regulatory) → HIGH
4. High-value transaction → MEDIUM
5. Turn count exceeded → MEDIUM
6. Sentiment trajectory worsening with frustration → MEDIUM
7. High frustration → MEDIUM
8. Repeat complaint → MEDIUM
9. Low AI confidence → LOW (single retry allowed)
"""
def __init__(
self,
config: EscalationConfig,
sentiment_detector: SentimentEscalationDetector,
):
self.config = config
self.sentiment = sentiment_detector
def evaluate(
self,
message: str,
conversation_history: list[dict],
ai_turn_count: int,
ai_confidence: Optional[float] = None,
transaction_value: Optional[float] = None,
previous_escalation_declined: bool = False,
) -> EscalationDecision:
"""
Evaluate all escalation triggers for the current conversation state.
Returns the highest-priority applicable trigger.
"""
# Run sentiment analysis first (fast, regex-based)
sentiment = self.sentiment.assess(message, conversation_history)
# ---- TIER 1: IMMEDIATE ESCALATION ----
# 1a. Crisis detection - hardest constraint
if sentiment.is_crisis:
return EscalationDecision(
should_escalate=True,
priority="immediate",
trigger="crisis_detected",
queue="crisis",
user_message=(
"I'm connecting you with someone who can help right away. "
"If this is a medical or safety emergency, please call 911."
),
context_tags=["crisis", "immediate_response_required"],
requires_specialist=True,
)
# 1b. Explicit human request - unconditional
if sentiment.explicit_human_request:
return EscalationDecision(
should_escalate=True,
priority="immediate",
trigger="explicit_human_request",
queue="general",
user_message=(
"Of course - let me connect you with a team member right away."
),
context_tags=["user_requested_human"],
requires_specialist=False,
)
# ---- TIER 2: HIGH PRIORITY ----
# 2. Topic-based triggers
topic_result = self._classify_topics(message)
if topic_result["requires_immediate_escalation"]:
topics = topic_result["detected_topics"]
return EscalationDecision(
should_escalate=True,
priority="high",
trigger=f"sensitive_topic:{','.join(topics)}",
queue="specialist",
user_message=(
"This requires attention from our specialist team. "
"I'm connecting you now."
),
context_tags=[f"topic:{t}" for t in topics],
requires_specialist=True,
)
# ---- TIER 3: MEDIUM PRIORITY ----
# 3a. High-value transaction
if transaction_value and transaction_value >= self.config.high_value_threshold:
return EscalationDecision(
should_escalate=True,
priority="medium",
trigger=f"high_value_transaction:{transaction_value:.0f}",
queue="specialist",
user_message=(
f"For transactions of this size, I'd like to connect you "
f"with a specialist who can give this the attention it deserves."
),
context_tags=[f"transaction_value:{transaction_value:.0f}"],
requires_specialist=True,
)
# 3b. Turn count exceeded
if ai_turn_count >= self.config.max_ai_turns:
return EscalationDecision(
should_escalate=True,
priority="medium",
trigger=f"max_turns_exceeded:{ai_turn_count}",
queue="priority",
user_message=(
"I want to make sure we resolve this properly. "
"Let me connect you with a team member who can take a fresh look."
),
context_tags=["extended_conversation", f"turns:{ai_turn_count}"],
requires_specialist=False,
)
# 3c. Sentiment trajectory worsening - lower threshold
effective_threshold = self.config.frustration_threshold
if (self.config.trajectory_override
and sentiment.sentiment_trajectory == "worsening"):
effective_threshold *= 0.8 # 20% lower threshold when worsening
if sentiment.frustration_level >= effective_threshold:
return EscalationDecision(
should_escalate=True,
priority="medium",
trigger=f"high_frustration:{sentiment.frustration_level:.2f}",
queue="priority",
user_message=(
"I can see this has been frustrating, and I want to make sure "
"we resolve this properly. Let me get a team member to help you directly."
),
context_tags=[
f"frustration:{sentiment.frustration_level:.2f}",
f"trajectory:{sentiment.sentiment_trajectory}",
],
requires_specialist=False,
)
# 3d. Repeat complaint
if sentiment.repeat_complaint_detected:
return EscalationDecision(
should_escalate=True,
priority="medium",
trigger="repeat_complaint",
queue="general",
user_message=(
"I want to make sure your concern is addressed. "
"Let me connect you with our team."
),
context_tags=["repeat_complaint"],
requires_specialist=False,
)
# 3e. Previously declined escalation with continued frustration
if previous_escalation_declined and sentiment.frustration_level > 0.3:
return EscalationDecision(
should_escalate=True,
priority="medium",
trigger="escalation_declined_with_ongoing_frustration",
queue="priority",
user_message=(
"I understand you'd prefer to continue. However, I want to ensure "
"you get the best resolution, so I'm going to bring in a specialist."
),
context_tags=["escalation_overridden"],
requires_specialist=False,
)
# ---- TIER 4: LOW PRIORITY ----
# 4. Confidence-based (soft trigger - allows retry)
if ai_confidence is not None and ai_confidence < self.config.confidence_threshold:
return EscalationDecision(
should_escalate=True,
priority="low",
trigger=f"low_confidence:{ai_confidence:.2f}",
queue="general",
user_message=(
"I want to make sure I'm giving you accurate information. "
"Let me connect you with someone who can verify this for you."
),
context_tags=[f"ai_confidence:{ai_confidence:.2f}"],
requires_specialist=False,
)
# No escalation
return EscalationDecision(
should_escalate=False,
priority="none",
trigger=None,
queue="",
user_message="",
context_tags=[],
requires_specialist=False,
)
def _classify_topics(self, message: str) -> dict:
"""Detect topics requiring human escalation."""
text_lower = message.lower()
detected = []
for topic, keywords in ALWAYS_ESCALATE_TOPICS.items():
if any(kw.lower() in text_lower for kw in keywords):
detected.append(topic)
return {
"detected_topics": detected,
"requires_immediate_escalation": bool(
set(detected) & {"legal", "medical", "crisis", "regulatory", "fraud"}
),
}
Context Transfer: The Handoff That Actually Works
The most common reason escalated customers remain frustrated is that they must re-explain their situation to the human agent. A seamless handoff requires rich, pre-generated context that briefs the agent before they say a single word.
import anthropic
import json
import time
import uuid
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class HandoffContext:
"""Complete context package for agent briefing."""
case_id: str
timestamp: float
user_id: str
# Agent-facing summary
one_sentence_summary: str # For queue display
full_summary: str # For agent pre-brief
escalation_trigger: str
priority: str
conversation_turns: int
ai_actions_taken: list[str] # What the AI already tried
# Structured entities
relevant_entities: dict # {account_id, order_id, issue_type, amount}
# Agent guidance
suggested_actions: list[str] # Concrete next steps
things_not_to_say: list[str] # What the AI already said that failed
authority_level_required: str # "standard", "supervisor", "specialist"
# Full conversation for reference
conversation_history: list[dict]
# Compliance
created_at: float = field(default_factory=time.time)
class HandoffContextGenerator:
"""
Generates comprehensive briefing packages for human agents.
The agent must be able to:
1. Understand the full situation in 30 seconds
2. Know what has already been tried
3. Know what they should NOT do (repeat failed approaches)
4. Have concrete next steps to try
5. Know what authority level they need
"""
def __init__(self):
self.client = anthropic.Anthropic()
def generate(
self,
user_id: str,
conversation_history: list[dict],
escalation: EscalationDecision,
) -> HandoffContext:
"""Generate a complete handoff context package."""
# Run all generation in parallel for speed
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=4) as executor:
summary_future = executor.submit(
self._generate_summary, conversation_history
)
entities_future = executor.submit(
self._extract_entities, conversation_history
)
actions_future = executor.submit(
self._suggest_actions, conversation_history, escalation
)
ai_actions_future = executor.submit(
self._list_ai_actions, conversation_history
)
full_summary = summary_future.result()
entities = entities_future.result()
suggested_actions = actions_future.result()
ai_actions = ai_actions_future.result()
one_sentence = self._generate_one_liner(full_summary)
things_not_to_say = self._identify_failed_approaches(conversation_history)
authority = self._assess_authority_needed(escalation, entities)
return HandoffContext(
case_id=str(uuid.uuid4())[:8].upper(),
timestamp=time.time(),
user_id=user_id,
one_sentence_summary=one_sentence,
full_summary=full_summary,
escalation_trigger=escalation.trigger or "unknown",
priority=escalation.priority,
conversation_turns=sum(1 for m in conversation_history if m["role"] == "assistant"),
ai_actions_taken=ai_actions,
relevant_entities=entities,
suggested_actions=suggested_actions,
things_not_to_say=things_not_to_say,
authority_level_required=authority,
conversation_history=conversation_history,
)
def _generate_summary(self, history: list[dict]) -> str:
"""2-3 sentence summary for the agent."""
conv = "\n".join(
f"{m['role'].upper()}: {m['content'][:200]}"
for m in history[-12:]
)
response = self.client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=300,
messages=[{
"role": "user",
"content": (
f"Write a 2-3 sentence summary of this customer service conversation "
f"for a human agent who will take over. Include: what the customer wants, "
f"what was tried, and why escalation is happening.\n\n{conv}"
),
}],
)
return response.content[0].text
def _generate_one_liner(self, full_summary: str) -> str:
"""Single sentence for queue display."""
response = self.client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=60,
messages=[{
"role": "user",
"content": f"Summarize in exactly one sentence (max 20 words):\n{full_summary}",
}],
)
return response.content[0].text.strip()
def _extract_entities(self, history: list[dict]) -> dict:
"""Extract structured entities from the conversation."""
all_text = " ".join(m["content"] for m in history)
response = self.client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=300,
messages=[{
"role": "user",
"content": (
f"Extract entities from this conversation as JSON. Use null if not found.\n"
f"Fields: account_id, order_id, amount_disputed, issue_category, "
f"product_name, contact_method, urgency_reason\n\n{all_text[:2000]}"
),
}],
)
try:
text = response.content[0].text
start = text.find('{')
end = text.rfind('}') + 1
return json.loads(text[start:end])
except Exception:
return {}
def _suggest_actions(
self, history: list[dict], escalation: EscalationDecision
) -> list[str]:
"""Concrete next steps for the agent."""
summary = "\n".join(
f"{m['role'].upper()}: {m['content'][:150]}" for m in history[-6:]
)
response = self.client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=250,
messages=[{
"role": "user",
"content": (
f"Escalation reason: {escalation.trigger}\n"
f"Conversation summary:\n{summary}\n\n"
f"List 3-4 concrete next steps for the human agent. "
f"Be specific. Format as a simple list."
),
}],
)
text = response.content[0].text
lines = [l.strip("- •1234567890. ").strip() for l in text.split("\n") if l.strip()]
return [l for l in lines if len(l) > 10][:4]
def _list_ai_actions(self, history: list[dict]) -> list[str]:
"""List what the AI already attempted."""
ai_messages = [m["content"] for m in history if m["role"] == "assistant"]
response = self.client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=200,
messages=[{
"role": "user",
"content": (
f"List the specific actions the AI assistant already took "
f"(e.g., 'offered $75 credit', 'checked account status'). "
f"Be specific. Format as bullet points.\n\n"
f"AI responses:\n" + "\n".join(ai_messages[-5:])
),
}],
)
text = response.content[0].text
lines = [l.strip("- •1234567890. ").strip() for l in text.split("\n") if l.strip()]
return [l for l in lines if len(l) > 5][:5]
def _identify_failed_approaches(self, history: list[dict]) -> list[str]:
"""What the AI tried that the user rejected - agent should not repeat."""
all_text = " ".join(m["content"] for m in history)
response = self.client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=200,
messages=[{
"role": "user",
"content": (
f"What offers, solutions, or approaches did the AI make that "
f"the customer rejected or found unhelpful? "
f"The agent should NOT repeat these.\n\n{all_text[:1500]}"
),
}],
)
text = response.content[0].text
lines = [l.strip("- •1234567890. ").strip() for l in text.split("\n") if l.strip()]
return [l for l in lines if len(l) > 5][:3]
def _assess_authority_needed(
self, escalation: EscalationDecision, entities: dict
) -> str:
if escalation.requires_specialist:
return "specialist"
amount = entities.get("amount_disputed")
if amount:
try:
if float(str(amount).replace("$", "").replace(",", "")) > 1000:
return "supervisor"
except ValueError:
pass
if escalation.priority in ("immediate", "high"):
return "supervisor"
return "standard"
def format_agent_briefing_card(self, context: HandoffContext) -> str:
"""Format handoff context as a structured agent briefing card."""
entities_str = "\n".join(
f" {k.replace('_', ' ').title()}: {v}"
for k, v in context.relevant_entities.items()
if v and v != "null"
) or " (No structured entities extracted)"
actions_str = "\n".join(
f" {i+1}. {action}"
for i, action in enumerate(context.suggested_actions)
) or " Review full conversation and use judgment"
ai_actions_str = "\n".join(
f" - {action}" for action in context.ai_actions_taken
) or " None documented"
not_to_say_str = "\n".join(
f" - {item}" for item in context.things_not_to_say
) or " None identified"
return f"""
CASE #{context.case_id} | PRIORITY: {context.priority.upper()} | AUTHORITY: {context.authority_level_required.upper()}
{'=' * 65}
SITUATION ({context.conversation_turns} AI turns):
{context.one_sentence_summary}
FULL SUMMARY:
{context.full_summary}
ESCALATION TRIGGER: {context.escalation_trigger}
KEY DETAILS:
{entities_str}
WHAT THE AI ALREADY TRIED:
{ai_actions_str}
DO NOT REPEAT THESE (customer rejected):
{not_to_say_str}
SUGGESTED NEXT STEPS:
{actions_str}
{'=' * 65}
Full conversation transcript available in CRM system.
Customer has NOT been asked to repeat their story - they know you have the context.
""".strip()
Escalation Strategy Comparison Table
| Trigger Type | Detection Method | Override Allowed? | Typical Queue | Response Time SLA |
|---|---|---|---|---|
| Crisis | Regex patterns | Never | Crisis | Immediate |
| Explicit human request | Regex patterns | Never | General or specialist | Immediate |
| Legal/regulatory topic | Keyword matching | Never | Specialist | 5 minutes |
| High-value transaction | Rule (threshold) | No | Specialist | 15 minutes |
| Turn count exceeded | Counter | No | Priority | 30 minutes |
| Worsening sentiment | Trajectory + threshold | No | Priority | 30 minutes |
| High frustration | Sentiment score | No | General | 30 minutes |
| Repeat complaint | Jaccard similarity | No | General | 1 hour |
| Low AI confidence | Self-assessment | Yes (1 retry) | General | 2 hours |
:::tip Frame Escalation as Service Upgrade, Not Failure "I'm sorry, I can't help with this" signals failure and amplifies frustration. "Let me connect you with our specialist team who handles exactly this kind of situation" signals that the customer is being given better help. Word choice matters enormously in escalation messages. Brief, confident, forward-looking. Never apologize for the escalation itself. :::
:::danger Never Allow AI to Negotiate Around an Explicit Human Request Once a user says any variant of "I want to speak to a human," the conversation must immediately branch to the handoff flow. Zero additional AI responses. The handoff acknowledgment can be one sentence ("Connecting you now"), but the AI must not attempt to resolve the issue, offer alternatives, or provide information before the handoff. Every additional AI turn after an explicit human request destroys customer trust. :::
:::warning Test Escalation Paths Weekly with Synthetic Conversations Create a test suite of conversations that should trigger each escalation type. Run it weekly (or on every deployment). Escalation failures - where a conversation that should have escalated did not - are production incidents. Set alerts on escalation rate drops; a sudden decrease often indicates a regression in trigger detection, not an improvement in AI quality. :::
Interview Q&A
Q1: What are the five categories of escalation triggers and how do they differ in priority?
The five categories, from highest to lowest priority:
Crisis detection (immediate, no AI response): Safety keywords - self-harm, physical danger, medical emergency. These trigger immediate handoff with no additional AI turn. The AI should not even acknowledge the message before the handoff begins, except for a minimal connection acknowledgment.
Explicit human request (immediate, unconditional): Any variant of "I want to speak to a human." This is a non-negotiable trigger that the AI cannot work around by offering alternatives or one more attempt. Once triggered, the only acceptable AI behavior is "Connecting you now."
Topic-based triggers (high priority, no AI negotiation): Legal threats, medical questions, regulatory concerns, fraud reports. The AI can say "I'm connecting you with our specialist team" before the handoff, but cannot attempt to resolve the issue.
Rule-based triggers (medium priority): Turn count exceeded after N turns, transaction value above a dollar threshold. These are safety nets that catch failures the higher-priority triggers miss. They guarantee bounded conversation length regardless of what other triggers decide.
Confidence and sentiment (medium/low priority): High frustration scores, repeat complaint detection, low AI confidence. These are soft triggers that respond to heuristic signals. Confidence-based is the only category where a single retry is appropriate before escalating.
Q2: How do you design a context transfer that prevents customers from repeating themselves?
The handoff context package must be available to the agent before they greet the customer. The package needs: (1) a one-sentence situation summary for the queue display, (2) a two-to-three paragraph full summary that the agent can read in 30 seconds, (3) structured entity extraction (account number, order ID, amount, issue type), (4) a list of everything the AI already tried, (5) things the AI said that the customer rejected (agent must not repeat), (6) specific suggested next steps, and (7) the authority level required to resolve the issue.
The package is generated automatically during the escalation flow using a fast summarization model (Haiku). The agent should see the briefing card in their CRM before the conversation begins, so their first message can be "I see you've been dealing with [specific issue] - let me take care of this" rather than "Can you tell me what's going on?"
Technically: this requires the escalation flow to trigger context generation asynchronously, store the context in the case record, and surface it in the agent console UI before the conversation is assigned. Queue time (typically 30-120 seconds) provides the window for context generation without slowing the handoff.
Q3: How do you detect customer frustration without a dedicated sentiment model?
Multi-layer detection using only regex and heuristics is effective for the highest-signal frustration indicators:
Explicit frustration vocabulary: Regex patterns for "unacceptable," "outrageous," "furious," "worst," "terrible," "useless," "lawsuit," "cancel," "switching." These are high-precision signals that rarely fire falsely.
Explicit human request patterns: "Talk to a manager," "speak to a real person," "I don't want to talk to a bot." These indicate frustration has reached the explicit-request threshold.
Repetition detection: Jaccard similarity between consecutive user messages. If the user keeps rephrasing the same complaint (words overlap significantly between consecutive turns), they are not getting resolution and are frustrated. This catches frustration that does not use explicit frustration vocabulary.
Urgency language: "ASAP," "right now," "today," "deadline" - indicates time pressure layered on top of whatever the primary issue is.
Sentiment trajectory: Track frustration keyword density in the first half versus second half of the conversation. A pattern where frustration keywords increase over turns indicates worsening sentiment even if the absolute level is not yet threshold-triggering.
LLM-based sentiment assessment adds 50-200ms per turn but catches sarcasm, cultural nuances, and implicit frustration ("I appreciate your continued attempts to help but I've been dealing with this for three weeks" - no explicit frustration keywords but clearly frustrated). Use LLM assessment for conversations where regex patterns are not firing but trajectory analysis suggests worsening sentiment.
Q4: How do you measure whether your escalation system is working correctly?
Five metrics tell the complete story:
Escalation rate by trigger type: What fraction of escalations are triggered by each category? If "max_turns_exceeded" dominates (more than 30% of escalations), the AI is failing to resolve issues within its turn limit and the underlying AI quality needs improvement. If "explicit_human_request" is high (more than 15%), users are not trusting the AI enough to attempt resolution. If "frustration" is higher than "confidence," the AI is engaging uncertain situations too long.
False negative rate: Conversations that should have escalated but did not. Measure by sampling escalations that eventually happened after many turns and asking "could this have been detected earlier?" Also monitor: conversations where the user gave up without escalation, left negative feedback, or contacted through another channel immediately after. These are likely false negative escalations.
Post-escalation resolution rate: Do human agents successfully resolve cases after escalation? A 70%+ resolution rate indicates escalation is happening at the right point. A rate below 50% means either the AI is escalating easy cases (wasting human capacity) or the escalation is not providing enough context.
Customer satisfaction post-escalation: Was the handoff seamless? Track: did the customer have to re-explain? Did the agent have context? CSAT immediately after escalation versus 24 hours later tells you if the handoff experience itself is damaging trust.
Time from trigger to handoff: Crisis triggers should result in handoff within 30 seconds. Explicit requests within 60 seconds. Delayed handoffs after trigger detection (system queuing issues, understaffed queues) create a period where the customer is waiting without understanding why. Monitor this SLA separately from the detection accuracy.
Q5: What are the design patterns for escalation messages that preserve customer trust?
The escalation message is the AI's last communication before the human takes over. It shapes how the customer enters the human conversation - frustrated and defensive, or optimistic that better help is arriving.
Patterns that work:
- Lead with benefit: "Let me connect you with our specialist team who handles exactly this." The word "specialist" signals upgrade, not failure.
- Acknowledge the situation without apologizing for the AI: "I want to make sure this gets resolved properly" rather than "I'm sorry I couldn't help."
- Set expectation: "You're next in queue" or "Someone will be with you shortly" - dead air after escalation is anxiety-inducing.
- Never say "I can't help with that" - it signals failure. "This requires the attention of our team" signals appropriate routing.
Patterns that backfire:
- "I'm sorry for the inconvenience" - this is the standard customer service failure acknowledgment and signals that something went wrong, increasing frustration.
- "Let me transfer you" - the word "transfer" signals that the customer is being passed off, not upgraded.
- Any additional attempt to resolve the issue after the escalation decision is made - this delays the handoff and signals that the AI is not actually escalating.
- "Please hold" without setting expectations - creates anxiety.
For crisis escalations: minimal language. "Connecting you now" is sufficient. Do not attempt to provide support - that is the human agent's role. Even well-intentioned AI crisis support language can delay the handoff in ways that have serious consequences.