Atención al cliente con IA: sistema de tickets, RAG sobre la base de conocimiento y soporte inteligente

El cliente escribe una pregunta
    ↓
Clasificación (Claude): típica / atípica / crítica
    ↓
Típica → RAG sobre la base de conocimiento → respuesta automática (la mayor parte, referencia 80%)
Atípica → borrador de la IA para el agente → el agente edita → envía (referencia 15%)
Crítica → escalamiento inmediato + aviso → el agente responde él mismo (referencia 5%)

knowledge-base/
  billing.md          — preguntas sobre pagos
  shipping.md         — envíos y devoluciones
  integrations.md     — conectar integraciones
  plans.md            — planes
  account.md          — administración de la cuenta

## Cómo cambiar de plan

Entra a tu cuenta → Configuración → Plan → elige el plan nuevo.
El cambio se aplica de inmediato. La diferencia se cobra de forma proporcional.

## Reembolsos

El reembolso es posible dentro de los 14 días posteriores al pago en el primer pedido.
Para pedirlo, escribe a [email protected] con el asunto "Reembolso" y el número de pedido.
Plazo de procesamiento: 5 días hábiles.

## En qué se diferencian los planes Basic y Pro

Basic: hasta 5 usuarios, 10 GB de almacenamiento, soporte por correo.
Pro: usuarios ilimitados, 100 GB de almacenamiento, soporte prioritario, acceso a la API.

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:
    """Carga la base de conocimiento desde archivos MD"""
    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:
    """Clasifica el ticket y responde o escala"""

    start_time = time.time()

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=800,
        system=f"""Eres un agente de IA de atención al cliente.

BASE DE CONOCIMIENTO DE LA EMPRESA:
{KB}

Reglas:
1. Si la respuesta está en la base de conocimiento, responde con exactitud usándola, no agregues nada propio
2. Al final de la respuesta indica: Fuente: [nombre de la sección de la base de conocimiento]
3. Si la respuesta no está en la base de conocimiento, escribe ESCALATE en la primera línea y explica por qué se necesita una persona
4. Si la pregunta es sobre un reembolso, escribe SIEMPRE ESCALATE (aunque la respuesta esté en la base)
5. Si la pregunta es sobre una falla técnica del cliente: ESCALATE
6. Tono: amable, concreto, sin relleno""",
        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):
    """Escribe las métricas en un 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")


# Ejemplo de uso
if __name__ == "__main__":
    questions = [
        "¿Cómo cambio de plan?",
        "Quiero que me devuelvan el dinero de la suscripción",
        "No me funciona la integración con Slack, se rompió todo"
    ]

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

        if result["needs_human"]:
            print(f"ESCALAMIENTO -> se pasa a un agente humano")
            print(f"Motivo: {result['answer']}")
        else:
            print(f"Respuesta automática ({result['response_time_sec']} s):")
            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"):
    """Asigna el ticket a un agente humano y le pone una etiqueta"""
    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  # lo asigna al equipo, no a un agente concreto
        }
    )

    # Ponemos la etiqueta de prioridad
    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"]

    # Responde la IA
    result = answer_support_ticket(message)
    track_metrics(result, message)

    if not result["needs_human"]:
        # Mandamos la respuesta automática a nombre del 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:
        # Le avisamos al cliente que vamos a conectar a una persona
        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": "Tu pregunta pasó a un especialista. Te respondemos en un plazo de 2 horas."
            }
        )
        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)

    # Indicador de que el bot está trabajando
    await update.message.reply_text("Revisando...")

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

    if not result["needs_human"]:
        await update.message.reply_text(result["answer"])
    else:
        # Al cliente: mensaje de que conectamos a una persona
        await update.message.reply_text(
            "Tu pregunta necesita la atención de un especialista. "
            "Te respondemos en un plazo de 2 horas en horario laboral."
        )
        # Al equipo: aviso con todo el contexto
        escalation_text = (
            f"Escalamiento de @{username} (id: {user_id})\n\n"
            f"Pregunta: {user_message}\n\n"
            f"Motivo del escalamiento: {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:
    """Calcula las métricas principales del periodo"""
    entries = []
    with open(metrics_file) as f:
        for line in f:
            entries.append(json.loads(line))

    if not entries:
        return {"error": "no hay datos"}

    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
    }


# Ejemplo de salida:
# {
#   "total_tickets": 150,
#   "escalated": 28,
#   "auto_resolved": 122,
#   "deflection_rate_pct": 81.3,
#   "avg_response_time_sec": 2.4
# }