Call Support AI — Vapi + Bland.ai, голосовые звонки в поддержке

Клиент звонит → Vapi перехватывает → STT (речь в текст) → LLM (думает) → TTS (текст в речь) → клиент слышит ответ

import requests
import json

VAPI_API_KEY = "your_vapi_key"

def create_support_agent(
    company_name: str,
    knowledge_base: str,
    transfer_phone: str  # Номер живого оператора
) -> dict:
    """Создаёт голосового агента поддержки через Vapi"""

    headers = {
        "Authorization": f"Bearer {VAPI_API_KEY}",
        "Content-Type": "application/json"
    }

    # Названия полей и моделей в Vapi меняются: перед запуском сверяй конфигурацию
    # с документацией docs.vapi.ai (модель, голос, распознавание, перевод звонка)
    agent_config = {
        "name": f"{company_name} Support Agent",
        "model": {
            "provider": "anthropic",
            "model": "claude-sonnet-5-5",
            "messages": [{"role": "system", "content": f"""Ты агент поддержки клиентов компании {company_name}.

БАЗА ЗНАНИЙ:
{knowledge_base}

ПРАВИЛА:
1. Будь дружелюбным и профессиональным
2. Отвечай только по базе знаний — не выдумывай
3. Если не знаешь ответа — честно скажи и предложи соединить с оператором
4. Всегда уточняй понял ли клиент
5. Если клиент злится — прояви сочувствие, не спорь

ПЕРЕВОД НА ОПЕРАТОРА (скажи это дословно):
"Позвольте соединить вас с нашим специалистом, который поможет с этим вопросом. 
Ожидайте, пожалуйста."

Говори естественно. Делай паузы. Не тараторь."""}],
            "temperature": 0.3,  # Меньше вариативности для поддержки
            # Перевод звонка на живого оператора: инструмент transferCall
            "tools": [{
                "type": "transferCall",
                "destinations": [{
                    "type": "number",
                    "number": transfer_phone,
                    "description": "Передать звонок оператору, если AI не знает ответа или клиент просит человека",
                    "message": "Соединяю вас со специалистом"
                }]
            }]
        },
        "voice": {
            "provider": "11labs",  # ElevenLabs
            "voiceId": "ID_голоса_из_ElevenLabs"  # выбери голос, который умеет по-русски
        },
        "transcriber": {
            "provider": "deepgram",
            "language": "ru"
        },
        "firstMessage": f"Здравствуйте! Вы позвонили в поддержку {company_name}. Чем могу помочь?",
        "endCallMessage": "Спасибо за обращение! Если возникнут вопросы — звоните. До свидания!",
        "maxDurationSeconds": 600  # Максимум 10 минут
    }

    response = requests.post(
        "https://api.vapi.ai/assistant",
        headers=headers,
        json=agent_config
    )

    return response.json()


def assign_phone_number(assistant_id: str, phone_number: str) -> dict:
    """Привязывает телефонный номер к агенту"""

    headers = {
        "Authorization": f"Bearer {VAPI_API_KEY}",
        "Content-Type": "application/json"
    }

    response = requests.post(
        "https://api.vapi.ai/phone-number",
        headers=headers,
        json={
            "assistantId": assistant_id,
            "number": phone_number,
            "provider": "twilio"  # или vonage, telnyx
        }
    )

    return response.json()


# Пример: агент для интернет-магазина
knowledge_base = """
ВОПРОСЫ О ЗАКАЗАХ:
- Статус заказа: попросите номер заказа, зайдите на track.example.com/[номер]
- Срок доставки: Москва 1-2 дня, регионы 3-5 дней, международно 7-14 дней
- Изменить адрес: возможно до момента отправки, после — нет

ВОПРОСЫ ОБ ОПЛАТЕ:
- Способы: карта Visa/MC, SberPay, YooMoney, наложенный платёж
- Возврат: 14 дней по нашей политике возврата
- Чек: отправляется на email автоматически

ТЕХНИЧЕСКИЕ ПРОБЛЕМЫ:
- Не работает сайт: попробуйте очистить кэш браузера
- Не пришёл email: проверьте папку Спам
- Не можете войти в аккаунт: сброс пароля на example.com/reset

НУЖЕН ОПЕРАТОР:
- Жалобы на качество товара
- Юридические вопросы
- Ситуации требующие компенсации
"""

agent = create_support_agent(
    company_name="Мой Магазин",
    knowledge_base=knowledge_base,
    transfer_phone="+79991234567"
)

print(f"Агент создан: {agent.get('id')}")
print(f"Имя: {agent.get('name')}")

import requests

BLAND_API_KEY = "your_bland_key"

