Build a Voice Agent using Vobiz

1
import base64
2
import io
3
import json
4
import os
5
import wave
6
from typing import AsyncIterator
7
8
import httpx
9
10
BASE_URL = "https://api.sarvam.ai"
11
STT_MODEL = os.getenv("SARVAM_STT_MODEL", "saaras:v3")
12
LLM_MODEL = os.getenv("SARVAM_LLM_MODEL", "sarvam-105b")
13
TTS_MODEL = os.getenv("SARVAM_TTS_MODEL", "bulbul:v3")
14
SAMPLE_WIDTH_BYTES = 2
15
16
17
class SarvamError(RuntimeError):
18
    pass
19
20
21
def pcm_to_wav(pcm_audio: bytes, sample_rate: int) -> bytes:
22
    """Wrap raw mono Linear16 samples in a WAV container for the STT upload."""
23
    buffer = io.BytesIO()
24
    with wave.open(buffer, "wb") as wav_file:
25
        wav_file.setnchannels(1)
26
        wav_file.setsampwidth(SAMPLE_WIDTH_BYTES)
27
        wav_file.setframerate(sample_rate)
28
        wav_file.writeframes(pcm_audio)
29
    return buffer.getvalue()
30
31
32
def raw_linear16(audio: bytes, expected_rate: int) -> bytes:
33
    """Vobiz playAudio needs container-free samples; Sarvam TTS may return a WAV."""
34
    if audio.startswith(b"RIFF"):
35
        with wave.open(io.BytesIO(audio), "rb") as wav_file:
36
            audio = wav_file.readframes(wav_file.getnframes())
37
    return audio
38
39
40
class SarvamClient:
41
    def __init__(self, api_key: str) -> None:
42
        self._headers = {"api-subscription-key": api_key}
43
        self._http = httpx.AsyncClient(base_url=BASE_URL, timeout=30.0)
44
45
    async def aclose(self) -> None:
46
        await self._http.aclose()
47
48
    async def transcribe(self, pcm_audio: bytes, sample_rate: int, language_code: str) -> str:
49
        wav_audio = pcm_to_wav(pcm_audio, sample_rate)
50
        response = await self._http.post(
51
            "/speech-to-text",
52
            headers=self._headers,
53
            files={"file": ("caller.wav", wav_audio, "audio/wav")},
54
            data={"model": STT_MODEL, "mode": "codemix", "language_code": language_code},
55
        )
56
        response.raise_for_status()
57
        return (response.json().get("transcript") or "").strip()
58
59
    async def stream_reply(
60
        self, messages: list[dict[str, str]], max_tokens: int = 2000
61
    ) -> AsyncIterator[str]:
62
        """Yield only reply text. These models also stream a private
63
        ``reasoning_content`` field that must never reach the caller's ear."""
64
        payload = {
65
            "model": LLM_MODEL,
66
            "messages": messages,
67
            "max_tokens": max_tokens,
68
            "stream": True,
69
        }
70
        async with self._http.stream(
71
            "POST", "/v1/chat/completions", headers=self._headers, json=payload
72
        ) as response:
73
            response.raise_for_status()
74
            async for line in response.aiter_lines():
75
                if not line.startswith("data:"):
76
                    continue
77
                data = line[len("data:"):].strip()
78
                if data in ("", "[DONE]"):
79
                    continue
80
                for choice in json.loads(data).get("choices", []):
81
                    delta = (choice.get("delta") or {}).get("content")
82
                    if delta:
83
                        yield delta
84
85
    async def synthesize(self, text: str, language_code: str, speaker: str, sample_rate: int) -> bytes:
86
        response = await self._http.post(
87
            "/text-to-speech",
88
            headers=self._headers,
89
            json={
90
                "text": text,
91
                "target_language_code": language_code,
92
                "speaker": speaker,
93
                "model": TTS_MODEL,
94
                "speech_sample_rate": sample_rate,
95
                "output_audio_codec": "linear16",
96
            },
97
        )
98
        response.raise_for_status()
99
        audios = response.json().get("audios") or []
100
        if not audios:
101
            raise SarvamError("Sarvam TTS returned no audio")
102
        return raw_linear16(base64.b64decode("".join(audios)), sample_rate)

