AI Customer Support: a ticket system, RAG over a knowledge base, smart support

Customer writes a question
    ↓
Classification (Claude): routine / unusual / urgent
    ↓
Routine → RAG over the knowledge base → automatic reply (most of them, roughly 80%)
Unusual → AI draft for the agent → agent edits → sends (roughly 15%)
Urgent → immediate escalation + notification → agent replies personally (roughly 5%)

knowledge-base/
  billing.md          — billing questions
  shipping.md         — shipping and returns
  integrations.md     — connecting integrations
  plans.md            — pricing plans
  account.md          — account management

## How to change your plan

Go to your account → Settings → Plan → choose a new plan.
The change takes effect immediately. Any extra charge is prorated.

## Refunds

Refunds are available within 14 days of payment on your first order.
To request a refund, email [email protected] with the subject "Refund" and your order number.
Processing takes 5 business days.

## Differences between the Basic and Pro plans

Basic: up to 5 users, 10 GB of storage, email support.
Pro: unlimited users, 100 GB of storage, priority support, API access.

import anthropic
import json
import time
from pathlib import Path
from datetime import datetime, timezone

client = anthropic.Anthropic()


def load_knowledge_base(kb_path: str) -> str:
    """Loads the knowledge base from MD files"""
    kb_content = []
    for md_file in Path(kb_path).glob("**/*.md"):
        kb_content.append(f"\n## {md_file.stem}\n{md_file.read_text()}")
    return "\n".join(kb_content)


KB = load_knowledge_base("knowledge-base/")


def answer_support_ticket(question: str) -> dict:
    """Classifies the ticket and either answers or escalates"""

    start_time = time.time()

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=800,
        system=f"""You are an AI customer support agent.

COMPANY KNOWLEDGE BASE:
{KB}

Rules:
1. If the answer is in the knowledge base, answer accurately using it and don't add anything of your own
2. At the end of the answer, write: Source: [name of the knowledge base section]
3. If the answer is not in the knowledge base, write ESCALATE on the first line and explain why a person is needed
4. If the question is about a refund, ALWAYS write ESCALATE (even if the answer is in the base)
5. If the question is about a technical problem on the customer's side, ESCALATE
6. Tone: friendly, specific, no filler""",
        messages=[{"role": "user", "content": question}]
    )

    answer = "".join(b.text for b in response.content if b.type == "text")
    response_time = time.time() - start_time

    return {
        "answer": answer,
        "needs_human": answer.strip().startswith("ESCALATE"),
        "response_time_sec": round(response_time, 2),
        "confidence": "low" if answer.strip().startswith("ESCALATE") else "high"
    }


def track_metrics(result: dict, question: str):
    """Writes metrics to the log"""
    log_entry = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "escalated": result["needs_human"],
        "response_time_sec": result["response_time_sec"],
        "confidence": result["confidence"],
        "question_length": len(question)
    }
    with open("support-metrics.jsonl", "a") as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")


# Usage example
if __name__ == "__main__":
    questions = [
        "How do I change my plan?",
        "I want a refund for my subscription",
        "My Slack integration isn't working, everything is broken"
    ]

    for question in questions:
        print(f"\nQuestion: {question}")
        result = answer_support_ticket(question)
        track_metrics(result, question)

        if result["needs_human"]:
            print(f"ESCALATION -> handing off to a live agent")
            print(f"Reason: {result['answer']}")
        else:
            print(f"Automatic reply ({result['response_time_sec']} sec):")
            print(result["answer"])

from flask import Flask, request
import requests
import os

app = Flask(__name__)

INTERCOM_TOKEN = os.environ["INTERCOM_TOKEN"]
AI_BOT_ID = os.environ["INTERCOM_BOT_ID"]


def assign_to_human_agent(conversation_id: str, priority: str = "normal"):
    """Assigns the ticket to a live agent and adds a tag"""
    requests.post(
        f"https://api.intercom.io/conversations/{conversation_id}/parts",
        headers={
            "Authorization": f"Bearer {INTERCOM_TOKEN}",
            "Content-Type": "application/json"
        },
        json={
            "type": "admin",
            "admin_id": AI_BOT_ID,
            "message_type": "assignment",
            "assignee_id": None  # assigns to the team, not a specific agent
        }
    )

    # Add the priority tag
    if priority == "high":
        requests.post(
            f"https://api.intercom.io/conversations/{conversation_id}/tags",
            headers={"Authorization": f"Bearer {INTERCOM_TOKEN}"},
            json={"id": os.environ["INTERCOM_HIGH_PRIORITY_TAG_ID"]}
        )


