agentlens/examples/customer_support_agent.py

"""
AgentLens Customer Support Example — Realistic support ticket workflow.

Demonstrates:
  - Ticket classification with ROUTING decisions
  - Specialist routing with TOOL_SELECTION
  - Escalation decisions with ESCALATION type
  - Error handling — traces capture exceptions automatically
  - Multiple real-world decision patterns

Usage:
    pip install vectry-agentlens
    python customer_support_agent.py
"""

import agentlens
import time
import random

# Initialize
agentlens.init(
    api_key="your-api-key-here",
    endpoint="http://localhost:4200",
)

# Simulated ticket data
TICKETS = [
    {
        "id": "TKT-4021",
        "subject": "Cannot access billing portal after password reset",
        "priority": "high",
        "customer_tier": "enterprise",
        "body": "After resetting my password, I get a 403 error on the billing page. "
        "I need to update our payment method before end of month.",
    },
    {
        "id": "TKT-4022",
        "subject": "Feature request: dark mode for dashboard",
        "priority": "low",
        "customer_tier": "free",
        "body": "Would love to have a dark mode option. My eyes hurt during late-night sessions.",
    },
    {
        "id": "TKT-4023",
        "subject": "API returning 500 errors intermittently",
        "priority": "critical",
        "customer_tier": "enterprise",
        "body": "Our production integration is failing ~20% of requests with 500 errors. "
        "Started about 2 hours ago. This is blocking our release.",
    },
]


def simulate_llm(prompt: str, delay: float = 0.15) -> str:
    """Fake LLM — replace with real calls."""
    time.sleep(delay)
    return f"[Response to: {prompt[:60]}]"


def process_ticket(ticket: dict) -> None:
    """Process a single support ticket through the agent pipeline."""

    with agentlens.trace(
        "customer-support-bot",
        tags=["support", ticket["priority"], ticket["customer_tier"]],
    ):
        # Step 1: Classify the ticket
        agentlens.log_decision(
            type="ROUTING",
            chosen={
                "name": "classify_ticket",
                "confidence": 0.91,
                "params": {
                    "ticket_id": ticket["id"],
                    "predicted_category": (
                        "billing"
                        if "billing" in ticket["subject"].lower()
                        else "bug"
                        if "error" in ticket["body"].lower() or "500" in ticket["body"]
                        else "feature_request"
                    ),
                },
            },
            alternatives=[
                {
                    "name": "ask_customer_for_clarification",
                    "confidence": 0.2,
                    "reason_rejected": "Ticket subject and body are clear enough",
                },
            ],
            reasoning=f"Ticket '{ticket['subject']}' clearly maps to a known category.",
        )

        classification = simulate_llm(f"Classify: {ticket['subject']}")

        # Step 2: Route to specialist
        is_critical = ticket["priority"] in ("critical", "high")
        is_enterprise = ticket["customer_tier"] == "enterprise"

        if is_critical and is_enterprise:
            specialist = "senior_engineer"
        elif is_critical:
            specialist = "engineer"
        elif "billing" in ticket["subject"].lower():
            specialist = "billing_team"
        else:
            specialist = "general_support"

        agentlens.log_decision(
            type="ROUTING",
            chosen={
                "name": specialist,
                "confidence": 0.87,
                "params": {
                    "ticket_id": ticket["id"],
                    "priority": ticket["priority"],
                    "sla_minutes": 30 if is_enterprise else 240,
                },
            },
            alternatives=[
                {
                    "name": "general_support",
                    "confidence": 0.4,
                    "reason_rejected": "Ticket requires specialized handling"
                    if specialist != "general_support"
                    else "This is general support already",
                },
            ],
            reasoning=f"Priority={ticket['priority']}, Tier={ticket['customer_tier']} -> route to {specialist}.",
        )

        # Step 3: Specialist handles ticket (nested trace)
        with agentlens.trace(f"specialist-{specialist}", tags=[specialist]):
            # Tool selection for the specialist
            agentlens.log_decision(
                type="TOOL_SELECTION",
                chosen={
                    "name": "search_knowledge_base",
                    "confidence": 0.82,
                    "params": {"query": ticket["subject"], "limit": 5},
                },
                alternatives=[
                    {
                        "name": "search_past_tickets",
                        "confidence": 0.7,
                        "reason_rejected": "KB is more authoritative for known issues",
                    },
                    {
                        "name": "check_status_page",
                        "confidence": 0.6,
                        "reason_rejected": "Already checked — no ongoing incidents posted",
                    },
                ],
                reasoning="Knowledge base has resolution guides for common issues.",
            )

            kb_result = simulate_llm(f"Search KB for: {ticket['subject']}")

            # Step 4: Escalation decision for critical tickets
            if ticket["priority"] == "critical":
                agentlens.log_decision(
                    type="ESCALATION",
                    chosen={
                        "name": "escalate_to_engineering",
                        "confidence": 0.94,
                        "params": {
                            "severity": "P1",
                            "team": "platform-reliability",
                            "ticket_id": ticket["id"],
                        },
                    },
                    alternatives=[
                        {
                            "name": "resolve_at_support_level",
                            "confidence": 0.15,
                            "reason_rejected": "500 errors suggest infrastructure issue beyond support scope",
                        },
                    ],
                    reasoning="Intermittent 500s on enterprise account = immediate P1 escalation.",
                )

                # Simulate escalation failure for the critical ticket (shows error handling)
                if random.random() < 0.3:
                    raise RuntimeError(
                        f"Escalation service unavailable for {ticket['id']}"
                    )

            # Generate response
            response = simulate_llm(
                f"Draft response for {ticket['id']}: {ticket['subject']}",
                delay=0.3,
            )

        print(f"  [{ticket['id']}] Processed -> routed to {specialist}")


# Process all tickets
print("Processing support tickets...\n")

for ticket in TICKETS:
    try:
        process_ticket(ticket)
    except Exception as e:
        # The trace context manager captures the error automatically
        print(f"  [{ticket['id']}] Error during processing: {e}")

# Shutdown
agentlens.shutdown()

print("\nDone! Check AgentLens dashboard for 'customer-support-bot' traces.")
print("Look for the ERROR trace — it shows how failures are captured.")