AI Agents in Practice: Customer Support Automation Agent

Introduction

In this AI agent pattern, we focus on a real-world Customer Support Automation Agent. This agent helps businesses automate the initial stages of customer support by reading customer inquiries, categorizing issues, proposing immediate responses, and escalating issues that require human intervention. This agent follows the principles of transactional autonomy—it only claims success when a tool action has been completed successfully and tracked with receipts, ensuring each step is auditable and compliant.

The Use Case

Customers often submit inquiries about products, services, or accounts, which need to be categorized and resolved quickly. The AI agent reads the customer's inquiry, categorizes it (e.g., billing, technical support, general inquiries), proposes a relevant response, and either provides a final answer or escalates the issue to a human agent when necessary. All decisions are backed by policy claims and real-time checks, ensuring compliance and correctness.

Prompt Contract (agent interface)

# file: contracts/customer_support_v1.yaml
role: "CustomerSupportAgent"
scope: >
  Automatically triage and respond to customer inquiries. Ask once if critical fields are missing (customer_id, issue_type, description).
  Propose tool calls; never assert success without a receipt.
output:
  type: object
  required: [summary, decision, response, citations, next_steps, tool_proposals]
  properties:
    summary: {type: string, maxWords: 100}
    decision: {type: string, enum: ["approve","reject","need_approval","need_more_info"]}
    response: {type: string, maxWords: 250}
    citations: {type: array, items: {type: string}}
    next_steps: {type: array, items: {type: string}, maxItems: 5}
    tool_proposals:
      type: array
      items:
        type: object
        required: [name, args, preconditions, idempotency_key]
        properties:
          name: {type: string, enum: [CategorizeIssue, RespondToInquiry, EscalateIssue]}
          args: {type: object}
          preconditions: {type: string}
          idempotency_key: {type: string}
policy_id: "support_policy.v3"
citation_rule: "1–2 minimal-span claim_ids per factual sentence"
decoding:
  narrative: {top_p: 0.92, temperature: 0.72, stop: ["\n\n## "]}
  bullets:   {top_p: 0.82, temperature: 0.45}

Example claims (context provided to the model)

[
  {"claim_id":"policy:support:technical","text":"Technical support inquiries must be responded to within 24 hours.",
   "effective_date":"2025-06-01","source_id":"doc:support_policy_v3","span":"responded to within 24 hours"},
  {"claim_id":"policy:support:billing","text":"Billing issues must be handled by the finance team.",
   "effective_date":"2025-06-01","source_id":"doc:support_policy_v3","span":"handled by the finance team"},
  {"claim_id":"policy:support:general_inquiries","text":"General inquiries are routed to the customer service team.",
   "effective_date":"2025-06-01","source_id":"doc:support_policy_v3","span":"routed to the customer service team"}
]

Tool Interfaces (typed, with receipts)

# tools.py
from pydantic import BaseModel
from datetime import datetime
from typing import List, Optional, Dict

class CategorizeIssueArgs(BaseModel):
    customer_id: str
    issue_description: str

class RespondToInquiryArgs(BaseModel):
    customer_id: str
    response_text: str

class EscalateIssueArgs(BaseModel):
    customer_id: str
    issue_description: str
    priority: str

class ToolReceipt(BaseModel):
    tool: str
    ok: bool
    ref: str
    message: str = ""
    data: Optional[Dict] = None

# adapters.py  (demo logic)
from tools import *

def categorize_issue(a: CategorizeIssueArgs) -> ToolReceipt:
    if "billing" in a.issue_description.lower():
        category = "Billing"
    elif "technical" in a.issue_description.lower():
        category = "Technical Support"
    else:
        category = "General Inquiry"
    return ToolReceipt(tool="CategorizeIssue", ok=True, ref=f"cat-{a.customer_id}",
                       message=f"Issue categorized as {category}", data={"category": category})

def respond_to_inquiry(a: RespondToInquiryArgs) -> ToolReceipt:
    return ToolReceipt(tool="RespondToInquiry", ok=True, ref=f"resp-{a.customer_id}",
                       message=f"Response sent to customer {a.customer_id}", data={"response": a.response_text})

def escalate_issue(a: EscalateIssueArgs) -> ToolReceipt:
    return ToolReceipt(tool="EscalateIssue", ok=True, ref=f"escalate-{a.customer_id}",
                       message=f"Issue escalated to priority {a.priority}", data={"priority": a.priority})

Agent Loop (proposal → verification → execution → receipts)

# agent_customer_support.py
import uuid, json
from typing import Any, Dict, List
from tools import *
from adapters import *

ALLOWED_TOOLS = {"CategorizeIssue","RespondToInquiry","EscalateIssue"}

def new_idem() -> str:
    return f"idem-{uuid.uuid4()}"

