feat(cambai): Add camb.ai realtime speech-to-speech translation

examples/other/translation/camb_realtime_translator.py

@@ -0,0 +1,121 @@
+"""Live speech-to-speech translation with Camb.ai realtime.
+
+Every participant who publishes audio gets a translated track published back into the
+room, named ``translated-<target language>``, plus the source transcript and the
+translated text on the console. One websocket per speaker does the whole job: no STT,
+no LLM, no TTS.
+
+Run it with a Camb.ai key and LiveKit credentials in the environment:
+
+    export CAMB_API_KEY=...
+    export LIVEKIT_URL=... LIVEKIT_API_KEY=... LIVEKIT_API_SECRET=...
+    python camb_realtime_translator.py dev
+
+The realtime endpoint speaks 24 kHz mono PCM16; room audio is resampled for you.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from dotenv import load_dotenv
+
+from livekit import rtc
+from livekit.agents import AgentServer, AutoSubscribe, JobContext, cli
+from livekit.plugins import cambai
+
+load_dotenv()
+
+logger = logging.getLogger("camb-translator")
+
+SOURCE_LANGUAGE = "en-US"
+TARGET_LANGUAGE = "fr-FR"
+
+# "fast" begins speaking sooner; "slow" covers a longer language list. Both translate
+# every complete utterance, so a live translator wants the lower latency.
+MODE = "fast"
+
+SAMPLE_RATE = 24000
+NUM_CHANNELS = 1
+
+# The event loop only holds a weak reference to a task, so keep them alive here.
+_tasks: set[asyncio.Task[None]] = set()
+
+
+def _spawn(coro: asyncio.coroutines) -> None:
+    task = asyncio.create_task(coro)
+    _tasks.add(task)
+    task.add_done_callback(_tasks.discard)
+
+
+async def translate_track(ctx: JobContext, track: rtc.Track, identity: str) -> None:
+    """Translate one participant's audio and publish the result as its own track."""
+    model = cambai.experimental.realtime.RealtimeModel(
+        source_language=SOURCE_LANGUAGE,
+        target_language=TARGET_LANGUAGE,
+        mode=MODE,
+    )
+    session = model.session()
+
+    source = rtc.AudioSource(SAMPLE_RATE, NUM_CHANNELS)
+    publication = await ctx.room.local_participant.publish_track(
+        rtc.LocalAudioTrack.create_audio_track(f"translated-{TARGET_LANGUAGE}", source),
+        rtc.TrackPublishOptions(source=rtc.TrackSource.SOURCE_MICROPHONE),
+    )
+    logger.info("translating %s into %s on %s", identity, TARGET_LANGUAGE, publication.sid)
+
+    @session.on("input_audio_transcription_completed")
+    def _on_transcript(ev: object) -> None:
+        logger.info("%s said: %s", identity, ev.transcript)  # type: ignore[attr-defined]
+
+    @session.on("error")
+    def _on_error(ev: object) -> None:
+        logger.error("translation failed for %s: %s", identity, ev.error)  # type: ignore[attr-defined]
+
+    @session.on("generation_created")
+    def _on_generation(ev: object) -> None:
+        async def show_text(message: object) -> None:
+            text = ""
+            async for chunk in message.text_stream:  # type: ignore[attr-defined]
+                text += chunk
+            if text:
+                logger.info("%s translated: %s", identity, text)
+
+        async def play_audio(message: object) -> None:
+            async for frame in message.audio_stream:  # type: ignore[attr-defined]
+                await source.capture_frame(frame)
+
+        async def deliver() -> None:
+            async for msg in ev.message_stream:  # type: ignore[attr-defined]
+                await asyncio.gather(show_text(msg), play_audio(msg))
+
+        _spawn(deliver())
+
+    try:
+        async for event in rtc.AudioStream(track):
+            session.push_audio(event.frame)
+    finally:
+        await session.aclose()
+        await model.aclose()
+
+
+server = AgentServer()
+
+
+@server.rtc_session()
+async def entrypoint(ctx: JobContext) -> None:
+    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
+
+    @ctx.room.on("track_subscribed")
+    def _on_track(
+        track: rtc.Track,
+        publication: rtc.RemoteTrackPublication,
+        participant: rtc.RemoteParticipant,
+    ) -> None:
+        if track.kind == rtc.TrackKind.KIND_AUDIO:
+            _spawn(translate_track(ctx, track, participant.identity))
+
+
+if __name__ == "__main__":
+    cli.run_app(server)

livekit-plugins/livekit-plugins-cambai/README.md

@@ -1,6 +1,6 @@
 # Camb.ai Plugin for LiveKit Agents
 
