Python 음성 인식: 파일, 스트리밍, 마이크 코드

Python으로 음성을 텍스트로 바꾸려면 음성 인식 모델에 오디오를 보내고 텍스트를 받으면 됩니다. 정말 정해야 할 것은 어떤 종류의 모델을 쓰느냐입니다. 녹음된 파일과 실시간 대화는 서로 다른 일이기 때문입니다. 파일은 클립이 끝날 때까지 기다릴 수 있습니다. 전화한 사람은 그럴 수 없습니다.

이 가이드는 세 가지 방법을, 그대로 실행되는 코드와 함께 다룹니다. 녹음 파일 전사, 자동 턴 감지가 있는 오디오 스트리밍, 마이크 입력의 실시간 전사입니다. 스트리밍 예제에는 저희가 음성 에이전트용으로 만드는 Cartesia Ink 2를 씁니다. 무료 로컬 모델이 더 맞는 경우에는 그렇다고 밝히고 그 코드도 보여 드립니다.

방법 고르기

대부분의 튜토리얼은 SpeechRecognition의 recognize_google()에서 끝납니다. 이 패키지는 십여 개 엔진을 감싼 편리한 래퍼지만, 기본 Google 백엔드는 Google이 언제든 폐기할 수 있는 공용 키를 쓰고, 문장이 끝날 때까지 아무것도 보내지 않습니다. 데모로는 괜찮습니다. 전화 응대용은 아닙니다.

사전 준비

• Python 3.9 이상
• Cartesia 예제용으로 Cartesia Playground에서 만든 API 키와 SDK:

python -m pip install "cartesia[websockets]"
export CARTESIA_API_KEY="your-api-key"

스트리밍 예제에는 websockets extra가 필요합니다. 키는 환경 변수나 시크릿 매니저에 두고, 커밋하는 코드에는 절대 넣지 마세요.

1. 녹음 파일 전사하기

이미 녹음된 오디오라면 파일 전체를 요청 한 번으로 보내세요. Cartesia의 배치 엔드포인트는 WAV, MP3 같은 컨테이너를 받고, 단어마다 타임스탬프도 돌려줄 수 있습니다. 다음을 transcribe_file.py로 저장합니다.

import os
import sys

from cartesia import Cartesia

client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])

with open(sys.argv[1], "rb") as audio_file:
    result = client.stt.transcribe(
        file=audio_file,
        model="ink-whisper",
        language="en",
        timestamp_granularities=["word"],
    )

print(result.text)
for word in result.words or []:
    print(f"{word.start:6.2f}s  {word.word}")

python transcribe_file.py meeting.mp3로 실행합니다. 배치 전사에는 약 100개 언어를 지원하는 ink-whisper를 씁니다. 오디오의 언어 코드를 넘겨 주세요. 비동기 버전은 문서의 배치 STT 예제에 있습니다.

로컬 대안

오디오를 컴퓨터 밖으로 보낼 수 없거나, 오디오가 많고 GPU가 있다면 Whisper를 직접 돌리세요. faster-whisper는 OpenAI의 Whisper를 CTranslate2 위에 다시 구현한 것으로, 노트북 CPU에서도 돌아갑니다.

python -m pip install faster-whisper

from faster_whisper import WhisperModel

model = WhisperModel("base.en", device="cpu", compute_type="int8")
segments, info = model.transcribe("jfk.flac", beam_size=5)

print(f"Detected language: {info.language} ({info.language_probability:.2f})")
for segment in segments:
    print(f"[{segment.start:.2f}s -> {segment.end:.2f}s] {segment.text}")

Whisper 저장소에 있는 11초짜리 JFK 샘플에서, base.en 모델은 4코어 서버 CPU로 약 2.5초 만에 인용문을 한 단어도 틀리지 않고 받아 적었습니다(처음 한 번 140MB 모델 다운로드). 더 큰 모델은 더 정확하고 더 느립니다. 얻을 수 없는 것은 스트리밍입니다. Whisper는 끝난 구간을 처리하므로 대화에는 맞지 않습니다.

2. 오디오를 스트리밍하고 턴 받기

음성 에이전트에는 배치 모델이 줄 수 없는 두 가지가 필요합니다. 상대가 아직 말하는 중에 나오는 텍스트, 그리고 말이 끝났는지에 대한 판단입니다. Ink 2는 WebSocket 하나로 둘 다 처리합니다. Ink 2의 턴 감지는 침묵뿐 아니라 말의 내용도 보기 때문에, 생각하느라 문장 중간에 잠시 멈춘 사람이 회선이 조용해졌다는 이유만으로 끊기지 않습니다.

