feat: add bytedance mllm extension

ai_agents/agents/ten_packages/extension/bytedance_mllm_python/README.md

@@ -0,0 +1,27 @@
+# ByteDance Realtime MLLM Extension
+
+This extension connects TEN `AsyncMLLMBaseExtension` to Volcengine Doubao
+end-to-end realtime speech model API.
+
+It sends mono PCM16 16 kHz audio to Doubao and requests mono PCM16 24 kHz audio
+back from the service.
+
+## Configuration
+
+Configure provider values under `params`:
+
+```json
+{
+  "params": {
+    "app_id": "${env:DOUBAO_REALTIME_APP_ID}",
+    "access_key": "${env:DOUBAO_REALTIME_ACCESS_KEY}",
+    "model": "1.2.1.1",
+    "speaker": "zh_female_vv_jupiter_bigtts",
+    "bot_name": "豆包",
+    "prompt": "You are a concise voice assistant."
+  }
+}
+```
+
+Optional nested `asr`, `tts`, and `dialog` objects are merged into the
+`StartSession` payload.

ai_agents/agents/ten_packages/extension/bytedance_mllm_python/__init__.py

@@ -0,0 +1 @@
+from . import addon

ai_agents/agents/ten_packages/extension/bytedance_mllm_python/addon.py

@@ -0,0 +1,10 @@
+from ten_runtime import Addon, TenEnv, register_addon_as_extension
+
+
+@register_addon_as_extension("bytedance_mllm_python")
+class BytedanceMLLMExtensionAddon(Addon):
+    def on_create_instance(self, ten_env: TenEnv, name: str, context) -> None:
+        from .extension import BytedanceMLLMExtension
+
+        ten_env.log_info("BytedanceMLLMExtensionAddon on_create_instance")
+        ten_env.on_create_instance_done(BytedanceMLLMExtension(name), context)

ai_agents/agents/ten_packages/extension/bytedance_mllm_python/config.py

@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import copy
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+from ten_ai_base.utils import encrypt
+
+
+class BytedanceMLLMConfig(BaseModel):
+    model_config = ConfigDict(protected_namespaces=())
+
+    params: dict[str, Any] = Field(default_factory=dict)
+    dump: bool = False
+    dump_path: str = ""
+
+    def _params(self) -> dict[str, Any]:
+        params = getattr(self, "params", {})
+        if isinstance(params, dict):
+            return params
+        return {}
+
+    def _param(self, key: str, default: Any = None) -> Any:
+        return self._params().get(key, default)
+
+    @model_validator(mode="after")
+    def validate_params(self) -> "BytedanceMLLMConfig":
+        required = {
+            "app_id": self.app_id,
+            "access_key": self.access_key,
+        }
+        missing = [k for k, v in required.items() if not v]
+        if missing:
+            raise ValueError(
+                f"Missing required params fields: {', '.join(missing)}"
+            )
+        if self.input_sample_rate <= 0:
+            raise ValueError("input_sample_rate must be greater than 0")
+        if self.output_sample_rate <= 0:
+            raise ValueError("output_sample_rate must be greater than 0")
+        return self
+
+    @property
+    def api_url(self) -> str:
+        return self._param(
+            "api_url",
+            "wss://openspeech.bytedance.com/api/v3/realtime/dialogue",
+        )
+
+    @property
+    def app_id(self) -> str:
+        return self._param("app_id", "")
+
+    @property
+    def access_key(self) -> str:
+        return self._param("access_key", "")
+
+    @property
+    def resource_id(self) -> str:
+        return self._param("resource_id", "volc.speech.dialog")
+
+    @property
+    def app_key(self) -> str:
+        return self._param("app_key", "PlgvMymc7f3tQnJ6")
+
+    @property
+    def model(self) -> str:
+        return self._param("model", "1.2.1.1")
+
+    @property
+    def speaker(self) -> str:
+        return self._param("speaker", "zh_female_vv_jupiter_bigtts")
+
+    @property
+    def bot_name(self) -> str:
+        return self._param("bot_name", "豆包")
+
+    @property
+    def prompt(self) -> str:
+        return self._param("prompt", "")
+
+    @property
+    def speaking_style(self) -> str:
+        return self._param("speaking_style", "")
+
+    @property
+    def dialog_id(self) -> str:
+        return self._param("dialog_id", "")
+
+    @property
+    def input_sample_rate(self) -> int:
+        return int(self._param("input_sample_rate", 16000))
+
+    @property
+    def output_sample_rate(self) -> int:
+        return int(self._param("output_sample_rate", 24000))
+
+    @property
+    def output_audio_format(self) -> str:
+        return self._param("output_audio_format", "pcm_s16le")
+
+    @property
+    def verbose(self) -> bool:
+        return bool(self._param("verbose", False))
+
+    def start_session_payload(
+        self, message_context: list[Any] | None = None
+    ) -> dict[str, Any]:
+        asr = copy.deepcopy(self._param("asr", {}))
+        tts = copy.deepcopy(self._param("tts", {}))
+        dialog = copy.deepcopy(self._param("dialog", {}))
+
+        asr.setdefault("audio_info", {})
+        asr["audio_info"].setdefault("format", "pcm_s16le")
+        asr["audio_info"].setdefault("sample_rate", self.input_sample_rate)
+        asr["audio_info"].setdefault("channel", 1)
+        asr.setdefault("extra", {})
+
+        tts.setdefault("speaker", self.speaker)
+        tts.setdefault("audio_config", {})
+        tts["audio_config"].setdefault("channel", 1)
+        tts["audio_config"].setdefault("format", self.output_audio_format)
+        tts["audio_config"].setdefault("sample_rate", self.output_sample_rate)
+        tts.setdefault("extra", {})
+
+        dialog.setdefault("bot_name", self.bot_name)
+        dialog.setdefault("dialog_id", self.dialog_id)
+        dialog.setdefault("extra", {})
+        dialog["extra"].setdefault("model", self.model)
+        if self.prompt and "system_role" not in dialog:
+            dialog["system_role"] = self.prompt
+        if self.speaking_style and "speaking_style" not in dialog:
+            dialog["speaking_style"] = self.speaking_style
+
+        if message_context and "dialog_context" not in dialog:
+            dialog["dialog_context"] = [
+                {"role": item.role, "text": item.content}
+                for item in message_context
+                if item.role in ("user", "assistant")
+            ]
+
+        return {
+            "asr": asr,
+            "tts": tts,
+            "dialog": dialog,
+        }
+
+    def to_str(self, sensitive_handling: bool = True) -> str:
+        if not sensitive_handling:
+            return f"{self}"
+
+        config = self.model_copy(deep=True)
+        params = getattr(config, "params", {})
+        if not isinstance(params, dict):
+            return f"{config}"
+
+        redacted_params: dict[str, Any] = dict(params)
+        for key in ("app_id", "access_key", "api_key"):
+            value = redacted_params.get(key)
+            if isinstance(value, str) and value:
+                redacted_params[key] = encrypt(value)
+        config.params = redacted_params
+        return f"{config}"

