🔖 Распознавание речи в реальном времени с Python и AssemblyAI

Создаём систему, которая транскрибирует речь с микрофона в текст в режиме реального времени, с автоматической расстановкой знаков препинания и заглавных букв.


🔸 Цель проекта

Каждую секунду микрофон передаёт аудио в AssemblyAI, которая распознаёт речь пословно. По завершении предложения ИИ понимает смысл и добавляет пунктуацию.


🔸 Установка

pip install pyaudio
pip install websockets

Mac: brew install portaudio — всё настроится автоматически

Windows: Скачайте wheel-файл с [неофициального сайта Python](https://www.lfd.uci.edu/~gohlke/pythonlibs/), затем:

pip install pyaudio‑0.2.11‑cp39‑cp39‑win_amd64.whl

Импорты:

import websockets
import asyncio
import base64
import json
from configure import auth_key
import pyaudio

🔸 Настройка конфигурации

Создайте файл configure.py:

auth_key = 'ваш_API_ключ_из_AssemblyAI'

Ключ можно получить на сайте AssemblyAI. Для real-time транскрибации нужен платный план.

🔸 Параметры аудио

FRAMES_PER_BUFFER = 3200
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000

p = pyaudio.PyAudio()

stream = p.open(
    format=FORMAT,
    channels=CHANNELS,
    rate=RATE,
    input=True,
    frames_per_buffer=FRAMES_PER_BUFFER
)

URL = "wss://api.assemblyai.com/v2/realtime/ws?sample_rate=16000"

Частота дискретизации 16000 синхронизирована с AssemblyAI API.


🔸 Асинхронные функции

Главная функция подключения:

async def send_receive():
    async with websockets.connect(
        URL,
        extra_headers=(("Authorization", auth_key),),
        ping_interval=5,
        ping_timeout=20
    ) as _ws:
        
        session_begins = await _ws.recv()
        print(session_begins)

Функция отправки аудио:

async def send():
    while True:
        try:
            data = stream.read(FRAMES_PER_BUFFER)
            data = base64.b64encode(data).decode("utf-8")
            json_data = json.dumps({"audio_data": str(data)})
            await _ws.send(json_data)
        except websockets.exceptions.ConnectionClosedError as e:
            break
        await asyncio.sleep(0.01)
    return True

Функция получения текста:

async def receive():
    while True:
        try:
            result_str = await _ws.recv()
            print(json.loads(result_str)['text'])
        except websockets.exceptions.ConnectionClosedError as e:
            break

🔸 Запуск

send_result, receive_result = await asyncio.gather(send(), receive())

while True:
    asyncio.run(send_receive())

Запустите командой:

python speech_recognition.py

🔸 Как это работает

1. Микрофон захватывает аудио каждые 3200 фреймов
2. Аудио кодируется в base64 и отправляется через WebSocket
3. AssemblyAI транскрибирует в реальном времени
4. Получаете два типа результатов:
- Частичные: посекундная транскрипция
- Финальные: с заглавными буквами и пунктуацией

Пример вывода:

hello
hello world
hello world this
Hello world, this is a test.

🔸 Особенности

- ИИ автоматически форматирует предложения
- Добавляет знаки препинания
- Исправляет грамматику
- Работает в реальном времени с минимальной задержкой


🔸 Ключевые технологии

- pyAudio — захват аудио с микрофона
- websockets — двусторонняя связь с API
- asyncio — асинхронная обработка
- AssemblyAI — ML-движок для распознавания речи

С развитием ИИ создание таких систем стало гораздо проще. Раньше точное распознавание речи было сложной задачей, теперь это реализуется несколькими десятками строк кода.


📎 Статья

🎙 Новости

📝 База вопросов