Telnyx와 OpenAI Whisper를 사용하여 실시간 전화 통화 전사 파이프라인을 구축하는 방법

이 예제가 수행하는 작업

이 예제는 가장 작으면서도 유용한 '전화 통화-AI 응답' 파이프라인을 연결합니다:

Telnyx Call Control을 통해 발신 전화가 이루어집니다.
Telnyx가 통화 오디오를 녹음하고 이를 호스팅된 URL에 저장합니다.
call.recording.saved 웹훅 (webhook)이 발생합니다.
앱이 오디오를 다운로드하고, OpenAI Whisper로 이를 전사(transcribe)하며, GPT-4로 문맥에 맞는 응답을 생성한 뒤, Telnyx TTS를 통해 이를 재생합니다.

전 과정은 약 245줄의 Python 코드로 구성됩니다. 이것은 **데모 (demo)**이며, 프로덕션용 코드가 아닙니다. 이 포스트에서는 실제 대규모 트래픽에서 실행하기 전에 어떤 특정 라인들을 강화해야 하는지 살펴볼 것입니다.

전체 코드는 telnyx-code-examples/call-whisper-monitoring-python에 있습니다. 이것은 인프라 (Infrastructure) 기둥의 예시입니다. 이는 네 개의 별도 벤더를 하나씩 이어 붙이는 대신, 단일 웹훅 뒤에서 Telnyx Call Control과 OpenAI 추론 (inference)을 체이닝함으로써 에이전트-플랫폼 논제를 입증합니다.

왜 Telnyx Call Recording과 OpenAI Whisper를 결합하는가?

Telnyx Call Control은 (포털의 Call Control Application에서 녹음이 활성화된 경우) 통화 오디오를 호스팅된 URL에 녹음합니다. OpenAI Whisper는 단 한 번의 API 호출로 해당 오디오를 텍스트로 변환합니다. 이 둘을 결합하면 미디어 서버, S3 버킷, 전사 워커 풀(transcription worker pool) 없이도 세 번의 웹훅 호출만으로 전화 통화-전사 파이프라인을 구축할 수 있습니다.

주의할 점: Telnyx는 녹음 파이프라인(캡처, 저장, 웹훅 전달)을 담당하고, OpenAI는 추론(Whisper + GPT-4 + 향후 모델들)을 담당합니다. 여러분은 이 둘을 잇는 접착제(glue) — 즉, 웹훅 핸들러(webhook handler), 재시도 정책(retry policy), 타이밍 예산(timing budget), 그리고 상태 관리(state management)를 담당하게 됩니다.

대부분의 "전화 통화 AI 구축" 튜토리얼은 이 접착제 부분을 생략합니다. 이 튜토리얼은 그렇지 않습니다.

4단계 파이프라인

  POST /calls/initiate
        │
        ▼
...

이 데모는 통화 제어를 위해 Telnyx를, 두 번의 AI 호출(Whisper + GPT-4)을 위해 OpenAI를, 그리고 TTS 재생을 위해 다시 Telnyx를 사용합니다. 두 개의 벤더, 세 개의 AI 시스템, 하나의 웹훅입니다.

프로젝트 설정

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/call-whisper-monitoring-python
cp .env.example .env    # 4개의 값을 채워주세요
...

.env 파일에 다음 네 가지 값이 필요합니다:

TELNYX_API_KEY — Portal에서 발급받은 Telnyx API v2 키
OPENAI_API_KEY — OpenAI 키 (Whisper와 GPT-4 모두 사용)
TELNYX_PHONE_NUMBER — 발신자 ID (Caller ID)로 사용될 Telnyx 전화번호
TELNYX_CONNECTION_ID — 해당 번호가 연결된 Call Control App ID

그 다음, 서버를 외부에 공개해야 합니다 (Telnyx가 서버에 접근할 수 있어야 함):

ngrok http 5000

그리고 ngrok URL을 Call Control Application의 웹훅(Webhook) URL로 설정하세요.

다음 curl 명령어를 사용하여 데모를 실행합니다:

curl -X POST http://localhost:5000/calls/initiate \
  -H "Content-Type: application/json" \
  -d '{"to": "+12125551234"}'

앱이 해당 번호로 전화를 걸고, 녹음 파일이 저장될 때까지 기다린 후, 전사(Transcribe) 및 응답을 수행합니다.

핵심 코드: 전사, 응답, 발화

세 개의 헬퍼 함수(Helper functions)가 모든 AI 작업을 수행합니다. 각 함수는 한 번에 읽을 수 있을 정도로 짧습니다.

OpenAI Whisper를 사용하여 오디오 전사 (Transcribe):

def transcribe_audio(audio_url: str) -> str:
    try:
        response = requests.get(audio_url, timeout=10)
...

오디오 파일은 Telnyx의 녹음 URL에서 다운로드된 후, 이진 블롭(Binary blob) 형태로 Whisper에 업로드됩니다. Whisper는 전사된 내용을 일반 텍스트로 반환합니다.

GPT-4를 사용하여 문맥에 맞는 응답 생성 (Generate a contextual response):

