Grok Voice Agent Builder: Python 실습 가이드

xAI가 음성 에이전트를 만들 수 있는 콘솔인 Voice Agent Builder를 공개했습니다. 통화 흐름을 설명하고, 문서와 도구를 연결하고, 음성을 선택하면 됩니다.

저는 음성 에이전트 콘솔을 테스트할 때 출시 공지보다 코드에 연결해야 하는 부분에 더 주목합니다. WebSocket 세션 구성, 오디오 전송 방식, 도구 호출 지점, 통화 비용, 다른 앱이 워크플로를 호출하는 방법 같은 것들입니다.

아래 코드는 그 흐름을 Voice Agent API에 직접 맞춰 재구성합니다. 구체적으로, 가용 시간을 확인하고 음성으로 응답하며, 비용을 추적하고, 도구 오류를 처리하며, FastAPI 엔드포인트를 노출하는 클리닉 예약 보조원을 사용합니다.

Grok Voice Agent Builder란?

Voice Agent Builder는 Grok Voice에서 음성 에이전트를 생성하고 배포하기 위한 xAI의 콘솔입니다. 2026년 7월 1일 베타로 출시되었습니다. 별도의 음성-텍스트 변환, 언어 모델, 텍스트-음성 변환 서비스를 조합하는 대신 하나의 음성 모델 경로를 사용합니다.

콘솔에는 전화(텔레포니), 문서 검색, 도구와 커넥터, 가드레일, 원격 MCP 서버, 그리고 녹음, 전사, 트레이스를 포함한 통화 로그가 포함되어 있습니다.

오디오는 분당 과금됩니다. 콘솔이 아직 베타이므로 여기서는 API를 직접 사용합니다.

Builder 아래에서 Grok Voice Agent API가 동작하는 방식

콘솔 아래에는 Builder와 동일한 런타임을 노출하는 실시간 WebSocket API인 Voice Agent API가 있습니다.

Builder는 Voice API 위에 있습니다. 이미지: 작성자.

여기서 사용하는 모델은 grok-voice-think-fast-1.0입니다. grok-voice-latest 별칭은 최신 모델을 가리킵니다. 여기서는 이를 사용하지만, 배포 앱에는 버전이 고정된 이름을 권합니다. xAI는 이 모델의 τ-voice Bench 리더보드 점수를 67.3%로 보고합니다. 저는 이를 보증이 아닌 한 가지 데이터 포인트로 취급합니다.

호환성 참고: 이 API는 OpenAI Realtime API와 호환됩니다. OpenAI의 실시간 엔드포인트와 통신하는 코드가 있다면, 주로 기본 URL과 키만 바꾸면 됩니다.

프로젝트 개요: 무엇을 만들 것인가

클리닉 보조원은 음성 입력을 받고, 생성된 음성으로 응답하며, 후속 질문을 하고, 슬롯을 제안하기 전에 가용 시간을 확인하고, 필요 시 사람에게 넘깁니다. 핵심 예시는 하나의 도구를 사용하고, Streamlit 데모는 예약, 상담원 연결, 통화 종료 동작을 추가합니다.

핵심 튜토리얼은 각자 하나의 역할을 가진 네 파일로 나뉩니다:

• voice_client.py — WebSocket 클라이언트, 오디오 헬퍼, 비용 추적

• tools.py — check_availability 및 Streamlit에서 사용하는 추가 데모 도구

• assistant.py — 시스템 프롬프트, 세션 구성, 워크플로우

• app.py — FastAPI로 전체 제공

이 네 파일이 글의 흐름입니다. 저장소에는 시각적 데모용 app_streamlit.py와 Windows용 실행기인 run.py도 포함되어 있지만, 핵심 흐름이 작동한 뒤에 살펴보겠습니다.

사전 준비

코드를 실행하기 전에 Python 3.10 이상, xAI 계정, console.x.ai에서 발급한 API 키, 선불 크레딧, 그리고 환경 변수, JSON, WebSocket에 대한 기본 이해가 필요합니다.

프로젝트 설정

폴더와 가상 환경을 만들고 패키지를 설치합니다:

mkdir appointment-agent
cd appointment-agent
python -m venv .venv
.venv\Scripts\activate       # macOS/Linux: source .venv/bin/activate
pip install websockets python-dotenv fastapi uvicorn pydantic httpx numpy streamlit

새 체크아웃에서도 동일한 설정을 사용하도록 requirements.txt에 이 패키지들을 고정하세요.

.env 파일을 Python 파일과 같은 위치에 만듭니다:

XAI_API_KEY=xai-your-key-here

