Triage Support Tickets with the Decisions API

mkdir decisions-api-ticket-triage && cd decisions-api-ticket-triage
# save requirements.txt here, then continue
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
export PERPLEXITY_API_KEY="your-api-key-here"

"""Triage inbound support tickets with the Decisions API.

One Decisions API request per ticket answers three questions with probabilities
(needs a human? which intent? how severe?). Threshold rules turn those numbers
into a route. Only tickets routed to `escalate` go on to the investigation
step: an Agent API request that drafts an investigation plan.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from dataclasses import asdict, dataclass
from typing import Any

import httpx
from perplexity import Perplexity, PerplexityError

# The Decisions API endpoint and model.
DECISIONS_URL = "https://api.perplexity.ai/v1/decisions"
DECISIONS_MODEL = "decider-27b"
AGENT_API_PRESET = "low"  # the investigation step; only escalated tickets reach it

# Thresholds. Tune these against your own labeled tickets; nothing else changes.
ESCALATE_P_CRITICAL = 0.35  # a one-in-three shot at the top level is an incident
ESCALATE_MIN_SCORE = 2.5  # halfway to "outage" is too hot for a queue
REVIEW_MIN_INTENT_CONFIDENCE = 0.60  # below this the runner-up intent is too close
CLOSE_SPAM_MAX_SEVERITY = 0.5  # real spam sits at level 0; anything higher gets a look
# T-1004 settles at 0.29 to 0.32; keep this cutoff clear of that cluster.
AUTO_REPLY_MAX_HUMAN = 0.35
AUTO_REPLY_MAX_SEVERITY = 1.5  # never auto-reply to a possible level 2 (blocks work)
AUTO_REPLY_INTENTS = {"how_to", "account_access", "feature_request"}  # self-service

QUESTIONS: dict[str, dict[str, Any]] = {
    "needs_human": {
        "type": "noul",
        "instructions": (
            "Does a person need to act on this ticket before an automated reply "
            "would be acceptable?"
        ),
        "criteria": {
            "true": (
                "Refunds, account recovery, security incidents, outages, complaints, "
                "or anything a canned answer cannot resolve."
            ),
            "false": "A documentation link or a self-service flow fully resolves it.",
        },
    },
    "intent": {
        "type": "choice",
        "instructions": "What is the customer asking for?",
        "criteria": {
            "outage_or_bug": "Something that used to work is broken or unavailable.",
            "billing": "Charges, refunds, invoices, plan changes.",
            "account_access": "Login, password, SSO, or permissions for an account.",
            "security": "Leaked credentials, suspicious access, or a data exposure.",
            "how_to": "A usage question a docs page could answer.",
            "feature_request": "Asking for something the product does not do.",
            "spam": "Unsolicited marketing or irrelevant to the product.",
        },
    },
    "severity": {
        "type": "score",
        "instructions": "How severe is the customer impact right now?",
        "criteria": [
            "No impact. A question, wish, or spam.",
            "Minor. Cosmetic or a workaround exists.",
            "Blocks a core workflow for this customer.",
            "Outage for many users, data loss, or security exposure.",
        ],
    },
}

class DecisionsError(Exception):
    """A Decisions API request that did not return an answer."""


def load_tickets(path: str) -> list[dict[str, str]]:
    """Read one JSON object per line: an `id`, a `subject`, and any fields you like.

    The whole object is sent as `state`, so the model sees every field you include.
    """
    with open(path, encoding="utf-8") as handle:
        return [json.loads(line) for line in handle if line.strip()]


@dataclass
class Decision:
    ticket_id: str
    route: str
    reason: str
    needs_human: float
    intent: str
    intent_confidence: float
    severity: float
    p_critical: float
    request_id: str  # the x-request-id header; quote it in support requests

def route(
    *,
    needs_human: float,
    intent: str,
    intent_confidence: float,
    severity: float,
    p_critical: float,
) -> tuple[str, str]:
    """Turn probabilities into a route: danger, then uncertainty, then automation."""
    if p_critical >= ESCALATE_P_CRITICAL or severity >= ESCALATE_MIN_SCORE:
        return "escalate", f"P(crit)={p_critical:.2f}, expected severity {severity:.2f}"
    if intent_confidence < REVIEW_MIN_INTENT_CONFIDENCE:
        return "review", f"intent unclear (confidence {intent_confidence:.2f})"
    if intent == "spam" and severity < CLOSE_SPAM_MAX_SEVERITY:
        return "close", f"spam, severity {severity:.2f}"
    if (
        needs_human <= AUTO_REPLY_MAX_HUMAN
        and severity < AUTO_REPLY_MAX_SEVERITY
        and intent in AUTO_REPLY_INTENTS
    ):
        return "auto_reply", f"{intent}, needs_human {needs_human:.2f}"
    return "queue", f"{intent}, needs_human {needs_human:.2f}, severity {severity:.2f}"

def classify(client: httpx.Client, ticket: dict[str, str]) -> Decision:
    """One request per ticket. All three questions share the ticket as `state`."""
    response = client.post(
        DECISIONS_URL,
        json={"model": DECISIONS_MODEL, "state": ticket, "questions": QUESTIONS},
    )
    request_id = response.headers.get("x-request-id", "none")
    if response.status_code != 200:
        detail = response.text.strip()  # the JSON error body
        status = response.status_code
        raise DecisionsError(f"{status} {detail} (x-request-id {request_id})")
    body = response.json()
    needs_human = body["answers"]["needs_human"]["noul"]
    intent = body["answers"]["intent"]
    severity = body["answers"]["severity"]
    top_level = str(len(QUESTIONS["severity"]["criteria"]) - 1)  # keys are strings
    p_critical = severity["probabilities"].get(top_level, 0.0)
    chosen, reason = route(
        needs_human=needs_human,
        intent=intent["choice"],
        intent_confidence=intent["confidence"],
        severity=severity["score"],
        p_critical=p_critical,
    )
    return Decision(
        ticket_id=ticket["id"],
        route=chosen,
        reason=reason,
        needs_human=needs_human,
        intent=intent["choice"],
        intent_confidence=intent["confidence"],
        severity=severity["score"],
        p_critical=p_critical,
        request_id=request_id,
    )

def draft_escalation(
    agent: Perplexity, ticket: dict[str, str], decision: Decision
) -> str:
    """The investigation step: one Agent API request, only for escalated tickets."""
    prompt = (
        "You are the on-call support engineer for Lumen Analytics, a SaaS dashboard "
        "product. Write a short investigation plan for this escalated ticket: what to "
        "check first, what to tell the customer in the first reply, and who to page. "
        "Plain text, under 150 words.\n\n"
        f"Ticket: {json.dumps(ticket)}\n"
        f"Triage: intent={decision.intent}, expected severity={decision.severity:.2f}, "
        f"P(crit)={decision.p_critical:.2f}"
    )
    response = agent.responses.create(preset=AGENT_API_PRESET, input=prompt)
    return response.output_text.strip() or "(no text in Agent API response)"

def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--draft-escalations",
        action="store_true",
        help="send escalations to the Agent API",
    )
    parser.add_argument("--tickets", default="tickets.jsonl", help="input file")
    parser.add_argument("--out", default="triage_results.jsonl", help="output file")
    args = parser.parse_args()

    api_key = os.environ.get("PERPLEXITY_API_KEY")
    if not api_key:
        print("Set PERPLEXITY_API_KEY first.", file=sys.stderr)
        return 2

    tickets = load_tickets(args.tickets)
    if not tickets:
        print(f"No tickets found in {args.tickets}.")
        return 0

    decisions: list[Decision] = []
    with httpx.Client(
        headers={"Authorization": f"Bearer {api_key}"}, timeout=30.0
    ) as client:
        for ticket in tickets:
            try:
                decisions.append(classify(client, ticket))
            except (DecisionsError, httpx.HTTPError) as error:
                print(f"{ticket['id']}: Decisions API failed: {error}", file=sys.stderr)
                return 1

    header = (
        f"{'ticket':<7} {'route':<10} {'human':>5} {'intent':<16} "
        f"{'conf':>4} {'sev':>4} {'P(crit)':>7}  reason"
    )
    print(header)
    print("-" * len(header))
    for d in decisions:
        print(
            f"{d.ticket_id:<7} {d.route:<10} {d.needs_human:>5.2f} {d.intent:<16} "
            f"{d.intent_confidence:>4.2f} {d.severity:>4.2f} {d.p_critical:>7.2f}  "
            f"{d.reason}"
        )

    with open(args.out, "w", encoding="utf-8") as handle:
        for d in decisions:
            handle.write(json.dumps(asdict(d)) + "\n")
    escalated = [d for d in decisions if d.route == "escalate"]
    print(f"\n{len(escalated)} of {len(decisions)} escalated. Results: {args.out}")

    if not args.draft_escalations:
        print("Add --draft-escalations to send the escalated tickets to the Agent API.")
        return 0
    agent = Perplexity(api_key=api_key)
    by_id = {t["id"]: t for t in tickets}
    for d in escalated:
        print(f"\n=== {d.ticket_id}: {by_id[d.ticket_id]['subject']} ===")
        try:
            print(draft_escalation(agent, by_id[d.ticket_id], d))
        except PerplexityError as error:  # the library's base class
            print(f"{d.ticket_id}: Agent API request failed: {error}", file=sys.stderr)
    return 0


if __name__ == "__main__":
    sys.exit(main())

python triage_tickets.py

ticket  route      human intent           conf  sev P(crit)  reason
-------------------------------------------------------------------
T-1001  escalate    1.00 outage_or_bug    0.99 2.97    0.97  P(crit)=0.97, expected severity 2.97
T-1002  queue       0.99 billing          0.99 1.17    0.02  billing, needs_human 0.99, severity 1.17
T-1003  auto_reply  0.08 how_to           0.97 0.42    0.00  how_to, needs_human 0.08
T-1004  auto_reply  0.31 feature_request  0.99 0.27    0.00  feature_request, needs_human 0.31
T-1005  close       0.42 spam             0.98 0.09    0.02  spam, severity 0.09
T-1006  escalate    0.99 security         0.99 2.99    0.99  P(crit)=0.99, expected severity 2.99
T-1007  escalate    0.96 outage_or_bug    0.51 2.84    0.85  P(crit)=0.85, expected severity 2.84
T-1008  queue       0.81 account_access   0.99 1.76    0.01  account_access, needs_human 0.81, severity 1.76
T-1009  queue       0.87 outage_or_bug    0.86 1.85    0.02  outage_or_bug, needs_human 0.87, severity 1.85
T-1010  queue       0.98 outage_or_bug    0.97 2.00    0.04  outage_or_bug, needs_human 0.98, severity 2.00
T-1011  review      0.96 account_access   0.55 2.19    0.24  intent unclear (confidence 0.55)
T-1012  queue       0.98 billing          0.99 0.87    0.01  billing, needs_human 0.98, severity 0.87

3 of 12 escalated. Results: triage_results.jsonl
Add --draft-escalations to send the escalated tickets to the Agent API.

python triage_tickets.py --draft-escalations

=== T-1001: Dashboards down for whole org ===
First: Treat this as a critical, org-wide outage. Check monitoring and status-page alerts, recent deploys, and API/gateway error rates around 08:40 UTC; verify whether dashboards fail across regions and capture request IDs and logs for the 502s.

First reply: “We’ve escalated this as a critical incident and are investigating the dashboard 502s affecting your organization. We understand the 10:00 UTC board meeting is time-sensitive. We’ll share the next update within 15 minutes, even if we’re still investigating.”

Page the incident commander and on-call dashboard/backend and SRE engineers immediately; notify the enterprise account owner/support lead.