def make_batch_calls(
    contacts: list[dict],  # [{"phone": "+7...", "name": "Иван", ...}]
    call_purpose: str,
    script_template: str
) -> dict:
    """Запускает массовый обзвон через Bland.ai"""

    # Схема запросов Bland меняется: сверяй поля и адрес с документацией docs.bland.ai
    headers = {
        "Authorization": BLAND_API_KEY,
        "Content-Type": "application/json"
    }

    # Формируем задачи для каждого контакта
    call_objects = []
    for contact in contacts:
        # Персонализируем скрипт для каждого человека
        personalized_script = script_template.format(**contact)

        call_objects.append({
            "phone_number": contact["phone"],
            "task": personalized_script
        })

    # Запускаем батч: общие настройки идут в global, звонки в call_objects
    response = requests.post(
        "https://api.bland.ai/v2/batches/create",
        headers=headers,
        json={
            "global": {
                "task": script_template,           # запасной сценарий
                "voice": "ID_голоса_из_кабинета",  # подбери русскоязычный голос в кабинете Bland
                "language": "ru",                  # коды языков смотри в документации
                "max_duration": 2,                 # максимум 2 минуты
                "record": True                     # записывать звонок (см. предупреждение ниже)
            },
            "call_objects": call_objects,
            "description": call_purpose[:60]
        }
    )

    return response.json()


# Пример: напоминания о записи
contacts_for_reminder = [
    {"phone": "+79991111111", "name": "Иван", "date": "завтра", "time": "14:00", "service": "стрижка"},
    {"phone": "+79992222222", "name": "Мария", "date": "в среду", "time": "11:30", "service": "маникюр"},
]

reminder_script = """
Добрый день, {name}! Звоним из салона Красота.
Напоминаем, что {date} в {time} у вас запись на {service}.
Если планы изменились — пожалуйста, позвоните нам или напишите в WhatsApp.
Ждём вас! До свидания.
"""

result = make_batch_calls(
    contacts=contacts_for_reminder,
    call_purpose="Напоминания о записи — 2026-10-15",
    script_template=reminder_script
)

print(f"Батч создан: {result.get('batch_id')}")
print(f"Звонков запланировано: {len(contacts_for_reminder)}")

import anthropic
from typing import Optional

# Простая RAG-система для голосового агента
def create_rag_support_handler(
    documents: list[str],  # Список документов базы знаний
) -> callable:
    """Создаёт обработчик вопросов с поиском по базе знаний"""

    client = anthropic.Anthropic()

    # В реальном проекте здесь будет векторная БД (Pinecone, Chroma)
    # Для простоты — передаём все документы в контекст
    knowledge_context = "\n\n---\n\n".join(documents)

    def handle_voice_query(
        user_query: str,
        conversation_history: list[dict],
        customer_info: Optional[dict] = None
    ) -> dict:
        """Обрабатывает запрос от голосового агента"""

        system_prompt = f"""Ты агент поддержки. Отвечай ТОЛЬКО на основе базы знаний.
Если информации нет — честно скажи что не знаешь и предложи соединить с оператором.

ВАЖНО ДЛЯ ГОЛОСОВОГО ОТВЕТА:
- Ответ должен звучать естественно при произношении вслух
- Не используй маркеры списков (•, -, *) — они не звучат хорошо
- Числа пиши словами: не "14:00", а "в два часа дня"
- Максимум 3-4 предложения на один ответ

БАЗА ЗНАНИЙ:
{knowledge_context}"""

        messages = conversation_history + [{
            "role": "user",
            "content": user_query
        }]

        response = client.messages.create(
            model="claude-sonnet-5-5",
            max_tokens=200,  # Голосовые ответы короткие
            system=system_prompt,
            messages=messages
        )

        answer = "".join(b.text for b in response.content if b.type == "text")

        # Детектируем нужен ли перевод на оператора
        transfer_signals = [
            "не знаю", "не могу помочь", "нужен специалист",
            "соединю с оператором", "передам вашу ситуацию"
        ]
        needs_transfer = any(signal in answer.lower() for signal in transfer_signals)

        return {
            "response_text": answer,
            "needs_human_transfer": needs_transfer,
            "transfer_reason": "AI не смог ответить" if needs_transfer else None
        }

    return handle_voice_query


# Пример использования
faq_documents = [
    """ВОЗВРАТ ТОВАРА
    Вернуть товар можно в течение четырнадцати дней с момента покупки.
    Товар должен быть в оригинальной упаковке и без следов использования.
    Деньги возвращаем в течение трёх рабочих дней на карту с которой оплачивали.""",

    """ДОСТАВКА
    По Москве доставляем от одного до двух дней.
    По России от трёх до семи дней.
    Доставка бесплатна при заказе от трёх тысяч рублей.
    Отслеживать можно на сайте в разделе Мои заказы.""",

    """ОПЛАТА
    Принимаем карты Visa, MasterCard и МИР.
    Также можно оплатить через СберПэй или наличными при получении.
    Рассрочку предоставляем через банк-партнёр на три, шесть или двенадцать месяцев."""
]

handler = create_rag_support_handler(faq_documents)

# Тест диалога
conversation = []
queries = [
    "хочу вернуть товар купленный неделю назад",
    "а если упаковка не сохранилась?"
]

for query in queries:
    print(f"\nКлиент: {query}")
    result = handler(query, conversation)
    print(f"AI: {result['response_text']}")
    if result['needs_human_transfer']:
        print("⚠️ ПЕРЕВОД НА ОПЕРАТОРА")

    # Обновляем историю разговора
    conversation.append({"role": "user", "content": query})
    conversation.append({"role": "assistant", "content": result['response_text']})

