feat(cambai): Add camb.ai realtime speech-to-speech translation - #7305
Open
adithyaraja-camb wants to merge 18 commits into
Open
adithyaraja-camb wants to merge 18 commits into
adithyaraja-camb wants to merge 18 commits into
Conversation
Adds cambai.realtime.RealtimeModel so a participant can speak one language and the room hears another in the speaker's voice, over a single connection instead of an STT + LLM + TTS chain. The existing TTS is untouched. The endpoint translates rather than converses, so it segments and responds on its own: commit_audio and clear_audio have nothing to do, and generate_reply returns the translation the next utterance produces. No server-side speech start/stop events exist, so turn_detection is False and turn taking stays with the session. Audio is 24kHz mono PCM16 both directions, with inbound frames resampled. The websocket is spoken directly, as tts.py speaks the REST API, so the plugin gains no new dependency.
version.py is bumped by the repo's release commits (livekit-agents@1.8.0, @1.7.1, @1.7.0), not by feature PRs -- livekit#7121 changed only a README and a model file. Bumping it here would collide with that.
No file in the repo uses them, including this plugin's tts.py.
Three defects, all found by driving this against realtime.camb.ai rather than reading the protocol: - update_instructions raised, breaking startup for any caller that sets instructions. A translation has none to steer, so it warns and continues. - a response was never closed. text.done and audio.done arrive together and end a response, so both now close the turn. The idle timer is only a fallback for a response the server never completes, and at 2s it fired inside a response -- measured intra-response audio gaps reach 2.25s -- so it is 6s. Without this a turn could hang and block everything behind it. - audio was forwarded as the server's 400ms blobs, which the room pipeline drops. It is chunked to 100ms frames through AudioByteStream, as the openai plugin does. The README documents the usage verified end to end: drive the session and publish the translated track. Both modes translate every complete utterance; a fragment cut mid-sentence is ignored, which is worth saying because it looks like data loss when replaying a clipped file.
AgentSession is the usual consumer of a RealtimeModel, and it does not work for a translator: its orchestration treats incoming speech as an interruption, which a continuously-speaking source triggers constantly. Shipping this as a plain realtime module would imply that drop-in works. aws and nvidia both keep their realtime models under experimental/, so this follows that convention: the import path says what it is, and the README says plainly what is and is not supported. Adds examples/other/translation/camb_realtime_translator.py, which publishes a translated track per speaker. Verified end to end against realtime.camb.ai in a local room: both utterances either side of a 5s pause were transcribed and translated, and a third participant received 5.32s of French audio.
The endpoint segments utterances itself, but the model declared turn_detection=False, which made AgentSession run its own barge-in detection. A translator's speaker never stops talking, so that fired constantly and cancelled the agent mid-translation -- audio reached the session's output and never reached the room. aws does the same thing correctly: turn_detection=True and no input_speech_started/stopped events, leaving turn taking to the server. Following that, plain AgentSession usage now delivers translated audio to the room in 5 of 5 runs against realtime.camb.ai, where it was 1 in 3 before. Direct-drive is unaffected: both modes still translate every utterance either side of a 5s pause. README now documents the AgentSession usage as well as the example.
…o fast models.py claimed "slow" translates more accurately at the cost of a cold boot. Neither was measured, and the first contradicted the README, which says translation quality was comparable across both -- that one came from real recordings, 3.9s to 12s. Replaced with the measured statement. The example and the README snippet also pinned mode="slow" while the plugin defaults to "fast", with no reason given, so the two things a reader copies disagreed with the default. Both now use "fast": quality is comparable and "fast" starts speaking sooner, which is what a live translator wants. No behaviour change beyond the example's own mode; both modes were exercised end to end and neither dropped a complete utterance.
The plugin opened its own aiohttp websocket and reimplemented what camb.realtime already does. Both sides were identical on the wire: the session.update payload matched connect()'s, and the input audio message matched send_audio() byte for byte, down to the base64 and the nested auth key. The plugin also hardcoded the six server event strings and repeated the binary-frame and base64 normalisation the SDK performs in its dispatcher. Owning that copy is the cost. camb.realtime's events.py carries an alias for the server's model -> mode rename with a note that a missing key failed the whole session.created parse and surfaced as an unexplained connection error. Nothing here was shielded from that: a wrong event string is a silently dead branch that connects fine and emits no text, where the SDK's enum raises at import. What stays is the part that is genuinely LiveKit's: the RealtimeSession and MessageGeneration mapping, the generation lifecycle, resampling, and the measured behaviours around the server's pacing. http_session is gone rather than kept as a no-op, since the SDK owns its transport and honouring that argument is no longer possible. The dependency is camb-sdk, whose import name is camb -- "camb" on PyPI is an unrelated cosmology library. Verified equivalent rather than assumed. Against dev and prod, both modes, 30s and 10-minute Spanish-to-English: 600.0s in produced 598.6s out on prod and 598.4s on dev with no drift. The same runs on the previous raw-websocket code produce the same output duration, generation count and gap profile. In a real LiveKit room with three participants -- one publishing Spanish, one running this model, one recording -- the recording transcribes back to the same English the text channel reported, so the room hears the complete translation and not merely audio of the right length.
It guarded one case: a response the server streams but never marks done would leave _current set, and since _ensure_generation reuses it, every later response would merge into a message that never closes -- the session going silent rather than losing a turn. That case was never reproduced. Measuring the gap between audio chunks over a 30s run gives max 1.08s, p99 0.65s, and nothing above 3s, against a 6s threshold, so the timer only ever fired on the trailing silence after the last response. Removing it changes nothing measurable: same output duration, same text, same translation, and the final generation still closes because _run's finally calls _finish_generation. The earlier claim that slow mode exceeded the 6s window was an artifact of measuring the gap across the drain period, which tracked whatever the timeout was set to rather than the server's pacing. _AUDIO_IDLE_TIMEOUT and _Generation.last_audio_at go with it; nothing else read them.
A connection to the realtime endpoint lives exactly 3600s: the load balancer in front of it caps every backend connection at that, in dev and prod alike, and severs it with a TCP FIN and no close frame. Measured on a 65-minute run, both sessions died at 3600s to the second, and the websockets keepalive never once timed out a pong -- so no client-side ping tuning can avoid it. Recycle at 50 minutes instead, the way the openai and aws realtime plugins handle their own providers' limits: race a timer against the session, let the utterance in flight finish, then hand over to a fresh connection. An unexpected drop takes the same path with backoff, and the input channel is kept rather than replaced, so queued speech is sent late instead of lost. Verified over 65 minutes against the live endpoint, one session per mode: zero abrupt drops, zero errors, one scheduled handover each at minute 50. base_url now defaults to the SDK's own endpoint rather than a second copy of it here.
The loop had no attempt limit, so an endpoint that never recovers was retried forever. It also reset the backoff on a successful connect rather than a working session, which is what let one bad endpoint produce a reconnect per second. A scheduled handover is the only ending that counts as success. Elapsed time and readiness both fail to separate a working session from a broken one: the production endpoint accepts the connection, sends session.created, and drops it 61 seconds later, so it looks ready and long-lived on every attempt. Verified both ways against live endpoints: three consecutive failures against an endpoint that behaves like that now give up after 188s with APIConnectionError, and an 8-minute run with the recycle forced to 90s took five scheduled handovers without spending any of the budget.
The previous commit counted any session that did not end in a scheduled handover as a failure. That is wrong: the endpoint closes an idle connection after about 60 seconds, so a speaker who stops talking produces three of those in three minutes and the session would have been abandoned while it was working. Measured on dev, same code, only difference is whether audio is flowing: idle reconnects at 63.3s and 127.1s, audio flowing has no reconnect at all in 150s. Both endpoints behave the same way, so this was never specific to production. A session that became ready did its job, whatever ended it. Only a session that never became ready, or a connect that raised, counts against the limit. Verified: an idle session survived 260s across four idle reconnects without giving up, an unreachable endpoint still gave up after three, and a six-minute run with audio and the handover forced to 90s took three handovers with no errors.
Counting failures by readiness left one loop unbounded: the endpoint closes an idle connection after about 60 seconds, so a room with nothing publishing reconnected once a minute forever and translated nothing. Audio is the signal that a session did something. A session that carried frames resets the count, so a conversation is never cut off however long it runs, and a scheduled handover is unaffected because the session before it carried audio. Verified: no audio at all stops after 189s with a clear error, a five-minute run with audio and the handover forced to 90s took three handovers with no errors, and a real LiveKit room publishing pure silence pushed 3600 frames over three minutes without a single reconnect -- the room keeps sending frames when nobody speaks, so this cannot end a live call.
Counting sessions that carried no audio conflated two things. A retry is for a connection that failed; a session that became ready and then closed did not fail, however it ended. The endpoint closes an idle connection after about 60 seconds, so a participant who mutes reconnects until they speak again, which is correct. A room with nobody in it is not this loop's problem: LiveKit closes the agent session when the participant disconnects (close_on_disconnect defaults to True), which closes the channel the loop runs on. Verified: three failed connections stop after 65s, and a room whose track is unpublished keeps reconnecting rather than abandoning a participant who may unmute.
A session is capped at an hour and the endpoint's load balancer closes a connection at exactly 3600s, so one connection covers a whole session and there is nothing to hand over to. The recycle timer, the retry counter and the loop around them were sized against a 65 minute test, which is longer than a session can be. What remains is what was asked for: base_url defaults to the SDK's own endpoint instead of a second copy of it here.
The dict literal inferred as dict[str, str], which mypy could bind to camb_connect's second positional parameter under the repo's strict config, so the call failed with two arg-type errors. Annotating it dict[str, Any] types the spread without changing behaviour.
…e-s2s # Conflicts: # livekit-plugins/livekit-plugins-cambai/pyproject.toml
|
Adithya Raja seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account. You have signed the CLA already but the status is still pending? Let us recheck it. |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Devin Review found 6 potential issues.
3 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
Comment on lines
+68
to
+70
| @session.on("input_audio_transcription_completed") | ||
| def _on_transcript(ev: object) -> None: | ||
| logger.info("%s said: %s", identity, ev.transcript) # type: ignore[attr-defined] |
Contributor
There was a problem hiding this comment.
🟨 Source speech enters unredacted logs
Every completed transcript logs the participant identity and spoken text in the message body. Standard redaction cannot protect these PII-sensitive values.
Was this helpful? React with 👍 or 👎 to provide feedback.
Author
There was a problem hiding this comment.
Showing the translated text alongside the source is the point of the example, and examples/other/translation/multi-user-translator.py:203 in this same directory logs the translated transcript already
Comment on lines
+82
to
+83
| if text: | ||
| logger.info("%s translated: %s", identity, text) |
Contributor
There was a problem hiding this comment.
🟨 Translated speech enters unredacted logs
Every translated utterance logs the participant identity and translated text in the message body. Standard redaction cannot protect these PII-sensitive values.
Was this helpful? React with 👍 or 👎 to provide feedback.
Author
There was a problem hiding this comment.
Since this is an example proof printing what the speaker said is what this example is demonstrating, examples/other/transcription/multi-user-transcriber.py:40 logs {participant_identity} -> {user_transcript} identically.
| 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) |
Contributor
There was a problem hiding this comment.
🟨 Participant identity enters unredacted logs
Starting a translation logs the remote participant identity in the message body. Standard redaction cannot protect this PII-sensitive value.
Was this helpful? React with 👍 or 👎 to provide feedback.
Author
There was a problem hiding this comment.
This is the example's startup line showing which participant is being translated and on which track. examples/other/transcription/multi-user-transcriber.py:67 logs starting session for {participant.identity} the same way
… close Three faults in the generation lifecycle, all reachable from a normal session. The function channel was constructed inline in the GenerationCreatedEvent and its only reference was dropped immediately, so nothing could close it. A generation ends only once both its message and function streams close, so AgentActivity waited forever after the first utterance and every later translation stayed queued behind a speech handle that never completed. The channel now lives on the generation and closes with the others. A generation handed to a waiting generate_reply() still carried user_initiated=False, so the framework both resolved the caller's future and scheduled the same event from the generation_created listener, leaving two tasks reading one pair of channels. The flag is now true exactly when the generation satisfies a pending reply, and spontaneous ones stay false. A pending generate_reply() was never resolved when the session ended, so a caller awaiting one hung after the websocket closed or aclose() returned. Both paths now fail it with a RealtimeError. Reported by Devin on the pull request. The three logging findings it raised against the example are left alone: examples/other/transcription/multi-user- transcriber.py and examples/other/translation/multi-user-translator.py log participant identity and transcript text the same way.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds a
RealtimeModeltolivekit-plugins-cambaithat translates speech to speech over a single Camb.ai websocket, with no STT, LLM or TTS in the path.What it adds
cambai.experimental.realtime.RealtimeModelimplementsllm.RealtimeModel, so it drops intoAgentSessionwherever a realtime model is accepted.experimentalnamespace, leaving the plugin's existing TTS surface untouched.source_languageandtarget_languageas BCP-47 tags, and optionally avoice_idto speak the translation in one of your cloned voices. Withvoice_idomitted the server picks a built-in voice for the target language.modeselects the pipeline:"fast"begins speaking sooner,"slow"covers a longer list of languages."fast"is the default.api_keyfalls back to theCAMB_API_KEYenvironment variable.base_urlis optional and only forwarded when set, so an unset value keeps whichever endpoint the installed SDK defaults to.turn_detectionanduser_transcriptiontrue,message_truncationandauto_tool_reply_generationfalse, matching how the server segments utterances and the absence of tool calls on this path.What's included
livekit-plugins/livekit-plugins-cambai/livekit/plugins/cambai/experimental/realtime/— theRealtimeModelandRealtimeSessionmodels.py—RealtimeMode,DEFAULT_REALTIME_MODEandREALTIME_SAMPLE_RATEpyproject.toml—camb-sdk>=1.6.0added, description and keywords updated to cover realtime alongside TTS, withuv.lockregeneratedexamples/other/translation/camb_realtime_translator.py— publishes atranslated-<target language>track back into the room for every participant who publishes audio, and prints both transcriptsREADME.md— realtime setup, the language and voice options, and a runnable snippetTest plan
uv run ruff check— cleanuv run ruff format --check— cleanuv run mypy -p livekit.plugins.cambai— clean under the repo's strict configuv run pytest --unit— 3384 passed, 5 skippedbase_urlset, and that setting it reaches a non-production deployment