Каждый день в российском бизнесе происходят миллионы телефонных звонков. Колл-центры, клиники, юридические конторы, отделы продаж — везде, где есть телефон, есть поток неструктурированных данных, который никто не обрабатывает. Менеджер повесил трубку, записал в 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 минуты, качество — типичный мобильный звонок):
Модель | WER (телефон, рус.) | Latency (3 мин) | Стоимость | Streaming |
|---|---|---|---|---|
Whisper large-v3 (локально, A100) | 14.2% | 18 сек | ~$0 (GPU) | Нет |
Whisper large-v3 (локально, RTX 4090) | 14.2% | 35 сек | ~$0 (GPU) | Нет |
Yandex SpeechKit | 8.1% | 12 сек | ~₽1.2/мин | Да |
Deepgram Nova-2 | 11.7% | 4 сек | ~$0.0043/мин | Да |
Несколько неочевидных наблюдений:
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 размеченных звонках (ручная разметка — золотой стандарт):
Модель | F1 (извлечение) | Latency | $/звонок |
|---|---|---|---|
GPT-4o | 0.91 | 2.8 сек | $0.03 |
Claude 3.5 Sonnet | 0.90 | 2.1 сек | $0.02 |
GPT-4o-mini | 0.84 | 0.9 сек | $0.003 |
Llama 3.1 70B (vLLM) | 0.82 | 1.8 сек | $0.001 |
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'^```json\s*', '', 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-обработка — пишите в комментариях.
