Agentes de voz con IA: Vapi, Retell, Bland.ai y llamadas reales

El cliente habla
  → STT (reconocimiento de voz): cientos de milisegundos
  → LLM (generación de la respuesta): cientos de milisegundos
  → TTS (síntesis de voz): cientos de milisegundos
  → El cliente escucha la respuesta
  
Total: hay que caber en más o menos un segundo; mientras más rápido, mejor

VAPI_KEY=your-vapi-key-here
VAPI_PHONE_NUMBER_ID=your-phone-number-id
ANTHROPIC_API_KEY=sk-ant-your-key-here

# voice_agent.py
import requests
import os
from dotenv import load_dotenv

load_dotenv()

VAPI_KEY = os.getenv("VAPI_KEY")
HEADERS = {"Authorization": f"Bearer {VAPI_KEY}"}

def create_lead_qualifier() -> str:
    """
    Crea un agente de voz para calificar leads.
    Devuelve el ID del assistant creado.
    """
    assistant = requests.post(
        "https://api.vapi.ai/assistant",
        headers=HEADERS,
        json={
            "name": "Lead Qualifier ES",
            "model": {
                "provider": "anthropic",
                "model": "claude-sonnet-5-5",   # revisa que el modelo esté en la lista de Vapi; modelos vigentes: página «Lo vigente»
                "messages": [{"role": "system", "content": """Eres un ejecutivo de ventas profesional.
Tu tarea es calificar un lead entrante en 3 a 5 minutos.

Haz estas cuatro preguntas a lo largo de la conversación:
1. ¿Qué presupuesto contempla el cliente?
2. ¿Qué plazos tiene: cuándo piensa tomar la decisión?
3. ¿Qué busca exactamente: requisitos concretos?
4. ¿Quién toma la decisión final?

Reglas de la conversación:
- Habla breve, amable, sin sonar a guion
- No hagas todas las preguntas seguidas: intégralas en el diálogo
- Si el cliente ya respondió una pregunta, no la repitas
- Al final, ofrece agendar una cita con un ejecutivo
- La conversación no debe durar más de 5 minutos
- Si te preguntan si eres una persona, di con honestidad que eres un asistente de IA"""}],
                "temperature": 0.7,
                "maxTokens": 150
            },
            "voice": {
                "provider": "11labs",
                "voiceId": "pNInz6obpgDQGcFmaJgB"
            },
            "firstMessage": "¡Buenas tardes! Le habla el asistente de IA de la empresa. Usted dejó una solicitud en nuestro sitio, ¿tiene un par de minutos para platicar?",
            "endCallMessage": "Perfecto, ya anoté todo lo necesario. Un ejecutivo se comunicará con usted en menos de una hora. ¡Que tenga buen día!",
            "maxDurationSeconds": 300
        }
    ).json()

    assistant_id = assistant["id"]
    print(f"Assistant creado: {assistant_id}")
    return assistant_id


def make_call(assistant_id: str, phone_number: str) -> dict:
    """
    Inicia una llamada al cliente.
    phone_number en formato E.164: +525512345678
    Solo puedes llamar a quien dio su consentimiento (ver la sección de leyes más abajo).
    """
    call = requests.post(
        "https://api.vapi.ai/call",
        headers=HEADERS,
        json={
            "assistantId": assistant_id,
            "phoneNumberId": os.getenv("VAPI_PHONE_NUMBER_ID"),   # el número desde el que llamamos
            "customer": {
                "number": phone_number,
                "name": "Cliente"  # opcional, para los logs
            }
        }
    ).json()

    print(f"Llamada iniciada: {call['id']}")
    print(f"Estado: {call['status']}")
    return call


def get_call_transcript(call_id: str) -> str:
    """
    Obtiene la transcripción de una llamada terminada.
    """
    call = requests.get(
        f"https://api.vapi.ai/call/{call_id}",
        headers=HEADERS
    ).json()

    if call.get("artifact", {}).get("transcript"):
        return call["artifact"]["transcript"]
    return ""


# Ejecución
if __name__ == "__main__":
    assistant_id = create_lead_qualifier()

    # Para probar, llámate a tu propio número
    call = make_call(assistant_id, "+525512345678")
    print(f"\nID de la llamada para obtener la transcripción: {call['id']}")

pip install requests python-dotenv

# knowledge_loader.py
from pathlib import Path

def load_knowledge_base(filename: str) -> str:
    """Lee el archivo con la base de conocimiento"""
    path = Path("knowledge") / filename
    if path.exists():
        return path.read_text(encoding="utf-8")
    return ""


