Голосовые AI-агенты — Vapi, Retell, Bland.ai, реальные звонки

Клиент говорит
  → STT (распознавание речи): сотни миллисекунд
  → LLM (генерация ответа): сотни миллисекунд
  → TTS (синтез голоса): сотни миллисекунд
  → Клиент слышит ответ
  
Итого: нужно уложиться примерно в секунду; чем быстрее, тем лучше

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:
    """
    Создаёт голосового агента для квалификации лидов.
    Возвращает ID созданного ассистента.
    """
    assistant = requests.post(
        "https://api.vapi.ai/assistant",
        headers=HEADERS,
        json={
            "name": "Lead Qualifier RU",
            "model": {
                "provider": "anthropic",
                "model": "claude-sonnet-5-5",   # проверь, что модель есть в списке Vapi; актуальные модели: страница «Актуальное сейчас»
                "messages": [{"role": "system", "content": """Ты профессиональный менеджер по продажам.
Твоя задача — квалифицировать входящий лид за 3-5 минут.

Задай эти четыре вопроса в ходе разговора:
1. Какой бюджет рассматривает клиент?
2. Какие сроки — когда планирует принять решение?
3. Что именно ищет — конкретные требования?
4. Кто принимает финальное решение?

Правила разговора:
- Говори кратко, дружелюбно, без скриптового звучания
- Не задавай все вопросы подряд — вплетай в диалог
- Если клиент уже ответил на вопрос — не повторяй его
- В конце предложи записаться на встречу с менеджером
- Разговор не дольше 5 минут
- Если спросят, ты ли человек, честно скажи, что ты AI-ассистент"""}],
                "temperature": 0.7,
                "maxTokens": 150
            },
            "voice": {
                "provider": "11labs",
                "voiceId": "pNInz6obpgDQGcFmaJgB"
            },
            "firstMessage": "Добрый день! Это AI-ассистент компании. Вы оставили заявку на нашем сайте — удобно поговорить пару минут?",
            "endCallMessage": "Отлично, я записал всё что нужно. Наш менеджер свяжется с вами в течение часа. Хорошего дня!",
            "maxDurationSeconds": 300
        }
    ).json()

    assistant_id = assistant["id"]
    print(f"Ассистент создан: {assistant_id}")
    return assistant_id


def make_call(assistant_id: str, phone_number: str) -> dict:
    """
    Инициирует звонок клиенту.
    phone_number в формате E.164: +79161234567
    Звонить можно только тем, кто дал согласие (см. раздел о законах ниже).
    """
    call = requests.post(
        "https://api.vapi.ai/call",
        headers=HEADERS,
        json={
            "assistantId": assistant_id,
            "phoneNumberId": os.getenv("VAPI_PHONE_NUMBER_ID"),   # номер, с которого звоним
            "customer": {
                "number": phone_number,
                "name": "Клиент"  # опционально, для логов
            }
        }
    ).json()

    print(f"Звонок инициирован: {call['id']}")
    print(f"Статус: {call['status']}")
    return call


def get_call_transcript(call_id: str) -> str:
    """
    Получает транскрипт завершённого звонка.
    """
    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 ""


# Запуск
if __name__ == "__main__":
    assistant_id = create_lead_qualifier()

    # Для теста — звоним на свой номер
    call = make_call(assistant_id, "+79161234567")
    print(f"\nID звонка для получения транскрипта: {call['id']}")

pip install requests python-dotenv

# knowledge_loader.py
from pathlib import Path

def load_knowledge_base(filename: str) -> str:
    """Читает файл с базой знаний"""
    path = Path("knowledge") / filename
    if path.exists():
        return path.read_text(encoding="utf-8")
    return ""


def build_system_prompt() -> str:
    """Собирает системный промпт с базой знаний"""

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

    return f"""Ты консультант агентства недвижимости в Эквадоре.

НАШИ УСЛУГИ И ЦЕНЫ:
{services}

ЧАСТО ЗАДАВАЕМЫЕ ВОПРОСЫ:
{faq}

КАК ОТВЕЧАТЬ НА ВОЗРАЖЕНИЯ:
{objections}

ПРАВИЛА:
- Не называй цены выше указанных — всегда "от X"
- Если вопрос не в FAQ — говори "уточню детали и перезвоним"
- Всегда предлагай записаться на встречу или видеоконсультацию
- Говори на русском, если клиент не переходит на другой язык
- Не читай списком — вплетай информацию в разговор"""


# Пример: создать ассистента с базой знаний
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": "Добрый день! Это AI-консультант агентства. Чем могу помочь?"
        }
    ).json()

    return assistant["id"]

knowledge/
  services.md    — список услуг с ценами
  faq.md         — 10-15 частых вопросов с ответами
  objections.md  — типичные возражения и как на них отвечать