1
import asyncio
2
import audioop
3
import base64
4
import json
5
import os
6
import re
7
from typing import Any
8
9
from sarvam import SarvamClient
10
11
OUTBOUND_SAMPLE_RATE = int(os.getenv("OUTBOUND_SAMPLE_RATE", "8000"))
12
SAMPLE_WIDTH_BYTES = 2
13
AGENT_LANGUAGE = os.getenv("AGENT_LANGUAGE", "en-IN")
14
TTS_SPEAKER = os.getenv("TTS_SPEAKER", "anand")
15
16
SYSTEM_PROMPT = (
17
    "You are a Sarvam AI voice assistant speaking with a caller on a phone "
18
    "line. You understand English and Indian languages, including code-mixed "
19
    "speech. Keep every reply to one or two short spoken sentences. Never use "
20
    "markdown, bullet points, emoji, or symbols, because your reply is read aloud."
21
)
22
GREETING = "Namaste! I am your Sarvam AI assistant. How can I help you today?"
23
24
# Split streamed LLM text on sentence enders, including the Devanagari danda,
25
# so the caller hears sentence one while the model is still writing sentence two.
26
SENTENCE_END = re.compile(r"[.!?|।॥]+[\s\"')\]]*|\n+")
27
MIN_TTS_CHARS = 24
28
29
SILENCE_THRESHOLD = 250       # RMS energy that counts as speech
30
END_OF_SPEECH_MS = 700        # trailing silence that ends an utterance
31
MIN_SPEECH_MS = 200           # shorter utterances are treated as noise
32
33
34
class CallSession:
35
    def __init__(self, websocket: Any) -> None:
36
        self.websocket = websocket
37
        self.stream_id: str | None = None
38
        self.input_sample_rate = 8000
39
        self.input_encoding = "audio/x-mulaw"
40
41
        self.sarvam = SarvamClient(os.getenv("SARVAM_API_KEY", ""))
42
        self.conversation = [{"role": "system", "content": SYSTEM_PROMPT}]
43
44
        self._audio = bytearray()
45
        self._speech_ms = 0.0
46
        self._silence_ms = 0.0
47
        self._speech_active = False
48
        self._playback_active = False
49
        self._tasks: set[asyncio.Task] = set()
50
51
    async def handle_message(self, raw_message: str) -> None:
52
        data = json.loads(raw_message)
53
        event = data.get("event")
54
55
        if event == "start":
56
            await self._handle_start(data)
57
        elif event == "media":
58
            await self._handle_media(data)
59
        elif event in {"playedStream", "clearedAudio"}:
60
            self._playback_active = False
61
62
    async def _handle_start(self, data: dict) -> None:
63
        start = data.get("start") or {}
64
        self.stream_id = data.get("streamId") or start.get("streamId")
65
66
        # Trust the negotiated format rather than assuming a rate: Vobiz sends
67
        # mu-law at 8 kHz or Linear16 at 16 kHz depending on <Stream contentType>.
68
        media_format = start.get("mediaFormat") or {}
69
        self.input_sample_rate = int(media_format.get("sampleRate", 8000))
70
        self.input_encoding = str(media_format.get("encoding", "audio/x-mulaw"))
71
72
        self._spawn(self._speak(GREETING))
73
74
    async def _handle_media(self, data: dict) -> None:
75
        media = data.get("media") or {}
76
        payload = media.get("payload")
77
        if not payload:
78
            return
79
80
        raw = base64.b64decode(payload)
81
        pcm = audioop.ulaw2lin(raw, 2) if self.input_encoding == "audio/x-mulaw" else raw
82
        chunk_ms = len(pcm) * 1000 / (self.input_sample_rate * SAMPLE_WIDTH_BYTES)
83
        rms = audioop.rms(pcm, SAMPLE_WIDTH_BYTES)
84
85
        if rms >= SILENCE_THRESHOLD:
86
            if self._playback_active:
87
                await self._clear_audio()  # barge-in
88
            self._speech_active = True
89
            self._speech_ms += chunk_ms
90
            self._silence_ms = 0
91
            self._audio.extend(pcm)
92
        elif self._speech_active:
93
            self._silence_ms += chunk_ms
94
            self._audio.extend(pcm)
95
96
        if self._speech_active and self._silence_ms >= END_OF_SPEECH_MS:
97
            utterance, speech_ms = bytes(self._audio), self._speech_ms
98
            self._audio.clear()
99
            self._speech_ms = self._silence_ms = 0
100
            self._speech_active = False
101
            if speech_ms >= MIN_SPEECH_MS:
102
                self._spawn(self._process_utterance(utterance))
103
104
    async def _process_utterance(self, pcm_audio: bytes) -> None:
105
        transcript = await self.sarvam.transcribe(pcm_audio, self.input_sample_rate, AGENT_LANGUAGE)
106
        if not transcript:
107
            return
108
        self.conversation.append({"role": "user", "content": transcript})
109
110
        pending = ""
111
        async for delta in self.sarvam.stream_reply(self.conversation):
112
            pending += delta
113
            match = SENTENCE_END.search(pending)
114
            if match and match.end() >= MIN_TTS_CHARS:
