Телефонный звонок → структурированный JSON: строим STT + LLM пайплайн на Python

Каждый день в российском бизнесе происходят миллионы телефонных звонков. Колл-центры, клиники, юридические конторы, отделы продаж — везде, где есть телефон, есть поток неструктурированных данных, который никто не обрабатывает. Менеджер повесил трубку, записал в CRM «клиент интересовался» — и 80% информации из разговора потерялось.

Я потратил полгода на то, чтобы построить пайплайн, который берёт аудиозапись телефонного звонка и выдаёт структурированный JSON: кто звонил, чего хотел, какие суммы называл, что договорились делать дальше. В процессе набил достаточно шишек, чтобы написать эту статью.

Здесь не будет теории из документации. Будут конкретные решения, рабочий код на Python и грабли, на которые я наступил, чтобы вам не пришлось.

Архитектура: четыре этапа, один сюрприз

Пайплайн выглядит просто:

Аудио (PCM 16kHz) → ASR → Диаризация → LLM → JSON

На практике «просто» заканчивается после первого pip install. Каждый этап имеет свои подводные камни, а самый неожиданный — взаимодействие между этапами. Ошибка STT на 3% каскадно снижает точность LLM-извлечения на 10-15%.

Этап 1. Аудио: почему 8 кГц — это боль

Источник аудио — SIP-транк облачной АТС. Большинство провайдеров (Mango, Zadarma, UIS) отдают записи через API или webhook. Формат на входе — обычно PCM 16kHz mono или G.711 (8kHz).

И вот тут первые грабли.

G.711 (8 кГц). Телефонный стандарт, придуманный в 1972 году. Частотный диапазон — 300-3400 Гц. Человеческая речь содержит информативные компоненты до 8000 Гц. Итог: STT-модели, обученные на широкополосном аудио, на 8 кГц показывают WER на 5-8 процентных пунктов хуже.

Апсемплинг. Наивный librosa.resample(audio, orig_sr=8000, target_sr=16000) не добавляет информации — он просто интерполирует сэмплы. Но! Модели типа Whisper обучены на 16 кГц, и подача 8 кГц напрямую ломает их внутренние фильтры. Апсемплинг даёт +2-3% к точности просто за счёт корректного формата.

Моно vs стерео. Если АТС отдаёт стерео (левый канал = оператор, правый = клиент) — считайте, что вам повезло. Задача диаризации решена бесплатно. На практике 70% АТС отдают моно.

Минимальный код для приёма записи:

from fastapi import FastAPI, Request
import aiohttp
import aiofiles

app = FastAPI()

@app.post("/webhook/call-recording")
async def receive_recording(request: Request):
    body = await request.json()
    audio_url = body["recording_url"]
    call_id = body["call_id"]

    async with aiohttp.ClientSession() as session:
        async with session.get(audio_url) as resp:
            audio_bytes = await resp.read()

    path = f"/data/calls/{call_id}.wav"
    async with aiofiles.open(path, "wb") as f:
        await f.write(audio_bytes)

    await pipeline.enqueue(call_id, path)
    return {"status": "accepted"}

Грабля, на которую я наступил: webhook может прийти раньше, чем запись дозаписалась на стороне АТС. Файл будет обрезан. Решение: retry с проверкой длительности через ffprobe — если аудио короче 5 секунд, ждём 10 секунд и скачиваем повторно.

Этап 2. Speech-to-Text: Whisper, SpeechKit и честные бенчмарки

Я протестировал четыре варианта на корпусе из 200 телефонных записей на русском языке (средняя длительность 3.5 минуты, качество — типичный мобильный звонок):

Несколько неочевидных наблюдений:

Whisper врёт красиво. Когда Whisper не уверен, он не ставит [inaudible] — он генерирует правдоподобный, но неверный текст. Фраза «давайте встретимся в среду в три» может превратиться в «давайте встретимся в среду утри». Выглядит похоже, но LLM на следующем этапе не поймёт «утри» и потеряет время встречи.

SpeechKit выигрывает на русском. Это ожидаемо — модель дообучена именно на русской речи. Разница в WER (8.1% vs 14.2%) выглядит небольшой, но на практике это означает, что SpeechKit правильно распознаёт «двадцать третье» как дату, а Whisper — как «двадцать третья».

Deepgram — компромисс. Latency 4 секунды на трёхминутный звонок — это почти реалтайм. Для сценариев, где важна скорость (уведомления, алерты), Deepgram вне конкуренции.

Конфиг Whisper для телефонного аудио (не дефолтный!):