# 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 анализирует транскрипт и извлекает структурированные данные.
    """
    response = "".join(b.text for b in claude.messages.create(
        model="claude-sonnet-5-5",   # актуальные модели: страница «Актуальное сейчас»
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"""Проанализируй разговор с клиентом и верни JSON.

ТРАНСКРИПТ:
{transcript}

Верни ТОЛЬКО валидный JSON без пояснений:
{{
  "qualified": true или false,
  "budget": "сумма или диапазон, или 'не уточнил'",
  "timeline": "когда планирует принять решение",
  "requirements": "что именно ищет (2-3 предложения)",
  "decision_maker": true или false,
  "next_action": "что делать дальше (например: позвонить менеджеру, отправить презентацию, убрать из воронки)",
  "sentiment": "позитивный, нейтральный или негативный",
  "summary": "2-3 предложения о разговоре"
}}"""
        }]
    ).content if b.type == "text")

    # Очищаем от markdown если модель добавила
    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": "Не удалось проанализировать транскрипт",
            "raw_response": response
        }


def save_to_crm(analysis: dict, customer: dict, call_id: str):
    """
    Сохраняет результат в вашу систему.
    Здесь замени на реальный CRM: Notion, Airtable, Google Sheets, AmoCRM.
    """
    record = {
        "call_id": call_id,
        "customer_phone": customer.get("number", "unknown"),
        "customer_name": customer.get("name", "unknown"),
        **analysis
    }

    # Простой вариант — сохранить в 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"Сохранено: {filename}")

    # Если квалифицирован — отправь уведомление менеджеру
    if analysis.get("qualified"):
        notify_manager(record)


def notify_manager(record: dict):
    """
    Отправляет уведомление менеджеру о квалифицированном лиде.
    Замени на реальный канал: Telegram, Slack, email.
    """
    # Пример: вывод в консоль (в продакшене — Telegram Bot API)
    print(f"""
=== КВАЛИФИЦИРОВАННЫЙ ЛИД ===
Телефон: {record['customer_phone']}
Бюджет: {record.get('budget', 'не уточнил')}
Сроки: {record.get('timeline', 'не уточнил')}
Следующий шаг: {record.get('next_action', '')}
Резюме: {record.get('summary', '')}
==============================
    """)


@app.post("/vapi-webhook")
def handle_vapi_event():
    """Основной обработчик событий от Vapi"""
    # Vapi кладёт событие внутрь ключа "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"Звонок завершён: {call_id}")
        print(f"Причина завершения: {message.get('endedReason', 'unknown')}")

        if transcript:
            print("Анализирую транскрипт...")
            analysis = analyze_call(transcript)
            save_to_crm(analysis, customer, call_id)
        else:
            print("Транскрипт пустой — звонок не состоялся или клиент не ответил")

    elif event_type == "status-update":
        call_id = message.get("call", {}).get("id", "unknown")
        print(f"Статус звонка {call_id}: {message.get('status', '?')}")

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


if __name__ == "__main__":
    # В продакшне закрой webhook секретом в заголовке и не включай debug
    app.run(port=5000, debug=False)

pip install flask anthropic requests python-dotenv

# Запусти webhook сервер
python webhook_handler.py

# В другом терминале — пробрось порт через ngrok для тестирования
ngrok http 5000
# ngrok даст публичный URL вроде https://abc123.ngrok.io

# Обнови ассистента чтобы он отправлял события на твой 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}}   # в старых примерах встречается поле serverUrl
)

# auto_caller.py
"""
Скрипт для автоматического обзвона.
Подключи к вебхуку вашей CRM или формы обратной связи.
"""
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:
    """
    Звонит новому лиду сразу после подачи заявки.
    Возвращает call_id для отслеживания.
    """
    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
            },
            # Передаём контекст агенту через overrides
            "assistantOverrides": {
                "variableValues": {
                    "lead_source": source,
                    "lead_name": name
                }
            }
        }
    ).json()

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


def batch_call(leads: list) -> list:
    """
    Обзванивает список лидов.
    leads = [{"phone": "+7...", "name": "...", "source": "..."}, ...]
    """
    results = []

    for lead in leads:
        print(f"Звоним: {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
        })

        # Пауза между звонками чтобы не перегружать
        import time
        time.sleep(2)

    return results


# Пример: обзвон после выгрузки из CRM
if __name__ == "__main__":
    today_leads = [
        {"phone": "+79161111111", "name": "Иван Иванов", "source": "Сайт"},
        {"phone": "+79162222222", "name": "Мария Петрова", "source": "Реклама"},
    ]

    results = batch_call(today_leads)
    print(f"\nЗапущено звонков: {len(results)}")
    for r in results:
        print(f"  {r['lead']['name']}: call_id = {r['call_id']}")