Spaces:
Sleeping
Sleeping
File size: 9,404 Bytes
b33a861 964103b b33a861 964103b b33a861 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 | """
Escalation Engine for the Smart Escalation API.
This module provides the EscalationEngine class that:
- Evaluates retrieval quality using relevance scores
- Decides whether to answer or escalate to human agents
- Generates confidence explanations for decisions
- Coordinates LLM calls when answering questions
"""
from typing import List, Literal, Optional
from dataclasses import dataclass
from src.llm_client import LLMClient, LLMResponse
from src.rag import RetrievedChunk
@dataclass
class EscalationDecision:
"""Decision result from the escalation engine."""
action: Literal["answer", "escalation"]
message: str
confidence_explanation: str
sources: Optional[List[str]] = None
class EscalationEngine:
"""
Engine that decides whether to answer customer questions or escalate to human agents.
The engine uses multiple signals to make escalation decisions:
- Relevance scores from retrieval
- LLM uncertainty signals
- Coverage diversity of retrieved chunks
"""
def __init__(
self,
llm_client: LLMClient,
relevance_threshold: float = 0.5,
min_chunks_for_answer: int = 1
):
"""
Initialize the escalation engine.
Args:
llm_client: LLM client for generating answers
relevance_threshold: Minimum relevance score to consider answering (default: 0.5)
min_chunks_for_answer: Minimum number of chunks needed to attempt answering
"""
self.llm_client = llm_client
self.relevance_threshold = relevance_threshold
self.min_chunks_for_answer = min_chunks_for_answer
def _should_escalate_based_on_retrieval(
self,
retrieved_chunks: List[RetrievedChunk]
) -> tuple[bool, str]:
"""
Determine if escalation is needed based on retrieval quality.
Args:
retrieved_chunks: Chunks retrieved from RAG system
Returns:
Tuple of (should_escalate, reason)
"""
# No chunks retrieved
if not retrieved_chunks:
return True, "No relevant information found in help articles"
# Insufficient chunks
if len(retrieved_chunks) < self.min_chunks_for_answer:
return True, f"Only {len(retrieved_chunks)} relevant chunk(s) found, insufficient for confident answer"
# Check best relevance score
max_score = max(chunk.score for chunk in retrieved_chunks)
if max_score < self.relevance_threshold:
return True, f"Best retrieval score was {max_score:.2f}, below confidence threshold of {self.relevance_threshold}"
# All checks passed
return False, ""
def _generate_confidence_explanation(
self,
action: Literal["answer", "escalation"],
retrieved_chunks: List[RetrievedChunk],
llm_uncertain: bool = False,
retrieval_reason: str = ""
) -> str:
"""
Generate human-readable confidence explanation.
Args:
action: Whether answering or escalating
retrieved_chunks: Retrieved chunks from RAG
llm_uncertain: Whether LLM signaled uncertainty
retrieval_reason: Reason for retrieval-based escalation
Returns:
Confidence explanation string
"""
if action == "escalation":
if llm_uncertain:
return "LLM indicated uncertainty in answering based on available context"
elif retrieval_reason:
return retrieval_reason
else:
return "Insufficient confidence to provide accurate answer"
# For answers, provide positive confidence signal
if not retrieved_chunks:
return "Answer generated with available context"
max_score = max(chunk.score for chunk in retrieved_chunks)
sources = list(set(chunk.source for chunk in retrieved_chunks))
if len(sources) == 1:
return f"High confidence - retrieved relevant content from '{sources[0]}' with similarity score {max_score:.2f}"
else:
return f"High confidence - retrieved relevant content from {len(sources)} help articles with best similarity score {max_score:.2f}"
def _extract_sources(self, retrieved_chunks: List[RetrievedChunk]) -> List[str]:
"""
Extract unique source article names from retrieved chunks.
Args:
retrieved_chunks: Retrieved chunks from RAG
Returns:
List of unique source filenames
"""
if not retrieved_chunks:
return []
sources = list(set(chunk.source for chunk in retrieved_chunks))
return sorted(sources)
def process_question(
self,
question: str,
retrieved_chunks: List[RetrievedChunk]
) -> EscalationDecision:
"""
Process a customer question and decide whether to answer or escalate.
This is the main entry point for the escalation engine. It:
1. Evaluates retrieval quality
2. If quality is sufficient, calls LLM to generate answer
3. Checks LLM response for uncertainty signals
4. Returns decision with confidence explanation
Args:
question: Customer question
retrieved_chunks: Retrieved chunks from RAG system
Returns:
EscalationDecision with action, message, and explanation
Example:
>>> engine = EscalationEngine(llm_client)
>>> decision = engine.process_question("How do I reset password?", chunks)
>>> decision.action
'answer'
>>> decision.message
'To reset your password, go to...'
"""
# First check: Should we escalate based on retrieval quality?
should_escalate, retrieval_reason = self._should_escalate_based_on_retrieval(
retrieved_chunks
)
if should_escalate:
# Escalate due to poor retrieval
return EscalationDecision(
action="escalation",
message="I'm not certain I can answer this accurately. Let me connect you with a human support agent who can help.",
confidence_explanation=self._generate_confidence_explanation(
action="escalation",
retrieved_chunks=retrieved_chunks,
retrieval_reason=retrieval_reason
),
sources=None
)
# Retrieval quality is sufficient, attempt to generate answer
try:
# Extract chunk contents for LLM
chunk_contents = [chunk.content for chunk in retrieved_chunks]
# Generate answer using LLM
llm_response: LLMResponse = self.llm_client.generate_answer(
question=question,
context_chunks=chunk_contents
)
# Second check: Did LLM signal uncertainty?
if llm_response.uncertain:
# Escalate due to LLM uncertainty
return EscalationDecision(
action="escalation",
message="I'm not certain I can answer this accurately. Let me connect you with a human support agent who can help.",
confidence_explanation=self._generate_confidence_explanation(
action="escalation",
retrieved_chunks=retrieved_chunks,
llm_uncertain=True
),
sources=None
)
# All checks passed - return answer
return EscalationDecision(
action="answer",
message=llm_response.answer,
confidence_explanation=self._generate_confidence_explanation(
action="answer",
retrieved_chunks=retrieved_chunks
),
sources=self._extract_sources(retrieved_chunks)
)
except Exception as e:
# If LLM call fails, escalate gracefully
error_msg = str(e)
# Log the actual error for debugging
print(f"ERROR in escalation engine: {error_msg}")
import traceback
traceback.print_exc()
# Provide specific error context in explanation
if "timeout" in error_msg.lower():
explanation = "Unable to generate response due to timeout, escalating to human agent"
elif "rate limit" in error_msg.lower():
explanation = "Service temporarily busy, escalating to human agent"
else:
explanation = f"Unable to generate response, escalating to human agent (Error: {error_msg})"
return EscalationDecision(
action="escalation",
message="I'm having trouble processing your question right now. Let me connect you with a human support agent who can help.",
confidence_explanation=explanation,
sources=None
)
|