키는 엔진 밖으로 안 나가

"클라이언트는 소켓 하나짜리 토큰은 쥐어도 돼. 계정 키는 절대 못 쥐어."

원치 않는 중계

실시간 전사는 모양이 좀 곤란해. 소리는 마이크에서 끊김 없이 흘러가야 하는데, 계정 키는 브라우저에도 폰에도 절대 닿으면 안 돼. 뻔한 답은 중계야. 클라이언트가 우리 서버로 흘려보내고, 키를 쥔 서버가 제공자로 다시 흘려보내는 거지. 돌아가긴 해. 대신 아빠 목소리 1초 1초가 한 단계를 더 거치고, 멈출 수 있는 프로세스가 하나 늘고, 날것의 소리를 보는 곳이 하나 더 생겨. 대역폭도 괜히 두 배로 들고.

일회용 토큰

ElevenLabs엔 더 깔끔한 모양이 있고, 클라이언트용으로 이걸 권해. 일회용 토큰이야. 키를 쥔 서버가 제공자에게 realtime_scribe 타입 토큰을 달라고 해. 이 토큰은 실시간 소켓 딱 하나를 열고, 처음 쓰는 순간 소진되고, 15분 뒤엔 만료돼. 클라이언트는 그 토큰을 URL에 넣어 제공자에 직접 붙어. 그래서 소리는 어떤 중계도 안 거치고, 키는 서버 밖으로 한 발짝도 안 나가. 토큰 발급은 공짜고, 요금은 소켓이 들은 소리만큼 매겨져.

층 셋, 주인 셋

가족 안에선 이게 서비스 둘로 나뉘는데, 이 나눔 자체가 교훈이야.

Bellows가 제공자를 맡아. 계정 키를 쥐고, 토큰을 발급해서 소켓 URL과 모델 이름을 같이 돌려주는 라우트 하나를 열어둬. 듣기에 관해 Bellows가 아는 건 그게 전부야.
cwkPippa가 대화의 정책을 맡아. /api/stt/realtime-session 라우트가 Bellows에서 토큰을 받아서, 그 둘레에 완성된 소켓 URL을 만들어. 모델, 토큰, pcm_16000 오디오, 침묵 기준을 단 vad 커밋 전략, 아빠가 고른 언어, 그리고 키텀까지. 그 URL을 샘플 레이트, 침묵 창, 토큰 수명과 함께 돌려줘.
클라이언트는 붙기만 해. 웹 페이지, 폰, 맥의 Firekeeper가 전부 똑같이 완성된 URL을 받아서 열어. 침묵 창도 어휘도 이 중 누구도 안 정하니까, 서로 어긋날 수가 없어.

키텀은 적는 게 아니라 끌어내는 것

Scribe는 keyterms를 받아. 나올 줄 미리 알고 그쪽으로 기울여 들을 단어들이야. 전사기는 "Bellows"도 똘이라는 소울도 들어본 적이 없어서, 도와주지 않으면 소리가 제일 비슷한 흔한 단어를 적어. 실시간 세션은 지금 소울이 볼 수 있는 소울들의 표시 이름을 넘겨. 세션을 만드는 순간 소울 레지스트리에서 읽어 오고, 절대 손으로 쓴 목록이 아니야. 이유는 둘이야. 손 목록은 새 소울이 태어나는 날 낡아. 그리고 손 목록은 비밀을 흘려. 다른 소울이 봐선 안 되는 소울은 이 집의 어떤 목록에도 안 나오는데, 키텀 목록도 같은 함수로 만드니까 같은 규칙을 따라.

Code

토큰은 엔진에서 발급, 소켓 URL은 브레인에서 조립·python

import os
from urllib.parse import urlencode

import httpx

API = "https://api.elevenlabs.io/v1"
REALTIME_URL = "wss://api.elevenlabs.io/v1/speech-to-text/realtime"
PROVIDER_SILENCE_MAX = 3.0   # Scribe's VAD refuses more than 3.0 s


# --- the engine: the only process that ever sees the key -------------------
def mint_realtime_token() -> dict:
    response = httpx.post(
        f"{API}/single-use-token/realtime_scribe",
        headers={"xi-api-key": os.environ["ELEVENLABS_API_KEY"]},
        timeout=10,
    )
    response.raise_for_status()
    return {"token": response.json()["token"], "websocket_url": REALTIME_URL,
            "model_id": "scribe_v2_realtime", "expires_in_seconds": 15 * 60}


# --- the brain: conversation policy, decided once for every client -----------
def build_session(grant: dict, *, language: str, silence: float,
                  keyterms: list[str]) -> dict:
    query = [
        ("model_id", grant["model_id"]),
        ("token", grant["token"]),
        ("audio_format", "pcm_16000"),
        ("commit_strategy", "vad"),
        ("vad_silence_threshold_secs", str(min(silence, PROVIDER_SILENCE_MAX))),
        ("language_code", language),
    ]
    query += [("keyterms", term) for term in sorted(set(keyterms))]  # repeated key
    return {
        "websocket_url": f"{grant['websocket_url']}?{urlencode(query)}",
        "sample_rate": 16000,
        "silence_seconds": silence,
        "provider_silence_seconds": min(silence, PROVIDER_SILENCE_MAX),
        "expires_in_seconds": grant["expires_in_seconds"],
    }


if __name__ == "__main__":
    fake_grant = {"token": "sut_example", "websocket_url": REALTIME_URL,
                  "model_id": "scribe_v2_realtime", "expires_in_seconds": 900}
    visible_souls = ["Pippa", "Ttori", "Feynman"]   # from the registry, not typed here
    session = build_session(fake_grant, language="ko", silence=3.0,
                            keyterms=visible_souls)
    print(session["websocket_url"])
    print(session["provider_silence_seconds"], session["expires_in_seconds"])

Exercise

코드를 돌려서 찍히는 URL을 읽어봐. 그다음 키텀을 쌍의 리스트 대신 딕셔너리로 넘기게 바꿔서 다시 돌리고, 어느 단어가 살아남았는지 봐. 마지막으로 둘 사이에 놓일 라우트를 써. 클라이언트에게서 언어와 선택적인 침묵 창을 받아 둘 다 검사하고, 엔진에서 토큰을 발급받아 세션을 돌려주는 거야.

Hint

딕셔너리는 키 하나에 값 하나만 남기니까 마지막 키텀만 살아남고, 전사기는 나머지 이름이 나올 거라고 예상하지 않게 돼. 아무 경고도 없이. 라우트에선 발급 전에 검사해. 발급은 공짜지만 발급된 토큰 하나하나가 15분짜리 살아 있는 자격 증명이고, 언어 코드가 틀린 요청은 토큰을 만들지 않고 실패해야 해.