115
                sentence, pending = pending[:match.end()].strip(), pending[match.end():]
116
                if sentence:
117
                    await self._speak(sentence)
118
        if pending.strip():
119
            await self._speak(pending.strip())
120
121
    async def _speak(self, text: str) -> None:
122
        pcm = await self.sarvam.synthesize(text, AGENT_LANGUAGE, TTS_SPEAKER, OUTBOUND_SAMPLE_RATE)
123
        await self._play(pcm)
124
125
    async def _play(self, pcm_audio: bytes) -> None:
126
        chunk_size = OUTBOUND_SAMPLE_RATE * SAMPLE_WIDTH_BYTES * 20 // 1000  # 20 ms chunks
127
        self._playback_active = True
128
        for offset in range(0, len(pcm_audio), chunk_size):
129
            await self.websocket.send_text(json.dumps({
130
                "event": "playAudio",
131
                "streamId": self.stream_id,
132
                "media": {
133
                    "contentType": "audio/x-l16",
134
                    "sampleRate": OUTBOUND_SAMPLE_RATE,
135
                    "payload": base64.b64encode(pcm_audio[offset:offset + chunk_size]).decode(),
136
                },
137
            }))
138
        await self.websocket.send_text(json.dumps({"event": "checkpoint", "streamId": self.stream_id, "name": "tts"}))
139
140
    async def _clear_audio(self) -> None:
141
        await self.websocket.send_text(json.dumps({"event": "clearAudio", "streamId": self.stream_id}))
142
        self._playback_active = False
143
144
    def _spawn(self, coroutine) -> None:
145
        task = asyncio.create_task(coroutine)
146
        self._tasks.add(task)
147
        task.add_done_callback(self._tasks.discard)
148
149
    async def close(self) -> None:
150
        for task in list(self._tasks):
151
            task.cancel()
152
        await self.sarvam.aclose()

1
import html
2
import os
3
from pathlib import Path
4
5
import uvicorn
6
from dotenv import load_dotenv
7
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
8
from fastapi.responses import Response
9
10
from agent import CallSession
11
12
load_dotenv(Path(__file__).resolve().parent / ".env")
13
14
PUBLIC_URL = os.getenv("PUBLIC_URL", "").rstrip("/")
15
STREAM_CONTENT_TYPE = os.getenv("STREAM_CONTENT_TYPE", "audio/x-mulaw;rate=8000")
16
17
app = FastAPI()
18
19
20
def websocket_url(public_url: str) -> str:
21
    if public_url.startswith("https://"):
22
        return "wss://" + public_url.removeprefix("https://") + "/ws"
23
    return "ws://" + public_url.removeprefix("http://") + "/ws"
24
25
26
@app.api_route("/answer", methods=["GET", "POST"])
27
async def answer(request: Request) -> Response:
28
    """The XML Vobiz executes once the call is answered."""
29
    base_url = html.escape(PUBLIC_URL, quote=True)
30
    ws_url = html.escape(websocket_url(PUBLIC_URL), quote=False)
31
    xml = f"""<?xml version="1.0" encoding="UTF-8"?>
32
<Response>
33
    <Stream bidirectional="true"
34
            audioTrack="inbound"
35
            keepCallAlive="true"
36
            contentType="{html.escape(STREAM_CONTENT_TYPE, quote=True)}"
37
            statusCallbackUrl="{base_url}/stream-status"
38
            statusCallbackMethod="POST">
39
        {ws_url}
40
    </Stream>
41
    <Hangup/>
42
</Response>"""
43
    return Response(content=xml, media_type="application/xml")
44
45
46
@app.websocket("/ws")
47
async def websocket_endpoint(websocket: WebSocket) -> None:
48
    await websocket.accept()
49
    session = CallSession(websocket)
50
    try:
51
        async for message in websocket.iter_text():
52
            await session.handle_message(message)
53
    except WebSocketDisconnect:
54
        pass
55
    finally:
56
        await session.close()
57
58
59
@app.post("/stream-status")
60
async def stream_status(request: Request) -> Response:
61
    await request.form()
62
    return Response(status_code=204)
63
64
65
@app.post("/hangup")
66
async def hangup(request: Request) -> Response:
67
    await request.form()
68
    return Response(status_code=204)
69
70
71
if __name__ == "__main__":
72
    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("HTTP_PORT", "8000")))

1
AGENT_LANGUAGE = "hi-IN"
2
TTS_SPEAKER = "simran"  # or: priya, ishita, kavya, aditya, anand, rohan

1
AGENT_LANGUAGE = "unknown"  # Sarvam auto-detects the caller's spoken language
2
TTS_SPEAKER = "anand"