@app.post("/intercom-webhook")
def handle_message():
    data = request.json

    if data.get("type") != "conversation.user.created":
        return {"status": "ignored"}

    item = data["data"]["item"]
    conversation_id = item["id"]
    parts = item["conversation_parts"]["conversation_parts"]

    if not parts:
        return {"status": "no_message"}

    message = parts[0]["body"]

    # AI answers
    result = answer_support_ticket(message)
    track_metrics(result, message)

    if not result["needs_human"]:
        # Send the automatic reply as the bot
        requests.post(
            f"https://api.intercom.io/conversations/{conversation_id}/reply",
            headers={
                "Authorization": f"Bearer {INTERCOM_TOKEN}",
                "Content-Type": "application/json"
            },
            json={
                "type": "admin",
                "admin_id": AI_BOT_ID,
                "message_type": "comment",
                "body": result["answer"]
            }
        )
    else:
        # Let the customer know a person is joining
        requests.post(
            f"https://api.intercom.io/conversations/{conversation_id}/reply",
            headers={
                "Authorization": f"Bearer {INTERCOM_TOKEN}",
                "Content-Type": "application/json"
            },
            json={
                "type": "admin",
                "admin_id": AI_BOT_ID,
                "message_type": "comment",
                "body": "Your question has been passed to a specialist. We'll reply within 2 hours."
            }
        )
        assign_to_human_agent(conversation_id, priority="high")

    return {"status": "ok"}


if __name__ == "__main__":
    app.run(port=5000)

import asyncio
import os
from telegram import Update, Bot
from telegram.ext import Application, MessageHandler, filters

TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
SUPPORT_TEAM_CHAT = os.environ["SUPPORT_TEAM_CHAT_ID"]

bot = Bot(token=TELEGRAM_BOT_TOKEN)


async def handle_support_message(update: Update, context):
    user_message = update.message.text
    user_id = update.effective_user.id
    username = update.effective_user.username or str(user_id)

    # Show that the bot is working
    await update.message.reply_text("Checking...")

    result = answer_support_ticket(user_message)
    track_metrics(result, user_message)

    if not result["needs_human"]:
        await update.message.reply_text(result["answer"])
    else:
        # To the customer: a message that a person is joining
        await update.message.reply_text(
            "Your question needs a specialist's attention. "
            "We'll reply within 2 business hours."
        )
        # To the team: a notification with the full context
        escalation_text = (
            f"Escalation from @{username} (id: {user_id})\n\n"
            f"Question: {user_message}\n\n"
            f"Escalation reason: {result['answer']}"
        )
        await bot.send_message(
            chat_id=SUPPORT_TEAM_CHAT,
            text=escalation_text
        )


def run_bot():
    application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
    application.add_handler(
        MessageHandler(filters.TEXT & ~filters.COMMAND, handle_support_message)
    )
    application.run_polling()


if __name__ == "__main__":
    run_bot()

def generate_support_report(metrics_file: str = "support-metrics.jsonl") -> dict:
    """Calculates the main metrics for the period"""
    entries = []
    with open(metrics_file) as f:
        for line in f:
            entries.append(json.loads(line))

    if not entries:
        return {"error": "no data"}

    total = len(entries)
    escalated = sum(1 for e in entries if e["escalated"])
    deflection_rate = round((total - escalated) / total * 100, 1)
    avg_response_time = round(
        sum(e["response_time_sec"] for e in entries) / total, 2
    )

    return {
        "total_tickets": total,
        "escalated": escalated,
        "auto_resolved": total - escalated,
        "deflection_rate_pct": deflection_rate,
        "avg_response_time_sec": avg_response_time
    }


# Example output:
# {
#   "total_tickets": 150,
#   "escalated": 28,
#   "auto_resolved": 122,
#   "deflection_rate_pct": 81.3,
#   "avg_response_time_sec": 2.4
# }