TTS and STT using gemini model

.env.example

@@ -7,6 +7,7 @@ DEEPGRAM_UTTERANCE_END_MS=
 
 SAMPLE_RATE=
 WAKE_WORD="Hai <agent name>"
+WAKE_WORD_ENABLED=false
 
 # Gemini TTS (optional — hanya dibutuhkan jika provider=gemini)
 GOOGLE_API_KEY=
@@ -17,3 +18,16 @@ GEMINI_TTS_LANGUAGE=id-ID
 # Chatbot agent service
 CHATBOT_BASE_URL=http://localhost:8000/chatbot-knowledgemanagement
 CHATBOT_TIMEOUT=30
+
+# Google Cloud Project (untuk referensi Vertex AI di masa depan)
+GOOGLE_PROJECT_NAME=
+GOOGLE_PROJECT_NUMBER=
+
+# Gemini STT
+GEMINI_STT_MODEL=gemini-2.0-flash
+GEMINI_LIVE_MODEL=gemini-live-2.5-flash-preview
+GEMINI_STT_LANGUAGE=id-ID
+
+# Provider default untuk WebSocket /ws/voice: "deepgram" atau "gemini"
+STT_PROVIDER=gemini
+TTS_PROVIDER=gemini

README.md

@@ -4,7 +4,7 @@ emoji: 🌍
 colorFrom: pink
 colorTo: pink
 sdk: docker
-pinned: false
+pinned: true
 ---
 
 # Voice Agent Service
@@ -52,16 +52,16 @@ WAKE_WORD=Hai EMA                    # Default: "Hai EMA"
 ## Run
 
 ```bash
-uv run uvicorn main:app --host 0.0.0.0 --port 7860
+uv run uvicorn main:app --host 0.0.0.0 --port 7861
 ```
 
-Server akan berjalan di `http://localhost:7860`.
+Server akan berjalan di `http://localhost:7861`.
 
 ## Test
 
 **Health check:**
 ```bash
-curl http://localhost:7860/health
+curl http://localhost:7861/health
 ```
 
 Expected response:
@@ -111,7 +111,7 @@ docker build -t voice-agent .
 
 **Run:**
 ```bash
-docker run -p 7860:7860 --env-file .env voice-agent
+docker run -p 7861:7861 --env-file .env voice-agent
 ```
 
 ## Wake Word
@@ -160,7 +160,7 @@ Client Audio Playback
 
 ## WebSocket Protocol
 
-**Endpoint:** `ws://localhost:7860/ws/voice`
+**Endpoint:** `ws://localhost:7861/ws/voice`
 
 **Client → Server:**
 

main.py

@@ -1,16 +1,17 @@
 import json
 import logging
 import uvicorn
-from fastapi import FastAPI, WebSocket, WebSocketDisconnect, File, UploadFile, HTTPException, Query
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect, File, Form, UploadFile, HTTPException, Query
 from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel
 from src.pipeline import VoicePipeline
 from typing import Literal
 from src.config import (
     DEEPGRAM_API_KEY, CARTESIA_API_KEY, CARTESIA_VOICE_ID,
-    SAMPLE_RATE, GOOGLE_API_KEY, CHATBOT_BASE_URL,
+    SAMPLE_RATE, GOOGLE_API_KEY, CHATBOT_BASE_URL, STT_PROVIDER, TTS_PROVIDER, WAKE_WORD_ENABLED,
 )
-from src.stt.deepgram_rest import transcribe_audio
+from src.stt.deepgram_rest import transcribe_audio as deepgram_transcribe
+from src.stt.gemini_stt import transcribe_audio as gemini_stt_transcribe
 from src.tts.cartesia_client import synthesize_stream as cartesia_synthesize
 from src.tts.gemini_client import synthesize_stream as gemini_synthesize, GEMINI_SAMPLE_RATE
 
@@ -36,6 +37,7 @@ async def health() -> JSONResponse:
         "stt_ready": stt_ready,
         "tts_ready": tts_ready,
         "gemini_tts_ready": gemini_ready,
+        "gemini_stt_ready": gemini_ready,
         "chatbot_ready": chatbot_ready,
     }
     if not all_ready:
@@ -46,16 +48,26 @@ async def health() -> JSONResponse:
 
 class TTSRequest(BaseModel):
     text: str
-    provider: Literal["cartesia", "gemini"] = "cartesia"
+    provider: Literal["cartesia", "gemini"] = "gemini"
 
 
 @app.post("/stt")
-async def speech_to_text(audio: UploadFile = File(...)) -> JSONResponse:
+async def speech_to_text(
+    audio: UploadFile = File(...),
+    provider: str = Form(default="gemini"),
+) -> JSONResponse:
     data = await audio.read()
     if not data:
         raise HTTPException(status_code=400, detail="Audio file is empty.")
     mimetype = audio.content_type or "audio/wav"
-    result = await transcribe_audio(data, mimetype=mimetype)
+
+    if provider == "gemini":
+        if not GOOGLE_API_KEY:
+            raise HTTPException(status_code=503, detail="Gemini STT not configured: missing GOOGLE_API_KEY.")
+        result = await gemini_stt_transcribe(data, mimetype=mimetype)
+    else:
+        result = await deepgram_transcribe(data, mimetype=mimetype)
+
     return JSONResponse(content=result)
 
 
@@ -94,9 +106,15 @@ async def voice_ws(
     site: str = Query(default="HO"),
     role: str = Query(default="engineer"),
     agent: str = Query(default="analysis"),
+    stt_provider: str = Query(default=STT_PROVIDER),
+    tts_provider: str = Query(default=TTS_PROVIDER),
+    wake_word_enabled: bool = Query(default=WAKE_WORD_ENABLED),
 ) -> None:
     await ws.accept()
-    logger.info("Client connected: %s (user_id=%s)", ws.client, user_id)
+    logger.info(
+        "Client connected: %s (user_id=%s, stt=%s, tts=%s)",
+        ws.client, user_id, stt_provider, tts_provider,
+    )
 
     async def send_audio(chunk: bytes) -> None:
         await ws.send_bytes(chunk)
@@ -104,6 +122,16 @@ async def voice_ws(
     async def send_event(event: dict) -> None:
         await ws.send_text(json.dumps(event))
 
+    tts_sample_rate = GEMINI_SAMPLE_RATE if tts_provider == "gemini" else SAMPLE_RATE
+    await send_event({
+        "event": "tts_config",
+        "tts_provider": tts_provider,
+        "stt_provider": stt_provider,
+        "sample_rate": tts_sample_rate,
+        "encoding": "pcm_s16le",
+        "channels": 1,
+    })
+
     pipeline = VoicePipeline(
         send_audio=send_audio,
         send_event=send_event,
@@ -114,6 +142,9 @@ async def voice_ws(
         site=site,
         role=role,
         agent=agent,
+        stt_provider=stt_provider,
+        tts_provider=tts_provider,
+        wake_word_enabled=wake_word_enabled,
     )
     pipeline.start()
 

src/config.py

@@ -9,12 +9,22 @@ 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()]
+WAKE_WORD_ENABLED: bool = os.getenv("WAKE_WORD_ENABLED", "false").lower() in ("1", "true", "yes")
 CARTESIA_MODEL: str = os.getenv("CARTESIA_MODEL", "sonic-3")
 
 GOOGLE_API_KEY: str = os.getenv("GOOGLE_API_KEY", "")
+GOOGLE_PROJECT_NAME: str = os.getenv("GOOGLE_PROJECT_NAME", "")
+GOOGLE_PROJECT_NUMBER: str = os.getenv("GOOGLE_PROJECT_NUMBER", "")
+
 GEMINI_TTS_MODEL: str = os.getenv("GEMINI_TTS_MODEL", "gemini-2.5-flash-preview-tts")
 GEMINI_TTS_VOICE: str = os.getenv("GEMINI_TTS_VOICE", "Autonoe")
 GEMINI_TTS_LANGUAGE: str = os.getenv("GEMINI_TTS_LANGUAGE", "id-ID")
+
+GEMINI_STT_MODEL: str = os.getenv("GEMINI_STT_MODEL", "gemini-2.0-flash")
+GEMINI_LIVE_MODEL: str = os.getenv("GEMINI_LIVE_MODEL", "gemini-live-2.5-flash-preview")
+GEMINI_STT_LANGUAGE: str = os.getenv("GEMINI_STT_LANGUAGE", "id-ID")
+STT_PROVIDER: str = os.getenv("STT_PROVIDER", "deepgram")
+TTS_PROVIDER: str = os.getenv("TTS_PROVIDER", "cartesia")
 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/pipeline.py

@@ -2,9 +2,10 @@ import uuid
 import asyncio
 import logging
 from typing import Callable, Awaitable
-from src.config import WAKE_WORDS
+from src.config import WAKE_WORDS, WAKE_WORD_ENABLED
 from src.stt.deepgram_client import DeepgramStreamer
-from src.tts.cartesia_client import synthesize_stream
+from src.tts.cartesia_client import synthesize_stream as cartesia_synthesize_stream
+from src.tts.gemini_client import synthesize_stream as gemini_synthesize_stream, GEMINI_SAMPLE_RATE
 from src.chatbot.client import call_agent
 
 logger = logging.getLogger(__name__)
@@ -33,10 +34,15 @@ class VoicePipeline:
         site: str = "HO",
         role: str = "engineer",
         agent: str = "analysis",
+        stt_provider: str = "deepgram",
+        tts_provider: str = "cartesia",
+        wake_word_enabled: bool = WAKE_WORD_ENABLED,
     ):
         self._send_audio = send_audio
         self._send_event = send_event
         self._loop = asyncio.get_event_loop()