def build_system_prompt() -> str:
    """Arma el prompt de sistema con la base de conocimiento"""

    services = load_knowledge_base("services.md")
    faq = load_knowledge_base("faq.md")
    objections = load_knowledge_base("objections.md")

    return f"""Eres asesor de una agencia inmobiliaria en Ecuador.

NUESTROS SERVICIOS Y PRECIOS:
{services}

PREGUNTAS FRECUENTES:
{faq}

CÓMO RESPONDER A LAS OBJECIONES:
{objections}

REGLAS:
- No menciones precios por encima de los indicados: siempre "desde X"
- Si la pregunta no está en las preguntas frecuentes, di "voy a confirmar los detalles y le devolvemos la llamada"
- Ofrece siempre agendar una cita o una videollamada de asesoría
- Habla en español, salvo que el cliente cambie de idioma
- No leas listas: integra la información en la conversación"""


# Ejemplo: crear un assistant con base de conocimiento
def create_realty_agent() -> str:
    import requests
    import os

    VAPI_KEY = os.getenv("VAPI_KEY")

    assistant = requests.post(
        "https://api.vapi.ai/assistant",
        headers={"Authorization": f"Bearer {VAPI_KEY}"},
        json={
            "name": "Realty Consultant",
            "model": {
                "provider": "anthropic",
                "model": "claude-sonnet-5-5",
                "messages": [{"role": "system", "content": build_system_prompt()}]
            },
            "voice": {
                "provider": "11labs",
                "voiceId": "pNInz6obpgDQGcFmaJgB"
            },
            "firstMessage": "¡Buenas tardes! Le habla el asesor de IA de la agencia. ¿En qué le puedo ayudar?"
        }
    ).json()

    return assistant["id"]

knowledge/
  services.md    — lista de servicios con precios
  faq.md         — 10 a 15 preguntas frecuentes con sus respuestas
  objections.md  — objeciones típicas y cómo responderlas

# webhook_handler.py
from flask import Flask, request, jsonify
import anthropic
import json
import os
from dotenv import load_dotenv

load_dotenv()

app = Flask(__name__)
claude = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))


def analyze_call(transcript: str) -> dict:
    """
    Claude analiza la transcripción y extrae datos estructurados.
    """
    response = "".join(b.text for b in claude.messages.create(
        model="claude-sonnet-5-5",   # modelos vigentes: página «Lo vigente»
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"""Analiza la conversación con el cliente y devuelve un JSON.

TRANSCRIPCIÓN:
{transcript}

Devuelve SOLO un JSON válido, sin explicaciones:
{{
  "qualified": true o false,
  "budget": "monto o rango, o 'no lo especificó'",
  "timeline": "cuándo piensa tomar la decisión",
  "requirements": "qué busca exactamente (2-3 oraciones)",
  "decision_maker": true o false,
  "next_action": "qué hacer después (por ejemplo: llamada de un ejecutivo, enviar presentación, sacar del embudo)",
  "sentiment": "positivo, neutral o negativo",
  "summary": "2-3 oraciones sobre la conversación"
}}"""
        }]
    ).content if b.type == "text")

    # Limpiamos el markdown si el modelo lo agregó
    if "```json" in response:
        response = response.split("```json")[1].split("```")[0]
    elif "```" in response:
        response = response.split("```")[1].split("```")[0]

    try:
        return json.loads(response.strip())
    except json.JSONDecodeError:
        return {
            "qualified": False,
            "summary": "No se pudo analizar la transcripción",
            "raw_response": response
        }


def save_to_crm(analysis: dict, customer: dict, call_id: str):
    """
    Guarda el resultado en tu sistema.
    Aquí cámbialo por tu CRM real: Notion, Airtable, Google Sheets, HubSpot.
    """
    record = {
        "call_id": call_id,
        "customer_phone": customer.get("number", "unknown"),
        "customer_name": customer.get("name", "unknown"),
        **analysis
    }

    # Opción sencilla: guardar en un archivo JSON
    import datetime
    filename = f"calls/call-{call_id}-{datetime.date.today()}.json"
    os.makedirs("calls", exist_ok=True)

    with open(filename, "w", encoding="utf-8") as f:
        json.dump(record, f, ensure_ascii=False, indent=2)

    print(f"Guardado: {filename}")

    # Si está calificado, avisa al ejecutivo
    if analysis.get("qualified"):
        notify_manager(record)