def generate_prompt_response(transcript: str) -> str:
    response = openai_client.chat.completions.create(
        model="gpt-4",
...

시스템 프롬프트(System prompt)는 작지만 중요한 방어 기제(Hardening)입니다. 이는 모델이 범위를 벗어나지 않고(통화 내용에 집중) 발신자의 전사 내용으로부터 발생하는 프롬프트 인젝션(Prompt injection)을 무시하도록 지시합니다. 이는 신뢰할 수 없는 입력에 노출되는 모든 AI 에이전트(AI agent)에 사용할 수 있는 동일한 패턴입니다.

Telnyx TTS를 통해 응답을 다시 발화 (Speak the response back):

def speak_response(call_control_id: str, text: str) -> dict:
    response = telnyx_client.calls.actions.speak(
        call_control_id=call_control_id,
...

Telnyx는 합성된 오디오를 실시간 통화 중에 재생합니다.

동기식 웹훅 (Synchronous Webhooks)이 숨겨진 함정인 이유

이 세 가지 함수를 하나로 묶는 웹훅 핸들러(webhook handler)는 작성된 그대로 프로덕션(production) 환경에 배포해서는 안 되는 코드 부분입니다:

if event_type == "call.recording.saved":
    recording_url = payload.get("data", {}).get("recording_urls", {}).get("wav")
    if recording_url and call_control_id in call_state:
...

웹훅 핸들러는 다음과 같이 동작합니다:

Telnyx에서 오디오 파일을 다운로드합니다.
Whisper를 호출하여 전사 (transcribe) 합니다.
GPT-4를 호출하여 응답을 생성합니다.
Telnyx TTS를 호출하여 응답을 재생합니다.
Telnyx에 200 응답을 반환합니다.

이는 웹훅을 승인(acknowledge)하기 전에 세 번의 순차적인 API 호출이 발생함을 의미합니다. 해당 체인 내의 어떤 느린 API 호출(네트워크 일시 오류, 모델 지연 시간 급증, 업스트림 속도 제한 등)이라도 Telnyx의 전달 타임아웃(delivery timeout)을 초과할 수 있습니다. 이 시점에서 Telnyx는 재시도(retry)를 수행하고, 당신은 동일한 오디오를 다시 전사하며, 이미 끊겨버린 통화에 응답을 재생하려고 시도하게 됩니다.

해결책은 웹훅을 먼저 승인하고 작업을 비동기적 (asynchronously)으로 처리하는 것입니다:

if event_type == "call.recording.saved":
    recording_url = payload.get("data", {}).get("recording_urls", {}).get("wav")
    call_control_id = payload.get("data", {}).get("call_control_id")
...

process_recording()은 자체적인 재시도 정책(retry policy)과 함께 백그라운드에서 전사 → 응답 → 재생 체인을 실행합니다. 웹훅은 즉시 200을 반환합니다. Telnyx는 재시도하지 않습니다. 발신자는 여전히 응답을 듣게 됩니다.

데모에서는 핵심 패턴을 흐리지 않기 위해 인프라(워커, 큐, 상태 저장소)를 추가하지 않았으므로 이 방식을 사용하지 않았습니다. 하지만 이 코드를 실제로 배포할 예정이라면, 이것이 가장 먼저 수행해야 할 변경 사항입니다.

프로덕션 강화 (Production Hardening)

동기식 웹훅 문제 외에도, 데모가 어떤 규모에서도 안정적으로 실행되려면 네 가지 사항이 더 필요합니다:

지속적인 상태(Persistent state). 예제는 call_state를 인메모리 딕셔너리에 저장합니다. 서버가 재시작되면 모든 활성 통화의 전사본과 상태가 손실됩니다. 이를 Redis나 Postgres로 옮기고 보존 정책에 맞는 TTL을 설정하세요.

/calls/initiate에서의 인증(Authentication). 해당 엔드포인트에 접근할 수 있는 사람은 누구나 귀하의 OpenAI 예산과 Telnyx 분을 소모할 수 있습니다. 아웃바운드 통화를 허용하기 전에 API 키 확인, JWT 유효성 검사 또는 세션 조회를 추가하세요.

녹음 웹훅에서의멱등성(Idempotency). Telnyx는 5xx 오류 및 타임아웃 발생 시 웹훅을 재시도합니다. 멱등성 키가 없으면 동일한 오디오를 다시 전사하고 OpenAI에 이중으로 청구됩니다. 녹음 ID와 결과를 상태 저장소에 저장하고 중복 이벤트 발생 시 처리를 중단(short-circuit)하세요.

TTS 응답을 웹훅 경로에서 분리하세요. 발신자가 TTS가 재생되기 전에 통화를 끊으면, Telnyx로부터 4xx 오류를 받고 발신자의

메모리 내 call_state가 여러 워커(worker)에 대해 안전한가요? 아니요. 이는 Flask 프로세스 내의 Python 딕셔너리(dict)입니다. 따라서 여러 개의 gunicorn 워커, 여러 개의 컨테이너, 또는 어떠한 수평적 확장(horizontal scaling)이 이루어지더라도 상태를 잃게 됩니다. 운영 환경(production)에서의 해결책은 각 call_control_id별로 키를 생성하고 각 항목에 TTL(Time To Live)을 설정한 Redis를 사용하는 것입니다.

리소스 (Resources)

Telnyx Call Control 가이드: https://developers.telnyx.com/docs/voice/call-control
Telnyx Speak (TTS) API: https://developers.telnyx.com/api-reference/call-commands/speak-text
OpenAI Whisper Speech-to-Text: https://platform.openai.com/docs/guides/speech-to-text
OpenAI Chat Completions: https://platform.openai.com/docs/guides/text-generation
Telnyx Portal: https://portal.telnyx.com
Telnyx Python SDK: https://github.com/team-telnyx/telnyx-python

Insights