diff --git a/agents-core/vision_agents/core/stt/stt.py b/agents-core/vision_agents/core/stt/stt.py index 5ef8f05e7..168846171 100644 --- a/agents-core/vision_agents/core/stt/stt.py +++ b/agents-core/vision_agents/core/stt/stt.py @@ -9,6 +9,7 @@ from vision_agents.core.edge.types import Participant from vision_agents.core.events.manager import EventManager from vision_agents.core.observability import MetricsCollector +from vision_agents.core.stt import events from vision_agents.core.stt.events import ( STTConnectedEvent, STTDisconnectedEvent, @@ -87,6 +88,7 @@ def __init__( self.provider_name = provider_name or self.__class__.__name__ self.events = EventManager() + self.events.register_events_from_module(events, ignore_not_compatible=True) self.metrics = MetricsCollector() self._output: Stream[Transcript | TurnEnded | TurnStarted] = Stream() diff --git a/plugins/cartesia/README.md b/plugins/cartesia/README.md index 925ee7dc6..f0bbfa355 100644 --- a/plugins/cartesia/README.md +++ b/plugins/cartesia/README.md @@ -2,7 +2,7 @@ [Cartesia](https://cartesia.ai) is a service that provides Speech-to-Text (STT) and Text-to-Speech (TTS) capabilities. It's designed for real-time voice applications, making it ideal for voice AI agents, transcription pipelines, and conversational interfaces. -The Cartesia plugin for the Stream Python AI SDK allows you to add TTS functionality to your project. +The Cartesia plugin for the Stream Python AI SDK allows you to add STT and TTS functionality to your project. ## Installation @@ -16,19 +16,20 @@ uv add vision-agents-plugins-cartesia ## Examples -Read on for some key details and check out our [Cartesia examples](https://github.com/GetStream/vision-agents/tree/main/examples/other_examples/plugins_examples/tts_cartesia) to see working code samples: +Read on for some key details and check out our [Cartesia examples](https://github.com/GetStream/Vision-Agents/tree/main/plugins/cartesia/example) to see working code samples: -- in [tts.py](https://github.com/GetStream/vision-agents/tree/main/examples/other_examples/plugins_examples/tts_cartesia/tts.py) we see a simple bot greeting users upon joining a call -- in [narrator-example.py](https://github.com/GetStream/vision-agents/tree/main/examples/other_examples/plugins_examples/tts_cartesia/narrator-example.py) we see a well-prompted combination of a STT -> LLM -> TTS flow that leverages the powers of Cartesia's Sonic 3 model to narrate a creative story from the user's input +- in [main.py](https://github.com/GetStream/Vision-Agents/blob/main/plugins/cartesia/example/main.py) we see a voice bot that uses Cartesia STT and TTS in a Stream call +- in [narrator-example.py](https://github.com/GetStream/Vision-Agents/blob/main/plugins/cartesia/example/narrator-example.py) we see a well-prompted combination of an STT -> LLM -> TTS flow that leverages Cartesia's Ink and Sonic models to narrate a creative story from the user's input ## Initialisation -The Cartesia plugin for Stream exists in the form of the `TTS` class: +The Cartesia plugin for Stream exposes `STT` and `TTS` classes: ```python from vision_agents.plugins import cartesia +stt = cartesia.STT() tts = cartesia.TTS() ``` @@ -39,17 +40,39 @@ tts = cartesia.TTS() ## Parameters -These are the parameters available in the CartesiaTTS plugin for you to customise: +These are the parameters available in the Cartesia STT plugin for you to customise: + +| Name | Type | Default | Description | +|---------------------|-----------------|----------------------------------------|---------------------------------------------------------------------------------------------------------------| +| `api_key` | `str` or `None` | `None` | Your Cartesia API key. If not provided, the plugin will look for the `CARTESIA_API_KEY` environment variable. | +| `model` | `str` | `"ink-2"` | ID of the Cartesia STT model to use. | +| `sample_rate` | `int` | `16000` | Sample rate (in Hz) sent to Cartesia. | +| `encoding` | `str` | `"pcm_s16le"` | PCM encoding sent to Cartesia. | +| `cartesia_version` | `str` | `"2026-03-01"` | Cartesia API version used for the turn-detection websocket. | + +These are the parameters available in the Cartesia TTS plugin for you to customise: | Name | Type | Default | Description | |---------------|-----------------|------------------------------------------|---------------------------------------------------------------------------------------------------------------| | `api_key` | `str` or `None` | `None` | Your Cartesia API key. If not provided, the plugin will look for the `CARTESIA_API_KEY` environment variable. | -| `model_id` | `str` | `"sonic-3"` | ID of the Cartesia STT or TTS model to use. Defaults to the recently released Sonic-3 | +| `model_id` | `str` | `"sonic-3.5"` | ID of the Cartesia TTS model to use. | | `voice_id` | `str` or `None` | `"f9836c6e-a0bd-460e-9d3c-f7299fa60f94"` | ID of the voice to use for TTS responses. | | `sample_rate` | `int` | `16000` | Sample rate (in Hz) used for audio processing. | ## Functionality +### Send audio to transcribe speech + +`STT` streams PCM audio to Cartesia Ink and emits transcript and turn events that Vision Agents can use for interruption and eager turn handling. + +```python theme={null} +agent = Agent( + ..., + stt=cartesia.STT(), + tts=cartesia.TTS(), +) +``` + ### Send text to convert to speech The `send_iter()` method sends the text passed in for the service to synthesize diff --git a/plugins/cartesia/example/README.md b/plugins/cartesia/example/README.md index 5a48460d5..f921e408c 100644 --- a/plugins/cartesia/example/README.md +++ b/plugins/cartesia/example/README.md @@ -1,20 +1,20 @@ -# Stream + Cartesia TTS Bot Example +# Stream + Cartesia Voice Bot Example -This example demonstrates how to build a text-to-speech bot that joins a Stream video call and greets participants -using [Cartesia's](https://cartesia.ai/?utm_medium=partner&utm_source=getstream) Sonic voices. +This example demonstrates how to build a voice bot that joins a Stream video call, transcribes participants with Cartesia STT, and speaks responses with Cartesia TTS. ## What it does -- 🤖 Creates a TTS bot that joins a Stream video call -- 🌐 Opens a browser interface for users to join the call -- 🔊 Greets users when they join using Cartesia TTS -- 🎙️ Sends audio directly to the call in real-time +- Creates a voice bot that joins a Stream video call +- Uses Cartesia Ink for realtime STT and turn detection +- Uses Cartesia Sonic for TTS responses +- Uses OpenAI for the LLM response ## Prerequisites 1. **Stream Account**: Get your API credentials from [Stream Dashboard](https://getstream.io/try-for-free/?utm_source=github.com&utm_medium=referral&utm_campaign=vision_agents) 2. **Cartesia Account**: Get your API key from [Cartesia](https://cartesia.ai/?utm_medium=partner&utm_source=getstream) -3. **Python 3.10+**: Required for running the example +3. **OpenAI Account**: Set an `OPENAI_API_KEY` for the example LLM. +4. **Python 3.10+**: Required for running the example ## Installation @@ -31,7 +31,7 @@ You can use your preferred package manager, but we recommend [`uv`](https://docs ``` 3. **Set up environment variables:** - Rename `env.example` to `.env` and fill in your actual credentials. + Create a `.env` file with `STREAM_API_KEY`, `STREAM_API_SECRET`, `CARTESIA_API_KEY`, and `OPENAI_API_KEY`. ## Usage @@ -40,3 +40,5 @@ Run the example: ```bash uv run main.py run ``` + +Join the generated call, speak into your microphone, and the bot should answer out loud. diff --git a/plugins/cartesia/example/main.py b/plugins/cartesia/example/main.py index 7ef26bc36..de271546a 100644 --- a/plugins/cartesia/example/main.py +++ b/plugins/cartesia/example/main.py @@ -1,22 +1,24 @@ #!/usr/bin/env python3 """ -Example: Text-to-Speech with Cartesia using Agent class +Example: Speech-to-Text and Text-to-Speech with Cartesia using Agent class This minimal example shows how to: -1. Create an Agent with TTS capabilities +1. Create an Agent with Cartesia STT and TTS 2. Join a Stream video call -3. Greet users when they join +3. Greet users and respond to spoken input -Run it, join the call in your browser, and hear the bot greet you 🗣️ +Run it, join the call in your browser, and speak to the bot. Usage:: - python main.py + uv run main.py run The script looks for the following env vars (see `env.example`): STREAM_API_KEY / STREAM_API_SECRET CARTESIA_API_KEY + OPENAI_API_KEY """ +import asyncio import logging from dotenv import load_dotenv @@ -31,11 +33,15 @@ async def create_agent(**kwargs) -> Agent: - # Create agent with TTS + """Create an agent with Cartesia STT and TTS.""" agent = Agent( edge=getstream.Edge(), - agent_user=User(name="TTS Bot", id="agent"), - instructions="I'm a TTS bot that greets users when they join.", + agent_user=User(name="Cartesia Voice Bot", id="agent"), + instructions=( + "You're a helpful voice AI assistant. " + "Keep replies short and conversational." + ), + stt=cartesia.STT(), llm=openai.LLM(model="gpt-4o-mini"), tts=cartesia.TTS(), ) @@ -44,12 +50,16 @@ async def create_agent(**kwargs) -> Agent: async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None: - # Create a call call = await agent.create_call(call_type, call_id) - # Join call and wait + logger.info("Starting Cartesia STT/TTS voice bot") + async with agent.join(call): - await agent.simple_response("tell me something interesting in a short sentence") + logger.info("Joined call") + await asyncio.sleep(3) + await agent.simple_response( + "Hello! I'm listening. What would you like to talk about?" + ) await agent.finish() diff --git a/plugins/cartesia/example/narrator-example.py b/plugins/cartesia/example/narrator-example.py index 5be378bfd..98880ecdc 100644 --- a/plugins/cartesia/example/narrator-example.py +++ b/plugins/cartesia/example/narrator-example.py @@ -24,7 +24,7 @@ from vision_agents.core import Runner from vision_agents.core.agents import Agent, AgentLauncher from vision_agents.core.edge.types import User -from vision_agents.plugins import cartesia, deepgram, getstream, openai +from vision_agents.plugins import cartesia, getstream, openai logger = logging.getLogger(__name__) @@ -37,7 +37,7 @@ async def create_agent(**kwargs) -> Agent: edge=getstream.Edge(), agent_user=User(name="Narrator", id="agent"), instructions="You're the narrator of a story. When you're given a topic start narrating a story and make heavy use of the audio markup tags to customize the speech output that are described in @sonic3-info.md.", - stt=deepgram.STT(), + stt=cartesia.STT(), llm=openai.LLM(model="gpt-4o-mini"), tts=cartesia.TTS(), ) diff --git a/plugins/cartesia/example/pyproject.toml b/plugins/cartesia/example/pyproject.toml index e774a3a8d..e76785644 100644 --- a/plugins/cartesia/example/pyproject.toml +++ b/plugins/cartesia/example/pyproject.toml @@ -9,12 +9,10 @@ dependencies = [ "vision-agents-plugins-cartesia", "vision-agents-plugins-getstream", "vision-agents-plugins-openai", - "vision-agents-plugins-deepgram", ] [tool.uv.sources] -vision-agents = { editable = true } -vision-agents-plugins-cartesia = { editable = true } -vision-agents-plugins-getstream = { editable = true } -vision-agents-plugins-openai = { editable = true } -vision-agents-plugins-deepgram = { editable = true } +"vision-agents" = { path = "../../../agents-core", editable = true } +"vision-agents-plugins-cartesia" = { path = "..", editable = true } +"vision-agents-plugins-getstream" = { path = "../../getstream", editable = true } +"vision-agents-plugins-openai" = { path = "../../openai", editable = true } diff --git a/plugins/cartesia/pyproject.toml b/plugins/cartesia/pyproject.toml index c46056e04..eb4c1a62e 100644 --- a/plugins/cartesia/pyproject.toml +++ b/plugins/cartesia/pyproject.toml @@ -5,14 +5,15 @@ build-backend = "hatchling.build" [project] name = "vision-agents-plugins-cartesia" dynamic = ["version"] -description = "Cartesia TTS integration for Vision Agents" +description = "Cartesia STT and TTS integration for Vision Agents" readme = "README.md" -keywords = ["cartesia", "TTS", "text-to-speech", "AI", "voice agents", "agents"] +keywords = ["cartesia", "STT", "TTS", "speech-to-text", "text-to-speech", "AI", "voice agents", "agents"] requires-python = ">=3.10" license = "MIT" dependencies = [ "vision-agents", "cartesia>=3.0.2,<3.1", + "websockets>=15.0.1,<16", ] [project.urls] diff --git a/plugins/cartesia/tests/test_stt.py b/plugins/cartesia/tests/test_stt.py new file mode 100644 index 000000000..f5c9fb135 --- /dev/null +++ b/plugins/cartesia/tests/test_stt.py @@ -0,0 +1,199 @@ +import asyncio +import os + +import pytest +import vision_agents.plugins.cartesia.stt as cartesia_stt_module + +from vision_agents.core.edge.types import Participant +from vision_agents.core.stt import Transcript +from vision_agents.core.turn_detection import TurnEnded, TurnStarted +from vision_agents.plugins import cartesia + + +def _require_cartesia_api_key() -> str: + api_key = os.getenv("CARTESIA_API_KEY") + if not api_key: + pytest.fail( + "Cartesia integration tests require CARTESIA_API_KEY. " + "Set CARTESIA_API_KEY in the environment or in a .env file before " + "running tests marked with @pytest.mark.integration.", + pytrace=False, + ) + return api_key + + +class TestCartesiaSTT: + @pytest.fixture + def participant(self) -> Participant: + return Participant({}, user_id="test-user", id="test-user") + + async def test_start_duplicate_guard_runs_before_websocket_connect( + self, monkeypatch + ): + class FakeConnection: + def __init__(self) -> None: + self.closed = False + self._closed = asyncio.Event() + + def __aiter__(self): + return self + + async def __anext__(self): + await self._closed.wait() + raise StopAsyncIteration + + async def close(self) -> None: + self.closed = True + self._closed.set() + + connections: list[FakeConnection] = [] + + async def fake_connect(*args, **kwargs) -> FakeConnection: + connection = FakeConnection() + connections.append(connection) + return connection + + monkeypatch.setattr(cartesia_stt_module.websockets, "connect", fake_connect) + + stt = cartesia.STT(api_key="fake") + await stt.start() + try: + with pytest.raises(ValueError, match="already started"): + await stt.start() + finally: + await stt.close() + + assert len(connections) == 1 + + async def test_start_closed_guard_runs_before_websocket_connect(self, monkeypatch): + connect_calls = 0 + + async def fake_connect(*args, **kwargs): + nonlocal connect_calls + connect_calls += 1 + raise AssertionError("websocket connect should not be called") + + monkeypatch.setattr(cartesia_stt_module.websockets, "connect", fake_connect) + + stt = cartesia.STT(api_key="fake") + await stt.close() + + with pytest.raises(ValueError, match="closed"): + await stt.start() + + assert connect_calls == 0 + + async def test_process_audio_fails_fast_after_listener_error( + self, silence_1s_16khz + ): + class FailingConnection: + def __aiter__(self): + return self + + async def __anext__(self): + raise RuntimeError("socket broke") + + stt = cartesia.STT(api_key="fake") + stt.started = True + stt.connection = FailingConnection() + stt._connection_ready.set() + + await stt._listen() + + assert stt.started is False + assert stt.connection is None + assert not stt._connection_ready.is_set() + + with pytest.raises(RuntimeError, match="websocket connection failed") as exc: + await asyncio.wait_for( + stt.process_audio(silence_1s_16khz), + timeout=0.1, + ) + + assert isinstance(exc.value.__cause__, RuntimeError) + assert str(exc.value.__cause__) == "socket broke" + + def test_defaults_to_latest_model_and_turn_detection(self): + stt = cartesia.STT(api_key="fake") + + assert stt.model == "ink-2" + assert stt.turn_detection is True + assert stt.eager_turn_detection is True + assert "model=ink-2" in stt._build_websocket_url() + assert "cartesia_version=2026-03-01" in stt._build_websocket_url() + + async def test_handle_turn_events(self, participant): + stt = cartesia.STT(api_key="fake") + stt._current_participant = participant + + stt._handle_message({"type": "turn.start", "confidence": 0.8}) + stt._handle_message( + { + "type": "turn.update", + "transcript": "hello", + "words": [{"word": "hello", "confidence": 0.9}], + "duration": 0.5, + } + ) + stt._handle_message( + { + "type": "turn.eager_end", + "transcript": "hello world", + "confidence": 0.7, + } + ) + stt._handle_message({"type": "turn.resume", "confidence": 0.6}) + stt._handle_message( + { + "type": "turn.end", + "transcript": "hello world again", + "confidence": 0.95, + "duration_ms": 1200, + "trailing_silence_ms": 250, + } + ) + + items = await stt.output.collect(timeout=0) + + transcripts = [i for i in items if isinstance(i, Transcript)] + assert [t.text for t in transcripts] == [ + "hello", + "hello world", + "hello world again", + ] + assert [t.final for t in transcripts] == [False, False, True] + assert transcripts[0].confidence == 0.9 + assert transcripts[-1].model_name == "ink-2" + + turns_started = [i for i in items if isinstance(i, TurnStarted)] + turns_ended = [i for i in items if isinstance(i, TurnEnded)] + assert len(turns_started) == 2 + assert turns_started[0].confidence == 0.8 + assert turns_started[1].confidence == 0.6 + assert [t.eager for t in turns_ended] == [True, False] + assert turns_ended[-1].duration_ms == 1200 + assert turns_ended[-1].trailing_silence_ms == 250 + + @pytest.mark.integration + async def test_transcribe_mia_audio_48khz( + self, + mia_audio_48khz, + silence_2s_48khz, + participant, + ): + stt = cartesia.STT(api_key=_require_cartesia_api_key()) + await stt.start() + try: + await stt.process_audio(mia_audio_48khz, participant=participant) + await stt.process_audio(silence_2s_48khz, participant=participant) + + items = await stt.output.collect(timeout=10.0) + finally: + await stt.close() + + transcripts = [i for i in items if isinstance(i, Transcript)] + finals = [t for t in transcripts if t.final] + assert finals, "No final Transcript emitted by Cartesia STT" + full_transcript = " ".join(t.text for t in finals) + assert "forgotten treasures" in full_transcript.lower() + assert any(isinstance(i, TurnEnded) for i in items) diff --git a/plugins/cartesia/tests/test_tts.py b/plugins/cartesia/tests/test_tts.py index 38d283a6f..7eaf52549 100644 --- a/plugins/cartesia/tests/test_tts.py +++ b/plugins/cartesia/tests/test_tts.py @@ -1,21 +1,32 @@ import os import pytest -from dotenv import load_dotenv from vision_agents.plugins import cartesia -# Load environment variables -load_dotenv() + +def _require_cartesia_api_key() -> str: + api_key = os.getenv("CARTESIA_API_KEY") + if not api_key: + pytest.fail( + "Cartesia integration tests require CARTESIA_API_KEY. " + "Set CARTESIA_API_KEY in the environment or in a .env file before " + "running tests marked with @pytest.mark.integration.", + pytrace=False, + ) + return api_key + + +class TestCartesiaTTS: + def test_defaults_to_sonic_35(self) -> None: + tts = cartesia.TTS(api_key="fake") + assert tts.model_id == "sonic-3.5" -@pytest.mark.skipif( - os.getenv("CARTESIA_API_KEY") is None, reason="CARTESIA_API_KEY not set" -) @pytest.mark.integration class TestCartesiaTTSIntegration: @pytest.fixture async def tts(self) -> cartesia.TTS: - return cartesia.TTS() + return cartesia.TTS(api_key=_require_cartesia_api_key()) async def test_cartesia_convert_text_to_audio(self, tts): out = [] diff --git a/plugins/cartesia/vision_agents/plugins/cartesia/__init__.py b/plugins/cartesia/vision_agents/plugins/cartesia/__init__.py index 4aaee0ff1..ebe903f19 100644 --- a/plugins/cartesia/vision_agents/plugins/cartesia/__init__.py +++ b/plugins/cartesia/vision_agents/plugins/cartesia/__init__.py @@ -1,6 +1,7 @@ +from .stt import STT from .tts import TTS # Re-export under the new namespace for convenience __path__ = __import__("pkgutil").extend_path(__path__, __name__) -__all__ = ["TTS"] +__all__ = ["STT", "TTS"] diff --git a/plugins/cartesia/vision_agents/plugins/cartesia/stt.py b/plugins/cartesia/vision_agents/plugins/cartesia/stt.py new file mode 100644 index 000000000..51b97d80e --- /dev/null +++ b/plugins/cartesia/vision_agents/plugins/cartesia/stt.py @@ -0,0 +1,343 @@ +import asyncio +import json +import logging +import os +import time +from typing import Any, Literal, Optional +from urllib.parse import urlencode + +import websockets +from getstream.video.rtc.track_util import PcmData +from vision_agents.core import stt +from vision_agents.core.edge.types import Participant +from vision_agents.core.stt import TranscriptResponse +from vision_agents.core.utils.utils import cancel_and_wait + +logger = logging.getLogger(__name__) + +DEFAULT_WEBSOCKET_URL = "wss://api.cartesia.ai/stt/turns/websocket" +DEFAULT_CARTESIA_VERSION = "2026-03-01" +CONNECTION_READY_TIMEOUT_SECONDS = 10.0 + + +class STT(stt.STT): + """Speech-to-Text plugin backed by Cartesia Ink.""" + + turn_detection: bool = True + eager_turn_detection: bool = True + + def __init__( + self, + api_key: Optional[str] = None, + model: str = "ink-2", + sample_rate: Literal[8000, 16000, 22050, 24000, 44100, 48000] = 16000, + encoding: Literal["pcm_s16le"] = "pcm_s16le", + cartesia_version: str = DEFAULT_CARTESIA_VERSION, + websocket_url: str = DEFAULT_WEBSOCKET_URL, + audio_chunk_duration_ms: int = 100, + ) -> None: + """Create a new Cartesia STT instance. + + Args: + api_key: Cartesia API key; falls back to ``CARTESIA_API_KEY``. + model: Cartesia STT model to use. Defaults to ``ink-2``. + sample_rate: PCM sample rate sent to Cartesia. + encoding: Audio encoding sent to Cartesia. + cartesia_version: Cartesia API version query parameter. + websocket_url: WebSocket endpoint, mainly useful for tests. + audio_chunk_duration_ms: Maximum duration per websocket audio frame. + """ + super().__init__(provider_name="cartesia") + + resolved_api_key = api_key or os.getenv("CARTESIA_API_KEY") + if not resolved_api_key: + raise ValueError("CARTESIA_API_KEY env var or api_key parameter required") + self.api_key = resolved_api_key + + self.model = model + self.sample_rate = sample_rate + self.encoding = encoding + self.cartesia_version = cartesia_version + self.websocket_url = websocket_url + self.audio_chunk_duration_ms = audio_chunk_duration_ms + + self.connection: websockets.ClientConnection | None = None + self._connection_ready = asyncio.Event() + self._listen_task: asyncio.Task[Any] | None = None + self._current_participant: Participant | None = None + self._audio_start_time: float | None = None + self._turn_in_progress = False + self._connection_error: Exception | None = None + + async def start(self) -> None: + """Open the Cartesia realtime STT websocket.""" + if self.closed: + raise ValueError("STT is closed and cannot be started") + + await super().start() + self._connection_error = None + + try: + url = self._build_websocket_url() + self.connection = await asyncio.wait_for( + websockets.connect( + url, + additional_headers={"X-API-Key": self.api_key}, + ), + timeout=10.0, + ) + + self._listen_task = asyncio.create_task(self._listen()) + self._connection_ready.set() + self._on_connected() + except Exception: + self.started = False + self.connection = None + self._connection_error = None + self._connection_ready.clear() + raise + + async def process_audio( + self, + pcm_data: PcmData, + participant: Optional[Participant] = None, + ) -> None: + """Send PCM audio to Cartesia for realtime transcription.""" + if self.closed: + logger.warning("Cartesia STT is closed, ignoring audio") + return + + if self._connection_error is not None: + raise RuntimeError( + "Cartesia STT websocket connection failed; call start() to reconnect" + ) from self._connection_error + + if not self.started: + raise RuntimeError("Cartesia STT is not started; call start() first") + + try: + await asyncio.wait_for( + self._connection_ready.wait(), + timeout=CONNECTION_READY_TIMEOUT_SECONDS, + ) + except asyncio.TimeoutError as exc: + raise RuntimeError( + "Timed out waiting for Cartesia STT websocket connection" + ) from exc + + connection = self.connection + if connection is None or not self._connection_ready.is_set(): + raise RuntimeError( + "Cartesia STT websocket connection is not ready; " + "call start() to reconnect" + ) + + self._current_participant = participant + if self._audio_start_time is None: + self._audio_start_time = time.perf_counter() + + resampled_pcm = pcm_data.resample(self.sample_rate, 1) + audio_bytes = resampled_pcm.samples.tobytes() + bytes_per_sample = 2 + frame_size = max( + bytes_per_sample, + int( + self.sample_rate + * bytes_per_sample + * self.audio_chunk_duration_ms + / 1000 + ), + ) + for offset in range(0, len(audio_bytes), frame_size): + await connection.send(audio_bytes[offset : offset + frame_size]) + + async def clear(self) -> None: + self._turn_in_progress = False + self._audio_start_time = None + await super().clear() + + async def close(self) -> None: + await super().close() + + if self._listen_task is not None: + await cancel_and_wait(self._listen_task) + self._listen_task = None + + if self.connection is not None: + try: + await self.connection.close() + except Exception as exc: + logger.warning("Error closing Cartesia STT websocket: %s", exc) + finally: + self.connection = None + self._connection_ready.clear() + self._on_disconnected(clean=True) + + def _build_websocket_url(self) -> str: + query = urlencode( + { + "model": self.model, + "encoding": self.encoding, + "sample_rate": str(self.sample_rate), + "cartesia_version": self.cartesia_version, + } + ) + separator = "&" if "?" in self.websocket_url else "?" + return f"{self.websocket_url}{separator}{query}" + + async def _listen(self) -> None: + assert self.connection is not None + try: + async for message in self.connection: + self._handle_message(message) + except asyncio.CancelledError: + raise + except Exception as exc: + if not self.closed: + logger.exception("Cartesia STT websocket error") + self._emit_error_event(exc, context="listen") + self._connection_error = exc + self.started = False + self.connection = None + self._connection_ready.clear() + self._on_disconnected(reason=str(exc), clean=False) + + def _handle_message(self, message: str | bytes | dict[str, Any]) -> None: + if isinstance(message, bytes): + message = message.decode("utf-8") + if isinstance(message, str): + data = json.loads(message) + else: + data = message + + event_type = data.get("type") + if event_type in {"turn.start", "turn_start"}: + self._handle_turn_started(data) + elif event_type in {"turn.update", "turn_update"}: + self._handle_transcript(data, final=False) + elif event_type in {"turn.eager_end", "turn_eager_end"}: + self._handle_transcript(data, final=False) + self._handle_turn_ended(data, eager=True) + elif event_type in {"turn.resume", "turn_resume"}: + self._handle_turn_resumed(data) + elif event_type in {"turn.end", "turn_end"}: + self._handle_transcript(data, final=True) + self._handle_turn_ended(data, eager=False) + elif event_type == "error" or data.get("error"): + error = RuntimeError(str(data.get("error") or data)) + self._emit_error_event(error, context="message") + else: + logger.debug("Unhandled Cartesia STT event: %s", event_type) + + def _handle_turn_started(self, data: dict[str, Any]) -> None: + participant = self._current_participant + if participant is None: + logger.warning("Received Cartesia turn start but no participant set") + return + + self._turn_in_progress = True + self._emit_turn_started_event( + participant, + confidence=self._confidence(data), + ) + + def _handle_turn_resumed(self, data: dict[str, Any]) -> None: + participant = self._current_participant + if participant is None: + logger.warning("Received Cartesia turn resume but no participant set") + return + + self._turn_in_progress = True + self._emit_turn_started_event( + participant, + confidence=self._confidence(data), + ) + + def _handle_turn_ended(self, data: dict[str, Any], *, eager: bool) -> None: + participant = self._current_participant + if participant is None: + logger.warning("Received Cartesia turn end but no participant set") + return + + if not eager: + self._turn_in_progress = False + self._audio_start_time = None + + self._emit_turn_ended_event( + participant, + confidence=self._confidence(data), + eager=eager, + duration_ms=self._duration_ms(data), + trailing_silence_ms=self._trailing_silence_ms(data), + ) + + def _handle_transcript(self, data: dict[str, Any], *, final: bool) -> None: + transcript_text = self._transcript_text(data) + if not transcript_text: + return + + participant = self._current_participant + if participant is None: + logger.warning("Received Cartesia transcript but no participant set") + return + + processing_time_ms: float | None = None + if self._audio_start_time is not None: + processing_time_ms = (time.perf_counter() - self._audio_start_time) * 1000 + + response = TranscriptResponse( + confidence=self._confidence(data), + language=data.get("language"), + audio_duration_ms=self._duration_ms(data), + model_name=self.model, + processing_time_ms=processing_time_ms, + other={"words": data.get("words")} if data.get("words") else None, + ) + self._emit_transcript_event( + transcript_text, + participant, + response, + mode="final" if final else "replacement", + ) + + @staticmethod + def _transcript_text(data: dict[str, Any]) -> str: + transcript = data.get("transcript") + if transcript is None: + transcript = data.get("text") + return str(transcript or "") + + @staticmethod + def _confidence(data: dict[str, Any]) -> float: + confidence = data.get("confidence") + if isinstance(confidence, int | float): + return float(confidence) + + words = data.get("words") or [] + confidences = [ + float(word[key]) + for word in words + for key in ("confidence", "score") + if isinstance(word, dict) and isinstance(word.get(key), int | float) + ] + if confidences: + return sum(confidences) / len(confidences) + return 0.0 + + @staticmethod + def _duration_ms(data: dict[str, Any]) -> float | None: + if isinstance(data.get("duration_ms"), int | float): + return float(data["duration_ms"]) + if isinstance(data.get("audio_duration_ms"), int | float): + return float(data["audio_duration_ms"]) + if isinstance(data.get("duration"), int | float): + return float(data["duration"]) * 1000 + return None + + @staticmethod + def _trailing_silence_ms(data: dict[str, Any]) -> float | None: + if isinstance(data.get("trailing_silence_ms"), int | float): + return float(data["trailing_silence_ms"]) + if isinstance(data.get("trailing_silence"), int | float): + return float(data["trailing_silence"]) * 1000 + return None diff --git a/plugins/cartesia/vision_agents/plugins/cartesia/tts.py b/plugins/cartesia/vision_agents/plugins/cartesia/tts.py index bf17a7bf5..24a14751a 100644 --- a/plugins/cartesia/vision_agents/plugins/cartesia/tts.py +++ b/plugins/cartesia/vision_agents/plugins/cartesia/tts.py @@ -1,5 +1,3 @@ -from __future__ import annotations - import logging import os from typing import AsyncIterator, Iterator, Literal, Optional @@ -19,7 +17,7 @@ class TTS(tts.TTS): def __init__( self, api_key: Optional[str] = None, - model_id: str = "sonic-3", + model_id: str = "sonic-3.5", voice_id: str | None = "6ccbfb76-1fc6-48f7-b71d-91ac6298247b", sample_rate: Literal[8000, 16000, 22050, 24000, 44100, 48000] = 16000, client: Optional[AsyncCartesia] = None, @@ -28,7 +26,7 @@ def __init__( Args: api_key: Cartesia API key – falls back to ``CARTESIA_API_KEY`` env var. - model_id: Which model to use (default ``sonic-3``). + model_id: Which model to use (default ``sonic-3.5``). voice_id: Cartesia voice ID. When ``None`` the model default is used. sample_rate: PCM sample-rate you want back (must match output track). """ diff --git a/tests/test_stt_events.py b/tests/test_stt_events.py new file mode 100644 index 000000000..bd5b6bd4b --- /dev/null +++ b/tests/test_stt_events.py @@ -0,0 +1,45 @@ +from getstream.video.rtc.track_util import PcmData + +from vision_agents.core.edge.types import Participant +from vision_agents.core.stt import STT +from vision_agents.core.stt.events import ( + STTConnectedEvent, + STTDisconnectedEvent, + STTErrorEvent, +) + + +class DummySTT(STT): + async def process_audio( + self, + pcm_data: PcmData, + participant: Participant, + ) -> None: + return None + + +async def test_stt_plugin_events_are_registered(): + stt = DummySTT(provider_name="dummy") + seen: list[object] = [] + + @stt.events.subscribe + async def on_connected(event: STTConnectedEvent) -> None: + seen.append(event) + + @stt.events.subscribe + async def on_disconnected(event: STTDisconnectedEvent) -> None: + seen.append(event) + + @stt.events.subscribe + async def on_error(event: STTErrorEvent) -> None: + seen.append(event) + + stt._on_connected() + stt._on_disconnected(reason="done", clean=True) + stt._emit_error_event(RuntimeError("boom"), context="test") + await stt.events.wait() + + assert [event.plugin_name for event in seen] == ["dummy", "dummy", "dummy"] + assert isinstance(seen[0], STTConnectedEvent) + assert isinstance(seen[1], STTDisconnectedEvent) + assert isinstance(seen[2], STTErrorEvent) diff --git a/uv.lock b/uv.lock index 019d0ea48..7f963067d 100644 --- a/uv.lock +++ b/uv.lock @@ -6766,6 +6766,7 @@ source = { editable = "plugins/cartesia" } dependencies = [ { name = "cartesia" }, { name = "vision-agents" }, + { name = "websockets" }, ] [package.dev-dependencies] @@ -6778,6 +6779,7 @@ dev = [ requires-dist = [ { name = "cartesia", specifier = ">=3.0.2,<3.1" }, { name = "vision-agents", editable = "agents-core" }, + { name = "websockets", specifier = ">=15.0.1,<16" }, ] [package.metadata.requires-dev]