이 스크립트는 WAV 파일을 실시간 오디오처럼 보냅니다. stream_file.py로 저장합니다.

import os
import sys
import time
import wave

from cartesia import Cartesia

with wave.open(sys.argv[1], "rb") as wav:
    if wav.getnchannels() != 1 or wav.getsampwidth() != 2:
        raise SystemExit("Convert the file to 16-bit mono WAV first.")
    sample_rate = wav.getframerate()
    chunks = []
    while data := wav.readframes(sample_rate // 10):  # 100 ms per chunk
        chunks.append(data)

client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])
transcript = ""

with client.stt.auto_finalize.websocket(
    model="ink-2",
    encoding="pcm_s16le",
    sample_rate=sample_rate,
) as ws:
    for chunk in chunks:
        ws.send_raw(chunk)
        time.sleep(0.1)  # send at the speed of speech
    ws.send({"type": "close"})  # flush buffered audio; the server then closes

    turn_text = ""
    for event in ws:
        if event.type == "turn.update":
            turn_text = event.transcript or turn_text  # the whole turn so far
            print("...", turn_text)
        elif event.type == "turn.end":
            text = event.transcript or turn_text  # fall back if turn.end is empty
            print("turn:", text)
            transcript += text  # join verbatim, no strip()
            turn_text = ""
        elif event.type == "error":
            raise SystemExit(event.message)

print(transcript)

파일이 16비트 모노 WAV가 아니라면 먼저 변환하세요.

ffmpeg -i input.mp3 -ac 1 -ar 16000 -c:a pcm_s16le input.wav

이 스크립트에는 틀리기 쉬운 부분이 세 가지 있습니다.

• 파일이 아니라 원시 샘플. 실시간 엔드포인트는 헤더 없는 오디오 바이트를 받습니다. 그래서 스크립트는 파일을 보내는 대신 WAV에서 프레임을 읽어 보냅니다. encoding과 sample_rate는 연결할 때 지정하며, 실제 바이트와 맞아야 합니다. 오디오 입력 가이드에 따르면 맞지 않아도 서버가 오류를 돌려주지 않을 수 있어서, 잘못된 샘플 레이트는 예외가 아니라 엉뚱한 전사 결과로 나타납니다.
• 실시간 속도로 보내기. 100밀리초 조각마다 넣은 sleep(0.1)은 장식이 아닙니다. 실시간 엔드포인트는 1초에 약 1초 분량의 오디오를 기대하며, 파일 전체를 소켓에 한꺼번에 쏟아부으면 서버 쪽 버퍼가 넘칠 수 있습니다. 실시간일 필요가 없는 파일이라면 1단계의 배치 호출을 쓰세요.
• 누적되는 전사 결과. 각 turn.update에는 새 조각이 아니라 그 턴의 지금까지 전체 내용이 들어 있습니다. 업데이트를 이어 붙이면 단어가 반복됩니다. 텍스트는 턴마다 한 번 turn.end에서 가져와, 도착한 그대로 이어 붙이세요. 이 스크립트는 마지막 turn.update도 보관해 두었다가, turn.end가 빈 전사 결과로 오면 그것을 씁니다. 저희 클라이언트들도 대비하는 경우입니다.

짧게 쓰기 위해 이 스크립트는 파일을 다 보낸 뒤에 이벤트를 읽으므로, 출력이 마지막에 한꺼번에 나옵니다. 아래 마이크 예제는 보내기와 읽기를 동시에 합니다.

Ink 2는 영어, 프랑스어, 힌디어, 일본어, 스페인어를 전사하고, 어떤 언어인지 자동으로 감지합니다. 다른 언어를 실시간으로 처리해야 한다면 음성 인식 모델 비교를 참고하세요.

3. 마이크 입력을 실시간으로 전사하기

엔드포인트는 같고 입력만 진짜입니다. 이 버전은 sounddevice로 16kHz 모노 오디오를 받고, 비동기 클라이언트로 보내기와 받기를 동시에 합니다.

python -m pip install sounddevice

Linux에서는 PortAudio도 필요할 수 있습니다(sudo apt install libportaudio2). 다음을 microphone.py로 저장합니다.

