Add Boson realtime plugin

livekit-agents/pyproject.toml

@@ -76,6 +76,7 @@ baseten = ["livekit-plugins-baseten>=1.6.10"]
 bey = ["livekit-plugins-bey>=1.6.10"]
 bithuman = ["livekit-plugins-bithuman>=1.6.10"]
 bland = ["livekit-plugins-bland>=1.6.10"]
+boson = ["livekit-plugins-boson>=1.6.10"]
 browser = ["livekit-plugins-browser>=0.2.10"]
 cambai = ["livekit-plugins-cambai>=1.6.10"]
 cartesia = ["livekit-plugins-cartesia>=1.6.10"]

livekit-plugins/livekit-plugins-boson/README.md

@@ -0,0 +1,183 @@
+# Boson plugin for LiveKit Agents
+
+This plugin lets LiveKit Agents use the Boson realtime WebSocket API as a
+LiveKit `RealtimeModel`.
+
+LiveKit owns the room connection, participant audio, interruption plumbing,
+transcript IO, and worker lifecycle. This plugin adapts LiveKit Agents realtime
+model calls to Boson's OpenAI-compatible realtime protocol subset.
+
+## Installation
+
+```bash
+pip install livekit-plugins-boson
+```
+
+## Usage
+
+Set `BOSON_API_KEY` in your environment. Get a key at
+[boson.ai/workspace/api-key](https://www.boson.ai/workspace/api-key).
+
+```python
+from livekit.agents import AgentSession
+from livekit.plugins import boson
+
+
+session = AgentSession(llm=boson.realtime.RealtimeModel())
+```
+
+The defaults connect to the hosted API (`wss://api.boson.ai/v1/realtime`) with
+the `higgs-realtime` model and the `default` voice. To point at another
+deployment, pass `url=` — and `api_key=None` if it does not authenticate.
+
+`examples/agent.py` is a complete agent with function tools that you can talk to
+over your own microphone, with no LiveKit server involved:
+
+```bash
+export BOSON_API_KEY=...
+python examples/agent.py console
+```
+
+## Model options
+
+`boson.realtime.RealtimeModel` accepts these options:
+
+| Option | Description |
+| --- | --- |
+| `url` | Realtime WebSocket endpoint. Defaults to `wss://api.boson.ai/v1/realtime`. `http`/`https` URLs are normalized to `ws`/`wss`; the path is passed through unchanged. |
+| `api_key` | Sent as an `Authorization: Bearer ...` header. Omit it to read `BOSON_API_KEY` from the environment; omitting both raises `ValueError`. Pass `None` explicitly to send no header, for a local dev server without auth. |
+| `model` | Sent as `session.model`. Defaults to `"higgs-realtime"`. |
+| `voice` | Sent as `session.audio.output.voice`. |
+| `instructions` | Sent as `session.instructions`. |
+| `output_modalities` | `["audio"]` by default. `["text"]` is also supported for text-only responses. |
+| `temperature` | Optional. Sent as `session.temperature` when set; otherwise the server default is used. |
+| `max_output_tokens` | Sent as `session.max_output_tokens`. |
+| `tool_choice` | Sent as `session.tool_choice`. |
+| `speed` | Sent as `session.audio.output.speed`. Not currently supported for output audio; kept for wire compatibility. |
+| `turn_detection` | Sent as `session.audio.input.turn_detection`. |
+| `input_audio_transcription` | Sent as `session.audio.input.transcription`. |
+| `input_audio_noise_reduction` | Sent as `session.audio.input.noise_reduction`. |
+| `truncation` | `"auto"` (default) or `"disabled"`. Sent as `session.truncation`. `"auto"` enables background context summarization where the model supports it. Do not pass `None` — it is rejected and ends the session. |
+| `query_params` | Extra query parameters added to the WebSocket URL. |
+
+If `turn_detection` is omitted, the plugin sends a default server VAD config:
+
+```python
+{
+    "type": "server_vad",
+    "create_response": True,
+    "interrupt_response": True,
+    "prefix_padding_ms": 300,
+    "silence_duration_ms": 500,
+    "threshold": 0.55,
+}
+```
+
+`create_response` and `interrupt_response` are accepted for OpenAI wire
+compatibility but are not configurable here: with server VAD enabled, a
+response is always created after user speech and the active response is always
+cancelled when new speech is detected. The plugin's own duplicate
+`response.cancel` suppression follows that behavior (keyed on whether server
+VAD is enabled at all), not on what these fields are set to.
+
+The dict is sent to the server as given, so `{"type": "semantic_vad", ...}` works
+too; the plugin does not filter or rewrite its keys.
+
+Passing `turn_detection=None`/`False` disables server-side turn detection and
+switches to client-driven `input_audio_buffer.commit`. This is exposed for
+interface completeness but isn't a heavily-exercised path yet — treat it as
+experimental rather than a recommended default.
+
+`input_audio_transcription` can be passed as a full dict, or built with:
+
+- `input_audio_transcription_model` (e.g. `"higgs-stt-3.1"`)
+- `input_audio_transcription_language`
+
+User transcription is enabled only when the transcription config contains a
+non-empty `model`. A transcription `prompt` is not supported; the plugin drops
+it even if present in a raw `input_audio_transcription` dict.
+
+### `generate_reply()` overrides
+
+`generate_reply(instructions=...)` sends the instructions in `response.create`.
+The server applies them to that response alone: they replace the session
+instructions for the turn, which still answers from the conversation so far.
+Because it is a replacement rather than an addition, the event carries the
+session instructions and the per-response ones together.
+
+`tools` and `tool_choice` are **not** applied per response — the server accepts
+and ignores them. `generate_reply()` therefore ignores them too (with a warning)
+rather than sending values that would appear to take effect. The framework
+scopes them at the session level around the call instead, since the plugin
+reports `per_response_tool_choice = False`; to scope them from your own code,
+use `update_tools()` and `update_options(tool_choice=...)` directly.
+
+## Protocol compatibility
+
+The plugin translates LiveKit realtime model operations to the following client
+events:
+
+- `session.update`
+- `input_audio_buffer.append`
+- `input_audio_buffer.commit`
+- `input_audio_buffer.clear`
+- `conversation.item.create`
+- `conversation.item.delete`
+- `conversation.item.truncate`
+- `response.create`
+- `response.cancel`
+
+The plugin handles Boson response audio, response text, audio transcript, input
+transcription, function call, interruption, and error events and maps them back
+to LiveKit realtime model streams.
+
+Function tools registered on the LiveKit `Agent` are sent in `session.update`.
+When Boson returns a `function_call` item, LiveKit Agents executes the Python
+tool locally and the plugin sends the result back as a
+`function_call_output` conversation item.
+
+Only function tools have a shape on the wire. A tool of any other kind is left
+out of `session.update` — the model never sees it — and the plugin logs a
+warning once per session saying how many were dropped.
+
+The plugin does not retry a WebSocket close that a reconnect cannot fix: an
+invalid API key (close code 3000) or a billing entitlement refusal (close code
+4429, e.g. `insufficient_quota`/`monthly_cap_reached`) both end the session
+instead of retrying. Server errors that reflect an expected client/server race
+rather than a real failure (`response_not_active`, `response_id_mismatch`,
+`voice_output_task_ongoing`, `invalid_previous_item_id`) are logged but not
+surfaced as recoverable `error` events.
+
+The server has no insert-at-head primitive for conversation items
+(`previous_item_id: null` always means append-at-tail). If `update_chat_ctx()`
+needs to insert a new item ahead of turns the server already has (e.g.
+prepending a context summary), the plugin deletes and recreates the entire
+remote conversation in the target order instead of silently misordering it by
+appending the new item at the tail.
+
+That rebuild can only put back what it can express as text. A user turn whose
+transcript has not arrived yet has none, and no audio is kept client-side, so it
+is deleted along with the rest and not recreated — the server answers without it
+until a later `update_chat_ctx()` recreates it (at the tail) once its transcript
+exists. The plugin logs a warning naming the affected item ids whenever this
+happens.
+
+The same limit applies without a rebuild. A change to a turn's text goes out as
+a delete plus a create under one id, so clearing that text leaves a create with
+nothing to send: the turn is deleted rather than emptied. The delete is kept —
+the caller asked for that text to leave the context — and the same warning names
+the item, which returns on a later sync once it has text again.
+
+## Limitations
+
+- The plugin expects the current realtime event names, such as
+  `response.output_audio.delta` and `response.output_text.delta`.
+- Video input is not supported yet.
+- Mixed `["text", "audio"]` output modalities are not supported; choose either
+  `["audio"]` or `["text"]`.
+- This plugin sends and receives 24 kHz PCM audio. That's the format this
+  integration uses, not the full range the Higgs Realtime API supports.
+- `system`/`developer`-role chat items are not supported (the server's
+  conversation store only accepts `assistant`/`user` items); they're silently
+  dropped when syncing `update_chat_ctx()`. Use `instructions` for persistent
+  directives instead.

livekit-plugins/livekit-plugins-boson/examples/agent.py

@@ -0,0 +1,96 @@
+"""A voice agent backed by the Boson (Higgs) realtime model.
+
+Get an API key at https://www.boson.ai/workspace/api-key and export it:
+
+    export BOSON_API_KEY=...
+
+Then talk to it through your own microphone and speakers, with no LiveKit
+server and no LIVEKIT_* credentials involved:
+
+    python agent.py console
+
+`console` needs PortAudio (`apt-get install libportaudio2` on Debian/Ubuntu).
+To run it against a LiveKit room instead, use `dev` or `start` and set the
+LIVEKIT_* variables as usual.
+
+Try asking what the weather is somewhere, or what time it is, to see a function
+call go out and come back. Interrupting mid-answer should stop it immediately.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from livekit.agents import (
+    Agent,
+    AgentServer,
+    AgentSession,
+    JobContext,
+    MetricsCollectedEvent,
+    RunContext,
+    cli,
+    metrics,
+)
+from livekit.agents.llm import function_tool
+from livekit.plugins import boson
+
+logger = logging.getLogger("boson-agent")
+
+
+class MyAgent(Agent):
+    def __init__(self) -> None:
+        super().__init__(
+            instructions=(
+                "You are a helpful assistant speaking to the user over voice. "
+                "Keep answers to one or two sentences. Do not use emoji, markdown, "
+                "or any other characters that do not read aloud."
+            )
+        )
+
+    async def on_enter(self) -> None:
+        self.session.generate_reply(instructions="Greet the user and offer your help.")
+
+    @function_tool
+    async def lookup_weather(self, context: RunContext, location: str) -> str:
+        """Look up the current weather for a location.
+
+        Args:
+            location: The city or region the user asked about.
+        """
+        logger.info("looking up weather for %s", location)
+        # A real integration would call a weather API here.
+        return f"It is sunny in {location}, 22 degrees celsius."
+
+    @function_tool
+    async def get_current_time(self, context: RunContext) -> str:
+        """Get the current UTC date and time."""
+        return datetime.now(timezone.utc).strftime("%A %d %B %Y, %H:%M UTC")
+
+
+server = AgentServer()
+
+
+@server.rtc_session()
+async def entrypoint(ctx: JobContext) -> None:
+    session: AgentSession = AgentSession(
+        # url defaults to the hosted endpoint and api_key falls back to
+        # BOSON_API_KEY, so a bare RealtimeModel() is enough to connect.
+        # input_audio_transcription_model is what turns user transcripts on;
+        # without it the server still runs ASR for the model's own use but
+        # sends nothing back to display.
+        llm=boson.realtime.RealtimeModel(
+            input_audio_transcription_model="higgs-stt-3.1",
+            input_audio_transcription_language="english",
+        ),
+    )
+
+    @session.on("metrics_collected")
+    def _on_metrics_collected(ev: MetricsCollectedEvent) -> None:
+        metrics.log_metrics(ev.metrics)
+
+    await session.start(agent=MyAgent(), room=ctx.room)
+
+
+if __name__ == "__main__":
+    cli.run_app(server)

livekit-plugins/livekit-plugins-boson/livekit/plugins/boson/__init__.py

@@ -0,0 +1,19 @@
+"""Boson plugin for LiveKit Agents."""
+
+from __future__ import annotations
+
+from livekit.agents import Plugin
+
+from . import realtime
+from .log import logger
+from .version import __version__
+
+__all__ = ["realtime", "__version__"]
+
+
+class BosonPlugin(Plugin):
+    def __init__(self) -> None:
+        super().__init__(__name__, __version__, __package__, logger)
+
+
+Plugin.register_plugin(BosonPlugin())

livekit-plugins/livekit-plugins-boson/livekit/plugins/boson/log.py

@@ -0,0 +1,3 @@
+import logging
+
+logger = logging.getLogger("livekit.plugins.boson")

livekit-plugins/livekit-plugins-boson/livekit/plugins/boson/realtime/__init__.py

@@ -0,0 +1,3 @@
+from .realtime_model import RealtimeModel, RealtimeSession
+
+__all__ = ["RealtimeModel", "RealtimeSession"]