.env를 .gitignore에 추가하세요. API 키는 서버에만 보관해야 합니다.

Voice Agent 만들기

이제 빌드를 시작해 보겠습니다.

WebSocket으로 Grok Voice Agent API에 연결하기

첫 단계는 연결을 여는 것입니다. 모델은 쿼리 매개변수로, 키는 핸드셰이크 시 베어러 토큰으로 전달합니다:

import asyncio
import json
import os
import websockets

async def voice_agent():
    url = "wss://api.x.ai/v1/realtime?model=grok-voice-latest"
    async with websockets.connect(
        url,
        additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
    ) as ws:
        async for message in ws:
            print(json.loads(message)["type"])

asyncio.run(voice_agent())

실제 키로 실행하면 가장 먼저 보이는 이벤트는 session.created로, 소켓이 열리고 구성 준비가 되었음을 의미합니다.

세션 생성 이벤트는 연결이 확인되었음을 보여줍니다. 이미지: 작성자.

음성 세션 구성하기

실시간 소켓이 곧바로 구성된 에이전트는 아닙니다. session.update 이벤트와 session 객체를 보내어 에이전트를 빚어갑니다.

음성, 오디오 포맷, 지시문

가장 자주 다루는 세 가지 설정은 음성, 오디오 포맷, 시스템 프롬프트입니다. 실시간 API는 명명된 다섯 가지 음성 eve, ara, rex, sal, leo를 제공하며, 커스텀 클론도 사용할 수 있습니다. 오디오는 기본적으로 audio/pcm (24000 Hz)이며 입력과 출력을 별도로 구성합니다.

다음은 assistant.py에서 조립하는 세션 구성입니다:

def build_session_config(voice="ara", instructions=SYSTEM_PROMPT, sample_rate=24000):
    # The model needs to know "today" or it guesses the year for a date like "July 6th".
    instructions = f"{instructions}\nToday's date is {date.today().isoformat()}."
    return {
        "voice": voice,
        "instructions": instructions,
        "turn_detection": None,  # manual turns for file-based input
        "audio": {
            "input": {"format": {"type": "audio/pcm", "rate": sample_rate}},
            "output": {"format": {"type": "audio/pcm", "rate": sample_rate}},
        },
        "tools": [CHECK_AVAILABILITY_TOOL],
    }

instructions 필드는 시스템 프롬프트입니다. 음성 통화에서 긴 답변은 따라가기 어려우므로, 이 클리닉 프롬프트는 짧게 유지합니다:

You are a voice appointment assistant for a small clinic. Help callers book,
reschedule, cancel, or ask questions about appointments, services, and hours.
Answer whatever the caller asks that relates to the clinic. Keep responses short
and natural for a phone conversation. Ask one question at a time. Confirm
important details before taking action. Use the availability tool before offering
a time slot. Escalate to a human for medical, urgent, sensitive, or unclear
requests. If a caller asks about something unrelated to the clinic, say briefly
that it is outside what you can help with, then steer back to booking. If you
cannot make out what the caller said, ask them to repeat it instead of repeating
your last message.

에스컬레이션 지시는 클리닉 에이전트가 의학적 조언을 하지 않게 해줍니다. 마지막 두 줄은 범위를 벗어나지 않도록 하고, 발화가 불명확할 때 반복 루프에 빠지지 않도록 합니다. 또한 구성에 오늘 날짜를 덧붙입니다. 실사용 테스트에서 "July 6th" 같은 날짜의 연도를 모델이 잘못 추정하는 경우가 있었기 때문입니다.

턴 감지 조정

턴 감지는 발화를 멈췄다고 판단하는 방식입니다. turn_detection.type을 server_vad로 설정하면 서버가 무음을 기준으로 턴을 종료합니다. null로 두면 오디오 버퍼를 커밋하여 직접 턴을 제어합니다. 파일 기반 흐름에서는 이 방식을 씁니다.

서버 VAD에는 알아둘 설정이 세 가지 있습니다. threshold는 음성이 인식될 최소 음량, silence_duration_ms는 어느 정도의 정적이 턴 종료로 간주되는지, prefix_padding_ms는 발화 시작 전 약간의 오디오를 보존합니다. 에이전트가 말을 자주 끊는다면 먼저 silence_duration_ms를 올려보세요.

에이전트로 오디오 보내기

이제 발신자의 음성을 전송합니다. 오디오는 세션 포맷과 일치해야 합니다: 모노 16비트 PCM, 24000 Hz, base64로 인코딩하여 청크로 전송합니다.

