diff --git a/CHANGELOG.md b/CHANGELOG.md index 817084927..9da0874de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ ## New Features +### `telnyx` plugin: LLM, STT and TTS (#620, #621, #622) + +The Telnyx plugin, until now a phone transport, also exposes `telnyx.LLM`, +`telnyx.STT`, and `telnyx.TTS`, so a phone agent can run end to end on Telnyx. +`telnyx.LLM` wraps Telnyx Inference's OpenAI-compatible Chat Completions +endpoint and defaults to `meta-llama/Llama-3.3-70B-Instruct`. `telnyx.STT` +streams `linear16` over WebSocket and takes a `sample_rate`, so telephony audio +from `TelnyxMediaStream` can be transcribed at 8 kHz without an upsample; pick +the engine with `transcription_engine`. `telnyx.TTS` streams MP3 over WebSocket +and decodes to `PcmData` as it arrives. All three read `TELNYX_API_KEY` from the +environment. See `plugins/telnyx/examples/voice_agent_call.py` for an inbound +call answered by an all-Telnyx pipeline. + ### `speechify` plugin: Speechify TTS Adds a new `speechify` plugin exposing `speechify.TTS`, backed by Speechify's streaming API. It streams raw PCM audio, defaults to the `simba-3.2` model with the `geffen_32` voice, and reads `SPEECHIFY_API_KEY` from the environment. Install with `vision-agents[speechify]`. diff --git a/plugins/telnyx/README.md b/plugins/telnyx/README.md index dd4564e8b..e2ac9b019 100644 --- a/plugins/telnyx/README.md +++ b/plugins/telnyx/README.md @@ -12,6 +12,9 @@ real-time bidirectional media streaming. - **Audio Conversion**: PCMU, PCMA, and L16 RTP payload conversion - **WebSocket Management**: Handle Telnyx WebSocket media events - **Stream Bridge**: Attach a Telnyx phone participant to a Stream call +- **LLM**: Telnyx Inference via the OpenAI-compatible Chat Completions API +- **STT**: Streaming speech to text over WebSocket +- **TTS**: Streaming text to speech over WebSocket ## Installation @@ -46,6 +49,75 @@ call.telnyx_stream = stream await stream.run() ``` +## LLM + +Telnyx Inference is OpenAI-compatible, so the LLM is a thin wrapper over +`ChatCompletionsLLM` pointed at `https://api.telnyx.com/v2/ai`. Streaming and +tool calling work the same as any other Chat Completions provider. + +```python +from vision_agents.plugins import telnyx + +llm = telnyx.LLM(model="openai/gpt-4o") +``` + +Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. + +Model ids come from the Telnyx catalogue at `GET /v2/ai/models` and are not +validated locally. The default is `meta-llama/Llama-3.3-70B-Instruct`. + +## STT + +```python +from vision_agents.plugins import telnyx + +# 8000 matches the PCMU telephony audio that TelnyxMediaStream decodes, +# so nothing is upsampled on the way to the transcriber. +stt = telnyx.STT(sample_rate=8000) +``` + +Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. + +Audio is resampled to `sample_rate` and sent as raw `linear16` frames. Pick the +engine with `transcription_engine`; the default is `Telnyx`. The engine +catalogue is served by Telnyx and is not validated locally. + +Telnyx does not send VAD signals on this endpoint, so the plugin emits +transcripts only and leaves turn detection to the agent. + +`interim_results` is honoured per engine rather than per endpoint, and defaults +to `False`. Measured against the live API with the same audio, `Speechmatics` +and `Soniox` stream partial transcripts, while `Telnyx` and `Deepgram` accept +the parameter and return finals only: + +```python +stt = telnyx.STT(transcription_engine="Speechmatics", interim_results=True) +``` + +## TTS + +```python +from vision_agents.plugins import telnyx + +tts = telnyx.TTS(voice="AWS.Polly.Danielle-Neural") +``` + +Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. + +Voice ids come from `GET /v2/text-to-speech/voices`. The default is +`Telnyx.KokoroTTS.af_heart`. + +Telnyx serves each synthesis on its own WebSocket and closes the socket after +the stop frame, so the plugin reconnects per `stream_audio` call. Audio arrives +as MP3 and is decoded to `PcmData` as it streams. The output sample rate follows +the voice, so it is taken from the decoder rather than configured. + +The endpoint takes an `audio_format` parameter, but it is honoured only by some +voices — `AWS.Polly.*` and `Telnyx.NaturalHD.*` serve raw PCM, while the default +`Telnyx.KokoroTTS.*` returns MP3 regardless. Since the PCM sample rate is not +reported on the wire and differs per voice, the plugin decodes MP3 for every +voice rather than carrying a voice-to-rate table that would go stale. + ## Examples See [examples/](examples/) for minimal inbound and outbound Telnyx phone @@ -62,6 +134,11 @@ uv run plugins/telnyx/examples/outbound_call.py \ uv run plugins/telnyx/examples/inbound_call.py \ --setup-telnyx \ --phone-number +15551234567 + +# Inbound call answered by an all-Telnyx STT/LLM/TTS pipeline +uv run plugins/telnyx/examples/voice_agent_call.py \ + --setup-telnyx \ + --phone-number +15551234567 ``` Telnyx phone calls require a Call Control App. The Call Control App is where @@ -189,5 +266,7 @@ payload = pcm_to_pcmu(pcm) ## Dependencies - vision-agents +- vision-agents-plugins-openai +- aiohttp - numpy - fastapi diff --git a/plugins/telnyx/examples/README.md b/plugins/telnyx/examples/README.md index c95826e13..4f39daa70 100644 --- a/plugins/telnyx/examples/README.md +++ b/plugins/telnyx/examples/README.md @@ -1,7 +1,12 @@ # Telnyx Phone Examples -Minimal inbound and outbound phone examples for the Telnyx plugin. These examples -use Telnyx Call Control, Telnyx Media Streaming, Stream, and Gemini Realtime. +Minimal inbound and outbound phone examples for the Telnyx plugin. + +- `outbound_call.py` and `inbound_call.py` use Telnyx Call Control, Telnyx Media + Streaming, Stream, and Gemini Realtime. +- `voice_agent_call.py` answers an inbound call with a pipeline that runs + entirely on Telnyx: `telnyx.STT`, `telnyx.LLM`, and `telnyx.TTS`. It needs no + `GOOGLE_API_KEY`. ## Requirements @@ -10,7 +15,7 @@ Create a `.env` file at the repo root or export these variables: ```bash STREAM_API_KEY= STREAM_API_SECRET= -GOOGLE_API_KEY= +GOOGLE_API_KEY= # not needed by voice_agent_call.py TELNYX_API_KEY= TELNYX_PUBLIC_KEY= ``` @@ -57,6 +62,14 @@ uv run plugins/telnyx/examples/inbound_call.py \ --phone-number +15551234567 ``` +Inbound, all Telnyx: + +```bash +uv run plugins/telnyx/examples/voice_agent_call.py \ + --setup-telnyx \ + --phone-number +15551234567 +``` + For inbound calls, `--setup-telnyx` also routes the Telnyx number to the temporary Call Control App and restores the previous routing on normal shutdown. diff --git a/plugins/telnyx/examples/voice_agent_call.py b/plugins/telnyx/examples/voice_agent_call.py new file mode 100644 index 000000000..137b66b35 --- /dev/null +++ b/plugins/telnyx/examples/voice_agent_call.py @@ -0,0 +1,248 @@ +"""Inbound Telnyx phone example running the whole pipeline on Telnyx. + +Unlike `inbound_call.py`, which bridges the call into a realtime model, this +example uses Telnyx for transport, speech to text, inference, and text to +speech. The only non-Telnyx pieces are Stream for the call and a turn detector, +which Telnyx's transcription endpoint does not provide. + +Run after starting ngrok and routing your Telnyx number to a Call Control App: + + NGROK_URL=example.ngrok-free.app uv run plugins/telnyx/examples/voice_agent_call.py +""" + +import argparse +import asyncio +import contextlib +import logging +import os +import uuid + +import uvicorn +from dotenv import load_dotenv +from fastapi import FastAPI, Request, WebSocket +from fastapi.responses import JSONResponse +from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware +from vision_agents.core import Agent, User +from vision_agents.plugins import getstream, smart_turn, telnyx +from vision_agents.plugins.getstream.stream_edge_transport import StreamEdge +from vision_agents.plugins.telnyx.example_helpers import ( + TelnyxClient, + TelnyxConfig, + TelnyxSetupError, + cleanup_telnyx_example_setup, + media_stream_url, + parse_verified_telnyx_webhook, + preflight_inbound, + prepare_telnyx_example_setup, + require_env, + require_telnyx_public_key, +) + + +logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.INFO) + +load_dotenv() + +app = FastAPI() +app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"]) +call_registry = telnyx.TelnyxCallRegistry() +telnyx_client: TelnyxClient | None = None +telnyx_config: TelnyxConfig | None = None +telnyx_public_key: str | None = None + + +@app.exception_handler(Exception) +async def global_exception_handler(_request: Request, exc: Exception): + logger.exception("Unhandled exception: %s", exc) + return JSONResponse(status_code=500, content={"detail": "Internal server error"}) + + +async def create_agent() -> tuple[Agent, StreamEdge]: + edge = getstream.Edge() + agent = Agent( + edge=edge, + agent_user=User(id="ai-agent", name="AI Assistant"), + instructions=( + "Speak English. Keep replies short and natural. You are answering " + "an inbound Telnyx test call through Vision Agents. Start by saying " + "this is a quick inbound Telnyx bridge test and ask whether the " + "audio is clear." + ), + # 8000 is what TelnyxMediaStream decodes PCMU telephony audio to, so + # the transcriber receives it without an upsample that adds nothing. + stt=telnyx.STT(sample_rate=8000), + llm=telnyx.LLM(), + tts=telnyx.TTS(), + # Telnyx's transcription endpoint sends no VAD signals, so turns are + # detected locally rather than by the STT. + turn_detection=smart_turn.TurnDetection(), + ) + return agent, edge + + +async def prepare_call(call_id: str): + agent, edge = await create_agent() + phone_user = User( + name=f"Inbound Telnyx call {call_id[:8]}", + id=f"phone-{call_id}", + ) + await edge.create_users([agent.agent_user, phone_user]) + stream_call = await agent.create_call("default", call_id) + logger.info("Prepared Stream call %s", call_id) + return agent, phone_user, stream_call + + +async def wait_for_start( + stream: telnyx.TelnyxMediaStream, timeout: float = 10.0 +) -> None: + deadline = asyncio.get_running_loop().time() + timeout + while not stream.has_started: + if asyncio.get_running_loop().time() >= deadline: + raise TimeoutError("Telnyx media stream did not start in time") + await asyncio.sleep(0.05) + + +@app.post("/telnyx/events") +async def telnyx_events(request: Request): + if telnyx_client is None or telnyx_config is None or telnyx_public_key is None: + raise RuntimeError("Telnyx example was not initialized") + + data = await parse_verified_telnyx_webhook(request, telnyx_public_key) + event_type = data.get("data", {}).get("event_type") + payload = data.get("data", {}).get("payload", {}) + logger.info("Telnyx webhook event: %s", event_type) + + if event_type == "call.initiated" and payload.get("direction") == "incoming": + call_control_id = payload["call_control_id"] + call_id = str(uuid.uuid4()) + telnyx_call = call_registry.create( + call_id, + webhook_data=data, + prepare=lambda: prepare_call(call_id), + ) + stream_url = media_stream_url( + telnyx_config.ngrok_url, + call_id, + telnyx_call.token, + ) + await asyncio.to_thread( + telnyx_client.answer_call, + call_control_id, + stream_url=stream_url, + ) + logger.info("Answered inbound Telnyx call %s", call_id) + + return {"ok": True} + + +@app.websocket("/telnyx/media/{call_id}/{token}") +async def media_stream(websocket: WebSocket, call_id: str, token: str): + telnyx_call = call_registry.validate(call_id, token) + logger.info("Media stream connected for inbound call %s", call_id) + + telnyx_stream = telnyx.TelnyxMediaStream(websocket) + await telnyx_stream.accept() + telnyx_call.telnyx_stream = telnyx_stream + stream_task = asyncio.create_task(telnyx_stream.run()) + + try: + agent, phone_user, stream_call = await telnyx_call.await_prepare() + telnyx_call.stream_call = stream_call + + await telnyx.attach_phone_to_call(stream_call, telnyx_stream, phone_user.id) + await wait_for_start(telnyx_stream) + + async with agent.join(stream_call, participant_wait_timeout=0): + await agent.simple_response( + text="Greet the caller and ask whether the inbound Telnyx audio is clear." + ) + await stream_task + finally: + if not stream_task.done(): + stream_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await stream_task + call_registry.remove(call_id) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run an inbound Telnyx call with a Telnyx STT/LLM/TTS pipeline." + ) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", default=8000, type=int) + parser.add_argument( + "--phone-number-id", + default=None, + help="Telnyx phone number resource ID. Defaults to TELNYX_PHONE_NUMBER_ID.", + ) + parser.add_argument( + "--phone-number", + default=None, + help="Telnyx inbound number. Defaults to TELNYX_PHONE_NUMBER.", + ) + parser.add_argument( + "--call-control-app-id", + default=None, + help="Existing Telnyx Call Control App ID. Defaults to TELNYX_CALL_CONTROL_APP_ID.", + ) + parser.add_argument( + "--ngrok-url", + default=None, + help="Public ngrok hostname. Defaults to NGROK_URL or local ngrok autodetection.", + ) + parser.add_argument( + "--setup-telnyx", + action="store_true", + help="Create a temporary Call Control App, route the phone number, and restore it on exit.", + ) + return parser.parse_args() + + +def main() -> None: + global telnyx_client, telnyx_config, telnyx_public_key + + args = parse_args() + values = require_env(["STREAM_API_KEY", "STREAM_API_SECRET", "TELNYX_API_KEY"]) + telnyx_public_key = require_telnyx_public_key() + telnyx_client = TelnyxClient(values["TELNYX_API_KEY"]) + setup = prepare_telnyx_example_setup( + telnyx_client, + api_key=values["TELNYX_API_KEY"], + phone_number=args.phone_number or os.environ.get("TELNYX_PHONE_NUMBER"), + ngrok_url=args.ngrok_url or os.environ.get("NGROK_URL"), + call_control_app_id=( + args.call_control_app_id or os.environ.get("TELNYX_CALL_CONTROL_APP_ID") + ), + phone_number_id=( + args.phone_number_id or os.environ.get("TELNYX_PHONE_NUMBER_ID") + ), + setup_telnyx=args.setup_telnyx, + route_phone_number=True, + ) + telnyx_config = setup.config + resolved_phone_number_id = setup.phone_number_id or ( + args.phone_number_id or os.environ.get("TELNYX_PHONE_NUMBER_ID") + ) + if not resolved_phone_number_id: + raise TelnyxSetupError( + "Missing TELNYX_PHONE_NUMBER_ID. Pass `--setup-telnyx` to discover " + "and route the Telnyx number automatically." + ) + + try: + preflight_inbound( + telnyx_client, + config=telnyx_config, + telnyx_phone_number_id=resolved_phone_number_id, + ) + + logger.info("Telnyx voice agent ready for call") + uvicorn.run(app, host=args.host, port=args.port) + finally: + cleanup_telnyx_example_setup(telnyx_client, setup) + + +if __name__ == "__main__": + main() diff --git a/plugins/telnyx/pyproject.toml b/plugins/telnyx/pyproject.toml index 961aedd0e..0ea9a1290 100644 --- a/plugins/telnyx/pyproject.toml +++ b/plugins/telnyx/pyproject.toml @@ -11,9 +11,11 @@ requires-python = ">=3.10" license = "MIT" dependencies = [ "vision-agents", + "vision-agents-plugins-openai", "numpy>=1.24.0", # capped at <2.0 via workspace override in root pyproject.toml "cryptography>=44.0.0", "fastapi>=0.135.1", + "aiohttp>=3.13.3", ] [project.urls] @@ -33,6 +35,7 @@ include = ["/vision_agents"] [tool.uv.sources] vision-agents = { workspace = true } +vision-agents-plugins-openai = { workspace = true } [dependency-groups] dev = [ diff --git a/plugins/telnyx/tests/test_telnyx_llm.py b/plugins/telnyx/tests/test_telnyx_llm.py new file mode 100644 index 000000000..897e595ae --- /dev/null +++ b/plugins/telnyx/tests/test_telnyx_llm.py @@ -0,0 +1,86 @@ +"""Tests for the Telnyx LLM plugin.""" + +import os + +import pytest +from dotenv import load_dotenv +from openai import AsyncOpenAI +from vision_agents.core.agents.conversation import InMemoryConversation +from vision_agents.plugins.telnyx import LLM +from vision_agents.testing import collect_simple_response + +load_dotenv() + + +class TestTelnyxLLM: + """Unit tests for Telnyx LLM configuration.""" + + def test_requires_api_key(self, monkeypatch): + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + with pytest.raises(ValueError, match="TELNYX_API_KEY"): + LLM() + + async def test_default_model(self): + llm = LLM(api_key="KEY_test") + assert llm.model == "meta-llama/Llama-3.3-70B-Instruct" + + async def test_custom_model(self): + llm = LLM(api_key="KEY_test", model="openai/gpt-4o") + assert llm.model == "openai/gpt-4o" + + async def test_provider_name(self): + llm = LLM(api_key="KEY_test") + assert llm.provider_name == "telnyx" + + async def test_explicit_client_skips_api_key_requirement(self, monkeypatch): + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + client = AsyncOpenAI(api_key="KEY_injected", base_url="https://example.invalid") + + assert LLM(client=client).model == "meta-llama/Llama-3.3-70B-Instruct" + + +@pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") +@pytest.mark.integration +class TestTelnyxLLMIntegration: + """Integration tests hitting the real Telnyx Inference endpoint.""" + + @pytest.fixture + async def llm(self): + llm = LLM() + llm.set_conversation(InMemoryConversation("be friendly", [])) + return llm + + async def test_simple_response(self, llm): + deltas, final = await collect_simple_response( + llm.simple_response("Greet the user in English") + ) + assert final.text + assert deltas + + async def test_streaming_chunks(self, llm): + deltas, final = await collect_simple_response( + llm.simple_response("List the first 3 prime numbers, separated by commas.") + ) + assert final.text + assert len(deltas) > 0, f"No chunks emitted. Response text: {final.text!r}" + + async def test_function_calling(self, llm): + calls: list[str] = [] + + @llm.register_function(description="Probe tool that records invocation") + async def probe_tool(ping: str) -> str: + calls.append(ping) + return f"probe_ok:{ping}" + + prompt = ( + "Call the tool named 'probe_tool' with the parameter ping='pong' now. " + "After receiving the tool result, reply by returning ONLY the tool result string." + ) + _, final = await collect_simple_response(llm.simple_response(prompt)) + + assert "pong" in calls, ( + f"probe_tool was not invoked with ping='pong' (got calls={calls})" + ) + assert "probe_ok:pong" in final.text, ( + f"Expected 'probe_ok:pong', got: {final.text}" + ) diff --git a/plugins/telnyx/tests/test_telnyx_stt.py b/plugins/telnyx/tests/test_telnyx_stt.py new file mode 100644 index 000000000..ef82d5c67 --- /dev/null +++ b/plugins/telnyx/tests/test_telnyx_stt.py @@ -0,0 +1,235 @@ +"""Tests for the Telnyx STT plugin.""" + +import asyncio +import os + +import aiohttp +import numpy as np +import pytest +from dotenv import load_dotenv +from getstream.video.rtc.track_util import PcmData +from vision_agents.core.edge.types import Participant +from vision_agents.core.stt import Transcript +from vision_agents.plugins.telnyx import STT + +load_dotenv() + + +class TestTelnyxSTT: + """Unit tests for Telnyx STT configuration and message handling.""" + + @pytest.fixture + def participant(self) -> Participant: + return Participant({}, user_id="test-user", id="test-user") + + @pytest.fixture + def stt_with_failing_socket(self) -> STT: + """An started STT whose socket rejects every send.""" + instance = STT(api_key="KEY_test") + + class FailingWS: + closed = False + + async def send_bytes(self, data: bytes) -> None: + raise aiohttp.ClientError("connection reset") + + instance.started = True + instance._ws = FailingWS() + instance._connection_ready.set() + return instance + + async def test_requires_api_key(self, monkeypatch): + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + with pytest.raises(ValueError, match="TELNYX_API_KEY"): + STT() + + async def test_default_configuration(self): + stt = STT(api_key="KEY_test") + assert stt.transcription_engine == "Telnyx" + assert stt.language == "en" + assert stt.sample_rate == 16000 + assert stt.interim_results is False + assert stt.provider_name == "telnyx" + assert stt.turn_detection is False + + async def test_unknown_engine_is_left_to_the_server(self): + """The engine catalogue is served by Telnyx, so it is not pinned here.""" + stt = STT(api_key="KEY_test", transcription_engine="SomeNewEngine") + assert "transcription_engine=SomeNewEngine" in stt._build_ws_url() + + async def test_url_carries_stream_parameters(self): + stt = STT(api_key="KEY_test", language="es", sample_rate=8000) + url = stt._build_ws_url() + assert url.startswith("wss://api.telnyx.com/v2/speech-to-text/transcription?") + assert "input_format=linear16" in url + assert "sample_rate=8000" in url + assert "language=es" in url + assert "transcription_engine=Telnyx" in url + + async def test_url_uses_interim_results_not_partial_results(self): + on = STT(api_key="KEY_test", interim_results=True)._build_ws_url() + off = STT(api_key="KEY_test")._build_ws_url() + assert "interim_results=true" in on + assert "interim_results=false" in off + assert "partial_results" not in on + + async def test_url_omits_model_when_unset(self): + assert "model=" not in STT(api_key="KEY_test")._build_ws_url() + assert ( + "model=whisper" in STT(api_key="KEY_test", model="whisper")._build_ws_url() + ) + + async def test_final_transcript_emitted(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message( + {"transcript": "hello world", "confidence": 0.9, "is_final": True} + ) + 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 world"] + assert transcripts[0].final + assert transcripts[0].confidence == 0.9 + assert transcripts[0].participant == participant + + async def test_interim_transcript_is_not_final(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message( + {"transcript": "hello", "confidence": None, "is_final": False} + ) + 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"] + assert not transcripts[0].final + assert transcripts[0].confidence is None + + async def test_empty_transcript_emits_nothing(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message({"transcript": "", "confidence": None, "is_final": True}) + stt._handle_message({"transcript": " ", "confidence": None, "is_final": True}) + + assert await stt.output.collect(timeout=0) == [] + + async def test_send_failure_does_not_propagate( + self, stt_with_failing_socket, participant + ): + pcm = PcmData( + samples=np.zeros(160, dtype=np.int16), sample_rate=16000, format="s16" + ) + await stt_with_failing_socket.process_audio(pcm, participant=participant) + + async def test_audio_before_start_is_dropped(self, participant): + """process_audio must not block forever when start() never ran.""" + stt = STT(api_key="KEY_test") + pcm = PcmData( + samples=np.zeros(160, dtype=np.int16), sample_rate=16000, format="s16" + ) + + await asyncio.wait_for( + stt.process_audio(pcm, participant=participant), timeout=1.0 + ) + + async def test_integer_confidence_is_kept(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message({"transcript": "hello", "confidence": 1, "is_final": True}) + items = await stt.output.collect(timeout=0) + + transcripts = [i for i in items if isinstance(i, Transcript)] + assert transcripts[0].confidence == 1.0 + + async def test_error_payload_emits_no_transcript(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message( + { + "errors": [ + { + "code": "40001", + "title": "Invalid Parameter", + "detail": "Unsupported input_format 'zzz'.", + } + ] + } + ) + + assert await stt.output.collect(timeout=0) == [] + + +@pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") +@pytest.mark.integration +class TestTelnyxSTTIntegration: + """Integration tests against the real Telnyx streaming STT.""" + + @pytest.fixture + def participant(self) -> Participant: + return Participant({}, user_id="test-user", id="test-user") + + @pytest.fixture + async def telnyx_stt(self): + stt = STT() + await stt.start() + yield stt + await stt.close() + + @pytest.fixture + async def telnyx_stt_8khz(self): + stt = STT(sample_rate=8000) + await stt.start() + yield stt + await stt.close() + + async def test_transcribe_mia_audio_16khz( + self, telnyx_stt, mia_audio_16khz, participant + ): + await telnyx_stt.process_audio(mia_audio_16khz, participant=participant) + + items = await telnyx_stt.output.collect(timeout=15.0) + + transcripts = [i for i in items if isinstance(i, Transcript)] + assert transcripts, "No Transcript emitted by Telnyx STT" + finals = [t for t in transcripts if t.final] + assert finals, "No final Transcript emitted by Telnyx STT" + assert "forgotten treasures" in " ".join(t.text for t in finals).lower() + assert transcripts[0].participant == participant + + async def test_transcribe_at_telephony_sample_rate( + self, telnyx_stt_8khz, mia_audio_16khz, participant + ): + """8 kHz is what TelnyxMediaStream produces from PCMU telephony audio.""" + await telnyx_stt_8khz.process_audio(mia_audio_16khz, participant=participant) + + items = await telnyx_stt_8khz.output.collect(timeout=15.0) + + finals = [i for i in items if isinstance(i, Transcript) and i.final] + assert finals, "No final Transcript emitted at 8kHz" + assert "forgotten treasures" in " ".join(t.text for t in finals).lower() + + async def test_interim_results_stream_partial_transcripts( + self, mia_audio_16khz, participant + ): + """Covers the replacement-mode path, which finals-only engines never hit. + + ``interim_results`` is honoured per engine: Speechmatics streams + partials, while the default Telnyx engine returns finals only. + """ + stt = STT(transcription_engine="Speechmatics", interim_results=True) + await stt.start() + try: + await stt.process_audio(mia_audio_16khz, participant=participant) + items = await stt.output.collect(timeout=15.0) + finally: + await stt.close() + + transcripts = [i for i in items if isinstance(i, Transcript)] + assert [t for t in transcripts if not t.final], "No interim Transcript emitted" + assert [t for t in transcripts if t.final], "No final Transcript emitted" diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py new file mode 100644 index 000000000..50209ede1 --- /dev/null +++ b/plugins/telnyx/tests/test_telnyx_tts.py @@ -0,0 +1,265 @@ +"""Tests for the Telnyx TTS plugin.""" + +import os +from types import SimpleNamespace + +import aiohttp +import av +import pytest +from dotenv import load_dotenv +from vision_agents.plugins.telnyx import TTS +from vision_agents.plugins.telnyx.tts import _Id3Stripper + +load_dotenv() + + +def id3_tag(payload_size: int) -> bytes: + """Build an ID3v2 tag header plus a body of ``payload_size`` bytes.""" + synchsafe = bytes( + [ + (payload_size >> 21) & 0x7F, + (payload_size >> 14) & 0x7F, + (payload_size >> 7) & 0x7F, + payload_size & 0x7F, + ] + ) + return b"ID3\x04\x00\x00" + synchsafe + b"\xaa" * payload_size + + +class TestId3Stripper: + """Unit tests for the streaming ID3v2 tag stripper.""" + + def test_untagged_data_passes_through(self): + stripper = _Id3Stripper() + assert stripper.feed(b"\xff\xf3audio") == b"\xff\xf3audio" + + def test_leading_tag_removed(self): + stripper = _Id3Stripper() + assert stripper.feed(id3_tag(34) + b"\xff\xf3audio") == b"\xff\xf3audio" + + def test_zero_length_tag_removed(self): + stripper = _Id3Stripper() + assert stripper.feed(id3_tag(0) + b"\xff\xf3") == b"\xff\xf3" + + def test_tag_body_spanning_frames(self): + stripper = _Id3Stripper() + blob = id3_tag(40) + b"\xff\xf3audio" + assert stripper.feed(blob[:20]) == b"" + assert stripper.feed(blob[20:]) == b"\xff\xf3audio" + + def test_tag_header_spanning_frames(self): + stripper = _Id3Stripper() + blob = id3_tag(12) + b"\xff\xf3audio" + assert stripper.feed(blob[:4]) == b"" + assert stripper.feed(blob[4:]) == b"\xff\xf3audio" + + def test_tag_at_head_of_later_frame(self): + stripper = _Id3Stripper() + assert stripper.feed(b"\xff\xf3first") == b"\xff\xf3first" + assert stripper.feed(id3_tag(8) + b"\xff\xf3second") == b"\xff\xf3second" + + def test_id3_bytes_inside_audio_are_kept(self): + stripper = _Id3Stripper() + audio = b"\xff\xf3 padding ID3 more audio" + assert stripper.feed(audio) == audio + + def test_tag_consuming_whole_frame(self): + stripper = _Id3Stripper() + blob = id3_tag(100) + b"\xff\xf3audio" + assert stripper.feed(blob[:50]) == b"" + assert stripper.feed(blob[50:]) == b"\xff\xf3audio" + + +@pytest.fixture +def tts() -> TTS: + return TTS(api_key="KEY_test") + + +@pytest.fixture +def dropped_socket_tts(request) -> TTS: + """A TTS whose socket is dropped mid-send, as stop_audio() would. + + Parametrised with ``stop_before_drop``: True models a concurrent + ``stop_audio()`` (barge-in), False a genuine connection failure. + """ + stop_before_drop = request.param + instance = TTS(api_key="KEY_test") + + class DroppedWS: + closed = True + + async def send_str(self, data: str) -> None: + if stop_before_drop: + instance._stop_event.set() + raise aiohttp.ClientConnectionResetError("Cannot write to closing") + + async def close(self) -> None: + return None + + class FakeSession: + closed = False + + async def ws_connect(self, url: str, headers: dict[str, str]): + return DroppedWS() + + instance._session = FakeSession() + return instance + + +@pytest.fixture +def ws_serving(): + """Build a websocket stand-in that replays ``payloads`` then closes.""" + + def build(payloads: list[str]) -> object: + messages = [ + SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=payload) + for payload in payloads + ] + [SimpleNamespace(type=aiohttp.WSMsgType.CLOSED, data=None)] + + class FakeWS: + def __init__(self) -> None: + self._queue = list(messages) + + async def receive(self): + return self._queue.pop(0) + + return FakeWS() + + return build + + +class TestTelnyxTTS: + """Unit tests for Telnyx TTS configuration and the receive loop.""" + + async def test_requires_api_key(self, monkeypatch): + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + with pytest.raises(ValueError, match="TELNYX_API_KEY"): + TTS() + + async def test_default_configuration(self, tts): + assert tts.voice == "Telnyx.KokoroTTS.af_heart" + assert tts.provider_name == "telnyx" + + async def test_custom_voice(self): + tts = TTS(api_key="KEY_test", voice="AWS.Polly.Danielle-Neural") + assert tts.voice == "AWS.Polly.Danielle-Neural" + + async def test_voice_is_reported_as_the_model(self): + """Agent metadata reads ``model`` off the component.""" + tts = TTS(api_key="KEY_test", voice="AWS.Polly.Danielle-Neural") + assert tts.model == "AWS.Polly.Danielle-Neural" + + @pytest.mark.parametrize("dropped_socket_tts", [True], indirect=True) + async def test_stop_during_synthesis_ends_quietly(self, dropped_socket_tts): + """A socket closed by a concurrent stop_audio() is a barge-in.""" + stream = await dropped_socket_tts.stream_audio("hello") + assert [chunk async for chunk in stream] == [] + + @pytest.mark.parametrize("dropped_socket_tts", [False], indirect=True) + async def test_connection_drop_without_stop_propagates(self, dropped_socket_tts): + """A stale stop must not silence a genuine failure in a new synthesis.""" + await dropped_socket_tts.stop_audio() + + stream = await dropped_socket_tts.stream_audio("hello") + with pytest.raises(aiohttp.ClientConnectionError): + [chunk async for chunk in stream] + + async def test_non_dict_payload_is_skipped(self, tts, ws_serving): + ws = ws_serving(['["not", "a", "dict"]']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + + async def test_invalid_base64_audio_is_skipped(self, tts, ws_serving): + ws = ws_serving(['{"audio": "!!!not base64!!!"}']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + + async def test_non_string_audio_is_skipped(self, tts, ws_serving): + ws = ws_serving(['{"audio": 12345}']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + + async def test_undecodable_audio_is_dropped(self, tts): + class FailingDecoder: + def parse(self, data: bytes): + raise av.InvalidDataError(1094995529, "Invalid data") + + assert ( + tts._decode(b"\xff\xf3junk", FailingDecoder(), None, _Id3Stripper()) == [] + ) + + +@pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") +@pytest.mark.integration +class TestTelnyxTTSIntegration: + """Integration tests against the real Telnyx streaming TTS.""" + + @pytest.fixture + async def tts(self): + instance = TTS(voice="AWS.Polly.Danielle-Neural") + try: + yield instance + finally: + await instance.close() + + async def test_stream_audio_yields_chunks(self, tts): + out = [] + async for item in tts.send_iter( + "This is a test of the Telnyx text to speech API." + ): + out.append(item) + + assert len(out) > 0 + assert out[0].data + assert out[-1].final + + async def test_decoded_audio_is_audible_and_long_enough(self, tts): + chunks = [ + item.data + async for item in tts.send_iter( + "The quick brown fox jumps over the lazy dog, and then keeps " + "running for a good while longer across the field." + ) + if item.data is not None + ] + + assert chunks + rates = {chunk.sample_rate for chunk in chunks} + assert len(rates) == 1 + total_samples = sum(len(chunk.samples) for chunk in chunks) + # A sentence this long spans more than one MP3 file, so a broken ID3 + # strip truncates the audio well before this threshold. + assert total_samples / rates.pop() > 3.0 + assert max(abs(int(chunk.samples.max())) for chunk in chunks) > 0 + + async def test_second_synthesis_reconnects(self, tts): + first = [item async for item in tts.send_iter("First utterance.")] + second = [item async for item in tts.send_iter("Second utterance.")] + + assert any(item.data is not None for item in first) + assert any(item.data is not None for item in second) + + async def test_stop_audio_ends_synthesis_promptly(self, tts): + """Barge-in must cut the audio, not just close the socket. + + Telnyx serves a synthesis in very few large frames, so one payload + decodes to hundreds of chunks. Checking the stop only per frame let the + whole utterance play out after a barge-in. + """ + text = ( + "This is a deliberately long utterance so that there is plenty of " + "audio still being streamed when the barge-in arrives, and it keeps " + "going for a while longer to make sure the socket is mid-flight." + ) + baseline = len([chunk async for chunk in await tts.stream_audio(text)]) + assert baseline > 50, f"need a long synthesis to test against, got {baseline}" + + received = 0 + async for _ in await tts.stream_audio(text): + received += 1 + if received == 3: + await tts.stop_audio() + + assert received < baseline / 2, ( + f"stop_audio() did not cut the synthesis: {received} of {baseline} chunks" + ) diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py index cf5b476c1..50ae8b4ea 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py @@ -13,20 +13,27 @@ telnyx_payload_to_pcm, ) from .call_registry import TelnyxCall, TelnyxCallRegistry +from .llm import TelnyxLLM as LLM from .media_stream import TelnyxMediaFormat, TelnyxMediaStream, attach_phone_to_call +from .stt import STT +from .tts import TTS, TelnyxTTSError CallRegistry = TelnyxCallRegistry MediaStream = TelnyxMediaStream __all__ = [ "CallRegistry", + "LLM", "MediaStream", + "STT", + "TTS", "TELNYX_DEFAULT_SAMPLE_RATE", "TELNYX_L16_SAMPLE_RATE", "TelnyxCall", "TelnyxCallRegistry", "TelnyxMediaFormat", "TelnyxMediaStream", + "TelnyxTTSError", "attach_phone_to_call", "l16_to_pcm", "pcma_to_pcm", diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/llm.py b/plugins/telnyx/vision_agents/plugins/telnyx/llm.py new file mode 100644 index 000000000..3a98ae687 --- /dev/null +++ b/plugins/telnyx/vision_agents/plugins/telnyx/llm.py @@ -0,0 +1,74 @@ +"""Telnyx Inference LLM using the OpenAI-compatible Chat Completions endpoint. + +Telnyx serves ``/v2/ai/chat/completions`` with the same request and response +shape as OpenAI, so we point an ``AsyncOpenAI`` client at the Telnyx base URL +and authenticate with the standard bearer token. Streaming, tool calling, and +conversation history are all inherited from :class:`ChatCompletionsLLM`. + +The catalogue of served models is fetched at runtime from ``/v2/ai/models`` +and changes over time, so model ids are not validated locally. + +Docs: https://developers.telnyx.com/api/inference/inference-embedding/post-chat-completions-public-chat-completions-post +""" + +import logging +import os +from typing import Optional + +from openai import AsyncOpenAI +from vision_agents.plugins.openai import ChatCompletionsLLM + +logger = logging.getLogger(__name__) + +TELNYX_BASE_URL = "https://api.telnyx.com/v2/ai" +DEFAULT_MODEL = "meta-llama/Llama-3.3-70B-Instruct" + + +class TelnyxLLM(ChatCompletionsLLM): + """Telnyx Inference Chat Completions LLM. + + Thin wrapper around :class:`ChatCompletionsLLM` that configures the OpenAI + client for Telnyx's OpenAI-compatible inference endpoint. + + Examples: + + from vision_agents.plugins import telnyx + llm = telnyx.LLM(model="openai/gpt-4o") + """ + + provider_name = "telnyx" + + def __init__( + self, + model: str = DEFAULT_MODEL, + api_key: Optional[str] = None, + base_url: str = TELNYX_BASE_URL, + client: Optional[AsyncOpenAI] = None, + tools_max_rounds: int = 3, + ) -> None: + """Initialize the Telnyx LLM. + + Args: + model: The model id as served by Telnyx Inference, for example + ``meta-llama/Llama-3.3-70B-Instruct`` or ``openai/gpt-4o``. + api_key: Telnyx API key. Defaults to the ``TELNYX_API_KEY`` env var. + base_url: API base URL. Defaults to ``https://api.telnyx.com/v2/ai``. + client: Optional pre-configured ``AsyncOpenAI`` client. Takes + precedence over ``api_key`` / ``base_url``. + tools_max_rounds: Max calling rounds for multi-hop tool calls. + """ + resolved_key = ( + api_key if api_key is not None else os.environ.get("TELNYX_API_KEY") + ) + if client is None and not resolved_key: + raise ValueError( + "TELNYX_API_KEY env var or api_key parameter required for Telnyx LLM" + ) + + super().__init__( + model=model, + api_key=resolved_key, + base_url=base_url, + client=client, + tools_max_rounds=tools_max_rounds, + ) diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/stt.py b/plugins/telnyx/vision_agents/plugins/telnyx/stt.py new file mode 100644 index 000000000..199956059 --- /dev/null +++ b/plugins/telnyx/vision_agents/plugins/telnyx/stt.py @@ -0,0 +1,286 @@ +"""Telnyx Speech-to-Text via WebSocket streaming. + +Docs: https://developers.telnyx.com/api/call-control/speech-to-text + +Audio is sent as raw binary frames in the encoding named by ``input_format``. +The default is ``linear16``, for which Telnyx requires an explicit +``sample_rate``. Transcripts come back as +``{"transcript": str, "confidence": float | None, "is_final": bool}``. + +The query parameter for partial transcripts is ``interim_results``. +``partial_results`` is accepted by the endpoint but ignored, which silently +yields finals only. + +``interim_results`` itself is honoured per engine rather than per endpoint. +Measured against the live API with the same audio: ``Speechmatics`` and +``Soniox`` stream partials, while ``Telnyx`` and ``Deepgram`` accept the +parameter and return finals only. +""" + +import asyncio +import json +import logging +import os +import time +from typing import Optional +from urllib.parse import urlencode + +import aiohttp +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 + +logger = logging.getLogger(__name__) + +WS_STT_URL = "wss://api.telnyx.com/v2/speech-to-text/transcription" + + +class STT(stt.STT): + """Telnyx streaming Speech-to-Text. + + Uses aiohttp for a fully async WebSocket connection to the Telnyx streaming + endpoint. Audio is resampled to the configured rate and sent as raw + ``linear16`` binary frames. + + Telnyx does not send VAD signals on this endpoint, so turn detection is + left to the agent's own turn detection. + + Examples: + + from vision_agents.plugins import telnyx + stt = telnyx.STT(sample_rate=8000) + """ + + turn_detection: bool = False + + def __init__( + self, + api_key: Optional[str] = None, + transcription_engine: str = "Telnyx", + language: str = "en", + sample_rate: int = 16000, + interim_results: bool = False, + model: str = "", + ) -> None: + """Initialize Telnyx STT. + + Args: + api_key: Telnyx API key. Falls back to the ``TELNYX_API_KEY`` env var. + transcription_engine: Engine to transcribe with, for example + ``Telnyx``, ``Deepgram`` or ``Speechmatics``. The catalogue is + served by Telnyx and grows over time, so the value is not + validated locally; an unknown engine is rejected by the server + with an explicit error. + language: Language code, for example ``en``. + sample_rate: Rate in Hz that audio is resampled to before being + sent. Use 8000 to pass telephony audio from + :class:`TelnyxMediaStream` through without upsampling. + interim_results: Emit partial transcripts as they are refined. + Honoured per engine, not per endpoint: ``Speechmatics`` and + ``Soniox`` stream partials, while the default ``Telnyx`` engine + accepts the parameter and returns finals only. Defaults to + ``False`` so the default configuration does not advertise + partials it will never emit. + model: Optional engine-specific model id. + """ + super().__init__(provider_name="telnyx") + + self._api_key = api_key or os.environ.get("TELNYX_API_KEY") + if not self._api_key: + raise ValueError( + "TELNYX_API_KEY env var or api_key parameter required for Telnyx STT" + ) + + self.transcription_engine = transcription_engine + self.language = language + self.sample_rate = sample_rate + self.interim_results = interim_results + self.model = model + + self._session: Optional[aiohttp.ClientSession] = None + self._ws: Optional[aiohttp.ClientWebSocketResponse] = None + self._receive_task: Optional[asyncio.Task[None]] = None + self._connection_ready = asyncio.Event() + self._current_participant: Optional[Participant] = None + self._audio_start_time: Optional[float] = None + self._error_reported = False + + async def start(self) -> None: + """Open the Telnyx WebSocket and start the receive loop.""" + await super().start() + + # aiohttp does not attach an Origin header to the handshake. Clients + # that do have to suppress it, because the Telnyx edge rejects a + # WebSocket handshake that carries one. + self._session = aiohttp.ClientSession() + try: + self._ws = await self._session.ws_connect( + self._build_ws_url(), + headers={"Authorization": f"Bearer {self._api_key}"}, + ) + except BaseException: + await self._session.close() + self._session = None + raise + + self._receive_task = asyncio.create_task(self._receive_loop()) + self._connection_ready.set() + self._on_connected() + + async def process_audio( + self, + pcm_data: PcmData, + participant: Participant, + ) -> None: + """Resample a PCM chunk to the configured rate and send it to Telnyx.""" + if self.closed: + logger.warning("Telnyx STT is closed, ignoring audio") + return + + if not self.started: + logger.warning("Telnyx STT is not started, dropping audio") + return + + await self._connection_ready.wait() + + if self._ws is None or self._ws.closed: + logger.warning("Telnyx STT WebSocket not open, dropping audio") + return + + resampled = pcm_data.resample(self.sample_rate, 1) + + self._current_participant = participant + if self._audio_start_time is None: + self._audio_start_time = time.perf_counter() + + try: + await self._ws.send_bytes(resampled.samples.tobytes()) + except (aiohttp.ClientError, ConnectionError) as exc: + self._emit_error_event(exc, context="telnyx_send_audio") + + async def close(self) -> None: + """Close the WebSocket and clean up.""" + await super().close() + + if self._ws is not None and not self._ws.closed: + await self._ws.close() + self._ws = None + + if self._receive_task is not None: + self._receive_task.cancel() + try: + await self._receive_task + except asyncio.CancelledError: + pass + self._receive_task = None + + if self._session is not None and not self._session.closed: + await self._session.close() + self._session = None + + self._connection_ready.clear() + self._on_disconnected() + self._audio_start_time = None + + def _build_ws_url(self) -> str: + params: dict[str, str] = { + "transcription_engine": self.transcription_engine, + "input_format": "linear16", + "sample_rate": str(self.sample_rate), + "language": self.language, + "interim_results": "true" if self.interim_results else "false", + } + if self.model: + params["model"] = self.model + return f"{WS_STT_URL}?{urlencode(params)}" + + async def _receive_loop(self) -> None: + ws = self._ws + if ws is None: + return + try: + async for msg in ws: + if msg.type == aiohttp.WSMsgType.TEXT: + try: + parsed = json.loads(msg.data) + except json.JSONDecodeError: + logger.warning("Telnyx STT sent non-JSON text: %s", msg.data) + continue + if not isinstance(parsed, dict): + logger.warning("Telnyx STT sent unexpected payload: %r", parsed) + continue + if logger.isEnabledFor(logging.DEBUG): + logger.debug("Telnyx STT message: %s", parsed) + self._handle_message(parsed) + elif msg.type in ( + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.ERROR, + ): + break + except asyncio.CancelledError: + raise + except aiohttp.ClientError: + logger.exception("Telnyx STT receive loop error") + + # The server closes the socket right after an error payload, so the + # close is that error's consequence rather than a separate failure. + if not self.closed and not self._error_reported: + self._emit_error_event( + ConnectionError("Telnyx STT WebSocket closed unexpectedly"), + context="telnyx_ws_closed", + ) + + def _handle_message(self, data: dict[str, object]) -> None: + """Dispatch a parsed Telnyx WebSocket message. + + Transcripts arrive as + ``{"transcript": str, "confidence": float | None, "is_final": bool}``. + Parameter rejections arrive as ``{"errors": [{"detail": ...}, ...]}`` + and are followed by the server closing the connection. + """ + errors = data.get("errors") + if isinstance(errors, list) and errors: + details = "; ".join( + str(err.get("detail") or err.get("title")) + for err in errors + if isinstance(err, dict) + ) + self._error_reported = True + self._emit_error_event( + RuntimeError(details or "Telnyx STT error"), + context="telnyx_streaming", + ) + return + + text = data.get("transcript") + if not isinstance(text, str) or not text.strip(): + return + + participant = self._current_participant + if participant is None: + logger.warning("Telnyx transcript received but no participant set") + return + + processing_time_ms: Optional[float] = None + if self._audio_start_time is not None: + processing_time_ms = (time.perf_counter() - self._audio_start_time) * 1000 + + confidence = data.get("confidence") + if isinstance(confidence, bool) or not isinstance(confidence, (int, float)): + confidence = None + is_final = bool(data.get("is_final", True)) + + response = TranscriptResponse( + confidence=float(confidence) if confidence is not None else None, + language=self.language, + model_name=self.model or self.transcription_engine, + processing_time_ms=processing_time_ms, + ) + + if is_final: + self._audio_start_time = None + self._emit_transcript_event(text, participant, response) + else: + self._emit_transcript_event(text, participant, response, mode="replacement") diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py new file mode 100644 index 000000000..a0e75774a --- /dev/null +++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py @@ -0,0 +1,299 @@ +"""Telnyx Text-to-Speech via WebSocket streaming. + +Docs: https://developers.telnyx.com/api/call-control/text-to-speech + +Two properties of the wire protocol drive this implementation: + +- A synthesis is primed with an init frame, then one or more text frames, then + an empty-text stop frame. Telnyx closes the socket once the stop frame has + been served, so a connection cannot be reused across ``stream_audio`` calls + the way a persistent-socket provider allows. +- The audio frames carry slices of MP3, and a synthesis can span several + concatenated MP3 files, each introduced by its own ID3v2 tag at the head of + a WebSocket frame. Those tags have to be dropped before the bytes reach the + decoder, otherwise decoding fails part way through the utterance. + +The decoded sample rate depends on the voice (Polly voices return 24 kHz, +Kokoro voices 22.05 kHz), so the rate reported by the decoder is used rather +than a configured one. +""" + +import asyncio +import base64 +import binascii +import json +import logging +import os +from typing import Any, AsyncIterator, Optional, cast +from urllib.parse import urlencode + +import aiohttp +import av +from getstream.video.rtc.track_util import PcmData +from vision_agents.core import tts + +logger = logging.getLogger(__name__) + +WS_TTS_URL = "wss://api.telnyx.com/v2/text-to-speech/speech" + +DEFAULT_VOICE = "Telnyx.KokoroTTS.af_heart" + +ID3_HEADER_SIZE = 10 + + +class TelnyxTTSError(Exception): + """Raised when Telnyx TTS returns an error frame over WebSocket.""" + + +class _Id3Stripper: + """Removes the ID3v2 tag heading each MP3 file in the audio stream. + + Telnyx concatenates one MP3 file per synthesised segment. Each file starts + with an ID3v2 tag at the head of a WebSocket frame, so only the head of a + frame is inspected. A tag whose header or body spans frames is carried + across calls. + """ + + def __init__(self) -> None: + self._skip = 0 + self._pending = b"" + + def feed(self, data: bytes) -> bytes: + """Return ``data`` with any leading ID3v2 tag removed.""" + if self._skip: + consumed = min(self._skip, len(data)) + data = data[consumed:] + self._skip -= consumed + if not data: + return b"" + + if self._pending: + data = self._pending + data + self._pending = b"" + + if data[:3] != b"ID3": + return data + + if len(data) < ID3_HEADER_SIZE: + self._pending = data + return b"" + + # ID3v2 stores the tag size as four synchsafe bytes (7 bits each). + size = data[6] << 21 | data[7] << 14 | data[8] << 7 | data[9] + body = data[ID3_HEADER_SIZE:] + consumed = min(size, len(body)) + self._skip = size - consumed + return body[consumed:] + + +class TTS(tts.TTS): + """Telnyx streaming Text-to-Speech. + + Opens one WebSocket per synthesis, streams MP3 slices back, and decodes + them into ``PcmData`` as they arrive. + + Examples: + + from vision_agents.plugins import telnyx + tts = telnyx.TTS(voice="AWS.Polly.Danielle-Neural") + """ + + def __init__( + self, + api_key: Optional[str] = None, + voice: str = DEFAULT_VOICE, + idle_timeout: float = 10.0, + connect_timeout: float = 10.0, + ) -> None: + """Initialize Telnyx TTS. + + Args: + api_key: Telnyx API key. Falls back to the ``TELNYX_API_KEY`` env var. + voice: Voice id as listed by ``GET /v2/text-to-speech/voices``, + for example ``Telnyx.KokoroTTS.af_heart`` or + ``AWS.Polly.Danielle-Neural``. + idle_timeout: Seconds of server silence before synthesis is treated + as finished. Normally the server marks the last frame with + ``isFinal``; this is a safety net. + connect_timeout: Seconds to wait for the WebSocket handshake. + """ + super().__init__(provider_name="telnyx") + + self._api_key = api_key or os.environ.get("TELNYX_API_KEY") + if not self._api_key: + raise ValueError( + "TELNYX_API_KEY env var or api_key parameter required for Telnyx TTS" + ) + + self.voice = voice + # The voice doubles as the model identifier reported in agent metadata; + # Telnyx voice ids carry the provider and model, e.g. AWS.Polly.Danielle-Neural. + self.model = voice + self._idle_timeout = idle_timeout + self._connect_timeout = connect_timeout + + self._session: Optional[aiohttp.ClientSession] = None + self._ws: Optional[aiohttp.ClientWebSocketResponse] = None + self._lock = asyncio.Lock() + self._stop_event = asyncio.Event() + # A socket is opened per synthesis, so connect/disconnect are reported + # once for the life of the plugin rather than once per utterance. + self._connected = False + + async def close(self) -> None: + """Close the current WebSocket and release the aiohttp session.""" + await super().close() + session = self._session + self._session = None + if session is not None and not session.closed: + await session.close() + if self._connected: + self._connected = False + self._on_disconnected() + + async def stream_audio( + self, text: str, *_: Any, **__: Any + ) -> AsyncIterator[PcmData]: + """Stream TTS audio chunks for ``text``. + + Returns: + Async iterator yielding ``PcmData`` chunks. + """ + + async def _stream() -> AsyncIterator[PcmData]: + async with self._lock: + # Cleared under the lock so a stop_audio() aimed at an + # in-flight synthesis cannot leave the event set for a call + # queued behind it. + self._stop_event.clear() + try: + ws = await self._connect() + # Telnyx rejects a text frame that is not preceded by an + # init frame with "Invalid message". + await ws.send_str(json.dumps({"text": " "})) + await ws.send_str(json.dumps({"text": text})) + await ws.send_str(json.dumps({"text": ""})) + async for chunk in self._receive_audio(ws): + yield chunk + except aiohttp.ClientConnectionError: + # stop_audio() closes the socket underneath us, which is a + # normal barge-in rather than a synthesis failure. + if not self._stop_event.is_set(): + raise + finally: + await self._close_ws() + + return _stream() + + async def stop_audio(self) -> None: + """Cancel any in-flight synthesis and drop the connection.""" + self._stop_event.set() + await self._close_ws() + + async def _connect(self) -> aiohttp.ClientWebSocketResponse: + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession() + + url = f"{WS_TTS_URL}?{urlencode({'voice': self.voice})}" + # aiohttp's default client timeout for the handshake is 300s, far past + # the point where a caller waiting on speech should give up. + ws = await asyncio.wait_for( + self._session.ws_connect( + url, headers={"Authorization": f"Bearer {self._api_key}"} + ), + timeout=self._connect_timeout, + ) + self._ws = ws + if not self._connected: + self._connected = True + self._on_connected() + logger.debug("Telnyx TTS websocket connected for voice %s", self.voice) + return ws + + async def _close_ws(self) -> None: + if self._ws is not None and not self._ws.closed: + try: + await self._ws.close() + except (aiohttp.ClientError, ConnectionError): + logger.debug("Error closing Telnyx TTS websocket", exc_info=True) + self._ws = None + + async def _receive_audio( + self, ws: aiohttp.ClientWebSocketResponse + ) -> AsyncIterator[PcmData]: + """Yield PcmData until the final frame, a stop, an idle timeout, or a close.""" + decoder = cast(av.AudioCodecContext, av.CodecContext.create("mp3", "r")) + resampler = av.AudioResampler(format="s16", layout="mono") + stripper = _Id3Stripper() + + while True: + if self._stop_event.is_set(): + break + try: + msg = await asyncio.wait_for(ws.receive(), timeout=self._idle_timeout) + except asyncio.TimeoutError: + logger.debug("Telnyx TTS idle timeout, ending synthesis") + break + + if msg.type in ( + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.ERROR, + ): + break + if msg.type != aiohttp.WSMsgType.TEXT: + continue + + try: + data = json.loads(msg.data) + except json.JSONDecodeError: + logger.warning("Telnyx TTS sent non-JSON text: %s", msg.data) + continue + + if not isinstance(data, dict): + logger.warning("Telnyx TTS sent unexpected payload: %r", data) + continue + + if data.get("error"): + raise TelnyxTTSError(str(data["error"])) + + encoded = data.get("audio") + if encoded: + try: + raw = base64.b64decode(encoded, validate=True) + except (binascii.Error, TypeError, ValueError): + logger.warning("Telnyx TTS sent audio that is not valid base64") + continue + # Telnyx serves a whole synthesis in very few large frames, so + # one payload decodes to hundreds of chunks. Checking the stop + # only per frame would let a barge-in play out the rest of the + # utterance. + for pcm in self._decode(raw, decoder, resampler, stripper): + if self._stop_event.is_set(): + return + yield pcm + + if data.get("isFinal"): + break + + def _decode( + self, + audio: bytes, + decoder: av.AudioCodecContext, + resampler: av.AudioResampler, + stripper: _Id3Stripper, + ) -> list[PcmData]: + """Decode one WebSocket audio payload into PcmData chunks. + + A corrupt payload is dropped rather than allowed to abort the + synthesis; the decoder recovers on the next packet boundary. + """ + chunks: list[PcmData] = [] + try: + for packet in decoder.parse(stripper.feed(audio)): + for frame in decoder.decode(packet): + for resampled in resampler.resample(frame): + chunks.append(PcmData.from_av_frame(resampled)) + except av.FFmpegError: + logger.warning("Telnyx TTS sent undecodable audio, dropping payload") + return chunks diff --git a/uv.lock b/uv.lock index d0db9afe3..fee97f593 100644 --- a/uv.lock +++ b/uv.lock @@ -7535,10 +7535,12 @@ dev = [ name = "vision-agents-plugins-telnyx" source = { editable = "plugins/telnyx" } dependencies = [ + { name = "aiohttp" }, { name = "cryptography" }, { name = "fastapi" }, { name = "numpy" }, { name = "vision-agents" }, + { name = "vision-agents-plugins-openai" }, ] [package.dev-dependencies] @@ -7549,10 +7551,12 @@ dev = [ [package.metadata] requires-dist = [ + { name = "aiohttp", specifier = ">=3.13.3" }, { name = "cryptography", specifier = ">=44.0.0" }, { name = "fastapi", specifier = ">=0.135.1" }, { name = "numpy", specifier = ">=1.24.0" }, { name = "vision-agents", editable = "agents-core" }, + { name = "vision-agents-plugins-openai", editable = "plugins/openai" }, ] [package.metadata.requires-dev]