[KM-501, KM-498, KM-499] Web Socket, STS and TTS

- develop STS module
- develop TTS module
- web socket protocol implementation

.dockerignore

@@ -0,0 +1,7 @@
+.env
+.venv/
+__pycache__/
+*.py[cod]
+output.pcm
+.gitignore
+.dockerignore

.env.example

@@ -0,0 +1,9 @@
+DEEPGRAM_API_KEY=
+CARTESIA_API_KEY=
+CARTESIA_VOICE_ID=
+CARTESIA_MODEL=
+DEEPGRAM_ENDPOINTING_MS=
+DEEPGRAM_UTTERANCE_END_MS=
+
+SAMPLE_RATE=
+WAKE_WORD="Hai <agent name>"

.gitignore

@@ -0,0 +1,24 @@
+# Environment
+.env
+
+# Python
+__pycache__/
+*.py[cod]
+*.pyo
+.venv/
+venv/
+env/
+
+# Test output
+output.pcm
+
+# IDE
+.idea/
+.vscode/
+*.iml
+
+# Others
+test_client.py
+/playground
+convert_audio.py
+API_CONTRACT.md

Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3.11-slim
+
+# Install uv
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+
+WORKDIR /app
+
+# Install dependencies (cached layer — only re-runs when pyproject.toml changes)
+COPY pyproject.toml .
+RUN uv sync --no-dev --no-install-project
+
+# Copy source
+COPY . .
+
+ENV PATH="/app/.venv/bin:$PATH"
+
+EXPOSE 7860
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,104 @@
 ---