ai_agents/agents/ten_packages/extension/bytedance_mllm_python/connection.py

@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+import json
+import uuid
+from typing import AsyncGenerator
+
+import websockets
+from ten_runtime import AsyncTenEnv
+from websockets.legacy.client import WebSocketClientProtocol
+
+from .config import BytedanceMLLMConfig
+from .protocol import (
+    ClientEvent,
+    MessageType,
+    Serialization,
+    ServerMessage,
+    build_client_event,
+    parse_server_message,
+)
+
+
+def _truncate_payload(payload: bytes, limit: int = 160) -> str:
+    if len(payload) <= limit:
+        return payload.decode("utf-8", errors="replace")
+    return payload[:limit].decode("utf-8", errors="replace") + "..."
+
+
+class BytedanceRealtimeConnection:
+    def __init__(
+        self,
+        ten_env: AsyncTenEnv,
+        config: BytedanceMLLMConfig,
+        session_id: str,
+    ):
+        self.ten_env = ten_env
+        self.config = config
+        self.session_id = session_id
+        self.websocket: WebSocketClientProtocol | None = None
+
+    async def connect(self) -> None:
+        connect_id = str(uuid.uuid4())
+        headers = {
+            "X-Api-App-ID": self.config.app_id,
+            "X-Api-Access-Key": self.config.access_key,
+            "X-Api-Resource-Id": self.config.resource_id,
+            "X-Api-App-Key": self.config.app_key,
+            "X-Api-Connect-Id": connect_id,
+        }
+
+        self.ten_env.log_info(
+            f"[Doubao] connecting to {self.config.api_url}, "
+            f"connect_id={connect_id}"
+        )
+        self.websocket = await websockets.connect(
+            self.config.api_url,
+            additional_headers=headers,
+            max_size=100_000_000,
+            compression=None,
+        )
+
+    async def send_start_connection(self) -> None:
+        await self._send(
+            build_client_event(ClientEvent.START_CONNECTION, payload={})
+        )
+
+    async def send_finish_connection(self) -> None:
+        await self._send(
+            build_client_event(ClientEvent.FINISH_CONNECTION, payload={})
+        )
+
+    async def send_start_session(self, payload: dict) -> None:
+        await self._send(
+            build_client_event(
+                ClientEvent.START_SESSION,
+                payload=payload,
+                session_id=self.session_id,
+            )
+        )
+
+    async def send_finish_session(self) -> None:
+        await self._send(
+            build_client_event(
+                ClientEvent.FINISH_SESSION,
+                payload={},
+                session_id=self.session_id,
+            )
+        )
+
+    async def send_audio_data(self, audio_data: bytes) -> None:
+        await self._send(
+            build_client_event(
+                ClientEvent.TASK_REQUEST,
+                payload=audio_data,
+                session_id=self.session_id,
+                message_type=MessageType.AUDIO_ONLY_REQUEST,
+                serialization=Serialization.RAW,
+            )
+        )
+
+    async def send_text_query(self, text: str) -> None:
+        await self._send(
+            build_client_event(
+                ClientEvent.CHAT_TEXT_QUERY,
+                payload={"content": text},
+                session_id=self.session_id,
+            )
+        )
+
+    async def send_client_interrupt(self) -> None:
+        await self._send(
+            build_client_event(
+                ClientEvent.CLIENT_INTERRUPT,
+                payload={},
+                session_id=self.session_id,
+            )
+        )
+
+    async def listen(self) -> AsyncGenerator[ServerMessage, None]:
+        if self.websocket is None:
+            raise RuntimeError("websocket is not connected")
+
+        async for message in self.websocket:
+            if isinstance(message, str):
+                raise RuntimeError(message)
+            parsed = parse_server_message(message)
+            if self.config.verbose:
+                self.ten_env.log_info(
+                    f"[Doubao] <- event={parsed.event} "
+                    f"payload={_truncate_payload(parsed.payload)}"
+                )
+            yield parsed
+
+    async def close(self) -> None:
+        if self.websocket is None:
+            return
+        await self.websocket.close()
+        self.websocket = None
+
+    async def _send(self, message: bytes) -> None:
+        if self.websocket is None:
+            raise RuntimeError("websocket is not connected")
+        if self.config.verbose:
+            self.ten_env.log_info(
+                f"[Doubao] -> {json.dumps(list(message[:32]))}"
+            )
+        await self.websocket.send(message)