+        self._tts_provider = tts_provider
+        self._wake_word_enabled = wake_word_enabled
 
         self._user_id = user_id
         self._fullname = fullname
@@ -53,10 +59,17 @@ class VoicePipeline:
         # Running chat history for multi-turn context
         self._chat_history: list[dict] = []
 
-        self._stt = DeepgramStreamer(
-            on_final_transcript=self._on_final_transcript,
-            loop=self._loop,
-        )
+        if stt_provider == "gemini":
+            from src.stt.gemini_stt import GeminiSTTStreamer
+            self._stt = GeminiSTTStreamer(
+                on_final_transcript=self._on_final_transcript,
+                loop=self._loop,
+            )
+        else:
+            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
 
@@ -69,12 +82,18 @@ class VoicePipeline:
     async def _on_final_transcript(self, text: str) -> None:
         await self._send_event({"event": "transcript", "text": text})
 
-        question = self._extract_question(text)
-        if question is None:
-            logger.debug("No wake word detected, ignoring: %s", text)
-            return
+        if self._wake_word_enabled:
+            question = self._extract_question(text)
+            if question is None:
+                logger.debug("No wake word detected, ignoring: %s", text)
+                return
+            logger.info("Wake word detected, question: %s", question)
+        else:
+            question = text.strip()
+            if not question:
+                return
+            logger.info("Processing transcript (wake word disabled): %s", question)
 
-        logger.info("Wake word detected, question: %s", question)
         self._tts_task = asyncio.create_task(self._answer_and_speak(question))
 
     def _extract_question(self, transcript: str) -> str | None:
@@ -124,7 +143,12 @@ class VoicePipeline:
     async def _speak(self, text: str) -> None:
         async with self._tts_lock:
             try:
-                async for audio_chunk in synthesize_stream(text):
+                stream = (
+                    gemini_synthesize_stream(text)
+                    if self._tts_provider == "gemini"
+                    else cartesia_synthesize_stream(text)
+                )
+                async for audio_chunk in stream:
                     await self._send_audio(audio_chunk)
                 await self._send_event({"event": "tts_end"})
             except asyncio.CancelledError:

src/stt/gemini_stt.py