-title: Demo Voice Agent Service
-emoji: 🏃
-colorFrom: red
-colorTo: gray
+title: Interview Agent Service Data Eyond
+emoji: 🌍
+colorFrom: pink
+colorTo: pink
 sdk: docker
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Voice Agent Service
+
+Real-time voice agent backend with WebSocket-based STT (AssemblyAI) and TTS (Cartesia). Accepts audio stream from client, detects wake word, and streams back synthesized speech.
+
+## Requirements
+
+- Python 3.11+
+- [uv](https://docs.astral.sh/uv/getting-started/installation/)
+- AssemblyAI API key (free tier)
+- Cartesia API key + Voice ID (free tier)
+
+## Setup
+
+**1. Clone & install dependencies**
+```bash
+uv sync
+```
+
+**2. Configure environment**
+```bash
+cp .env.example .env
+```
+
+Edit `.env` dan isi API keys:
+```env
+ASSEMBLYAI_API_KEY=your_key
+CARTESIA_API_KEY=your_key
+CARTESIA_VOICE_ID=your_voice_id
+```
+
+## Run
+
+```bash
+uv run uvicorn main:app --host 0.0.0.0 --port 7860
+```
+
+Server akan berjalan di `http://localhost:7860`.
+
+## Test
+
+**Health check:**
+```bash
+curl http://localhost:7860/health
+```
+
+Expected response:
+```json
+{
+  "status": "ok",
+  "version": "1.1.0",
+  "stt_ready": true,
+  "tts_ready": true
+}
+```
+
+**WebSocket test (kirim audio WAV, terima TTS response):**
+```bash
+uv run python test_client.py path/to/audio.wav
+```
+
+> File WAV harus dalam format: 16kHz, 16-bit, mono PCM.
+
+Output audio response akan disimpan ke `output.pcm`. Untuk memutarnya:
+```bash
+ffplay -f s16le -ar 16000 -ac 1 output.pcm
+```
+
+**Connectivity check (tanpa file audio):**
+```bash
+uv run python test_client.py
+```
+
+Mengirim 3 detik silence untuk memverifikasi koneksi WebSocket berhasil.
+
+## Docker
+
+**Build:**
+```bash
+docker build -t voice-agent .
+```
+
+**Run:**
+```bash
+docker run -p 7860:7860 --env-file .env voice-agent
+```
+
+## Wake Word
+
+Default wake word: **"hi voice agent"** (case-insensitive)
+
+Contoh: ucapkan _"Hi Voice Agent, what time is it?"_ → agent akan membalas dengan TTS _"what time is it"_.
+
+## API Contract
+
+Lihat [API_CONTRACT.md](API_CONTRACT.md) untuk dokumentasi lengkap WebSocket protocol.

main.py

@@ -0,0 +1,73 @@
+import json
+import logging
+import uvicorn
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from fastapi.responses import JSONResponse
+from src.pipeline import EchoPipeline
+from src.config import DEEPGRAM_API_KEY, CARTESIA_API_KEY, CARTESIA_VOICE_ID
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
+logger = logging.getLogger(__name__)
+
+VERSION = "1.1.0"
+
+app = FastAPI(title="Voice Agent Service", version=VERSION)
+
+
+@app.get("/health")
+async def health() -> JSONResponse:
+    stt_ready = bool(DEEPGRAM_API_KEY)
+    tts_ready = bool(CARTESIA_API_KEY and CARTESIA_VOICE_ID)
+    all_ready = stt_ready and tts_ready
+
+    body: dict = {
+        "status": "ok" if all_ready else "degraded",
+        "version": VERSION,
+        "stt_ready": stt_ready,
+        "tts_ready": tts_ready,
+    }
+    if not all_ready:
+        body["message"] = "One or more API keys are missing. Check your .env file."
+
+    return JSONResponse(status_code=200 if all_ready else 503, content=body)
+
+
+@app.websocket("/ws/voice")
+async def voice_ws(ws: WebSocket) -> None:
+    await ws.accept()
+    logger.info("Client connected: %s", ws.client)
+
+    async def send_audio(chunk: bytes) -> None:
+        await ws.send_bytes(chunk)
+
+    async def send_event(event: dict) -> None:
+        await ws.send_text(json.dumps(event))
+
+    pipeline = EchoPipeline(send_audio=send_audio, send_event=send_event)
+    pipeline.start()
+
+    try:
+        while True:
+            data = await ws.receive()
+            if "bytes" in data and data["bytes"]:
+                pipeline.feed_audio(data["bytes"])
+            elif "text" in data and data["text"]:
+                try:
+                    msg = json.loads(data["text"])
+                    action = msg.get("action")
+                    if action == "stop":
+                        break
+                    elif action == "ping":
+                        await ws.send_text(json.dumps({"event": "pong"}))
+                    elif action == "interrupt":
+                        await pipeline.interrupt()
+                except json.JSONDecodeError:
+                    pass
+    except WebSocketDisconnect:
+        logger.info("Client disconnected: %s", ws.client)
+    finally:
+        pipeline.stop()
+
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host="0.0.0.0", port=7860, reload=False)

pyproject.toml