def notify_manager(record: dict):
    """
    Envía un aviso al ejecutivo sobre un lead calificado.
    Cámbialo por un canal real: WhatsApp Business, Slack, email.
    """
    # Ejemplo: imprimir en consola (en producción, la API de tu canal de mensajería)
    print(f"""
=== LEAD CALIFICADO ===
Teléfono: {record['customer_phone']}
Presupuesto: {record.get('budget', 'no lo especificó')}
Plazos: {record.get('timeline', 'no lo especificó')}
Siguiente paso: {record.get('next_action', '')}
Resumen: {record.get('summary', '')}
==============================
    """)


@app.post("/vapi-webhook")
def handle_vapi_event():
    """Manejador principal de los eventos de Vapi"""
    # Vapi pone el evento dentro de la clave "message"
    message = (request.json or {}).get("message", {})
    event_type = message.get("type")

    if event_type == "end-of-call-report":
        call = message.get("call", {})
        call_id = call.get("id", "unknown")
        customer = call.get("customer", {})
        transcript = message.get("artifact", {}).get("transcript", "")

        print(f"Llamada terminada: {call_id}")
        print(f"Motivo de término: {message.get('endedReason', 'unknown')}")

        if transcript:
            print("Analizando la transcripción...")
            analysis = analyze_call(transcript)
            save_to_crm(analysis, customer, call_id)
        else:
            print("Transcripción vacía: la llamada no se concretó o el cliente no contestó")

    elif event_type == "status-update":
        call_id = message.get("call", {}).get("id", "unknown")
        print(f"Estado de la llamada {call_id}: {message.get('status', '?')}")

    return jsonify({"status": "ok"})


if __name__ == "__main__":
    # En producción protege el webhook con un secreto en el encabezado y no actives debug
    app.run(port=5000, debug=False)

pip install flask anthropic requests python-dotenv

# Inicia el servidor del webhook
python webhook_handler.py

# En otra terminal, expón el puerto con ngrok para probar
ngrok http 5000
# ngrok te dará una URL pública como https://abc123.ngrok.io

# Actualiza el assistant para que envíe los eventos a tu webhook
import requests

VAPI_KEY = "your-key"
ASSISTANT_ID = "your-assistant-id"
WEBHOOK_URL = "https://abc123.ngrok.io/vapi-webhook"

requests.patch(
    f"https://api.vapi.ai/assistant/{ASSISTANT_ID}",
    headers={"Authorization": f"Bearer {VAPI_KEY}"},
    json={"server": {"url": WEBHOOK_URL}}   # en ejemplos viejos aparece el campo serverUrl
)

# auto_caller.py
"""
Script para llamadas automáticas.
Conéctalo al webhook de tu CRM o de tu formulario de contacto.
"""
import requests
import os
from dotenv import load_dotenv

load_dotenv()

VAPI_KEY = os.getenv("VAPI_KEY")
ASSISTANT_ID = os.getenv("VAPI_ASSISTANT_ID")
PHONE_NUMBER_ID = os.getenv("VAPI_PHONE_NUMBER_ID")


def call_new_lead(phone: str, name: str = "", source: str = "") -> str:
    """
    Llama a un lead nuevo justo después de que envía la solicitud.
    Devuelve el call_id para darle seguimiento.
    """
    call = requests.post(
        "https://api.vapi.ai/call",
        headers={"Authorization": f"Bearer {VAPI_KEY}"},
        json={
            "assistantId": ASSISTANT_ID,
            "phoneNumberId": PHONE_NUMBER_ID,
            "customer": {
                "number": phone,
                "name": name
            },
            # Pasamos contexto al agente con overrides
            "assistantOverrides": {
                "variableValues": {
                    "lead_source": source,
                    "lead_name": name
                }
            }
        }
    ).json()

    return call.get("id", "")


def batch_call(leads: list) -> list:
    """
    Llama a una lista de leads.
    leads = [{"phone": "+52...", "name": "...", "source": "..."}, ...]
    """
    results = []

    for lead in leads:
        print(f"Llamando: {lead['name']} ({lead['phone']})")
        call_id = call_new_lead(
            phone=lead["phone"],
            name=lead.get("name", ""),
            source=lead.get("source", "")
        )
        results.append({
            "lead": lead,
            "call_id": call_id
        })

        # Pausa entre llamadas para no saturar
        import time
        time.sleep(2)

    return results


# Ejemplo: llamadas después de exportar del CRM
if __name__ == "__main__":
    today_leads = [
        {"phone": "+525511111111", "name": "Juan Pérez", "source": "Sitio web"},
        {"phone": "+525522222222", "name": "María González", "source": "Publicidad"},
    ]

    results = batch_call(today_leads)
    print(f"\nLlamadas iniciadas: {len(results)}")
    for r in results:
        print(f"  {r['lead']['name']}: call_id = {r['call_id']}")