@@ -0,0 +1,137 @@
+import asyncio
+import logging
+from typing import Callable, Awaitable
+
+from google import genai
+from google.genai import types
+
+from src.config import (
+    GOOGLE_API_KEY,
+    GEMINI_STT_MODEL,
+    GEMINI_LIVE_MODEL,
+    GEMINI_STT_LANGUAGE,
+    DEEPGRAM_UTTERANCE_END_MS,
+)
+
+logger = logging.getLogger(__name__)
+
+OnTranscriptCallback = Callable[[str], Awaitable[None]]
+
+
+async def transcribe_audio(data: bytes, mimetype: str = "audio/wav") -> dict:
+    """
+    Transcribes a full audio file using Gemini multimodal API.
+    Returns dict with 'text', 'language', and 'duration'.
+    """
+    client = genai.Client(api_key=GOOGLE_API_KEY)
+    response = await client.aio.models.generate_content(
+        model=GEMINI_STT_MODEL,
+        contents=[
+            types.Part.from_bytes(data=data, mime_type=mimetype),
+            f"Transcribe this audio accurately in {GEMINI_STT_LANGUAGE}. "
+            "Return only the transcription text, no commentary or explanation.",
+        ],
+    )
+    transcript = (response.text or "").strip()
+    logger.info("Gemini STT transcript: %s", transcript[:80])
+    return {"text": transcript, "language": GEMINI_STT_LANGUAGE, "duration": None}
+
+
+class GeminiSTTStreamer:
+    """
+    Real-time streaming STT using the Gemini Live API.
+
+    Interface matches DeepgramStreamer: start(), send_audio(chunk), stop().
+    Audio chunks must be PCM Linear16 16kHz mono bytes.
+
+    Final transcript segments are buffered and flushed after DEEPGRAM_UTTERANCE_END_MS
+    of silence (reusing the same configurable delay).
+    """
+
+    def __init__(self, on_final_transcript: OnTranscriptCallback, loop: asyncio.AbstractEventLoop):
+        self._on_final_transcript = on_final_transcript
+        self._loop = loop
+        self._audio_queue: asyncio.Queue[bytes | None] = None  # type: ignore[assignment]
+        self._session_task: asyncio.Task | None = None
+        self._transcript_buffer: list[str] = []
+        self._flush_handle: asyncio.TimerHandle | None = None
+
+    def start(self) -> None:
+        self._audio_queue = asyncio.Queue()
+        self._session_task = asyncio.ensure_future(self._run_session(), loop=self._loop)
+        logger.info("Gemini Live STT starting...")
+
+    def send_audio(self, chunk: bytes) -> None:
+        if self._audio_queue is not None:
+            self._audio_queue.put_nowait(chunk)
+
+    def stop(self) -> None:
+        if self._session_task and not self._session_task.done():
+            self._session_task.cancel()
+        if self._audio_queue is not None:
+            # Unblock the send loop if it's waiting on the queue
+            self._audio_queue.put_nowait(None)
+        if self._flush_handle is not None:
+            self._flush_handle.cancel()
+            self._flush_handle = None
+        logger.info("Gemini Live STT stopped")
+
+    async def _run_session(self) -> None:
+        client = genai.Client(api_key=GOOGLE_API_KEY)
+        config = types.LiveConnectConfig(
+            response_modalities=["TEXT"],
+            input_audio_transcription=types.AudioTranscriptionConfig(),
+        )
+        try:
+            async with client.aio.live.connect(model=GEMINI_LIVE_MODEL, config=config) as session:
+                logger.info("Gemini Live STT connected")
+                await asyncio.gather(
+                    self._send_loop(session),
+                    self._receive_loop(session),
+                )
+        except asyncio.CancelledError:
+            pass
+        except Exception:
+            logger.exception("Gemini Live session error")
+
+    async def _send_loop(self, session) -> None:
+        try:
+            while True:
+                chunk = await self._audio_queue.get()
+                if chunk is None:
+                    break
+                await session.send_realtime_input(
+                    audio=types.Blob(data=chunk, mime_type="audio/pcm;rate=16000")
+                )
+        except asyncio.CancelledError:
+            raise
+
+    async def _receive_loop(self, session) -> None:
+        try:
+            async for response in session.receive():
+                if response.server_content is None:
+                    continue
+                transcription = response.server_content.input_transcription
+                if transcription and transcription.text:
+                    self._buffer_transcript(transcription.text)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.exception("Gemini Live receive error")
+
+    def _buffer_transcript(self, text: str) -> None:
+        self._transcript_buffer.append(text)
+        logger.debug("Buffered segment: %s", text)
+        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:
+        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)