From de9768b72d2147630dc06e441388c67d952dfc7b Mon Sep 17 00:00:00 2001 From: 3DRX <3drxkjy@gmail.com> Date: Thu, 11 Jun 2026 13:41:48 +0800 Subject: [PATCH] feat: bytedance mllm extension --- .../extension/bytedance_mllm_python/README.md | 27 ++ .../bytedance_mllm_python/__init__.py | 1 + .../extension/bytedance_mllm_python/addon.py | 10 + .../extension/bytedance_mllm_python/config.py | 163 ++++++++ .../bytedance_mllm_python/connection.py | 146 +++++++ .../bytedance_mllm_python/extension.py | 358 ++++++++++++++++++ .../bytedance_mllm_python/manifest.json | 105 +++++ .../bytedance_mllm_python/property.json | 25 ++ .../bytedance_mllm_python/protocol.py | 262 +++++++++++++ .../bytedance_mllm_python/requirements.txt | 3 + 10 files changed, 1100 insertions(+) create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/README.md create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/__init__.py create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/addon.py create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/config.py create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/connection.py create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/extension.py create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/manifest.json create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/property.json create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/protocol.py create mode 100644 ai_agents/agents/ten_packages/extension/bytedance_mllm_python/requirements.txt diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/README.md b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/README.md new file mode 100644 index 0000000000..76394f0971 --- /dev/null +++ b/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. diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/__init__.py b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/__init__.py new file mode 100644 index 0000000000..f3c731cdd5 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/__init__.py @@ -0,0 +1 @@ +from . import addon diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/addon.py b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/addon.py new file mode 100644 index 0000000000..6de47ee19e --- /dev/null +++ b/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) diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/config.py b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/config.py new file mode 100644 index 0000000000..2035cbd3d8 --- /dev/null +++ b/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}" diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/connection.py b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/connection.py new file mode 100644 index 0000000000..820bbb0880 --- /dev/null +++ b/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) diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/extension.py b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/extension.py new file mode 100644 index 0000000000..fba8946aa7 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/extension.py @@ -0,0 +1,358 @@ +from __future__ import annotations + +import asyncio +import json +import traceback +import uuid + +from ten_ai_base.const import LOG_CATEGORY_KEY_POINT, LOG_CATEGORY_VENDOR +from ten_ai_base.message import ( + ModuleError, + ModuleErrorCode, + ModuleErrorVendorInfo, +) +from ten_ai_base.mllm import AsyncMLLMBaseExtension +from ten_ai_base.struct import ( + MLLMClientFunctionCallOutput, + MLLMClientMessageItem, + MLLMServerInputTranscript, + MLLMServerInterrupt, + MLLMServerOutputTranscript, + MLLMServerSessionReady, +) +from ten_ai_base.types import LLMToolMetadata +from ten_ai_base.utils import encrypt +from ten_runtime import AudioFrame, AsyncTenEnv, Data + +from .config import BytedanceMLLMConfig +from .connection import BytedanceRealtimeConnection +from .protocol import ServerEvent, ServerMessage + + +class BytedanceMLLMExtension(AsyncMLLMBaseExtension): + def __init__(self, name: str): + super().__init__(name) + self.ten_env: AsyncTenEnv | None = None + self.config: BytedanceMLLMConfig | None = None + self.conn: BytedanceRealtimeConnection | None = None + + self.connected = False + self.stopped = False + self.doubao_session_id = str(uuid.uuid4()) + self.request_transcript = "" + self.response_transcript = "" + self._session_started = False + self._response_final_sent = True + + async def on_init(self, ten_env: AsyncTenEnv) -> None: + await super().on_init(ten_env) + self.ten_env = ten_env + + properties, _ = await ten_env.get_property_to_json(None) + self.config = BytedanceMLLMConfig.model_validate_json(properties) + ten_env.log_info( + f"config: {self.config.to_str(sensitive_handling=True)}", + category=LOG_CATEGORY_KEY_POINT, + ) + + async def on_stop(self, ten_env: AsyncTenEnv) -> None: + await super().on_stop(ten_env) + self.stopped = True + if self.conn: + await self.conn.close() + + def vendor(self) -> str: + return "bytedance_doubao" + + def input_audio_sample_rate(self) -> int: + return self.config.input_sample_rate + + def synthesize_audio_sample_rate(self) -> int: + return self.config.output_sample_rate + + def synthesize_audio_channels(self) -> int: + return 1 + + def synthesize_audio_sample_width(self) -> int: + return 2 + + def is_connected(self) -> bool: + return self.connected + + async def start_connection(self) -> None: + try: + self.conn = BytedanceRealtimeConnection( + ten_env=self.ten_env, + config=self.config, + session_id=self.doubao_session_id, + ) + await self.conn.connect() + await self.conn.send_start_connection() + + self.ten_env.log_info("[Doubao] client loop started") + async for message in self.conn.listen(): + try: + await self._handle_server_message(message) + except Exception as e: + traceback.print_exc() + self.ten_env.log_error( + f"[Doubao] error processing message {message}: {e}" + ) + + self.ten_env.log_info("[Doubao] client loop finished") + except Exception as e: + traceback.print_exc() + self.ten_env.log_error(f"[Doubao] connection failed: {e}") + + await self._handle_reconnect() + + async def stop_connection(self) -> None: + self.connected = False + self._session_started = False + if self.conn: + try: + await self.conn.send_finish_session() + await self.conn.send_finish_connection() + except Exception as e: + self.ten_env.log_warn(f"[Doubao] finish failed: {e}") + await self.conn.close() + + async def _handle_reconnect(self) -> None: + await self.stop_connection() + if not self.stopped: + await asyncio.sleep(1) + self.doubao_session_id = str(uuid.uuid4()) + await self.start_connection() + + async def _handle_server_message(self, message: ServerMessage) -> None: + event = self._server_event(message.event) + if event is None: + self.ten_env.log_debug(f"[Doubao] unhandled event {message.event}") + return + + if event == ServerEvent.CONNECTION_STARTED: + self.ten_env.log_info( + f"[Doubao] connection started: {message.connection_id}", + category=LOG_CATEGORY_VENDOR, + ) + await self._start_session() + elif event == ServerEvent.CONNECTION_FAILED: + await self._send_vendor_error( + "CONNECTION_FAILED", + self._payload_error(message), + fatal=True, + ) + elif event == ServerEvent.SESSION_STARTED: + self.connected = True + self._session_started = True + self.ten_env.log_info( + f"[Doubao] session started: {message.session_id}", + category=LOG_CATEGORY_VENDOR, + ) + await self.send_server_session_ready(MLLMServerSessionReady()) + elif event == ServerEvent.SESSION_FAILED: + await self._send_vendor_error( + "SESSION_FAILED", + self._payload_error(message), + fatal=True, + ) + elif event == ServerEvent.SESSION_FINISHED: + self.connected = False + self._session_started = False + elif event == ServerEvent.ASR_INFO: + await self.send_server_interrupted(MLLMServerInterrupt()) + elif event == ServerEvent.ASR_RESPONSE: + await self._handle_asr_response(message) + elif event == ServerEvent.ASR_ENDED: + await self._finalize_input_transcript() + elif event == ServerEvent.CHAT_RESPONSE: + await self._handle_chat_response(message) + elif event == ServerEvent.CHAT_ENDED: + await self._finalize_output_transcript() + elif event == ServerEvent.TTS_RESPONSE: + if message.payload: + await self.send_server_output_audio_data(message.payload) + elif event == ServerEvent.TTS_ENDED: + await self._finalize_output_transcript() + elif event == ServerEvent.DIALOG_COMMON_ERROR: + await self._send_vendor_error( + "DIALOG_COMMON_ERROR", + self._payload_error(message), + fatal=False, + ) + elif event == ServerEvent.USAGE_RESPONSE: + self.ten_env.log_debug(f"[Doubao] usage: {message.payload_json}") + else: + self.ten_env.log_debug( + f"[Doubao] event={event.name} payload={message.payload_json}" + ) + + async def send_audio( + self, frame: AudioFrame, session_id: str | None + ) -> bool: + self.session_id = session_id + if not self.connected or not self.conn: + return False + if frame.get_sample_rate() != self.input_audio_sample_rate(): + self.ten_env.log_warn( + "[Doubao] dropping audio frame with unsupported sample rate " + f"{frame.get_sample_rate()}Hz; expected " + f"{self.input_audio_sample_rate()}Hz" + ) + return False + await self.conn.send_audio_data(frame.get_buf()) + return True + + async def on_data(self, ten_env: AsyncTenEnv, data: Data) -> None: + await super().on_data(ten_env, data) + + async def send_client_message_item( + self, item: MLLMClientMessageItem, session_id: str | None = None + ) -> None: + if item.role != "user": + self.ten_env.log_warn( + f"[Doubao] message role {item.role} is not supported" + ) + return + if not self.conn: + return + await self.conn.send_text_query(item.content) + + async def send_client_create_response( + self, session_id: str | None = None + ) -> None: + self.ten_env.log_debug("[Doubao] create_response is implicit") + + async def send_client_register_tool(self, _tool: LLMToolMetadata) -> None: + self.ten_env.log_warn("[Doubao] tool calling is not supported") + + async def send_client_function_call_output( + self, _function_call_output: MLLMClientFunctionCallOutput + ) -> None: + self.ten_env.log_warn("[Doubao] tool calling is not supported") + + async def _start_session(self) -> None: + payload = self.config.start_session_payload(self.message_context) + self.ten_env.log_info( + "[Doubao] start session payload: " + + json.dumps(self._redact_payload(payload), ensure_ascii=False), + category=LOG_CATEGORY_VENDOR, + ) + await self.conn.send_start_session(payload) + + async def _handle_asr_response(self, message: ServerMessage) -> None: + payload = message.payload_json or {} + results = payload.get("results", []) + if not results: + return + + text = "".join(str(result.get("text", "")) for result in results) + is_interim = any(bool(result.get("is_interim")) for result in results) + if not text: + return + + self.request_transcript = text + await self.send_server_input_transcript( + MLLMServerInputTranscript( + content=text, + delta=text, + final=not is_interim, + metadata={"session_id": self.session_id or "-1"}, + ) + ) + if not is_interim: + self.request_transcript = "" + + async def _finalize_input_transcript(self) -> None: + if not self.request_transcript: + return + await self.send_server_input_transcript( + MLLMServerInputTranscript( + content=self.request_transcript, + delta="", + final=True, + metadata={"session_id": self.session_id or "-1"}, + ) + ) + self.request_transcript = "" + + async def _handle_chat_response(self, message: ServerMessage) -> None: + payload = message.payload_json or {} + delta = str(payload.get("content", "")) + if not delta: + return + self.response_transcript += delta + self._response_final_sent = False + await self.send_server_output_text( + MLLMServerOutputTranscript( + content=self.response_transcript, + delta=delta, + final=False, + metadata={"session_id": self.session_id or "-1"}, + ) + ) + + async def _finalize_output_transcript(self) -> None: + if self._response_final_sent: + return + await self.send_server_output_text( + MLLMServerOutputTranscript( + content=self.response_transcript, + delta="", + final=True, + metadata={"session_id": self.session_id or "-1"}, + ) + ) + self.response_transcript = "" + self._response_final_sent = True + + async def _send_vendor_error( + self, code: str, message: str, fatal: bool + ) -> None: + self.ten_env.log_error(f"[Doubao] {code}: {message}") + await self.send_mllm_error( + ModuleError( + code=( + ModuleErrorCode.FATAL_ERROR.value + if fatal + else ModuleErrorCode.NON_FATAL_ERROR.value + ), + message=message, + ), + ModuleErrorVendorInfo( + vendor=self.vendor(), + code=code, + message=message, + ), + ) + + def _payload_error(self, message: ServerMessage) -> str: + if message.payload_json: + return json.dumps(message.payload_json, ensure_ascii=False) + if message.payload: + return message.payload.decode("utf-8", errors="replace") + if message.error_code: + return f"error_code={message.error_code}" + return "unknown error" + + def _server_event(self, event: int) -> ServerEvent | None: + try: + return ServerEvent(event) + except ValueError: + return None + + def _redact_payload(self, payload): + if isinstance(payload, dict): + result = {} + for key, value in payload.items(): + lower_key = key.lower() + if "key" in lower_key or "token" in lower_key: + result[key] = ( + encrypt(value) if isinstance(value, str) else "***" + ) + else: + result[key] = self._redact_payload(value) + return result + if isinstance(payload, list): + return [self._redact_payload(value) for value in payload] + return payload diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/manifest.json b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/manifest.json new file mode 100644 index 0000000000..675fe5c388 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/manifest.json @@ -0,0 +1,105 @@ +{ + "type": "extension", + "name": "bytedance_mllm_python", + "version": "0.1.0", + "dependencies": [ + { + "type": "system", + "name": "ten_runtime_python", + "version": "0.11" + }, + { + "type": "system", + "name": "ten_ai_base", + "version": "0.7" + } + ], + "package": { + "include": [ + "manifest.json", + "property.json", + "requirements.txt", + "**.py", + "README.md" + ] + }, + "api": { + "interface": [ + { + "import_uri": "../../system/ten_ai_base/api/mllm-interface.json" + } + ], + "property": { + "properties": { + "params": { + "type": "object", + "properties": { + "api_url": { + "type": "string" + }, + "app_id": { + "type": "string" + }, + "access_key": { + "type": "string" + }, + "resource_id": { + "type": "string" + }, + "app_key": { + "type": "string" + }, + "model": { + "type": "string" + }, + "speaker": { + "type": "string" + }, + "bot_name": { + "type": "string" + }, + "prompt": { + "type": "string" + }, + "speaking_style": { + "type": "string" + }, + "dialog_id": { + "type": "string" + }, + "input_sample_rate": { + "type": "int32" + }, + "output_sample_rate": { + "type": "int32" + }, + "output_audio_format": { + "type": "string" + }, + "verbose": { + "type": "bool" + }, + "asr": { + "type": "object", + "properties": {} + }, + "tts": { + "type": "object", + "properties": {} + }, + "dialog": { + "type": "object", + "properties": {} + } + } + }, + "dump": { + "type": "bool" + }, + "dump_path": { + "type": "string" + } + } + } + } +} diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/property.json b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/property.json new file mode 100644 index 0000000000..48b60a6ece --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/property.json @@ -0,0 +1,25 @@ +{ + "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": "", + "input_sample_rate": 16000, + "output_sample_rate": 24000, + "output_audio_format": "pcm_s16le", + "asr": { + "extra": {} + }, + "tts": { + "extra": {} + }, + "dialog": { + "extra": { + "model": "1.2.1.1" + } + } + } +} + diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/protocol.py b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/protocol.py new file mode 100644 index 0000000000..93d74891f7 --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/protocol.py @@ -0,0 +1,262 @@ +from __future__ import annotations + +import gzip +import json +from dataclasses import dataclass +from enum import IntEnum +from typing import Any + + +PROTOCOL_VERSION = 0b0001 +HEADER_SIZE = 0b0001 + + +class MessageType(IntEnum): + FULL_CLIENT_REQUEST = 0b0001 + AUDIO_ONLY_REQUEST = 0b0010 + FULL_SERVER_RESPONSE = 0b1001 + AUDIO_ONLY_RESPONSE = 0b1011 + ERROR = 0b1111 + + +class MessageFlag(IntEnum): + NO_SEQUENCE = 0b0000 + POSITIVE_SEQUENCE = 0b0001 + LAST_NO_SEQUENCE = 0b0010 + NEGATIVE_SEQUENCE = 0b0011 + WITH_EVENT = 0b0100 + + +class Serialization(IntEnum): + RAW = 0b0000 + JSON = 0b0001 + + +class Compression(IntEnum): + NONE = 0b0000 + GZIP = 0b0001 + + +class ClientEvent(IntEnum): + START_CONNECTION = 1 + FINISH_CONNECTION = 2 + START_SESSION = 100 + FINISH_SESSION = 102 + TASK_REQUEST = 200 + UPDATE_CONFIG = 201 + SAY_HELLO = 300 + END_ASR = 400 + CHAT_TTS_TEXT = 500 + CHAT_TEXT_QUERY = 501 + CLIENT_INTERRUPT = 515 + + +class ServerEvent(IntEnum): + CONNECTION_STARTED = 50 + CONNECTION_FAILED = 51 + CONNECTION_FINISHED = 52 + SESSION_STARTED = 150 + SESSION_FINISHED = 152 + SESSION_FAILED = 153 + USAGE_RESPONSE = 154 + CONFIG_UPDATED = 251 + TTS_SENTENCE_START = 350 + TTS_SENTENCE_END = 351 + TTS_RESPONSE = 352 + TTS_ENDED = 359 + ASR_INFO = 450 + ASR_RESPONSE = 451 + ASR_ENDED = 459 + CHAT_RESPONSE = 550 + CHAT_TEXT_QUERY_CONFIRMED = 553 + CHAT_ENDED = 559 + DIALOG_COMMON_ERROR = 599 + + +SESSION_EVENTS = { + ServerEvent.SESSION_STARTED, + ServerEvent.SESSION_FINISHED, + ServerEvent.SESSION_FAILED, + ServerEvent.USAGE_RESPONSE, + ServerEvent.CONFIG_UPDATED, + ServerEvent.TTS_SENTENCE_START, + ServerEvent.TTS_SENTENCE_END, + ServerEvent.TTS_RESPONSE, + ServerEvent.TTS_ENDED, + ServerEvent.ASR_INFO, + ServerEvent.ASR_RESPONSE, + ServerEvent.ASR_ENDED, + ServerEvent.CHAT_RESPONSE, + ServerEvent.CHAT_TEXT_QUERY_CONFIRMED, + ServerEvent.CHAT_ENDED, + ServerEvent.DIALOG_COMMON_ERROR, +} + + +@dataclass +class Header: + message_type: int + flags: int = MessageFlag.WITH_EVENT + serialization: int = Serialization.JSON + compression: int = Compression.NONE + + def to_bytes(self) -> bytes: + return bytes( + [ + (PROTOCOL_VERSION << 4) | HEADER_SIZE, + (self.message_type << 4) | self.flags, + (self.serialization << 4) | self.compression, + 0x00, + ] + ) + + +@dataclass +class ServerMessage: + message_type: int + flags: int + serialization: int + compression: int + event: int = 0 + session_id: str | None = None + connection_id: str | None = None + error_code: int = 0 + payload: bytes = b"" + payload_json: dict[str, Any] | None = None + + +def _bytes_with_size(value: str) -> bytes: + encoded = value.encode("utf-8") + return len(encoded).to_bytes(4, "big", signed=False) + encoded + + +def build_client_event( + event: int, + payload: bytes | dict[str, Any] | None = None, + session_id: str | None = None, + message_type: int = MessageType.FULL_CLIENT_REQUEST, + serialization: int = Serialization.JSON, +) -> bytes: + header = Header( + message_type=message_type, + serialization=serialization, + ).to_bytes() + + optional = bytearray() + optional.extend(int(event).to_bytes(4, "big", signed=False)) + if session_id is not None: + optional.extend(_bytes_with_size(session_id)) + + if payload is None: + payload_bytes = b"{}" if serialization == Serialization.JSON else b"" + elif isinstance(payload, (bytes, bytearray, memoryview)): + payload_bytes = bytes(payload) + else: + payload_bytes = json.dumps( + payload, ensure_ascii=False, separators=(",", ":") + ).encode("utf-8") + + return b"".join( + [ + header, + bytes(optional), + len(payload_bytes).to_bytes(4, "big", signed=False), + payload_bytes, + ] + ) + + +def read_sized_text(data: bytes, offset: int) -> tuple[str, int]: + if offset + 4 > len(data): + return "", offset + size = int.from_bytes(data[offset : offset + 4], "big", signed=False) + offset += 4 + if offset + size > len(data): + return "", offset + value = data[offset : offset + size].decode("utf-8") + return value, offset + size + + +def read_payload(data: bytes, offset: int) -> tuple[bytes, int]: + if offset + 4 > len(data): + return b"", offset + size = int.from_bytes(data[offset : offset + 4], "big", signed=False) + offset += 4 + if offset + size > len(data): + return b"", offset + return data[offset : offset + size], offset + size + + +def parse_server_message(data: bytes) -> ServerMessage: + if len(data) < 4: + raise ValueError("server message is shorter than header") + + header_size = data[0] & 0x0F + message_type = data[1] >> 4 + flags = data[1] & 0x0F + serialization = data[2] >> 4 + compression = data[2] & 0x0F + offset = header_size * 4 + + message = ServerMessage( + message_type=message_type, + flags=flags, + serialization=serialization, + compression=compression, + ) + + if message_type == MessageType.ERROR: + if offset + 4 <= len(data): + message.error_code = int.from_bytes( + data[offset : offset + 4], "big", signed=False + ) + offset += 4 + message.payload, _ = read_payload(data, offset) + _decode_payload(message) + return message + + if flags & MessageFlag.POSITIVE_SEQUENCE: + offset += 4 + + if flags & MessageFlag.WITH_EVENT: + if offset + 4 > len(data): + return message + message.event = int.from_bytes( + data[offset : offset + 4], "big", signed=False + ) + offset += 4 + + event = _safe_server_event(message.event) + if event == ServerEvent.CONNECTION_STARTED: + message.connection_id, offset = read_sized_text(data, offset) + elif event == ServerEvent.CONNECTION_FAILED: + message.payload, offset = read_payload(data, offset) + elif event in SESSION_EVENTS: + message.session_id, offset = read_sized_text(data, offset) + message.payload, offset = read_payload(data, offset) + else: + message.payload, offset = read_payload(data, offset) + + _decode_payload(message) + return message + + +def _safe_server_event(event: int) -> ServerEvent | None: + try: + return ServerEvent(event) + except ValueError: + return None + + +def _decode_payload(message: ServerMessage) -> None: + if message.compression == Compression.GZIP and message.payload: + message.payload = gzip.decompress(message.payload) + + if message.serialization != Serialization.JSON or not message.payload: + return + + try: + payload = message.payload.decode("utf-8") + message.payload_json = json.loads(payload) + except (UnicodeDecodeError, json.JSONDecodeError): + message.payload_json = None diff --git a/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/requirements.txt b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/requirements.txt new file mode 100644 index 0000000000..55754a186c --- /dev/null +++ b/ai_agents/agents/ten_packages/extension/bytedance_mllm_python/requirements.txt @@ -0,0 +1,3 @@ +websockets +pydantic +