@@ -0,0 +1,26 @@
+[project]
+name = "voice-agent"
+version = "1.1.0"
+requires-python = ">=3.11"
+dependencies = [
+    "fastapi==0.115.0",
+    "uvicorn[standard]==0.30.6",
+    "websockets==13.1",
+    "httpx==0.27.2",
+    "python-dotenv==1.0.1",
+    "assemblyai==0.33.0",
+    "cartesia==1.3.1",
+    "deepgram-sdk>=6.1.1",
+]
+
+[project.optional-dependencies]
+dev = [
+    "websockets==13.1",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src"]

server.log

@@ -0,0 +1,95 @@
+INFO:     Started server process [3804]
+INFO:     Waiting for application startup.
+INFO:     Application startup complete.
+ERROR:    [Errno 10048] error while attempting to bind on address ('0.0.0.0', 7860): [winerror 10048] only one usage of each socket address (protocol/network address/port) is normally permitted
+INFO:     Waiting for application shutdown.
+INFO:     INFO:     127.0.0.1:62731 - "GET /health HTTP/1.1" 200 OK
+INFO:     127.0.0.1:58019 - "GET /health HTTP/1.1" 200 OK
+INFO:     127.0.0.1:58328 - "GET /health HTTP/1.1" 200 OK
+INFO:     ('127.0.0.1', 64692) - "WebSocket /ws/voice" [accepted]
+2026-04-19 10:55:42,299 [INFO] main: Client connected: Address(host='127.0.0.1', port=64692)
+2026-04-19 10:55:43,272 [ERROR] src.stt.assemblyai_client: AssemblyAI error: Could not connect to the real-time service: server rejected WebSocket connection: HTTP 404
+2026-04-19 10:55:43,280 [INFO] src.stt.assemblyai_client: AssemblyAI STT connected
+INFO:     connection open
+2026-04-19 10:55:43,287 [INFO] src.stt.assemblyai_client: AssemblyAI STT closed
+INFO:     connection closed
+INFO:     ('127.0.0.1', 64730) - "WebSocket /ws/voice" [accepted]
+2026-04-19 10:55:45,331 [INFO] main: Client connected: Address(host='127.0.0.1', port=64730)
+2026-04-19 10:55:45,932 [ERROR] src.stt.assemblyai_client: AssemblyAI error: Could not connect to the real-time service: server rejected WebSocket connection: HTTP 404
+2026-04-19 10:55:45,932 [INFO] src.stt.assemblyai_client: AssemblyAI STT connected
+INFO:     connection open
+2026-04-19 10:55:45,934 [INFO] src.pipeline: TTS interrupted by client
+2026-04-19 10:55:45,935 [INFO] src.stt.assemblyai_client: AssemblyAI STT closed
+ERROR:    Exception in ASGI application
+Traceback (most recent call last):
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\uvicorn\protocols\websockets\websockets_impl.py", line 244, in run_asgi
+    result = await self.app(self.scope, self.asgi_receive, self.asgi_send)  # type: ignore[func-returns-value]
+             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\uvicorn\middleware\proxy_headers.py", line 70, in __call__
+    return await self.app(scope, receive, send)
+           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\fastapi\applications.py", line 1054, in __call__
+    await super().__call__(scope, receive, send)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\applications.py", line 113, in __call__
+    await self.middleware_stack(scope, receive, send)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\middleware\errors.py", line 152, in __call__
+    await self.app(scope, receive, send)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\middleware\exceptions.py", line 62, in __call__
+    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\_exception_handler.py", line 62, in wrapped_app
+    raise exc
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\_exception_handler.py", line 51, in wrapped_app
+    await app(scope, receive, sender)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\routing.py", line 715, in __call__
+    await self.middleware_stack(scope, receive, send)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\routing.py", line 735, in app
+    await route.handle(scope, receive, send)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\routing.py", line 362, in handle
+    await self.app(scope, receive, send)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\routing.py", line 95, in app
+    await wrap_app_handling_exceptions(app, session)(scope, receive, send)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\_exception_handler.py", line 62, in wrapped_app
+    raise exc
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\_exception_handler.py", line 51, in wrapped_app
+    await app(scope, receive, sender)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\routing.py", line 93, in app
+    await func(session)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\fastapi\routing.py", line 383, in app
+    await dependant.call(**solved_result.values)
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\main.py", line 51, in voice_ws
+    data = await ws.receive()
+           ^^^^^^^^^^^^^^^^^^
+  File "C:\Users\HarryyantoIshaqAgasi\Documents\HarryProjects\DataEyond\Interview-Agent-Service-Data-Eyond\.venv\Lib\site-packages\starlette\websockets.py", line 56, in receive
+    raise RuntimeError('Cannot call "receive" once a disconnect message has been received.')
+RuntimeError: Cannot call "receive" once a disconnect message has been received.
+INFO:     connection closed
+INFO:     ('127.0.0.1', 64763) - "WebSocket /ws/voice" [accepted]
+2026-04-19 10:55:47,981 [INFO] main: Client connected: Address(host='127.0.0.1', port=64763)
+2026-04-19 10:55:48,609 [ERROR] src.stt.assemblyai_client: AssemblyAI error: Could not connect to the real-time service: server rejected WebSocket connection: HTTP 404
+2026-04-19 10:55:48,609 [INFO] src.stt.assemblyai_client: AssemblyAI STT connected
+INFO:     connection open
+2026-04-19 10:55:48,611 [INFO] src.stt.assemblyai_client: AssemblyAI STT closed
+INFO:     connection closed
+INFO:     127.0.0.1:64988 - "GET /health HTTP/1.1" 200 OK
+INFO:     ('127.0.0.1', 65015) - "WebSocket /ws/voice" [accepted]
+2026-04-19 10:56:08,787 [INFO] main: Client connected: Address(host='127.0.0.1', port=65015)
+2026-04-19 10:56:09,406 [ERROR] src.stt.assemblyai_client: AssemblyAI error: Could not connect to the real-time service: server rejected WebSocket connection: HTTP 404
+2026-04-19 10:56:09,406 [INFO] src.stt.assemblyai_client: AssemblyAI STT connected
+INFO:     connection open
+2026-04-19 10:56:09,408 [INFO] src.stt.assemblyai_client: AssemblyAI STT closed
+INFO:     connection closed
+INFO:     ('127.0.0.1', 65047) - "WebSocket /ws/voice" [accepted]
+2026-04-19 10:56:11,439 [INFO] main: Client connected: Address(host='127.0.0.1', port=65047)
+2026-04-19 10:56:12,038 [ERROR] src.stt.assemblyai_client: AssemblyAI error: Could not connect to the real-time service: server rejected WebSocket connection: HTTP 404
+2026-04-19 10:56:12,039 [INFO] src.stt.assemblyai_client: AssemblyAI STT connected
+INFO:     connection open
+2026-04-19 10:56:12,040 [INFO] src.pipeline: TTS interrupted by client
+2026-04-19 10:56:12,040 [INFO] src.stt.assemblyai_client: AssemblyAI STT closed
+INFO:     connection closed
+INFO:     ('127.0.0.1', 54874) - "WebSocket /ws/voice" [accepted]
+2026-04-19 10:56:14,069 [INFO] main: Client connected: Address(host='127.0.0.1', port=54874)
+2026-04-19 10:56:14,681 [ERROR] src.stt.assemblyai_client: AssemblyAI error: Could not connect to the real-time service: server rejected WebSocket connection: HTTP 404
+2026-04-19 10:56:14,681 [INFO] src.stt.assemblyai_client: AssemblyAI STT connected
+INFO:     connection open
+2026-04-19 10:56:14,682 [INFO] src.stt.assemblyai_client: AssemblyAI STT closed
+INFO:     connection closed

src/config.py

@@ -0,0 +1,15 @@
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+DEEPGRAM_API_KEY: str = os.environ["DEEPGRAM_API_KEY"]
+CARTESIA_API_KEY: str = os.environ["CARTESIA_API_KEY"]
+CARTESIA_VOICE_ID: str = os.environ["CARTESIA_VOICE_ID"]
+
+SAMPLE_RATE: int = int(os.getenv("SAMPLE_RATE", "16000"))
+WAKE_WORDS: list[str] = [w.strip().lower() for w in os.getenv("WAKE_WORD", "Hai EMA").split(",") if w.strip()]
+CARTESIA_MODEL: str = os.getenv("CARTESIA_MODEL", "sonic-3")
+DEEPGRAM_LANGUAGE: str = os.getenv("DEEPGRAM_LANGUAGE", "id")
+DEEPGRAM_ENDPOINTING_MS: int = int(os.getenv("DEEPGRAM_ENDPOINTING_MS", "300"))
+DEEPGRAM_UTTERANCE_END_MS: int = int(os.getenv("DEEPGRAM_UTTERANCE_END_MS", "2000"))

src/knowledge/loader.py

@@ -0,0 +1,6 @@
+# Phase 2 placeholder: PDF knowledge base loader
+# Will use: pypdf or pdfplumber for extraction, then chunk + embed for RAG
+#
+# Interface (to be implemented):
+#   async def load_pdf(path: str) -> list[str]: ...
+#   async def retrieve(query: str, top_k: int = 3) -> list[str]: ...

src/llm/answerer.py

@@ -0,0 +1,5 @@
+# Phase 2 placeholder: LLM answer generation
+# Will use: Anthropic Claude API with retrieved PDF context
+#
+# Interface (to be implemented):
+#   async def answer(question: str, context_chunks: list[str]) -> str: ...

src/pipeline.py

@@ -0,0 +1,96 @@
+import asyncio
+import logging
+from typing import Callable, Awaitable
+from src.config import WAKE_WORDS
+from src.stt.deepgram_client import DeepgramStreamer
+from src.tts.cartesia_client import synthesize_stream
+
+logger = logging.getLogger(__name__)
+
+SendAudioCallback = Callable[[bytes], Awaitable[None]]
+SendEventCallback = Callable[[dict], Awaitable[None]]
+
+
+class EchoPipeline:
+    """
+    MVP Phase 1: Echo pipeline.
+    Audio in → STT → wake word check → TTS (echo) → audio out.
+
+    Phase 2 will replace the echo step with LLM answer generation.
+    """
+
+    def __init__(self, send_audio: SendAudioCallback, send_event: SendEventCallback):
+        self._send_audio = send_audio
+        self._send_event = send_event
+        self._loop = asyncio.get_event_loop()
+        self._stt = DeepgramStreamer(
+            on_final_transcript=self._on_final_transcript,
+            loop=self._loop,
+        )
+        self._tts_lock = asyncio.Lock()
+        self._tts_task: asyncio.Task | None = None
+
+    def start(self) -> None:
+        self._stt.start()
+
+    def feed_audio(self, chunk: bytes) -> None:
+        self._stt.send_audio(chunk)
+
+    async def _on_final_transcript(self, text: str) -> None:
+        await self._send_event({"event": "transcript", "text": text})
+
+        reply_text = self._extract_reply(text)
+        if reply_text is None:
+            logger.debug("No wake word detected, ignoring: %s", text)
+            return
+
+        logger.info("Wake word detected, replying with: %s", reply_text)
+        await self._send_event({"event": "reply", "text": reply_text})
+
+        # Run as a Task so it can be cancelled via interrupt()
+        self._tts_task = asyncio.create_task(self._speak(reply_text))
+
+    def _extract_reply(self, transcript: str) -> str | None:
+        lower = transcript.lower()
+        match = min(
+            ((lower.index(w), w) for w in WAKE_WORDS if w in lower),
+            key=lambda x: x[0],
+            default=None,
+        )
+        if match is None:
+            return None
+        idx = match[0] + len(match[1])
+        reply = transcript[idx:].strip(" ,.")
+        return reply if reply else "Hello! How can I help you?"
+
+    async def _speak(self, text: str) -> None:
+        async with self._tts_lock:
+            try:
+                async for audio_chunk in synthesize_stream(text):
+                    await self._send_audio(audio_chunk)
+                await self._send_event({"event": "tts_end"})
+            except asyncio.CancelledError:
+                # Interrupted mid-stream — do not send tts_end
+                raise
+            except Exception:
+                logger.exception("TTS error for text: %s", text)
+                await self._send_event({
+                    "event": "error",
+                    "code": "TTS_ERROR",
+                    "message": "TTS generation failed",
+                })
+
+    async def interrupt(self) -> None:
+        if self._tts_task and not self._tts_task.done():
+            self._tts_task.cancel()
+            try:
+                await self._tts_task
+            except asyncio.CancelledError:
+                pass
+        await self._send_event({"event": "interrupted"})
+        logger.info("TTS interrupted by client")
+
+    def stop(self) -> None:
+        if self._tts_task and not self._tts_task.done():
+            self._tts_task.cancel()
+        self._stt.stop()

src/stt/assemblyai_client.py

@@ -0,0 +1,55 @@
+import asyncio
+import logging
+from typing import Callable, Awaitable
+import assemblyai as aai
+from src.config import ASSEMBLYAI_API_KEY, SAMPLE_RATE
+
+logger = logging.getLogger(__name__)
+
+OnTranscriptCallback = Callable[[str], Awaitable[None]]
+
+
+class AssemblyAIStreamer:
+    """
+    Wraps AssemblyAI real-time streaming STT.
+    Audio chunks (PCM 16kHz 16-bit mono bytes) are fed via `send_audio()`.
+    When a final transcript arrives, `on_final_transcript` callback is awaited.
+    """
+
+    def __init__(self, on_final_transcript: OnTranscriptCallback, loop: asyncio.AbstractEventLoop):
+        self._on_final_transcript = on_final_transcript
+        self._loop = loop
+        self._transcriber: aai.RealtimeTranscriber | None = None
+
+    def start(self) -> None:
+        aai.settings.api_key = ASSEMBLYAI_API_KEY
+
+        self._transcriber = aai.RealtimeTranscriber(
+            sample_rate=SAMPLE_RATE,
+            on_data=self._on_data,
+            on_error=self._on_error,
+        )
+        self._transcriber.connect()
+        logger.info("AssemblyAI STT connected")
+
+    def _on_data(self, transcript: aai.RealtimeTranscript) -> None:
+        if not isinstance(transcript, aai.RealtimeFinalTranscript):
+            return
+        text = transcript.text.strip()
+        if not text:
+            return
+        logger.info("Final transcript: %s", text)
+        asyncio.run_coroutine_threadsafe(self._on_final_transcript(text), self._loop)
+
+    def _on_error(self, error: aai.RealtimeError) -> None:
+        logger.error("AssemblyAI error: %s", error)
+
+    def send_audio(self, chunk: bytes) -> None:
+        if self._transcriber:
+            self._transcriber.stream(chunk)
+
+    def stop(self) -> None:
+        if self._transcriber:
+            self._transcriber.close()
+            self._transcriber = None
+            logger.info("AssemblyAI STT closed")

src/stt/deepgram_client.py

@@ -0,0 +1,109 @@
+import asyncio
+import logging
+import threading
+from typing import Callable, Awaitable
+
+from deepgram import DeepgramClient
+from deepgram.core.events import EventType
+from src.config import DEEPGRAM_API_KEY, SAMPLE_RATE, DEEPGRAM_LANGUAGE, DEEPGRAM_ENDPOINTING_MS, DEEPGRAM_UTTERANCE_END_MS
+
+logger = logging.getLogger(__name__)
+
+OnTranscriptCallback = Callable[[str], Awaitable[None]]
+
+
+class DeepgramStreamer:
+    """
+    Wraps Deepgram real-time streaming STT.
+    Audio chunks (PCM 16kHz 16-bit mono bytes) are fed via `send_audio()`.
+
+    Final transcript segments are buffered. A flush timer (DEEPGRAM_UTTERANCE_END_MS)
+    is reset on each new segment. When the timer fires, all buffered segments are
+    joined and emitted as one complete utterance.
+    """
+
+    def __init__(self, on_final_transcript: OnTranscriptCallback, loop: asyncio.AbstractEventLoop):
+        self._on_final_transcript = on_final_transcript
+        self._loop = loop
+        self._connection = None
+        self._cm = None
+        self._ready = threading.Event()
+        self._transcript_buffer: list[str] = []
+        self._buffer_lock = threading.Lock()
+        self._flush_handle: asyncio.TimerHandle | None = None
+
+    def start(self) -> None:
+        client = DeepgramClient(api_key=DEEPGRAM_API_KEY)
+
+        connect_kwargs: dict = dict(
+            model="nova-2",
+            encoding="linear16",
+            sample_rate=SAMPLE_RATE,
+            endpointing=DEEPGRAM_ENDPOINTING_MS,
+        )
+        if DEEPGRAM_LANGUAGE:
+            connect_kwargs["language"] = DEEPGRAM_LANGUAGE
+
+        self._cm = client.listen.v1.connect(**connect_kwargs)
+        self._connection = self._cm.__enter__()
+
+        self._connection.on(EventType.OPEN, self._on_open)
+        self._connection.on(EventType.MESSAGE, self._on_message)
+        self._connection.on(EventType.ERROR, self._on_error)
+
+        # start_listening() is blocking — run in background thread
+        thread = threading.Thread(target=self._connection.start_listening, daemon=True)
+        thread.start()
+
+        # Wait until connection is open before returning
+        self._ready.wait(timeout=5)
+        logger.info("Deepgram STT connected")
+
+    def _on_open(self, _result) -> None:
+        self._ready.set()
+
+    def _on_message(self, result) -> None:
+        try:
+            transcript = result.channel.alternatives[0].transcript.strip()
+        except (AttributeError, IndexError):
+            return
+        if not transcript or not result.is_final:
+            return
+        with self._buffer_lock:
+            self._transcript_buffer.append(transcript)
+        logger.debug("Buffered segment: %s", transcript)
+        # Reset flush timer on the event loop thread
+        self._loop.call_soon_threadsafe(self._reset_flush_timer)
+
+    def _reset_flush_timer(self) -> None:
+        if self._flush_handle is not None:
+            self._flush_handle.cancel()
+        delay = DEEPGRAM_UTTERANCE_END_MS / 1000.0
+        self._flush_handle = self._loop.call_later(delay, self._flush_buffer)
+
+    def _flush_buffer(self) -> None:
+        with self._buffer_lock:
+            if not self._transcript_buffer:
+                return
+            full_text = " ".join(self._transcript_buffer)
+            self._transcript_buffer.clear()
+        self._flush_handle = None
+        logger.info("Final transcript: %s", full_text)
+        asyncio.ensure_future(self._on_final_transcript(full_text), loop=self._loop)
+
+    def _on_error(self, error) -> None:
+        logger.error("Deepgram error: %s", error)
+
+    def send_audio(self, chunk: bytes) -> None:
+        if self._connection:
+            self._connection.send_media(chunk)
+
+    def stop(self) -> None:
+        if self._cm:
+            try:
+                self._cm.__exit__(None, None, None)
+            except Exception:
+                pass
+            self._connection = None
+            self._cm = None
+            logger.info("Deepgram STT closed")

src/tts/cartesia_client.py

@@ -0,0 +1,44 @@
+import logging
+from typing import AsyncIterator
+import httpx
+from src.config import CARTESIA_API_KEY, CARTESIA_VOICE_ID, CARTESIA_MODEL, SAMPLE_RATE
+
+logger = logging.getLogger(__name__)
+
+CARTESIA_TTS_URL = "https://api.cartesia.ai/tts/bytes"
+
+# PCM 16-bit signed little-endian, matching our pipeline sample rate
+OUTPUT_FORMAT = {
+    "container": "raw",
+    "encoding": "pcm_s16le",
+    "sample_rate": SAMPLE_RATE,
+}
+
+
+async def synthesize_stream(text: str) -> AsyncIterator[bytes]:
+    """
+    Calls Cartesia TTS and yields raw PCM audio chunks as they arrive.
+    Yields bytes immediately for lowest possible latency.
+    """
+    payload = {
+        "model_id": CARTESIA_MODEL,
+        "transcript": text,
+        "voice": {"mode": "id", "id": CARTESIA_VOICE_ID},
+        "output_format": OUTPUT_FORMAT,
+    }
+    headers = {
+        "X-API-Key": CARTESIA_API_KEY,
+        "Cartesia-Version": "2025-04-16",
+        "Content-Type": "application/json",
+    }
+
+    logger.info("Cartesia TTS: synthesizing '%s'", text[:60])
+    async with httpx.AsyncClient(timeout=30) as client:
+        async with client.stream("POST", CARTESIA_TTS_URL, json=payload, headers=headers) as response:
+            if response.status_code >= 400:
+                body = await response.aread()
+                logger.error("Cartesia TTS %d: %s", response.status_code, body.decode())
+            response.raise_for_status()
+            async for chunk in response.aiter_bytes(chunk_size=4096):
+                if chunk:
+                    yield chunk