클라이언트는 파일을 슬라이스로 스트리밍한 뒤, 턴 종료를 표시하기 위해 버퍼를 커밋합니다:

async def send_audio(self, pcm_bytes, chunk_ms=100, commit=True):
    bytes_per_chunk = int(self._sample_rate * 2 * chunk_ms / 1000)
    for start in range(0, len(pcm_bytes), bytes_per_chunk):
        chunk = pcm_bytes[start:start + bytes_per_chunk]
        await self._t.send({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(chunk).decode(),
        })
    if commit:
        await self._t.send({"type": "input_audio_buffer.commit"})
    self.cost.audio_seconds += pcm_seconds(pcm_bytes, self._sample_rate)

샘플레이트나 인코딩이 session.update와 일치하지 않으면 명확한 오류 대신 지직거림이나 무음이 나올 수 있습니다. 오디오는 input_audio_buffer.append를 통해 전송되므로, 메시지당이 아니라 재생 시간 기준으로 과금됩니다.

음성 응답 받기

응답을 요청하면 오디오는 response.output_audio.delta로, 전사는 response.output_audio_transcript.delta로 도착하고, response.done이 턴을 닫습니다.

클라이언트는 이를 하나의 async 루프에서 수집합니다:

async def _collect_response(self):
    audio = bytearray()
    transcript, calls = [], []
    while True:
        event = await self._recv()
        etype = event["type"]
        if etype == "response.output_audio.delta":
            audio += base64.b64decode(event["delta"])
        elif etype == "response.output_audio_transcript.delta":
            transcript.append(event.get("delta", ""))
        elif etype == "response.function_call_arguments.done":
            calls.append(event)
        elif etype == "response.done":
            break
    return bytes(audio), "".join(transcript), calls

오디오 델타를 디코드해 순서대로 붙이고, 결과를 response.wav 파일로 저장하세요. 발신자의 발화를 캡처하려면 audio.input.transcription을 설정하고 conversation.item.input_audio_transcription.completed를 읽습니다.

예약 보조원 워크플로우 구성

이제 요소들이 대화가 됩니다: 예약 요청, 명확화 질문, 가용 시간 확인, 슬롯 제안, 확인. 턴 간 컨텍스트를 유지하려면 각 새 턴에서 대화 ID로 재연결하고 세션 재개를 선택합니다.

음성 에이전트에 도구 호출 추가

클리닉에서는 시간을 약속하기 전에 가용 시간을 확인해야 합니다. 커스텀 도구는 모델이 귀하의 코드에 접근하는 방법입니다. 모델이 요청을 내보내면 애플리케이션이 함수를 실행하고, 그 결과를 다시 전송합니다.

도구는 평범한 함수와, 세션 구성에 넣는 JSON 스키마로 이뤄집니다. 다음은 tools.py의 스키마입니다:

CHECK_AVAILABILITY_TOOL = {
    "type": "function",
    "name": "check_availability",
    "description": "Look up open appointment slots for a service on a given date. "
                   "Always call this before offering the caller a time.",
    "parameters": {
        "type": "object",
        "properties": {
            "service": {"type": "string", "description": "Service requested."},
            "date": {"type": "string", "description": "Requested date as YYYY-MM-DD."},
        },
        "required": ["service", "date"],
    },
}

루프는 정해진 형태를 가집니다. 모델이 도구를 원할 때, 인자를 담은 response.function_call_arguments.done을 보냅니다. 함수를 실행해 function_call_output을 반환한 뒤, 에이전트가 계속할 수 있도록 response.create 를 전송합니다. 마지막 response.create를 누락하면 에이전트가 침묵합니다.

도구 호출 왕복 과정 설명. 이미지: 작성자.

이런 커스텀 함수는 귀하의 코드에서 실행됩니다. Streamlit 데모는 같은 파일에서 book_appointment, transfer_to_human, end_call 세 가지를 더 등록합니다. 웹 검색, X 검색, 컬렉션 검색, 원격 MCP 도구 같은 내장 도구는 xAI 서버에서 실행됩니다.

도구 실패 처리

도구는 실패할 수 있으며, 성공을 가정하는 음성 에이전트는 존재하지 않는 슬롯을 약속할 수 있습니다. 제 ToolRegistry.execute는 예외를 던지지 않습니다. 조회 실패는 {"error": ...} 딕셔너리로 반환됩니다.

def execute(self, name, arguments):
    handler = self._handlers.get(name)
    if handler is None:
        return {"error": f"unknown tool: {name}"}
    try:
        return handler(**arguments)
    except ToolError as exc:
        return {"error": str(exc)}