-Text-to-Speech plugin for [Camb.ai](https://camb.ai) TTS API, powered by MARS technology.
+Text-to-Speech and realtime speech-to-speech translation for [Camb.ai](https://camb.ai), powered by MARS technology.
 
 ## Features
 
@@ -10,6 +10,7 @@ Text-to-Speech plugin for [Camb.ai](https://camb.ai) TTS API, powered by MARS te
 - Support for 140+ languages
 - Real-time HTTP streaming
 - Pre-built voice library
+- Realtime speech-to-speech translation: speech in one language, speech in another, in the speaker's voice
 
 ## Installation
 
@@ -234,6 +235,104 @@ Coming soon:
 - [LiveKit Agents Documentation](https://docs.livekit.io/agents/)
 - [GitHub Repository](https://github.com/livekit/agents)
 
+## Realtime speech-to-speech translation
+
+`cambai.experimental.realtime.RealtimeModel` translates speech to speech: the participant speaks one
+language and the model returns the same utterance spoken in another, along with a
+transcript of what was said and the translated text. It replaces the usual
+STT + LLM + TTS chain with a single connection.
+
+Drop it into an `AgentSession` like any other realtime model:
+
+```python
+from livekit.agents import AgentSession
+from livekit.plugins import cambai
+
+session = AgentSession(
+    llm=cambai.experimental.realtime.RealtimeModel(
+        source_language="en-US",
+        target_language="fr-FR",
+    ),
+)
+```
+
+No VAD is needed: the endpoint segments utterances itself, so the model reports
+server-side turn detection and the session does not run its own barge-in detection. That
+matters for translation, where the speaker never stops talking and would otherwise be
+treated as interrupting the agent.
+
+To publish a translated track per speaker instead, drive the session directly — see
+`examples/other/translation/camb_realtime_translator.py`:
+
+```python
+from livekit import rtc
+from livekit.plugins import cambai
+
+model = cambai.experimental.realtime.RealtimeModel(
+    source_language="en-US",   # what the speaker says
+    target_language="fr-FR",   # what the room hears
+    mode="fast",
+)
+session = model.session()
+
+translated = rtc.AudioSource(24000, 1)
+await ctx.room.local_participant.publish_track(
+    rtc.LocalAudioTrack.create_audio_track("translated-fr-FR", translated),
+    rtc.TrackPublishOptions(source=rtc.TrackSource.SOURCE_MICROPHONE),
+)
+
+
+async def forward(track: rtc.Track) -> None:
+    async for ev in rtc.AudioStream(track):
+        session.push_audio(ev.frame)
+
+
+@session.on("input_audio_transcription_completed")
+def _on_transcript(ev) -> None:
+    print("source:", ev.transcript)
+
+
+@session.on("generation_created")
+def _on_generation(ev) -> None:
+    async def play() -> None:
+        async for msg in ev.message_stream:
+            async for frame in msg.audio_stream:
+                await translated.capture_frame(frame)
+
+    asyncio.create_task(play())
+```
+
+Each generation also carries `msg.text_stream`, the translated text, which pairs with the
+source transcript above for captions.
+
+Audio is 24 kHz mono PCM16 in both directions; frames at any other rate are resampled for
+you. `voice_id` synthesizes the translation with one of your cloned voices instead of a
+built-in one, and `base_url` points the session at a non-production deployment.
+
+### Choosing a mode
+
+`mode="fast"` starts speaking sooner; `mode="slow"` covers a longer language list. Both
+translate every complete utterance they are given — measured against `realtime.camb.ai`
+on English recordings from 3.9s to 12s, neither mode dropped a finished sentence, and
+translation quality was comparable in both.
+
+What both modes ignore is an *incomplete* utterance. Feeding audio that stops mid-sentence
+leaves that fragment untranslated, which is correct but surprising if you are replaying a
+file you cut at an arbitrary offset: cut on pauses, or accept that the trailing fragment
+goes nowhere. A live microphone raises this only at the very end of a call.
+
+### Turn taking
+
+The endpoint segments utterances itself and streams translations continuously; it emits no
+speech-start or speech-stop events, so `capabilities.turn_detection` is `False`. Nothing
+needs committing and no reply needs requesting — `commit_audio`, `clear_audio` and
+`interrupt` are inert, and `generate_reply` hands back the translation the next utterance
+produces.
+
+Note that a conversational orchestrator is a poor fit for a translator: the speaker never
+stops talking, so anything that treats incoming speech as an interruption will cancel the
+translation mid-playback. Drive the session directly, as above.
+
 ## License
 
 Apache License 2.0

livekit-plugins/livekit-plugins-cambai/livekit/plugins/cambai/__init__.py

@@ -20,6 +20,7 @@
 
 from livekit.agents import APIStatusError, Plugin
 
+from . import experimental
 from .log import logger
 from .tts import API_BASE_URL, API_KEY_HEADER, TTS
 from .version import __version__
@@ -95,4 +96,4 @@ def __init__(self) -> None:
 
 Plugin.register_plugin(CambaiPlugin())
 
-__all__ = ["TTS", "list_voices", "__version__"]
+__all__ = ["TTS", "experimental", "list_voices", "__version__"]

livekit-plugins/livekit-plugins-cambai/livekit/plugins/cambai/experimental/__init__.py

@@ -0,0 +1,17 @@
+# Copyright 2023 LiveKit, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from . import realtime
+
+__all__ = ["realtime"]

livekit-plugins/livekit-plugins-cambai/livekit/plugins/cambai/experimental/realtime/__init__.py

@@ -0,0 +1,17 @@
+# Copyright 2023 LiveKit, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .realtime_model import RealtimeModel, RealtimeSession
+
+__all__ = ["RealtimeModel", "RealtimeSession"]