import whisper

model = whisper.load_model("large-v3", device="cuda")

def transcribe(audio_path: str) -> dict:
    result = model.transcribe(
        audio_path,
        language="ru",
        condition_on_previous_text=True,
        # Для телефонного аудио — снижаем порог тишины,
        # иначе модель «глотает» короткие реплики
        no_speech_threshold=0.45,
        # Повышаем порог сжатия — телефонный шум
        # даёт ложные повторы
        compression_ratio_threshold=2.8,
        # beam search вместо greedy — +2% точности,
        # но +40% времени
        beam_size=5,
    )
    return result

Почему no_speech_threshold=0.45, а не дефолтные 0.6? На телефонных записях фоновый шум (улица, машина, кафе) создаёт высокий no_speech probability. С дефолтным порогом Whisper пропускает 15-20% реплик, считая их шумом. С 0.45 — пропускает 3-5%, но появляется 2-3% ложных распознаваний тишины. Трейдофф в пользу полноты.

Этап 3. Диаризация: кто это сказал?

Если аудио в моно — нужна speaker diarization. Я использую pyannote-audio 3.1:

from pyannote.audio import Pipeline

diarization_pipeline = Pipeline.from_pretrained(
    "pyannote/speaker-diarization-3.1",
    use_auth_token="YOUR_HF_TOKEN"
)

def diarize(audio_path: str) -> list:
    result = diarization_pipeline(audio_path, num_speakers=2)
    segments = []
    for turn, _, speaker in result.itertracks(yield_label=True):
        segments.append({
            "start": round(turn.start, 2),
            "end": round(turn.end, 2),
            "speaker": speaker
        })
    return segments

Проблема перехлёстов. Люди перебивают друг друга. pyannote честно размечает overlapping speech — но как совместить это с транскриптом, где Whisper выдаёт один поток текста?

Мой подход — привязка по средней точке сегмента:

def merge_transcript_speakers(
    whisper_segments: list,
    diarization: list
) -> str:
    lines = []
    for seg in whisper_segments:
        mid = (seg["start"] + seg["end"]) / 2
        speaker = "?"
        for d in diarization:
            if d["start"] <= mid <= d["end"]:
                speaker = d["speaker"]
                break
        lines.append(f"[{speaker}]: {seg['text'].strip()}")
    return "n".join(lines)

Это работает в 85% случаев. В оставшихся 15% — когда оба говорят одновременно — спикер определяется неверно. Я пробовал более сложные алгоритмы (взвешенное пересечение, voting по субсегментам), но выигрыш — 3-4%, а сложность кода растёт кратно. Для продакшена оставил простой вариант.

Грабля: pyannote иногда разбивает одного спикера на два, если человек меняет тон (начал спокойно, потом стал говорить громче). Решение — постобработка: если SPEAKER_00 и SPEAKER_02 никогда не пересекаются по времени, это скорее всего один человек.

def merge_fragmented_speakers(segments: list) -> list:
    """Объединяет спикеров, которые никогда не говорят одновременно."""
    from itertools import combinations

    speakers = set(s["speaker"] for s in segments)
    merge_map = {}

    for s1, s2 in combinations(speakers, 2):
        s1_intervals = [(s["start"], s["end"]) for s in segments if s["speaker"] == s1]
        s2_intervals = [(s["start"], s["end"]) for s in segments if s["speaker"] == s2]

        has_overlap = any(
            a_start < b_end and b_start < a_end
            for a_start, a_end in s1_intervals
            for b_start, b_end in s2_intervals
        )

        if not has_overlap:
            # Объединяем менее частого спикера в более частого
            if len(s1_intervals) >= len(s2_intervals):
                merge_map[s2] = s1
            else:
                merge_map[s1] = s2

    for seg in segments:
        while seg["speaker"] in merge_map:
            seg["speaker"] = merge_map[seg["speaker"]]

    return segments

Этап 4. LLM-извлечение сущностей: prompt engineering на стероидах

Самая интересная часть. Берём размеченный транскрипт и просим LLM извлечь структурированные данные.

Промпт, который работает (после 40 итераций)

Первый промпт был наивный: «Извлеки из диалога контактные данные и суть обращения». LLM радостно галлюцинировал — додумывал имена, суммы и даты, которых в разговоре не было.

Вот версия, к которой я пришёл:

EXTRACTION_PROMPT = """Проанализируй транскрипт телефонного разговора.
Извлеки ТОЛЬКО данные, которые ЯВНО прозвучали.

ПРАВИЛА:
1. Если информация НЕ упоминалась — поле = null. НЕ додумывай.
2. Числа → цифры: "восемнадцать миллионов" → 18000000
3. Адреса → нормализуй: "Ленинский сто двадцать" → "Ленинский проспект, 120"
4. Если обсуждается несколько тем/запросов — отдельный объект на каждую

ФОРМАТ (JSON):
{
  "contacts": [{"name": "str|null", "phone": "str|null", "role": "str"}],
  "requests": [
    {
      "topic": "str — суть запроса в 1 предложении",
      "details": {"ключ": "значение — только то, что прозвучало"},
      "amounts": [{"value": number, "context": "str"}],
      "confidence": 0.0-1.0
    }
  ],
  "action_items": [
    {"action": "str", "who": "operator|client", "when": "ISO8601|null"}
  ],
  "summary": "2-3 предложения"
}

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

Почему confidence — это спасение

Поле confidence — не декорация. Я использую его для автоматической фильтрации:

def validate_extraction(result: dict, transcript: str) -> dict:
    """Отсекаем данные с низкой уверенностью + проверяем по транскрипту."""
    for req in result.get("requests", []):
        if req.get("confidence", 0) < 0.6:
            req["amounts"] = []  # Не доверяем суммам
            req["_warning"] = "low_confidence"

        # Проверяем, что суммы реально звучали в разговоре
        for amount in req.get("amounts", []):
            value = amount.get("value", 0)
            if value and not _number_mentioned(value, transcript):
                amount["_hallucinated"] = True

    return result


def _number_mentioned(n: int, text: str) -> bool:
    """Проверяет, упоминалось ли число в тексте (цифрами или словами)."""
    text_lower = text.lower()
    # Прямое вхождение
    if str(n) in text_lower:
        return True
    # Сокращения: "18 млн", "2.5 тыс"
    millions = n / 1_000_000
    if millions >= 1 and (
        f"{int(millions)} млн" in text_lower
        or f"{int(millions)} миллион" in text_lower
    ):
        return True
    return False

Выбор модели: не всегда нужен GPT-4

Тестировал на 500 размеченных звонках (ручная разметка — золотой стандарт):

GPT-4o-mini — мой выбор для продакшена. F1 0.84 — значит 84% сущностей извлечены верно. Оставшиеся 16% — это в основном неявные данные (клиент не назвал сумму прямо, но она вычисляется из контекста). Для таких случаев есть поле confidence < 0.6.

Llama 3.1 70B — если данные не должны покидать контур. Self-hosted на 2×A100 через vLLM. Точность чуть ниже, но нулевые затраты на API и полный контроль.

Собираем пайплайн

import asyncio
import time
from dataclasses import dataclass

@dataclass
class CallResult:
    call_id: str
    transcript: str
    extracted: dict
    duration_ms: int

async def process_call(call_id: str, audio_path: str) -> CallResult:
    t0 = time.monotonic()

    # STT и диаризация — параллельно (не зависят друг от друга)
    transcript_fut = asyncio.to_thread(transcribe, audio_path)
    diarize_fut = asyncio.to_thread(diarize, audio_path)

    transcript_result, diarization = await asyncio.gather(
        transcript_fut, diarize_fut
    )

    # Объединяем
    diarization = merge_fragmented_speakers(diarization)
    full_text = merge_transcript_speakers(
        transcript_result["segments"], diarization
    )

    # LLM-извлечение
    extracted = await call_llm(EXTRACTION_PROMPT.format(transcript=full_text))
    validated = validate_extraction(extracted, full_text)

    return CallResult(
        call_id=call_id,
        transcript=full_text,
        extracted=validated,
        duration_ms=int((time.monotonic() - t0) * 1000),
    )

Оптимизация: с 40 секунд до 8

Первая версия обрабатывала 3-минутный звонок за 40 секунд. Вот что помогло:

Параллелизация STT + диаризация. Они независимы — запускаем одновременно. Whisper: 18 сек, pyannote: 12 сек. Параллельно: 18 сек (вместо 30). Экономия: 12 секунд.

Кэширование моделей в памяти. Whisper large-v3 загружается ~30 секунд. Держим модель в GPU memory, переиспользуем между запросами. То же для pyannote.

Оптимизация промпта. Первый промпт: 800 токенов. Финальный: 350 токенов. Меньше токенов → быстрее ответ LLM. Убрал многословные инструкции, оставил чёткие правила.

Chunking для длинных звонков. Звонки > 10 минут разбиваем на чанки по 5 минут с перекрытием 30 секунд. Каждый чанк обрабатывается отдельно, результаты мержатся. Без этого LLM начинает «забывать» начало разговора.

Итог: 8-12 секунд на 3-минутный звонок (RTX 4090 + GPT-4o-mini API).

Грабли, которые сэкономят вам время

1. LLM генерирует невалидный JSON. В ~3% случаев модель возвращает JSON с незакрытой скобкой или markdown-обёрткой ```json...```. Решение — не json.loads(), а парсер с fallback:

import json
import re

def safe_parse_json(text: str) -> dict:
    # Убираем markdown-обёртку
    text = re.sub(r'^```jsons*', '', text.strip())
    text = re.sub(r's*```$', '', text.strip())

    try:
        return json.loads(text)
    except json.JSONDecodeError:
        # Пытаемся починить незакрытые скобки
        for closer in ["}", "]}", "]}}", '"}']:
            try:
                return json.loads(text + closer)
            except json.JSONDecodeError:
                continue
        raise

2. Whisper повторяет фразы. На тихих участках записи Whisper large-v3 иногда зацикливается: «да да да да да да да да». Это известный баг. Детектирую через compression_ratio:

def detect_repetition(segments: list) -> list:
    """Убираем сегменты с подозрительными повторами."""
    cleaned = []
    for seg in segments:
        words = seg["text"].split()
        if len(words) > 3:
            unique_ratio = len(set(words)) / len(words)
            if unique_ratio < 0.3:  # Более 70% слов повторяются
                continue
        cleaned.append(seg)
    return cleaned

3. Диаризация путает спикеров между звонками. pyannote присваивает SPEAKER_00 и SPEAKER_01 произвольно — в одном звонке оператор = SPEAKER_00, в другом = SPEAKER_01. Решение — эвристика: спикер, который говорит первым и произносит приветственную формулу («добрый день», «алло, компания…»), — оператор.

4. Кодировка номеров телефонов. «Плюс семь девятьсот пять триста двадцать один сорок два двенадцать» — STT может выдать как текст, а может как +79053214212. Нужен нормализатор, который обрабатывает оба варианта.

Мониторинг: как понять, что всё сломалось

Пайплайн без мониторинга — бомба замедленного действия.

import logging
from prometheus_client import Histogram, Counter

PROCESSING_TIME = Histogram(
    'call_processing_seconds', 'Время обработки звонка',
    buckets=[5, 10, 15, 20, 30, 60]
)
EXTRACTION_CONFIDENCE = Histogram(
    'extraction_confidence', 'Средняя уверенность извлечения',
    buckets=[0.3, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
)
ERRORS = Counter('call_processing_errors', 'Ошибки обработки', ['stage'])

async def monitored_process(call_id: str, path: str):
    try:
        with PROCESSING_TIME.time():
            result = await process_call(call_id, path)

        avg_conf = sum(
            r.get("confidence", 0) for r in result.extracted.get("requests", [])
        ) / max(len(result.extracted.get("requests", [])), 1)
        EXTRACTION_CONFIDENCE.observe(avg_conf)

        return result
    except Exception as e:
        stage = "stt" if "whisper" in str(e).lower() else "llm"
        ERRORS.labels(stage=stage).inc()
        raise

Ключевые алерты:

• avg_confidence < 0.6 за последний час → промпт деградировал или изменился формат звонков

• processing_time > 30 сек → GPU под нагрузкой или API тормозит

• error_rate > 5% → что-то сломалось, нужна ручная проверка

Раз в неделю — ручная выборка 20-30 звонков, сравнение с золотым стандартом. Если F1 падает ниже 0.80 — пересматриваем промпт или дообучаем STT.

Итого

Весь пайплайн — ~400 строк Python без учёта инфраструктуры. Ключевые решения:

• STT: Whisper large-v3 для self-hosted, Yandex SpeechKit если нужна точность на русском, Deepgram если нужна скорость

• Диаризация: pyannote-audio 3.1 — лучший open-source вариант

• LLM: GPT-4o-mini для продакшена (цена/качество), GPT-4o для критичных данных

• Валидация: обязательна — LLM галлюцинирует в ~5% случаев

Подход не привязан к домену. Замените JSON-схему в промпте — и пайплайн заработает для колл-центра, юридической консультации, медицинского приёма или любого бизнеса с телефонными переговорами.

Если есть вопросы по конкретному этапу — STT на русском, борьба с галлюцинациями, streaming-обработка — пишите в комментариях.