dvgateway 1.4.4

Python SDK for DVGateway — real-time voice AI integration

dvgateway

DVGateway Python SDK — AI 음성 서비스를 실시간 전화 통화에 연결합니다.

설치

# 기본 SDK
pip install dvgateway

# AI 어댑터 포함 (Anthropic, OpenAI)
pip install dvgateway[adapters]

빠른 시작

import asyncio
import os
from dvgateway import DVGatewayClient
from dvgateway.adapters.stt import DeepgramAdapter
from dvgateway.adapters.llm import AnthropicAdapter
from dvgateway.adapters.tts import ElevenLabsAdapter

async def main():
    gw = DVGatewayClient(
        base_url="http://localhost:8080",
        auth={"type": "apiKey", "api_key": os.environ["DV_API_KEY"]},
    )

    await (
        gw.pipeline()
        .stt(DeepgramAdapter(api_key=os.environ["DEEPGRAM_API_KEY"], language="ko"))
        .llm(AnthropicAdapter(api_key=os.environ["ANTHROPIC_API_KEY"], model="claude-sonnet-4-6"))
        .tts(ElevenLabsAdapter(api_key=os.environ["ELEVENLABS_API_KEY"]))
        .start()
    )

asyncio.run(main())

Comfort Noise — AI 처리 중 Dead Air 방지

게이트웨이에서 GW_COMFORT_NOISE_ENABLED=true를 설정하면, AI 처리 중 무음 구간에 배경 소음이 자동으로 주입됩니다.

파이프라인 빌더 사용 시 자동 동작 (별도 코드 불필요):

# thinking:start/stop 시그널이 자동 전송됩니다
await (
    gw.pipeline()
    .stt(DeepgramAdapter(api_key="...", language="ko"))
    .llm(AnthropicAdapter(api_key="...", model="claude-sonnet-4-6"))
    .tts(ElevenLabsAdapter(api_key="..."))
    .start()
)

수동 제어:

# WebSocket 시그널 (저레이턴시)
audio_stream = gw.stream_audio(linked_id)
await audio_stream.send_thinking_start()  # 배경 소음 시작
# ... AI 처리 ...
await audio_stream.send_thinking_stop()   # 배경 소음 중단

# REST API
await gw.start_thinking(linked_id)
await gw.stop_thinking(linked_id)

TTS 재생 완료 이벤트 — on_tts_complete (v1.4+)

gw.inject_tts() 는 오디오 iterator 소진 시점에 리턴하지만, 게이트웨이가 실제로 Asterisk에 프레임을 다 넣은 시점이 아닙니다. tts:complete 이벤트가 authoritative 완료 신호입니다.

주요 활용:

• 순차 TTS 재생 (인사말 → 메뉴 체이닝)
• AI 응답 종료 시점 정확히 포착 (S2S 모드)
• 고객 VAD 리오픈 타이밍 제어

import asyncio
from dvgateway import DVGatewayClient, TTSCompleteEvent

# 인사말 재생
await gw.inject_tts(linked_id, welcome_tts.synthesize("안녕하세요"))

# 인사말 끝나면 메뉴 안내로 전환
def on_done(ev: TTSCompleteEvent) -> None:
    if ev.linked_id != linked_id:
        return
    asyncio.ensure_future(
        gw.inject_tts(linked_id, menu_tts.synthesize("1번을 눌러주세요"))
    )

unsub = gw.on_tts_complete(on_done)
# 필요시 unsub() 호출로 구독 해지

발생 시점: Player.Play() 세션 정상 EOF / 명시적 Stop / 동시 호출 선점. Stale 세션은 중복 발행 안 함.

전체 이벤트 카탈로그:

이벤트	발생 시점	페이로드
call:new	새 통화 시작	session (CallSession)
call:ended	통화 종료	linked_id, duration_sec
conf:join / conf:leave / conf:ended	회의 이벤트	linked_id, conf_id
tts:complete	TTS 재생 완료	linked_id, tenant_id, server_id, timestamp

자세한 가이드: docs/sdk-guide/05-events-fallback.md

요구사항

• Python 3.10+
• aiohttp >= 3.9.0
• websockets >= 12.0

라이선스

MIT