def verify_proposal(p: Dict[str, Any]) -> str:
    required = {"name","args","preconditions","idempotency_key"}
    if not required.issubset(p): return "Missing proposal fields"
    if p["name"] not in ALLOWED_TOOLS: return "Tool not allowed"
    return ""

def execute(p: Dict[str, Any]) -> ToolReceipt:
    n, a = p["name"], p["args"]
    if n == "CategorizeIssue":       return categorize_issue(CategorizeIssueArgs(**a))
    if n == "RespondToInquiry":      return respond_to_inquiry(RespondToInquiryArgs(**a))
    if n == "EscalateIssue":         return escalate_issue(EscalateIssueArgs(**a))
    return ToolReceipt(tool=n, ok=False, ref="none", message="Unknown tool")

# --- Model shim returning a plan per contract (replace with your LLM call) ---
def call_model(contract_yaml: str, claims: List[Dict[str,Any]], inquiry: Dict[str,Any]) -> Dict[str,Any]:
    return {
      "summary": "Customer inquiry categorized and responded.",
      "decision": "approve",
      "response": "Thank you for your inquiry. Our team will assist you within 24 hours.",
      "citations": ["policy:support:technical", "policy:support:billing", "policy:support:general_inquiries"],
      "next_steps": ["Categorize issue", "Respond to inquiry", "Escalate if necessary"],
      "tool_proposals": [
        {"name":"CategorizeIssue","args":{"customer_id":inquiry["customer_id"],"issue_description":inquiry["issue_description"]},
         "preconditions":"Categorize inquiry by issue type.","idempotency_key": new_idem()},
        {"name":"RespondToInquiry","args":{"customer_id":inquiry["customer_id"],"response_text":"Thank you for your inquiry. Our team will assist you within 24 hours."},
         "preconditions":"Respond to customer within policy time limits.","idempotency_key": new_idem()},
        {"name":"EscalateIssue","args":{"customer_id":inquiry["customer_id"],"issue_description":inquiry["issue_description"],"priority":"high"},
         "preconditions":"Escalate if response exceeds SLA.","idempotency_key": new_idem()}
      ]
    }

def render_response(model_json: Dict[str,Any], receipts: List[ToolReceipt]) -> str:
    idx = {r.tool:r for r in receipts}
    lines = [model_json["summary"], ""]
    lines.append(f"Decision: {model_json['decision']}")
    lines.append(f"Response: {model_json['response']}")
    lines.append("")
    lines.append("Next steps:")
    for s in model_json["next_steps"]:
        lines.append(f"• {s}")
    if idx.get("RespondToInquiry") and idx["RespondToInquiry"].ok:
        lines.append(f"\nResponse sent: {idx['RespondToInquiry'].message}")
    if idx.get("EscalateIssue") and idx["EscalateIssue"].ok:
        lines.append(f"Issue escalated: {idx['EscalateIssue'].message}")
    lines.append("\nCitations: " + ", ".join(model_json["citations"]))
    return "\n".join(lines)

def handle(inquiry: Dict[str,Any]) -> str:
    contract = open("contracts/customer_support_v1.yaml").read()
    claims: List[Dict[str,Any]] = []  # load real claims from policy
    plan = call_model(contract, claims, inquiry)

    receipts: List[ToolReceipt] = []
    for prop in plan["tool_proposals"]:
        reason = verify_proposal(prop)
        if reason:
            receipts.append(ToolReceipt(tool=prop["name"], ok=False, ref="blocked", message=reason)); continue
        rec = execute(prop)
        receipts.append(rec)
        if not rec.ok and prop["name"] in {"RespondToInquiry"}:
            break
    return render_response(plan, receipts)

if __name__ == "__main__":
    example_inquiry = {
      "customer_id": "C123",
      "issue_description": "My billing statement is incorrect. Can you help me resolve this?",
    }
    print(handle(example_inquiry))

The Prompt You’d Send to the Model (concise and testable)

System:
You are CustomerSupportAgent. Follow the contract:
- Ask once if customer_id or issue_description is missing.
- Cite 1–2 claim_ids per factual sentence using provided claims.
- Propose tools; never assert success without a receipt.
- Output JSON with keys: summary, decision, response, citations[], next_steps[], tool_proposals[].

Claims (eligible only):
[ ... JSON array of policy claims like above ... ]

User:
Customer inquiry: "My billing statement is incorrect. Can you help me resolve this?"

How to adapt quickly

Replace the mock support ticketing system and customer inquiry classification with your real support tools, CRM systems, or knowledge bases. Implement idempotency for ticket updates, time-bound responses, and SLA enforcement. Load claims from company policies regarding issue categories, response times, and escalation rules. Add a validation layer to ensure correct format, lexicon, and compliance. Deploy the contract, policy bundle, and decoding settings behind a feature flag for easy testing, canary deployment, and rollback support.