명시적인 오류 상태는 에이전트가 실패한 도구 호출을 성공으로 오인하는 것을 막습니다.

비용 추적 추가

누군가에게 제공하기 전에 통화 비용을 파악하세요. 오디오는 분당 $0.05로 과금되며, 전송과 수신 모두를 합산합니다. 텍스트 입력 이벤트는 개당 $0.004입니다. function_call_output 결과와 response.create 이벤트는 과금되지 않습니다.

클라이언트는 진행 중에 비용을 추적하므로, 언제든 속성으로 읽을 수 있습니다:

@property
def audio_usd(self):
    rate = 0.05 + (0.01 if self.telephony else 0.0)
    return self.audio_seconds / 60 * rate

@property
def total_usd(self):
    return self.audio_usd + self.text_usd + self.tool_usd

xAI에서 제공한 전화번호를 사용하면 분당 $0.01의 텔레포니 할증이 붙으며, 이는 telephony=True로 설정 시 헬퍼가 적용합니다. xAI가 호스팅하는 도구는 별도 과금됩니다. 웹/X 검색은 천 건당 약 $5, 파일 검색은 약 $2.50입니다.

오류와 엣지 케이스 처리

대부분의 실패는 다음과 같은 짧은 목록에 속합니다:

• 누락되었거나 잘못된 API 키는 핸드셰이크에서 401을 반환하므로 먼저 키를 확인하세요

• 차단된 팀은 403, 레이트 리밋은 429를 반환하며, 후자는 백오프로 재시도합니다

• 잘못된 세션 구성은 400을 반환합니다. 대개 필드 이름 오타입니다

• 지원되지 않는 오디오 포맷은 오류 대신 지직거림을 내므로 세션 레이트를 맞추세요

• response.create 누락(도구 결과 이후)은 에이전트를 멈춘 상태로 둡니다

• 중복 예약 시도는 실제 문제를 일으킬 수 있으니 무작정 재시도하지 마세요

읽기 작업인 check_availability 재시도는 안전하지만, 실제 예약 같은 쓰기 작업 재시도는 이중 예약을 일으킬 수 있습니다. 데이터를 변경하는 작업은 반드시 멱등성 검사를 먼저 하세요.

클라이언트 앱을 위한 임시 토큰 사용

여기까지는 API 키가 있어야 하는 서버에서 코드가 실행된다고 가정했습니다. 브라우저나 모바일 앱이 직접 연결한다면 임시 토큰을 사용하세요.

서버는 POST https://api.x.ai/v1/realtime/client_secrets를 API 키와 함께 호출해 토큰 응답을 받고, 그 값(value)을 클라이언트에 전달합니다. 제 실행에서는 응답에 value와 expires_at이 포함되었습니다:

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            CLIENT_SECRETS_URL,
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()

브라우저는 사용자 지정 WebSocket 헤더를 설정할 수 없으므로, 토큰은 sec-websocket-protocol 헤더에 xai-client-secret. 접두사와 함께 실어 보냅니다.

워크플로우를 FastAPI 엔드포인트로 바꾸기

엔드포인트를 만들면 프론트엔드나 다른 서비스가 이 워크플로를 호출할 수 있습니다. 라우트는 Pydantic 모델로 요청 본문을 검증하고, 텍스트 메시지 또는 오디오 경로를 받아 전사, 응답 오디오, 도구 로그, 지연 시간, 추정 비용을 반환합니다.

@app.post("/appointments/voice")
async def appointments_voice(body: VoiceRequest):
    fail = {"check_availability"} if body.simulate_tool_failure else None
    assistant = AppointmentAssistant(voice=body.voice, telephony=body.telephony, fail_tools=fail)
    if body.text:
        result = await assistant.run_live(text=body.text, conversation_id=body.conversation_id)
    else:
        pcm = load_wav_as_pcm(body.audio_path, 24000)
        result = await assistant.run_live(pcm, conversation_id=body.conversation_id)
    return {
        "transcript": result.transcript,
        "audio_wav_base64": base64.b64encode(encode_wav_bytes(result.audio, 24000)).decode(),
        "tool_calls": result.tool_calls,
        "latency_seconds": round(result.latency_s, 3),
        "estimated_cost_usd": round(result.cost.total_usd, 6),
        "audio_seconds": round(result.cost.audio_seconds, 2),
        "conversation_id": result.conversation_id,
    }

uvicorn app:app --reload로 실행하고 http://localhost:8000/docs를 여세요. XAI_API_KEY 는 서버 환경에서 읽고, 요청 본문으로 받지 마세요.