How to Build an AI Customer Support Agent That Resolves Tickets Automatically

ai-support-agent/
├── app/
│   ├── __init__.py
│   ├── schemas.py
│   ├── tools.py
│   └── resolver.py
├── tests/
│   └── test_support_agent.py
└── requirements.txt

from enum import Enum
from typing import List, Optional
from pydantic import BaseModel, Field

class TicketIntent(str, Enum):
    ORDER_STATUS = "ORDER_STATUS"
    REFUND_REQUEST = "REFUND_REQUEST"
    TECHNICAL_HOWTO = "TECHNICAL_HOWTO"
    SECURITY_OR_LEGAL = "SECURITY_OR_LEGAL"
    HUMAN_REQUESTED = "HUMAN_REQUESTED"

class CustomerSentiment(str, Enum):
    CALM = "CALM"
    FRUSTRATED = "FRUSTRATED"
    ANGRY = "ANGRY"

class TicketTriage(BaseModel):
    intent: TicketIntent
    sentiment: CustomerSentiment
    priority: int = Field(..., ge=1, le=5)
    extracted_order_id: Optional[str] = None

class KBChunk(BaseModel):
    article_id: str
    title: str
    content: str
    similarity_score: float

class SupportResolution(BaseModel):
    action: str = Field(..., pattern="^(AUTO_REPLY|ESCALATE_TO_HUMAN)$")
    customer_reply: Optional[str] = None
    cited_article_ids: List[str] = Field(default_factory=list)
    internal_handoff_note: Optional[str] = None
    escalation_reason: Optional[str] = None

from typing import Dict, Any

MOCK_ORDERS_DB: Dict[str, Dict[str, Any]] = {
    "ORD-1001": {"customer_email": "user@example.com", "status": "DELIVERED", "amount_usd": 35.0, "age_days": 12},
    "ORD-1002": {"customer_email": "vip@example.com", "status": "PROCESSING", "amount_usd": 240.0, "age_days": 5},
}

def lookup_order_status(order_id: str, authenticated_email: str) -> Dict[str, Any]:
    """Looks up an order only if it belongs to the authenticated customer email."""
    order = MOCK_ORDERS_DB.get(order_id)
    if not order:
        return {"found": False, "error": "Order ID not found."}
    if order["customer_email"].lower() != authenticated_email.lower():
        raise PermissionError("Order does not belong to authenticated customer.")
    return {"found": True, "order_id": order_id, **order}

def process_policy_gated_refund(order_id: str, authenticated_email: str) -> Dict[str, Any]:
    """Executes an automated refund strictly within hard-coded policy limits (<= $50 and <= 30 days)."""
    order_info = lookup_order_status(order_id, authenticated_email)
    if not order_info.get("found"):
        return {"approved": False, "reason": "ORDER_NOT_FOUND"}
    if order_info["age_days"] > 30:
        return {"approved": False, "reason": "EXCEEDS_30_DAY_WINDOW"}
    if order_info["amount_usd"] > 50.0:
        return {"approved": False, "reason": "AMOUNT_REQUIRES_HUMAN_APPROVAL"}
    return {
        "approved": True,
        "refunded_amount_usd": order_info["amount_usd"],
        "transaction_ref": f"REF-{order_id}",
    }

from typing import List
from app.schemas import TicketTriage, TicketIntent, CustomerSentiment, KBChunk, SupportResolution

MIN_KB_SIMILARITY_THRESHOLD = 0.78

def evaluate_escalation_rules(
    triage: TicketTriage,
    kb_chunks: List[KBChunk],
) -> SupportResolution | None:
    """Deterministically checks whether a ticket must bypass auto-reply and escalate to a human agent."""
    if triage.intent in (TicketIntent.SECURITY_OR_LEGAL, TicketIntent.HUMAN_REQUESTED):
        return SupportResolution(
            action="ESCALATE_TO_HUMAN",
            escalation_reason=f"Mandatory escalation for intent: {triage.intent.value}",
            internal_handoff_note=f"Customer intent={triage.intent.value}, priority={triage.priority}.",
        )
    if triage.sentiment == CustomerSentiment.ANGRY:
        return SupportResolution(
            action="ESCALATE_TO_HUMAN",
            escalation_reason="High customer frustration detected (ANGRY sentiment).",
            internal_handoff_note="Route to senior support specialist immediately.",
        )
    if triage.intent == TicketIntent.TECHNICAL_HOWTO:
        best_score = max((c.similarity_score for c in kb_chunks), default=0.0)
        if best_score < MIN_KB_SIMILARITY_THRESHOLD:
            return SupportResolution(
                action="ESCALATE_TO_HUMAN",
                escalation_reason=f"Low KB retrieval confidence ({best_score:.2f} < {MIN_KB_SIMILARITY_THRESHOLD}).",
                internal_handoff_note="Knowledge base lacks a high-confidence article for this question.",
            )
    return None

import pytest
from app.schemas import TicketTriage, TicketIntent, CustomerSentiment, KBChunk
from app.tools import lookup_order_status, process_policy_gated_refund
from app.resolver import evaluate_escalation_rules

def test_cross_tenant_order_lookup_blocked():
    with pytest.raises(PermissionError):
        lookup_order_status("ORD-1001", authenticated_email="attacker@example.com")

def test_policy_gated_refund_limits():
    small_refund = process_policy_gated_refund("ORD-1001", "user@example.com")
    assert small_refund["approved"] is True

    large_refund = process_policy_gated_refund("ORD-1002", "vip@example.com")
    assert large_refund["approved"] is False
    assert large_refund["reason"] == "AMOUNT_REQUIRES_HUMAN_APPROVAL"

def test_low_kb_confidence_triggers_escalation():
    triage = TicketTriage(
        intent=TicketIntent.TECHNICAL_HOWTO,
        sentiment=CustomerSentiment.CALM,
        priority=2,
    )
    weak_chunks = [KBChunk(article_id="KB-12", title="Billing FAQ", content="...", similarity_score=0.61)]
    resolution = evaluate_escalation_rules(triage, weak_chunks)
    assert resolution is not None
    assert resolution.action == "ESCALATE_TO_HUMAN"