def analyze_call_transcripts(transcripts: list[str]) -> dict:
    """Анализирует пачку транскриптов звонков и находит паттерны"""

    client = anthropic.Anthropic()

    # Объединяем транскрипты с разделителями
    combined = "\n\n=== ЗВОНОК ===\n\n".join(transcripts[:20])  # Первые 20

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": f"""Проанализируй транскрипты звонков в поддержку.

ТРАНСКРИПТЫ:
{combined}

Нужно:
1. **Топ-10 вопросов** которые задают чаще всего (с процентом)
2. **Топ-5 проблем** которые AI не смог решить (нужен человек)
3. **Среднее время разговора** по категориям
4. **Эмоциональный тон**: сколько злых/довольных/нейтральных клиентов
5. **Рекомендации**: что добавить в базу знаний чтобы AI решал больше
6. **Красные флаги**: что срочно нужно исправить в продукте/сервисе

Формат: структурированный отчёт с цифрами."""
        }]
    )

    return "".join(b.text for b in response.content if b.type == "text")


# Стоимость vs ROI
def calculate_roi(
    current_calls_per_month: int,
    avg_call_duration_minutes: float,
    operator_hourly_cost: float,
    ai_cost_per_minute: float,  # сумма пяти статей стоимости минуты, посчитай по своему стеку
    ai_deflection_rate: float = 0.75  # допущение для примера: замерь на своих звонках
) -> dict:
    """Считает ROI от внедрения голосового AI (все числа в примере вымышленные)"""

    # Текущие расходы
    total_minutes = current_calls_per_month * avg_call_duration_minutes
    current_cost = (total_minutes / 60) * operator_hourly_cost

    # После AI
    ai_handled = current_calls_per_month * ai_deflection_rate
    human_handled = current_calls_per_month * (1 - ai_deflection_rate)

    ai_cost = ai_handled * avg_call_duration_minutes * ai_cost_per_minute
    human_cost = (human_handled * avg_call_duration_minutes / 60) * operator_hourly_cost

    new_total_cost = ai_cost + human_cost
    savings = current_cost - new_total_cost

    return {
        "current_monthly_cost": round(current_cost, 2),
        "new_monthly_cost": round(new_total_cost, 2),
        "monthly_savings": round(savings, 2),
        "annual_savings": round(savings * 12, 2),
        "ai_handled_calls": int(ai_handled),
        "human_calls_freed": int(ai_handled),
        "roi_percent": round((savings / new_total_cost) * 100, 1)
    }

# Пример расчёта
roi = calculate_roi(
    current_calls_per_month=2000,
    avg_call_duration_minutes=4.5,
    operator_hourly_cost=15,   # допущение для примера: подставь реальную ставку
    ai_cost_per_minute=0.20    # допущение для примера: подставь сумму по своему стеку
)

print("=== ROI АНАЛИЗ ===")
for k, v in roi.items():
    print(f"{k}: {v}")

# Идём на vapi.ai
# Создаём аккаунт
# Получаем API ключ в Dashboard → API Keys
# Сохраняем в .env:

echo "VAPI_API_KEY=your_key_here" >> .env

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

load_dotenv()
VAPI_API_KEY = os.environ["VAPI_API_KEY"]

# Создаём агента
agent = requests.post(
    "https://api.vapi.ai/assistant",
    headers={"Authorization": f"Bearer {VAPI_API_KEY}"},
    json={
        "name": "Тестовый агент поддержки",
        "model": {
            "provider": "anthropic",
            "model": "claude-sonnet-5-5",
            "messages": [{"role": "system", "content": """Ты агент поддержки тестового магазина.

Ты знаешь:
- Доставка: 2-5 дней по России, бесплатно от 3000 рублей
- Возврат: 14 дней, товар в упаковке
- Оплата: карты, СберПэй, наличные при получении
- График: пн-пт 9-18, сб 10-15

Если вопрос не по теме — скажи что соединяешь с оператором.
Говори кратко и по делу. Будь дружелюбным."""}]
        },
        "voice": {
            "provider": "11labs",
            "voiceId": "ID_голоса_из_ElevenLabs"
        },
        "firstMessage": "Добрый день! Тестовая поддержка. Как могу помочь?"
    }
).json()

print(f"ID агента: {agent['id']}")
print(f"Ссылка для тестового звонка через браузер:")
print(f"Используйте Dashboard → Test Phone Call")

# get_call_analytics.py
import requests
import os

VAPI_API_KEY = os.environ["VAPI_API_KEY"]

# Получаем список последних звонков
calls = requests.get(
    "https://api.vapi.ai/call",
    headers={"Authorization": f"Bearer {VAPI_API_KEY}"},
    params={"limit": 10}
).json()

for call in calls.get("results", []):
    print(f"\nЗвонок: {call.get('id')}")
    print(f"Длительность: {call.get('duration', 0)} сек")
    print(f"Статус: {call.get('status')}")
    if call.get('transcript'):
        print(f"Транскрипт:\n{call['transcript'][:300]}...")