aishastr.live

Agentic Design Pattern: Escalation

Written by

admin

in


    """
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

  1. Agent Class: Main implementation of the pattern logic
  2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
  3. Tool Integration: Optional tools for extended functionality
  4. Memory Management: State handling across interactions
  5. 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 .env files 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

  1. Prompt Optimization: Minimize tokens while maintaining clarity
  2. Caching: Cache repeated queries/responses
  3. Model Selection: Use smaller models when appropriate
  4. Batch Processing: Group similar requests
  5. 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

  1. Monitoring: Add logging, metrics, tracing
  2. Validation: Input/output schema validation
  3. Rate Limiting: Implement request throttling
  4. Caching: Redis or in-memory cache for responses
  5. Async Support: Non-blocking operations for scalability

Feature Extensions

  1. Multi-Model Support: Switch between providers
  2. Fallback Models: Graceful degradation strategy
  3. Custom Tools: Domain-specific function calling
  4. Advanced Memory: Vector stores for long-term memory
  5. Evaluation: Automated testing and benchmarking

Code Quality

  1. Testing: Unit tests, integration tests
  2. Documentation: API docs, usage examples
  3. Type Safety: Stricter type hints
  4. 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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts