"""
Pattern 140: Escalation Pattern
Description:
The Escalation Pattern implements mechanisms for identifying situations that
require human intervention or higher-level decision-making and routing them
appropriately through an escalation hierarchy. This pattern recognizes when an
agent encounters scenarios beyond its capabilities, authority, or confidence
level and escalates to appropriate handlers based on severity and context.
Escalation is critical for AI safety and reliability, ensuring that complex,
high-stakes, or ambiguous situations receive appropriate human oversight. The
pattern supports multiple escalation levels, severity assessment, automatic
triggering based on rules or confidence thresholds, and routing to appropriate
stakeholders. It balances agent autonomy with human control.
This implementation provides a comprehensive escalation system with severity
assessment, trigger conditions, escalation routing, stakeholder management,
escalation tracking, and resolution workflows. It supports both rule-based
and confidence-based escalation with customizable escalation paths.
Components:
- Severity Assessment: Evaluate situation severity
- Trigger Detection: Identify when to escalate
- Escalation Routing: Route to appropriate handler
- Stakeholder Management: Manage escalation recipients
- Escalation Tracking: Monitor escalation status
- Resolution Workflow: Handle escalation lifecycle
Use Cases:
- High-stakes decisions (medical, financial, legal)
- Low-confidence predictions
- Policy violations
- Security incidents
- Customer complaints
- System errors
- Ethical dilemmas
- Edge cases
LangChain Implementation:
This implementation uses:
- LLM for severity assessment and decision complexity analysis
- Rule-based escalation triggers
- Confidence-threshold escalation
- Hierarchical escalation routing
- Escalation event tracking
- Integration with human-in-the-loop systems
Benefits:
- Ensures human oversight for critical decisions
- Prevents autonomous errors in high-stakes scenarios
- Provides safety net for uncertain situations
- Enables graduated response based on severity
- Supports compliance requirements
- Facilitates learning from escalations
- Maintains audit trail
Trade-offs:
- Reduces full autonomy
- Introduces latency for escalated cases
- Requires human availability
- May create bottlenecks
- Needs clear escalation criteria
- Can interrupt workflow
Production Considerations:
- Define clear escalation criteria
- Set appropriate confidence thresholds
- Implement escalation routing rules
- Ensure stakeholder availability
- Track escalation metrics
- Provide escalation status updates
- Support escalation history
- Enable escalation feedback
- Monitor escalation patterns
- Reduce false escalations
- Support emergency escalations
- Document escalation procedures
"""
import os
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
from enum import Enum
from dataclasses import dataclass, field
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
load_dotenv()
class EscalationLevel(Enum):
"""Escalation severity levels."""
CRITICAL = "critical"
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
class EscalationStatus(Enum):
"""Status of escalation."""
PENDING = "pending"
IN_PROGRESS = "in_progress"
RESOLVED = "resolved"
CANCELLED = "cancelled"
class EscalationTrigger(Enum):
"""Reasons for escalation."""
LOW_CONFIDENCE = "low_confidence"
POLICY_VIOLATION = "policy_violation"
HIGH_RISK = "high_risk"
SECURITY_INCIDENT = "security_incident"
ERROR = "error"
AMBIGUOUS = "ambiguous"
SENSITIVE_DATA = "sensitive_data"
MANUAL_REQUEST = "manual_request"
THRESHOLD_EXCEEDED = "threshold_exceeded"
@dataclass
class Stakeholder:
"""Escalation stakeholder."""
stakeholder_id: str
name: str
role: str
email: str
escalation_levels: List[EscalationLevel]
availability: bool = True
priority: int = 0
@dataclass
class EscalationCase:
"""Escalation case record."""
case_id: str
level: EscalationLevel
trigger: EscalationTrigger
title: str
description: str
context: Dict[str, Any]
agent_id: str
confidence_score: Optional[float]
created_at: datetime
assigned_to: Optional[str] = None
status: EscalationStatus = EscalationStatus.PENDING
resolution: Optional[str] = None
resolved_at: Optional[datetime] = None
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class EscalationRule:
"""Rule for triggering escalation."""
rule_id: str
name: str
condition: str # e.g., "confidence None:
"""Add escalation stakeholder."""
self.stakeholders[stakeholder.stakeholder_id] = stakeholder
print(f"Added stakeholder: {stakeholder.name} ({stakeholder.role})")
def add_rule(self, rule: EscalationRule) -> None:
"""Add escalation rule."""
self.rules[rule.rule_id] = rule
print(f"Added escalation rule: {rule.name}")
def should_escalate(
self,
context: Dict[str, Any],
confidence: Optional[float] = None
) -> tuple[bool, Optional[EscalationTrigger], Optional[EscalationLevel]]:
"""
Determine if situation should be escalated.
Returns:
Tuple of (should_escalate, trigger, level)
"""
# Check confidence threshold
if confidence is not None and confidence bool:
"""Evaluate if rule condition is met."""
condition = rule.condition.lower()
# Simple condition evaluation
if "sensitive_data" in condition:
return context.get("has_sensitive_data", False)
if "high_risk" in condition:
risk_score = context.get("risk_score", 0)
return risk_score > self.risk_threshold
if "error" in condition:
return context.get("error_occurred", False)
return False
def create_escalation(
self,
title: str,
description: str,
context: Dict[str, Any],
trigger: EscalationTrigger,
level: Optional[EscalationLevel] = None,
confidence: Optional[float] = None
) -> EscalationCase:
"""Create an escalation case."""
# Assess severity if not provided
if level is None:
level = self._assess_severity(title, description, context, trigger)
case = EscalationCase(
case_id=f"esc_{len(self.cases) + 1}_{datetime.now().timestamp()}",
level=level,
trigger=trigger,
title=title,
description=description,
context=context,
agent_id=self.agent_id,
confidence_score=confidence,
created_at=datetime.now()
)
# Route to stakeholder
stakeholder = self._route_escalation(case)
if stakeholder:
case.assigned_to = stakeholder.stakeholder_id
self.cases.append(case)
return case
def _assess_severity(
self,
title: str,
description: str,
context: Dict[str, Any],
trigger: EscalationTrigger
) -> EscalationLevel:
"""Assess severity using LLM."""
try:
result = self.severity_chain.invoke({
"title": title,
"description": description,
"context": str(context),
"trigger": trigger.value
})
# Parse severity
for line in result.split('\n'):
if line.startswith("SEVERITY:"):
severity = line.replace("SEVERITY:", "").strip().lower()
if severity == "critical":
return EscalationLevel.CRITICAL
elif severity == "high":
return EscalationLevel.HIGH
elif severity == "medium":
return EscalationLevel.MEDIUM
else:
return EscalationLevel.LOW
return EscalationLevel.MEDIUM
except Exception as e:
print(f"Error assessing severity: {e}")
return EscalationLevel.MEDIUM
def _route_escalation(
self,
case: EscalationCase
) -> Optional[Stakeholder]:
"""Route escalation to appropriate stakeholder."""
# Find available stakeholders for this level
candidates = [
s for s in self.stakeholders.values()
if case.level in s.escalation_levels and s.availability
]
if not candidates:
return None
# Sort by priority and return highest
candidates.sort(key=lambda s: s.priority, reverse=True)
return candidates[0]
def update_case_status(
self,
case_id: str,
status: EscalationStatus,
resolution: Optional[str] = None
) -> bool:
"""Update escalation case status."""
case = next((c for c in self.cases if c.case_id == case_id), None)
if not case:
return False
case.status = status
if resolution:
case.resolution = resolution
if status == EscalationStatus.RESOLVED:
case.resolved_at = datetime.now()
return True
def get_pending_escalations(
self,
level: Optional[EscalationLevel] = None,
stakeholder_id: Optional[str] = None
) -> List[EscalationCase]:
"""Get pending escalation cases."""
pending = [c for c in self.cases if c.status == EscalationStatus.PENDING]
if level:
pending = [c for c in pending if c.level == level]
if stakeholder_id:
pending = [c for c in pending if c.assigned_to == stakeholder_id]
return pending
def get_escalation_metrics(
self,
days: int = 7
) -> Dict[str, Any]:
"""Get escalation metrics."""
cutoff = datetime.now() - timedelta(days=days)
recent_cases = [c for c in self.cases if c.created_at >= cutoff]
# Resolution time for resolved cases
resolved = [c for c in recent_cases if c.status == EscalationStatus.RESOLVED]
avg_resolution_time = None
if resolved:
times = [
(c.resolved_at - c.created_at).total_seconds() / 3600
for c in resolved if c.resolved_at
]
avg_resolution_time = sum(times) / len(times) if times else None
return {
"period_days": days,
"total_escalations": len(recent_cases),
"pending": len([c for c in recent_cases if c.status == EscalationStatus.PENDING]),
"in_progress": len([c for c in recent_cases if c.status == EscalationStatus.IN_PROGRESS]),
"resolved": len(resolved),
"by_level": {
level.value: len([c for c in recent_cases if c.level == level])
for level in EscalationLevel
},
"by_trigger": {
trigger.value: len([c for c in recent_cases if c.trigger == trigger])
for trigger in EscalationTrigger
},
"avg_resolution_hours": avg_resolution_time
}
def demonstrate_escalation():
"""Demonstrate the escalation pattern."""
print("=" * 80)
print("Escalation Pattern Demonstration")
print("=" * 80)
agent = EscalationAgent(agent_id="agent_001")
# Demonstration 1: Add Stakeholders
print("\n" + "=" * 80)
print("Demonstration 1: Configure Escalation Stakeholders")
print("=" * 80)
support = Stakeholder(
stakeholder_id="stake_001",
name="Support Team",
role="support",
email="support@company.com",
escalation_levels=[EscalationLevel.LOW, EscalationLevel.MEDIUM],
priority=10
)
manager = Stakeholder(
stakeholder_id="stake_002",
name="Engineering Manager",
role="manager",
email="manager@company.com",
escalation_levels=[EscalationLevel.MEDIUM, EscalationLevel.HIGH],
priority=20
)
director = Stakeholder(
stakeholder_id="stake_003",
name="Director",
role="director",
email="director@company.com",
escalation_levels=[EscalationLevel.HIGH, EscalationLevel.CRITICAL],
priority=30
)
agent.add_stakeholder(support)
agent.add_stakeholder(manager)
agent.add_stakeholder(director)
# Demonstration 2: Add Escalation Rules
print("\n" + "=" * 80)
print("Demonstration 2: Define Escalation Rules")
print("=" * 80)
sensitive_rule = EscalationRule(
rule_id="rule_001",
name="Sensitive Data Rule",
condition="contains sensitive_data",
level=EscalationLevel.HIGH,
trigger=EscalationTrigger.SENSITIVE_DATA
)
risk_rule = EscalationRule(
rule_id="rule_002",
name="High Risk Rule",
condition="high_risk",
level=EscalationLevel.HIGH,
trigger=EscalationTrigger.HIGH_RISK
)
agent.add_rule(sensitive_rule)
agent.add_rule(risk_rule)
# Demonstration 3: Low Confidence Escalation
print("\n" + "=" * 80)
print("Demonstration 3: Low Confidence Escalation")
print("=" * 80)
should_escalate, trigger, level = agent.should_escalate(
context={"task": "medical_diagnosis"},
confidence=0.45
)
print(f"Should escalate: {should_escalate}")
print(f"Trigger: {trigger.value if trigger else None}")
print(f"Level: {level.value if level else None}")
if should_escalate:
case1 = agent.create_escalation(
title="Low Confidence Medical Diagnosis",
description="Agent confidence below threshold for medical diagnosis",
context={"task": "medical_diagnosis", "symptoms": "fever, cough"},
trigger=trigger,
level=level,
confidence=0.45
)
print(f"\nCreated escalation case: {case1.case_id}")
print(f"Assigned to: {case1.assigned_to}")
print(f"Status: {case1.status.value}")
# Demonstration 4: Security Incident Escalation
print("\n" + "=" * 80)
print("Demonstration 4: Security Incident Escalation")
print("=" * 80)
case2 = agent.create_escalation(
title="Unauthorized Access Attempt",
description="Multiple failed login attempts detected",
context={
"security_incident": True,
"source_ip": "192.168.1.100",
"attempts": 10
},
trigger=EscalationTrigger.SECURITY_INCIDENT,
level=EscalationLevel.CRITICAL
)
print(f"Created critical escalation: {case2.case_id}")
print(f"Level: {case2.level.value}")
print(f"Trigger: {case2.trigger.value}")
print(f"Assigned to: {case2.assigned_to}")
# Demonstration 5: Rule-Based Escalation
print("\n" + "=" * 80)
print("Demonstration 5: Rule-Based Escalation")
print("=" * 80)
should_escalate, trigger, level = agent.should_escalate(
context={
"has_sensitive_data": True,
"data_type": "PII"
}
)
if should_escalate:
case3 = agent.create_escalation(
title="Sensitive Data Access",
description="Request to access personally identifiable information",
context={"has_sensitive_data": True, "data_type": "PII"},
trigger=trigger,
level=level
)
print(f"Created escalation: {case3.case_id}")
print(f"Trigger: {trigger.value}")
print(f"Level: {level.value}")
# Demonstration 6: Manual Escalation Request
print("\n" + "=" * 80)
print("Demonstration 6: Manual Escalation Request")
print("=" * 80)
case4 = agent.create_escalation(
title="Complex Customer Complaint",
description="Customer requesting refund for premium service",
context={
"escalate": True,
"customer_id": "cust_12345",
"amount": 5000
},
trigger=EscalationTrigger.MANUAL_REQUEST
)
print(f"Manual escalation: {case4.case_id}")
print(f"Level: {case4.level.value}")
print(f"Assigned to: {case4.assigned_to}")
# Demonstration 7: Case Resolution
print("\n" + "=" * 80)
print("Demonstration 7: Escalation Resolution")
print("=" * 80)
print(f"\nResolving case: {case3.case_id}")
agent.update_case_status(
case_id=case3.case_id,
status=EscalationStatus.RESOLVED,
resolution="Access granted with additional monitoring"
)
resolved_case = next(c for c in agent.cases if c.case_id == case3.case_id)
print(f"Status: {resolved_case.status.value}")
print(f"Resolution: {resolved_case.resolution}")
# Demonstration 8: Pending Escalations
print("\n" + "=" * 80)
print("Demonstration 8: Pending Escalations")
print("=" * 80)
pending = agent.get_pending_escalations()
print(f"\nTotal pending escalations: {len(pending)}")
for case in pending:
print(f"\n Case: {case.case_id}")
print(f" Title: {case.title}")
print(f" Level: {case.level.value}")
print(f" Assigned to: {case.assigned_to}")
print(f" Age: {(datetime.now() - case.created_at).seconds // 60} minutes")
# Demonstration 9: Critical Escalations Only
print("\n" + "=" * 80)
print("Demonstration 9: Critical Escalations")
print("=" * 80)
critical = agent.get_pending_escalations(level=EscalationLevel.CRITICAL)
print(f"\nCritical pending escalations: {len(critical)}")
for case in critical:
print(f" - {case.title}")
# Demonstration 10: Escalation Metrics
print("\n" + "=" * 80)
print("Demonstration 10: Escalation Metrics")
print("=" * 80)
metrics = agent.get_escalation_metrics(days=7)
print(f"\nEscalation Metrics (last {metrics['period_days']} days):")
print(f" Total Escalations: {metrics['total_escalations']}")
print(f" Pending: {metrics['pending']}")
print(f" In Progress: {metrics['in_progress']}")
print(f" Resolved: {metrics['resolved']}")
print(f"\n By Level:")
for level, count in metrics['by_level'].items():
if count > 0:
print(f" {level}: {count}")
print(f"\n By Trigger:")
for trigger, count in metrics['by_trigger'].items():
if count > 0:
print(f" {trigger}: {count}")
if metrics['avg_resolution_hours']:
print(f"\n Avg Resolution Time: {metrics['avg_resolution_hours']:.1f} hours")
# Summary
print("\n" + "=" * 80)
print("Summary: Escalation Pattern")
print("=" * 80)
print("""
The Escalation Pattern ensures appropriate human oversight for critical situations:
Key Features Demonstrated:
1. Stakeholder Management - Configure escalation recipients by level
2. Escalation Rules - Define automatic escalation triggers
3. Confidence-Based Escalation - Escalate low-confidence decisions
4. Security Escalation - Handle security incidents
5. Rule-Based Escalation - Trigger based on context rules
6. Manual Escalation - Support explicit escalation requests
7. Case Resolution - Track escalation lifecycle
8. Pending Cases - Monitor unresolved escalations
9. Level Filtering - Focus on critical cases
10. Metrics Tracking - Monitor escalation patterns
Benefits:
- Ensures human oversight for critical decisions
- Prevents autonomous errors
- Provides safety net for uncertainty
- Graduated response by severity
- Supports compliance
- Facilitates learning
- Maintains audit trail
Best Practices:
- Define clear escalation criteria
- Set appropriate thresholds
- Implement routing rules
- Ensure stakeholder availability
- Track metrics
- Provide status updates
- Support escalation history
- Enable feedback loops
- Monitor patterns
- Reduce false escalations
- Support emergency escalations
- Document procedures
This pattern is essential for safe AI systems requiring human oversight,
especially in high-stakes domains like healthcare, finance, and security.
""")
if __name__ == "__main__":
demonstrate_escalation()
Agentic Design Pattern: Escalation
Pattern 140: Escalation – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Escalation pattern implementation.
Pattern Description
Pattern 140: Escalation Pattern
Description: The Escalation Pattern implements mechanisms for identifying situations that require human intervention or higher-level decision-making and routing them appropriately through an escalation hierarchy. This pattern recognizes when an agent encounters scenarios beyond its capabilities, authority, or confidence level and escalates to appropriate handlers based on severity and context.
Escalation is critical for AI safety and reliability, ensuring that complex,
high-stakes, or ambiguous situations receive appropriate human oversight. The
pattern supports multiple escalation levels, severity assessment, automatic
triggering based on rules or confidence thresholds, and routing to appropriate
stakeholders. It balances agent autonomy with human control.
This implementation provides a comprehensive escalation system with severity
assessment, trigger conditions, escalation routing, stakeholder management,
escalation tracking, and resolution workflows. It supports both rule-based
and confidence-based escalation with customizable escalation paths.
Components: – Severity Assessment: Evaluate situation severity – Trigger Detection: Identify when to escalate – Escalation Routing: Route to appropriate handler – Stakeholder Management: Manage escalation recipients – Escalation Tracking: Monitor escalation status – Resolution Workflow: Handle escalation lifecycle
Use Cases: – High-stakes decisions (medical, financial, legal) – Low-confidence predictions – Policy violations – Security incidents – Customer complaints – System errors – Ethical dilemmas – Edge cases
LangChain Implementation: This implementation uses: – LLM for severity assessment and decision complexity analysis – Rule-based escalation triggers – Confidence-threshold escalation – Hierarchical escalation routing – Escalation event tracking – Integration with human-in-the-loop systems
Benefits: – Ensures human oversight for critical decisions – Prevents autonomous errors in high-stakes scenarios – Provides safety net for uncertain situations – Enables graduated response based on severity – Supports compliance requirements – Facilitates learning from escalations – Maintains audit trail
Trade-offs: – Reduces full autonomy – Introduces latency for escalated cases – Requires human availability – May create bottlenecks – Needs clear escalation criteria – Can interrupt workflow
Production Considerations: – Define clear escalation criteria – Set appropriate confidence thresholds – Implement escalation routing rules – Ensure stakeholder availability – Track escalation metrics – Provide escalation status updates – Support escalation history – Enable escalation feedback – Monitor escalation patterns – Reduce false escalations – Support emergency escalations – Document escalation procedures
File Statistics
- Total Lines: 709
- Classes: 7
- Functions: 1
- Constants: 0
Module Documentation
"""
Pattern 140: Escalation Pattern
Description:
The Escalation Pattern implements mechanisms for identifying situations that
require human intervention or higher-level decision-making and routing them
appropriately through an escalation hierarchy. This pattern recognizes when an
agent encounters scenarios beyond its capabilities, authority, or confidence
level and escalates to appropriate handlers based on severity and context.
Escalation is critical for AI safety and reliability, ensuring that complex,
high-stakes, or ambiguous situations receive appropriate human oversight. The
pattern supports multiple escalation levels, severity assessment, automatic
triggering based on rules or confidence thresholds, and routing to appropriate
stakeholders. It balances agent autonomy with human control.
This implementation provides a comprehensive escalation system with severity
assessment, trigger conditions, escalation routing, stakeholder management,
escalation tracking, and resolution workflows. It supports both rule-based
and confidence-based escalation with customizable escalation paths.
Components:
- Severity Assessment: Evaluate situation severity
- Trigger Detection: Identify when to escalate
- Escalation Routing: Route to appropriate handler
- Stakeholder Management: Manage escalation recipients
- Escalation Tracking: Monitor escalation status
- Resolution Workflow: Handle escalation lifecycle
Use Cases:
- High-stakes decisions (medical, financial, legal)
- Low-confidence predictions
- Policy violations
- Security incidents
- Customer complaints
- System errors
- Ethical dilemmas
- Edge cases
LangChain Implementation:
This implementation uses:
- LLM for severity assessment and decision complexity analysis
- Rule-based escalation triggers
- Confidence-threshold escalation
- Hierarchical escalation routing
- Escalation event tracking
- Integration with human-in-the-loop systems
Benefits:
- Ensures human oversight for critical decisions
- Prevents autonomous errors in high-stakes scenarios
- Provides safety net for uncertain situations
- Enables graduated response based on severity
- Supports compliance requirements
- Facilitates learning from escalations
- Maintains audit trail
Trade-offs:
- Reduces full autonomy
- Introduces latency for escalated cases
- Requires human availability
- May create bottlenecks
- Needs clear escalation criteria
- Can interrupt workflow
Production Considerations:
- Define clear escalation criteria
- Set appropriate confidence thresholds
- Implement escalation routing rules
- Ensure stakeholder availability
- Track escalation metrics
- Provide escalation status updates
- Support escalation history
- Enable escalation feedback
- Monitor escalation patterns
- Reduce false escalations
- Support emergency escalations
- Document escalation procedures
"""
Purpose: This module implements the Escalation pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import datetime.timedelta
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
Purpose: Core Python functionality:
– dataclasses.dataclass: Simplified class definitions with @dataclass
– dataclasses.field: Simplified class definitions with @dataclass
– datetime.datetime: Date and time handling
– datetime.timedelta: Date and time handling
– enum.Enum: Enumeration support
– langchain_core.output_parsers.StrOutputParser: Regular expression operations
– langchain_core.prompts.ChatPromptTemplate: Regular expression operations
– os: Environment variable access and system operations
– typing.Any: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.List: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)
LangChain Imports
from langchain_openai.ChatOpenAI
Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows
Other Imports
from dotenv.load_dotenv
Purpose: Additional dependencies for specialized functionality.
Classes
Class: EscalationLevel
Lines 95-100
Inherits from: Enum
Purpose:
Escalation severity levels.
Class: EscalationStatus
Lines 103-108
Inherits from: Enum
Purpose:
Status of escalation.
Class: EscalationTrigger
Lines 111-121
Inherits from: Enum
Purpose:
Reasons for escalation.
Class: Stakeholder
Lines 125-133
Purpose:
Escalation stakeholder.
Class: EscalationCase
Lines 137-152
Purpose:
Escalation case record.
Class: EscalationRule
Lines 156-163
Purpose:
Rule for triggering escalation.
Class: EscalationAgent
Lines 166-432
Purpose:
Agent that manages escalation of decisions and situations.
This agent identifies when to escalate, assesses severity,
routes to appropriate stakeholders, and tracks resolution.
Methods:
__init__(self, agent_id, temperature)
Lines 174-208
Purpose:
Initialize the escalation agent.
Parameters:
– agent_id: Agent id
– temperature: Sampling temperature (0-1)
add_stakeholder(self, stakeholder)
Lines 210-213
Purpose:
Add escalation stakeholder.
Parameters:
– stakeholder: Stakeholder
Returns: None
add_rule(self, rule)
Lines 215-218
Purpose:
Add escalation rule.
Parameters:
– rule: Rule
Returns: None
should_escalate(self, context, confidence)
Lines 220-253
Purpose:
Determine if situation should be escalated.
Returns:
Tuple of (should_escalate, trigger, level)
Parameters:
– context: Context
– confidence: Confidence
Returns: <ast.Subscript object at 0x000001A06D649A90>
_evaluate_rule(self, rule, context)
Lines 255-274
Purpose:
Evaluate if rule condition is met.
Parameters:
– rule: Rule
– context: Context
Returns: bool
create_escalation(self, title, description, context, trigger, level, confidence)
Lines 276-308
Purpose:
Create an escalation case.
Parameters:
– title: Title
– description: Description
– context: Context
– trigger: Trigger
– level: Level
– confidence: Confidence
Returns: EscalationCase
_assess_severity(self, title, description, context, trigger)
Lines 310-343
Purpose:
Assess severity using LLM.
Parameters:
– title: Title
– description: Description
– context: Context
– trigger: Trigger
Returns: EscalationLevel
_route_escalation(self, case)
Lines 345-361
Purpose:
Route escalation to appropriate stakeholder.
Parameters:
– case: Case
Returns: <ast.Subscript object at 0x000001A06D8890D0>
update_case_status(self, case_id, status, resolution)
Lines 363-381
Purpose:
Update escalation case status.
Parameters:
– case_id: Case id
– status: Status
– resolution: Resolution
Returns: bool
get_pending_escalations(self, level, stakeholder_id)
Lines 383-397
Purpose:
Get pending escalation cases.
Parameters:
– level: Level
– stakeholder_id: Stakeholder id
Returns: <ast.Subscript object at 0x000001A06D824C90>
get_escalation_metrics(self, days)
Lines 399-432
Purpose:
Get escalation metrics.
Parameters:
– days: Days
Returns: <ast.Subscript object at 0x000001A06D683110>
Functions
demonstrate_escalation()
Lines 435-704
Purpose:
Demonstrate the escalation pattern.
Key Concepts Summary
Pattern Overview
The Escalation pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.
Core Components
- Agent Class: Main implementation of the pattern logic
- Prompt Engineering: Carefully crafted prompts to guide LLM behavior
- Tool Integration: Optional tools for extended functionality
- Memory Management: State handling across interactions
- Output Parsing: Structured response processing
When to Use
✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability
❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)
Best Practices Demonstrated
1. Environment Configuration
- ✅ Use
.envfiles for API keys - ✅ Never hardcode credentials
- ✅ Validate environment variables on startup
2. Error Handling
- ✅ Graceful degradation on failures
- ✅ Informative error messages
- ✅ Logging for debugging
3. Prompt Engineering
- ✅ Clear, specific instructions
- ✅ Structured output formats
- ✅ Few-shot examples when beneficial
4. Code Organization
- ✅ Separation of concerns (class-based design)
- ✅ Type hints for clarity
- ✅ Docstrings for documentation
5. Performance Optimization
- ✅ Appropriate temperature settings
- ✅ Token budget management
- ✅ Caching where applicable
LangChain Components Used
- ChatOpenAI: OpenAI model interface
- Chains: Sequential component composition
Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development
Execution Flow
1. Initialize agent with configuration
↓
2. Load environment and dependencies
↓
3. Prepare prompt with input data
↓
4. Invoke LLM through LangChain
↓
5. Parse and validate response
↓
6. Return formatted result
Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation
Performance Considerations
Computational Cost
- LLM Calls: Primary cost factor (based on tokens)
- Latency: Depends on model and input size
- Memory: Minimal for most patterns
Optimization Strategies
- Prompt Optimization: Minimize tokens while maintaining clarity
- Caching: Cache repeated queries/responses
- Model Selection: Use smaller models when appropriate
- Batch Processing: Group similar requests
- Streaming: For long responses, stream tokens
Cost Estimates (Approximate)
- GPT-3.5-turbo: ~$0.002 per request
- GPT-4: ~$0.03 per request
- GPT-4-turbo: ~$0.01 per request
Actual costs vary based on input/output token counts
Common Pitfalls to Avoid
1. Prompt Design Issues
❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts
2. Error Handling
❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation
3. Token Management
❌ Exceeding model token limits
✅ Truncate inputs, chunk processing
4. Security
❌ Exposing API keys in code
✅ Use environment variables
5. Testing
❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs
6. Cost Management
❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring
Further Improvements
Production Enhancements
- Monitoring: Add logging, metrics, tracing
- Validation: Input/output schema validation
- Rate Limiting: Implement request throttling
- Caching: Redis or in-memory cache for responses
- Async Support: Non-blocking operations for scalability
Feature Extensions
- Multi-Model Support: Switch between providers
- Fallback Models: Graceful degradation strategy
- Custom Tools: Domain-specific function calling
- Advanced Memory: Vector stores for long-term memory
- Evaluation: Automated testing and benchmarking
Code Quality
- Testing: Unit tests, integration tests
- Documentation: API docs, usage examples
- Type Safety: Stricter type hints
- Configuration: Externalized config files
Related Patterns
- Pattern 139
- Pattern 141
Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…
Summary
The Escalation pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:
✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns
🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability
📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/
File: 140_escalation.py
Total Lines: 709
Pattern Category: AI Agent Design Pattern #140
Generated from actual implementation code
For educational and reference purposes