import asyncio
import os

import sounddevice as sd
from cartesia import AsyncCartesia

SAMPLE_RATE = 16000
BLOCK = SAMPLE_RATE // 10  # 100 ms of audio per message


async def main():
    loop = asyncio.get_running_loop()
    audio = asyncio.Queue(maxsize=50)  # about 5 s of audio

    def enqueue(block):
        if audio.full():  # the network fell behind: drop the oldest block
            audio.get_nowait()
        audio.put_nowait(block)

    def on_audio(indata, frames, time_info, status):
        loop.call_soon_threadsafe(enqueue, bytes(indata))

    client = AsyncCartesia(api_key=os.environ["CARTESIA_API_KEY"])
    async with client.stt.auto_finalize.websocket(
        model="ink-2",
        encoding="pcm_s16le",
        sample_rate=SAMPLE_RATE,
    ) as ws:

        async def send():
            with sd.RawInputStream(
                samplerate=SAMPLE_RATE,
                blocksize=BLOCK,
                channels=1,
                dtype="int16",
                callback=on_audio,
            ):
                while True:
                    await ws.send_raw(await audio.get())

        async def receive():
            turn_text = ""
            async for event in ws:
                if event.type == "turn.start":
                    print("(listening)")
                elif event.type == "turn.update":
                    turn_text = event.transcript or turn_text
                elif event.type == "turn.end":
                    print("You said:", event.transcript or turn_text)
                    turn_text = ""
                elif event.type == "error":
                    raise RuntimeError(event.message)

        await asyncio.gather(send(), receive())


asyncio.run(main())

python microphone.py를 실행하고 무언가 말한 뒤 잠시 멈추면 그 턴의 전사 결과가 출력됩니다. 멈추려면 Ctrl+C를 누르세요.

오디오 콜백은 별도 스레드에서 돌기 때문에, 소켓을 직접 건드리지 않고 call_soon_threadsafe로 각 블록을 이벤트 루프에 넘깁니다. 조용히 있는 동안에도 마이크는 계속 보내는데, 이게 중요합니다. 서버는 오디오가 오지 않으면 침묵이 아니라 네트워크 지연으로 보기 때문에, 음소거나 일시 정지 중인 입력은 아무것도 안 보내는 대신 0을 보내야 합니다. 큐는 오디오 약 5초 분량으로 제한됩니다. 연결이 밀리면 오래된 블록이 메모리에 쌓이는 대신 버려집니다. 1분 전에 한 말의 전사 결과는 실시간 대화에 쓸모가 없기 때문입니다.

음성 에이전트에서 turn.start는 봇의 말을 멈추는 지점이고, turn.end는 텍스트를 LLM에 보내는 지점입니다. 답변을 더 빨리 시작하려면 turn.eager_end도 처리하세요. 이 이벤트가 오면 생성을 시작하고, 뒤이어 turn.resume이 오면 초안을 버립니다. 전화한 사람이 자주 멈추거나 거의 멈추지 않는다면 턴 감지 문서에서 임계값을 확인하세요.

자주 생기는 문제

비용

로컬 모델의 비용은 연산뿐입니다. Cartesia에서 Ink 2는 오디오 1초에 3크레딧을 쓰며, Scale 플랜에서는 오디오 1시간에 0.39달러입니다. 무료 플랜의 크레딧으로 Ink 2를 약 1시간 50분 쓸 수 있습니다. 최신 숫자는 요금 페이지에서 확인하세요.

파이프라인을 직접 만들고 싶지 않다면, Cartesia Managed Agents가 Ink, LLM, Sonic 음성 합성을 전화번호에 연결해 줍니다. 녹음 몇 개만 전사하면 된다면 노트북의 faster-whisper로 충분할 겁니다.

다음 단계

1. 코드 없이 브라우저에서 내 목소리로 Ink 2를 써 보세요: Ink 페이지.
2. 전사 결과를 LLM에 보내고 Python 음성 합성으로 답을 말하게 하세요.
3. 프레임워크로 조립하세요: Cartesia로 Pipecat 음성 에이전트 만들기.

관련 문서

• 실시간 음성 인식 빠른 시작
• STT 엔드포인트 비교: 자동, 수동, 배치
• 모델이 정확히 받아 적어야 하는 이름과 전문 용어를 위한 키텀 프롬프팅
• 음성 활동 감지(VAD)와, 턴 감지가 이를 대신하는 이유