diff --git a/.env.example b/.env.example index f8140657..5c9037e1 100644 --- a/.env.example +++ b/.env.example @@ -18,7 +18,7 @@ WORKERS_MAX=100 # Worker quit timeout in seconds WORKER_QUIT_TIMEOUT_SECONDES=60 -# Agora App ID +# Agora App ID # Agora App Certificate(only required if enabled in the Agora Console) AGORA_APP_ID= AGORA_APP_CERTIFICATE= @@ -55,10 +55,18 @@ AZURE_STT_REGION= AZURE_TTS_KEY= AZURE_TTS_REGION= +# Extension: cartesia_tts +# Cartesia TTS key +CARTESIA_API_KEY= + # Extension: cosy_tts # Cosy TTS key COSY_TTS_KEY= +# Extension: deepgram_asr_python +# Deepgram ASR key +DEEPGRAM_API_KEY= + # Extension: elevenlabs_tts # ElevenLabs TTS key ELEVENLABS_TTS_KEY= @@ -106,4 +114,13 @@ WEATHERAPI_API_KEY= # Extension: bingsearch_tool_python # Bing search API key -BING_API_KEY= \ No newline at end of file +BING_API_KEY= + +# Extension: tsdb_firestore +# Firestore certifications +FIRESTORE_PROJECT_ID= +FIRESTORE_PRIVATE_KEY_ID= +FIRESTORE_PRIVATE_KEY= +FIRESTORE_CLIENT_EMAIL= +FIRESTORE_CLIENT_ID= +FIRESTORE_CERT_URL= diff --git a/agents/.gitignore b/agents/.gitignore index 175ad523..7e8da43e 100644 --- a/agents/.gitignore +++ b/agents/.gitignore @@ -2,6 +2,7 @@ ten_packages/extension_group/ ten_packages/extension/agora_rtc ten_packages/extension/azure_tts +ten_packages/extension/agora_sess_ctrl ten_packages/extension/py_init_extension_cpp ten_packages/system .ten diff --git a/agents/manifest-lock.json b/agents/manifest-lock.json index 1398fb18..01873ebb 100644 --- a/agents/manifest-lock.json +++ b/agents/manifest-lock.json @@ -67,6 +67,24 @@ } ] }, + { + "type": "extension", + "name": "agora_sess_ctrl", + "version": "0.2.0", + "hash": "52f26dee2fb8fbd22d55ecc6bb197176a51f4a3bd2268788f75582f68cf1270b", + "dependencies": [ + { + "type": "system", + "name": "ten_runtime" + } + ], + "supports": [ + { + "os": "linux", + "arch": "x64" + } + ] + }, { "type": "system", "name": "azure_speech_sdk", diff --git a/agents/manifest.json b/agents/manifest.json index f92fe394..814cd3c6 100644 --- a/agents/manifest.json +++ b/agents/manifest.json @@ -18,6 +18,11 @@ "name": "agora_rtc", "version": "=0.8.0-rc2" }, + { + "type": "extension", + "name": "agora_sess_ctrl", + "version": "0.2.0" + }, { "type": "system", "name": "azure_speech_sdk", diff --git a/agents/property.json b/agents/property.json index 18968b0c..5bfe55a1 100644 --- a/agents/property.json +++ b/agents/property.json @@ -2971,6 +2971,692 @@ ] } ] + }, + { + "name": "va_openai_v2v_storage", + "auto_start": true, + "nodes": [ + { + "type": "extension", + "extension_group": "rtc", + "addon": "agora_rtc", + "name": "agora_rtc", + "property": { + "app_id": "${env:AGORA_APP_ID}", + "token": "", + "channel": "ten_agent_test", + "stream_id": 1234, + "remote_stream_id": 123, + "subscribe_audio": true, + "publish_audio": true, + "publish_data": true, + "subscribe_audio_sample_rate": 24000 + } + }, + { + "type": "extension", + "extension_group": "llm", + "addon": "openai_v2v_python", + "name": "openai_v2v_python", + "property": { + "api_key": "${env:OPENAI_REALTIME_API_KEY}", + "temperature": 0.9, + "model": "gpt-4o-realtime-preview", + "max_tokens": 2048, + "voice": "alloy", + "language": "en-US", + "server_vad": true, + "dump": true, + "history": 10, + "enable_storage": true + } + }, + { + "type": "extension", + "extension_group": "transcriber", + "addon": "message_collector", + "name": "message_collector" + }, + { + "type": "extension", + "extension_group": "tools", + "addon": "weatherapi_tool_python", + "name": "weatherapi_tool_python", + "property": { + "api_key": "${env:WEATHERAPI_API_KEY}" + } + }, + { + "type": "extension", + "extension_group": "tools", + "addon": "bingsearch_tool_python", + "name": "bingsearch_tool_python", + "property": { + "api_key": "${env:BING_API_KEY}" + } + }, + { + "type": "extension", + "extension_group": "context", + "addon": "tsdb_firestore", + "name": "tsdb_firestore", + "property": { + "credentials": { + "type": "service_account", + "project_id": "${env:FIRESTORE_PROJECT_ID}", + "private_key_id": "${env:FIRESTORE_PRIVATE_KEY_ID}", + "private_key": "${env:FIRESTORE_PRIVATE_KEY}", + "client_email": "${env:FIRESTORE_CLIENT_EMAIL}", + "client_id": "${env:FIRESTORE_CLIENT_ID}", + "auth_uri": "https://accounts.google.com/o/oauth2/auth", + "token_uri": "https://oauth2.googleapis.com/token", + "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs", + "client_x509_cert_url": "${env:FIRESTORE_CERT_URL}", + "universe_domain": "googleapis.com" + }, + "channel_name": "ten_agent_test", + "collection_name": "llm_context" + } + } + ], + "connections": [ + { + "extension_group": "rtc", + "extension": "agora_rtc", + "audio_frame": [ + { + "name": "pcm_frame", + "dest": [ + { + "extension_group": "realtime", + "extension": "openai_v2v_python" + } + ] + } + ] + }, + { + "extension_group": "tools", + "extension": "weatherapi_tool_python", + "cmd": [ + { + "name": "tool_register", + "dest": [ + { + "extension_group": "realtime", + "extension": "openai_v2v_python" + } + ] + } + ] + }, + { + "extension_group": "tools", + "extension": "bingsearch_tool_python", + "cmd": [ + { + "name": "tool_register", + "dest": [ + { + "extension_group": "realtime", + "extension": "openai_v2v_python" + } + ] + } + ] + }, + { + "extension_group": "realtime", + "extension": "openai_v2v_python", + "audio_frame": [ + { + "name": "pcm_frame", + "dest": [ + { + "extension_group": "rtc", + "extension": "agora_rtc" + } + ] + } + ], + "data": [ + { + "name": "append", + "dest": [ + { + "extension_group": "context", + "extension": "tsdb_firestore" + } + ] + }, + { + "name": "text_data", + "dest": [ + { + "extension_group": "transcriber", + "extension": "message_collector" + } + ] + } + ], + "cmd": [ + { + "name": "flush", + "dest": [ + { + "extension_group": "rtc", + "extension": "agora_rtc" + } + ] + }, + { + "name": "retrieve", + "dest": [ + { + "extension_group": "context", + "extension": "tsdb_firestore" + } + ] + }, + { + "name": "tool_call_get_current_weather", + "dest": [ + { + "extension_group": "tools", + "extension": "weatherapi_tool_python", + "msg_conversion": { + "type": "per_property", + "keep_original": true, + "rules": [ + { + "path": "_ten.name", + "conversion_mode": "fixed_value", + "value": "tool_call" + } + ] + } + } + ] + }, + { + "name": "tool_call_get_past_weather", + "dest": [ + { + "extension_group": "tools", + "extension": "weatherapi_tool_python", + "msg_conversion": { + "type": "per_property", + "keep_original": true, + "rules": [ + { + "path": "_ten.name", + "conversion_mode": "fixed_value", + "value": "tool_call" + } + ] + } + } + ] + }, + { + "name": "tool_call_get_future_weather", + "dest": [ + { + "extension_group": "tools", + "extension": "weatherapi_tool_python", + "msg_conversion": { + "type": "per_property", + "keep_original": true, + "rules": [ + { + "path": "_ten.name", + "conversion_mode": "fixed_value", + "value": "tool_call" + } + ] + } + } + ] + }, + { + "name": "tool_call_bing_search", + "dest": [ + { + "extension_group": "tools", + "extension": "weatherapi_tool_python", + "msg_conversion": { + "type": "per_property", + "keep_original": true, + "rules": [ + { + "path": "_ten.name", + "conversion_mode": "fixed_value", + "value": "tool_call" + } + ] + } + } + ] + } + ] + }, + { + "extension_group": "transcriber", + "extension": "message_collector", + "data": [ + { + "name": "data", + "dest": [ + { + "extension_group": "rtc", + "extension": "agora_rtc" + } + ] + } + ] + } + ] + }, + { + "name": "va_deepgram_openai_cartesia", + "auto_start": false, + "nodes": [ + { + "type": "extension", + "extension_group": "default", + "addon": "agora_rtc", + "name": "agora_rtc", + "property": { + "app_id": "${env:AGORA_APP_ID}", + "token": "", + "channel": "ten_agent_test", + "stream_id": 1234, + "remote_stream_id": 123, + "subscribe_audio": true, + "publish_audio": true, + "publish_data": true, + "enable_agora_asr": false, + "agora_asr_vendor_name": "microsoft", + "agora_asr_language": "en-US", + "agora_asr_vendor_key": "${env:AZURE_STT_KEY}", + "agora_asr_vendor_region": "${env:AZURE_STT_REGION}", + "agora_asr_session_control_file_path": "session_control.conf" + } + }, + { + "type": "extension", + "extension_group": "asr", + "addon": "deepgram_asr_python", + "name": "deepgram_asr", + "property": { + "api_key": "${env:DEEPGRAM_API_KEY}", + "language": "en-US", + "model": "nova-2", + "sample_rate": "16000" + } + }, + { + "type": "extension", + "extension_group": "chatgpt", + "addon": "openai_chatgpt", + "name": "openai_chatgpt", + "property": { + "base_url": "", + "api_key": "${env:OPENAI_API_KEY}", + "frequency_penalty": 0.9, + "model": "gpt-4o-mini", + "max_tokens": 512, + "prompt": "", + "proxy_url": "${env:OPENAI_PROXY_URL}", + "greeting": "TEN Agent connected. How can I help you today?", + "max_memory_length": 10 + } + }, + { + "type": "extension", + "extension_group": "tts", + "addon": "cartesia_tts", + "name": "cartesia_tts", + "property": { + "api_key": "${env:CARTESIA_API_KEY}", + "cartesia_version": "2024-06-10", + "model_id": "sonic-english", + "voice_id": "f9836c6e-a0bd-460e-9d3c-f7299fa60f94", + "sample_rate": "16000" + } + }, + { + "type": "extension", + "extension_group": "default", + "addon": "interrupt_detector_python", + "name": "interrupt_detector" + }, + { + "type": "extension", + "extension_group": "transcriber", + "addon": "message_collector", + "name": "message_collector" + } + ], + "connections": [ + { + "extension_group": "default", + "extension": "agora_rtc", + "audio_frame": [ + { + "name": "pcm_frame", + "dest": [ + { + "extension_group": "asr", + "extension": "deepgram_asr" + }, + { + "extension_group": "transcriber", + "extension": "message_collector" + } + ] + } + ], + "cmd": [ + { + "name": "on_user_joined", + "dest": [ + { + "extension_group": "asr", + "extension": "deepgram_asr" + } + ] + }, + { + "name": "on_user_left", + "dest": [ + { + "extension_group": "asr", + "extension": "deepgram_asr" + } + ] + }, + { + "name": "on_connection_failure", + "dest": [ + { + "extension_group": "asr", + "extension": "deepgram_asr" + } + ] + } + ] + }, + { + "extension_group": "asr", + "extension": "deepgram_asr", + "data": [ + { + "name": "text_data", + "dest": [ + { + "extension_group": "default", + "extension": "interrupt_detector" + }, + { + "extension_group": "transcriber", + "extension": "message_collector" + } + ] + } + ] + }, + { + "extension_group": "chatgpt", + "extension": "openai_chatgpt", + "data": [ + { + "name": "text_data", + "dest": [ + { + "extension_group": "tts", + "extension": "cartesia_tts" + }, + { + "extension_group": "transcriber", + "extension": "message_collector" + } + ] + } + ], + "cmd": [ + { + "name": "flush", + "dest": [ + { + "extension_group": "tts", + "extension": "cartesia_tts" + } + ] + } + ] + }, + { + "extension_group": "transcriber", + "extension": "message_collector", + "data": [ + { + "name": "data", + "dest": [ + { + "extension_group": "default", + "extension": "agora_rtc" + } + ] + } + ] + }, + { + "extension_group": "tts", + "extension": "cartesia_tts", + "audio_frame": [ + { + "name": "pcm_frame", + "dest": [ + { + "extension_group": "default", + "extension": "agora_rtc" + } + ] + } + ], + "cmd": [ + { + "name": "flush", + "dest": [ + { + "extension_group": "default", + "extension": "agora_rtc" + } + ] + } + ] + }, + { + "extension_group": "default", + "extension": "interrupt_detector", + "data": [ + { + "name": "text_data", + "dest": [ + { + "extension_group": "chatgpt", + "extension": "openai_chatgpt" + } + ] + } + ], + "cmd": [ + { + "name": "flush", + "dest": [ + { + "extension_group": "chatgpt", + "extension": "openai_chatgpt" + } + ] + } + ] + } + ] + }, + { + "name": "va_minimax_v2v", + "auto_start": false, + "nodes": [ + { + "type": "extension", + "extension_group": "rtc", + "addon": "agora_rtc", + "name": "agora_rtc", + "property": { + "app_id": "${env:AGORA_APP_ID}", + "token": "", + "channel": "ten_agent_test", + "stream_id": 1234, + "remote_stream_id": 123, + "subscribe_audio": true, + "publish_audio": true, + "publish_data": true + } + }, + { + "type": "extension", + "extension_group": "agora_sess_ctrl", + "addon": "agora_sess_ctrl", + "name": "agora_sess_ctrl", + "property": { + "wait_for_eos": true + } + }, + { + "type": "extension", + "extension_group": "llm", + "addon": "minimax_v2v_python", + "name": "minimax_v2v_python", + "property": { + "in_sample_rate": 16000, + "token": "${env:MINIMAX_TOKEN}" + } + }, + { + "type": "extension", + "extension_group": "message_collector", + "addon": "message_collector", + "name": "message_collector" + } + ], + "connections": [ + { + "extension_group": "rtc", + "extension": "agora_rtc", + "audio_frame": [ + { + "name": "pcm_frame", + "dest": [ + { + "extension_group": "agora_sess_ctrl", + "extension": "agora_sess_ctrl" + } + ] + } + ] + }, + { + "extension_group": "agora_sess_ctrl", + "extension": "agora_sess_ctrl", + "audio_frame": [ + { + "name": "pcm_frame", + "dest": [ + { + "extension_group": "llm", + "extension": "minimax_v2v_python" + } + ] + } + ], + "cmd": [ + { + "name": "start_of_sentence", + "dest": [ + { + "extension_group": "llm", + "extension": "minimax_v2v_python", + "msg_conversion": { + "type": "per_property", + "keep_original": true, + "rules": [ + { + "path": "_ten.name", + "conversion_mode": "fixed_value", + "value": "flush" + } + ] + } + } + ] + } + ] + }, + { + "extension_group": "llm", + "extension": "minimax_v2v_python", + "data": [ + { + "name": "text_data", + "dest": [ + { + "extension_group": "message_collector", + "extension": "message_collector" + } + ] + } + ], + "audio_frame": [ + { + "name": "pcm_frame", + "dest": [ + { + "extension_group": "rtc", + "extension": "agora_rtc" + } + ] + } + ], + "cmd": [ + { + "name": "flush", + "dest": [ + { + "extension_group": "rtc", + "extension": "agora_rtc" + } + ] + } + ] + }, + { + "extension_group": "message_collector", + "extension": "message_collector", + "data": [ + { + "name": "data", + "dest": [ + { + "extension_group": "rtc", + "extension": "agora_rtc" + } + ] + } + ] + } + ] } ] } diff --git a/agents/ten_packages/extension/cartesia_tts/__init__.py b/agents/ten_packages/extension/cartesia_tts/__init__.py new file mode 100644 index 00000000..f6bb8f4c --- /dev/null +++ b/agents/ten_packages/extension/cartesia_tts/__init__.py @@ -0,0 +1,6 @@ +from . import cartesia_tts_addon +from .extension import EXTENSION_NAME +from .log import logger + + +logger.info(f"{EXTENSION_NAME} extension loaded") diff --git a/agents/ten_packages/extension/cartesia_tts/cartesia_tts_addon.py b/agents/ten_packages/extension/cartesia_tts/cartesia_tts_addon.py new file mode 100644 index 00000000..f633d7e3 --- /dev/null +++ b/agents/ten_packages/extension/cartesia_tts/cartesia_tts_addon.py @@ -0,0 +1,24 @@ +# +# +# Agora Real Time Engagement +# Created by XinHui Li in 2024-07. +# Copyright (c) 2024 Agora IO. All rights reserved. +# +# + +from ten import ( + Addon, + register_addon_as_extension, + TenEnv, +) +from .extension import EXTENSION_NAME +from .log import logger + + +@register_addon_as_extension(EXTENSION_NAME) +class CartesiaTTSExtensionAddon(Addon): + def on_create_instance(self, ten: TenEnv, addon_name: str, context) -> None: + logger.info("on_create_instance") + from .cartesia_tts_extension import CartesiaTTSExtension + + ten.on_create_instance_done(CartesiaTTSExtension(addon_name), context) diff --git a/agents/ten_packages/extension/cartesia_tts/cartesia_tts_extension.py b/agents/ten_packages/extension/cartesia_tts/cartesia_tts_extension.py new file mode 100644 index 00000000..f18c9af3 --- /dev/null +++ b/agents/ten_packages/extension/cartesia_tts/cartesia_tts_extension.py @@ -0,0 +1,197 @@ +# cartesia_tts_extension.py + +import queue +import threading +from datetime import datetime +import asyncio +import re +from ten import ( + Extension, + TenEnv, + Cmd, + AudioFrameDataFmt, + AudioFrame, + Data, + StatusCode, + CmdResult, +) +from .cartesia_wrapper import CartesiaWrapper, CartesiaConfig, CartesiaError +from .log import logger + +class CartesiaCallback: + # Handles audio processing and interrupt checks + def __init__(self, ten: TenEnv, sample_rate: int, need_interrupt_callback): + self.ten = ten + self.sample_rate = sample_rate + self.need_interrupt_callback = need_interrupt_callback + self.ts = datetime.now() + + def set_input_ts(self, ts: datetime): + # Updates timestamp for the current input + self.ts = ts + + def need_interrupt(self) -> bool: + # Checks if current task should be interrupted + return self.need_interrupt_callback(self.ts) + + def create_audio_frame(self, audio_data): + # Creates an AudioFrame from raw audio data + frame = AudioFrame.create("pcm_frame") + frame.set_sample_rate(self.sample_rate) + frame.set_bytes_per_sample(2) # s16le is 2 bytes per sample + frame.set_number_of_channels(1) + frame.set_data_fmt(AudioFrameDataFmt.INTERLEAVE) + frame.set_samples_per_channel(len(audio_data) // 2) + frame.alloc_buf(len(audio_data)) + buff = frame.lock_buf() + buff[:] = audio_data + frame.unlock_buf(buff) + return frame + + def process_audio(self, audio_data): + # Processes audio data if not interrupted + if self.need_interrupt(): + return + audio_frame = self.create_audio_frame(audio_data) + self.ten.send_audio_frame(audio_frame) + +class CartesiaTTSExtension(Extension): + def __init__(self, name: str): + super().__init__(name) + self.cartesia = None + self.loop = None + self.queue = queue.Queue() + self.outdate_ts = datetime.now() + self.stopped = False + self.thread = None + self.callback = None + self.skip_patterns = [r'\bssml_\w+\b'] # List of patterns to skip + self.ten = None + + def on_start(self, ten: TenEnv) -> None: + self.ten = ten + try: + # Initialize Cartesia config and wrapper + cartesia_config = CartesiaConfig( + api_key=ten.get_property_string("api_key"), + model_id=ten.get_property_string("model_id"), + voice_id=ten.get_property_string("voice_id"), + sample_rate=int(ten.get_property_string("sample_rate")), + cartesia_version=ten.get_property_string("cartesia_version") + ) + self.cartesia = CartesiaWrapper(cartesia_config) + self.callback = CartesiaCallback(ten, cartesia_config.sample_rate, self.need_interrupt) + + # Set up asyncio event loop + self.loop = asyncio.new_event_loop() + asyncio.set_event_loop(self.loop) + + # Connect to Cartesia API + self.loop.run_until_complete(self.cartesia.connect()) + logger.info("Successfully connected to Cartesia API") + + # Start async handling thread + self.thread = threading.Thread(target=self.async_handle, args=[ten]) + self.thread.start() + + ten.on_start_done() + except Exception as e: + logger.error(f"Failed to start CartesiaTTSExtension: {e}") + ten.on_start_done() + + def on_stop(self, ten: TenEnv) -> None: + # Clean up resources and stop thread + self.stopped = True + self.flush() + self.queue.put(None) + if self.thread is not None: + self.thread.join() + self.thread = None + + if self.cartesia: + self.loop.run_until_complete(self.cartesia.close()) + if self.loop: + self.loop.close() + ten.on_stop_done() + + def need_interrupt(self, ts: datetime) -> bool: + # Check if task is outdated + return self.outdate_ts > ts + + def process_input_text(self, input_text: str) -> str: + # Process input text to remove parts that should be skipped + for pattern in self.skip_patterns: + input_text = re.sub(pattern, '', input_text, flags=re.IGNORECASE) + return input_text.strip() + + def create_pause_text(self, duration_ms: int) -> str: + # Create pause text + return f"PAUSE_{duration_ms}_MS" + + def on_data(self, ten: TenEnv, data: Data) -> None: + # Queue incoming text for processing + input_text = data.get_property_string("text") + if not input_text: + return + + # Handle the case of just a period or comma + if input_text.strip() in ['.', ',']: + pause_duration = 150 if input_text.strip() == '.' else 150 + pause_text = self.create_pause_text(pause_duration) + self.queue.put(("PAUSE", pause_text, datetime.now())) + return + + processed_text = self.process_input_text(input_text) + + if processed_text.strip(): + self.queue.put(("TEXT", processed_text, datetime.now())) + else: + logger.info("Processed text is empty. Skipping synthesis.") + + def async_handle(self, ten: TenEnv): + # Process queue items asynchronously + while not self.stopped: + try: + value = self.queue.get() + if value is None: + break + + item_type, content, ts = value + + self.callback.set_input_ts(ts) + + if self.callback.need_interrupt(): + logger.info("Drop outdated input") + continue + + try: + audio_data = self.loop.run_until_complete(self.cartesia.synthesize(content)) + self.callback.process_audio(audio_data) + except CartesiaError as e: + logger.error(f"Failed to synthesize: {str(e)}. Moving to next item.") + # Optionally, you could add some fallback behavior here, like playing an error sound + + except Exception as e: + logger.exception(f"Error in async_handle: {e}") + # Continue processing the next item instead of breaking the loop + + def on_cmd(self, ten: TenEnv, cmd: Cmd) -> None: + # Handle incoming commands + cmd_name = cmd.get_name() + + if cmd_name == "flush": + self.outdate_ts = datetime.now() + self.flush() + cmd_result = CmdResult.create(StatusCode.OK) + cmd_result.set_property_string("detail", "Flush command executed") + else: + logger.warning(f"Unknown command received: {cmd_name}") + cmd_result = CmdResult.create(StatusCode.ERROR) + cmd_result.set_property_string("detail", f"Unknown command: {cmd_name}") + + ten.return_result(cmd_result, cmd) + + def flush(self): + # Clear the queue + while not self.queue.empty(): + self.queue.get() diff --git a/agents/ten_packages/extension/cartesia_tts/cartesia_wrapper.py b/agents/ten_packages/extension/cartesia_tts/cartesia_wrapper.py new file mode 100644 index 00000000..37a3d7c4 --- /dev/null +++ b/agents/ten_packages/extension/cartesia_tts/cartesia_wrapper.py @@ -0,0 +1,112 @@ +# cartesia_wrapper.py + +import asyncio +import websockets +import json +import base64 +import logging +from urllib.parse import urlparse + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +class CartesiaError(Exception): + """Custom exception class for Cartesia-related errors.""" + pass + +class CartesiaConfig: + # Configuration class for Cartesia API + def __init__(self, api_key, model_id, voice_id, sample_rate, cartesia_version): + self.api_key = api_key + self.model_id = model_id + self.voice_id = voice_id + self.sample_rate = sample_rate + self.cartesia_version = cartesia_version + +class CartesiaWrapper: + # Wrapper class for Cartesia API interactions + def __init__(self, config: CartesiaConfig): + self.config = config + self.websocket = None + self.context_id = 0 + + async def connect(self): + # Establish WebSocket connection to Cartesia API + ws_url = f"wss://api.cartesia.ai/tts/websocket?api_key={self.config.api_key}&cartesia_version={self.config.cartesia_version}" + try: + self.websocket = await websockets.connect(ws_url) + logger.info("Connected to Cartesia WebSocket") + except Exception as e: + logger.error(f"Failed to connect to Cartesia API: {str(e)}") + raise CartesiaError(f"Connection failed: {str(e)}") + + async def synthesize(self, text: str): + # Synthesize speech from text using Cartesia API + if not self.websocket: + await self.connect() + + if text.startswith("PAUSE_"): + # Handle custom pause marker + try: + duration_ms = int(text.split("_")[1]) + return self.generate_silence(duration_ms) + except (IndexError, ValueError): + logger.error(f"Invalid pause format: {text}") + raise CartesiaError(f"Invalid pause format: {text}") + + self.context_id += 1 + request = { + "context_id": f"context_{self.context_id}", + "model_id": self.config.model_id, + "transcript": text, + "voice": {"mode": "id", "id": self.config.voice_id}, + "output_format": { + "container": "raw", + "encoding": "pcm_s16le", + "sample_rate": int(self.config.sample_rate) + }, + "language": "en", + "add_timestamps": False + } + + try: + # Send synthesis request + await self.websocket.send(json.dumps(request)) + + # Receive and process audio chunks + audio_data = bytearray() + while True: + response = await self.websocket.recv() + message = json.loads(response) + + if message['type'] == 'chunk': + chunk_data = base64.b64decode(message['data']) + audio_data.extend(chunk_data) + elif message['type'] == 'done': + break + elif message['type'] == 'error': + raise CartesiaError(f"Synthesis error: {message.get('error', 'Unknown error')}") + else: + logger.warning(f"Unknown message type: {message['type']}") + + return audio_data + except websockets.exceptions.ConnectionClosed: + # Handle connection errors and retry + logger.error("WebSocket connection closed unexpectedly. Attempting to reconnect...") + await self.connect() + return await self.synthesize(text) # Retry the synthesis after reconnecting + except Exception as e: + logger.error(f"Error during synthesis: {str(e)}") + raise CartesiaError(f"Synthesis failed: {str(e)}") + + def generate_silence(self, duration_ms: int) -> bytes: + # Generate silent audio data + sample_rate = self.config.sample_rate + num_samples = int(sample_rate * duration_ms / 1000) + return b"\x00" * (num_samples * 2) # Assuming 16-bit audio + + async def close(self): + # Close WebSocket connection + if self.websocket: + await self.websocket.close() + logger.info("Closed WebSocket connection to Cartesia API") diff --git a/agents/ten_packages/extension/cartesia_tts/extension.py b/agents/ten_packages/extension/cartesia_tts/extension.py new file mode 100644 index 00000000..4883c11c --- /dev/null +++ b/agents/ten_packages/extension/cartesia_tts/extension.py @@ -0,0 +1 @@ +EXTENSION_NAME = "cartesia_tts" diff --git a/agents/ten_packages/extension/cartesia_tts/log.py b/agents/ten_packages/extension/cartesia_tts/log.py new file mode 100644 index 00000000..fad21710 --- /dev/null +++ b/agents/ten_packages/extension/cartesia_tts/log.py @@ -0,0 +1,12 @@ +import logging +from .extension import EXTENSION_NAME + +logger = logging.getLogger(EXTENSION_NAME) +logger.setLevel(logging.INFO) + +formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(process)d - [%(filename)s:%(lineno)d] - %(message)s") + +console_handler = logging.StreamHandler() +console_handler.setFormatter(formatter) + +logger.addHandler(console_handler) diff --git a/agents/ten_packages/extension/cartesia_tts/manifest.json b/agents/ten_packages/extension/cartesia_tts/manifest.json new file mode 100644 index 00000000..24269a64 --- /dev/null +++ b/agents/ten_packages/extension/cartesia_tts/manifest.json @@ -0,0 +1,56 @@ +{ + "type": "extension", + "name": "cartesia_tts", + "version": "0.1.0", + "dependencies": [ + { + "type": "system", + "name": "ten_runtime_python", + "version": "0.3" + } + ], + "api": { + "property": { + "api_key": { + "type": "string" + }, + "cartesia_version": { + "type": "string" + }, + "model_id": { + "type": "string" + }, + "sample_rate": { + "type": "string" + }, + "voice_id": { + "type": "string" + } + }, + "data_in": [ + { + "name": "text_data", + "property": { + "text": { + "type": "string" + } + } + } + ], + "cmd_in": [ + { + "name": "flush" + } + ], + "cmd_out": [ + { + "name": "flush" + } + ], + "audio_frame_out": [ + { + "name": "pcm_frame" + } + ] + } +} diff --git a/agents/ten_packages/extension/cartesia_tts/property.json b/agents/ten_packages/extension/cartesia_tts/property.json new file mode 100644 index 00000000..9e26dfee --- /dev/null +++ b/agents/ten_packages/extension/cartesia_tts/property.json @@ -0,0 +1 @@ +{} \ No newline at end of file diff --git a/agents/ten_packages/extension/deepgram_asr_python/__init__.py b/agents/ten_packages/extension/deepgram_asr_python/__init__.py new file mode 100644 index 00000000..71578b73 --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/__init__.py @@ -0,0 +1,5 @@ +from . import deepgram_asr_addon +from .extension import EXTENSION_NAME +from .log import logger + +logger.info(f"{EXTENSION_NAME} extension loaded") diff --git a/agents/ten_packages/extension/deepgram_asr_python/deepgram_asr_addon.py b/agents/ten_packages/extension/deepgram_asr_python/deepgram_asr_addon.py new file mode 100644 index 00000000..8551dd87 --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/deepgram_asr_addon.py @@ -0,0 +1,14 @@ +from ten import ( + Addon, + register_addon_as_extension, + TenEnv, +) +from .extension import EXTENSION_NAME +from .log import logger +from .deepgram_asr_extension import DeepgramASRExtension + +@register_addon_as_extension(EXTENSION_NAME) +class DeepgramASRExtensionAddon(Addon): + def on_create_instance(self, ten: TenEnv, addon_name: str, context) -> None: + logger.info("on_create_instance") + ten.on_create_instance_done(DeepgramASRExtension(addon_name), context) diff --git a/agents/ten_packages/extension/deepgram_asr_python/deepgram_asr_extension.py b/agents/ten_packages/extension/deepgram_asr_python/deepgram_asr_extension.py new file mode 100644 index 00000000..fbaccbdc --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/deepgram_asr_extension.py @@ -0,0 +1,104 @@ +from ten import ( + Extension, + TenEnv, + Cmd, + AudioFrame, + StatusCode, + CmdResult, +) + +import asyncio +import threading + +from .log import logger +from .deepgram_wrapper import AsyncDeepgramWrapper, DeepgramConfig + +PROPERTY_API_KEY = "api_key" # Required +PROPERTY_LANG = "language" # Optional +PROPERTY_MODEL = "model" # Optional +PROPERTY_SAMPLE_RATE = "sample_rate" # Optional + + +class DeepgramASRExtension(Extension): + def __init__(self, name: str): + super().__init__(name) + + self.stopped = False + self.queue = asyncio.Queue(maxsize=3000) # about 3000 * 10ms = 30s input + self.deepgram = None + self.thread = None + + self.loop = asyncio.new_event_loop() + asyncio.set_event_loop(self.loop) + + def on_start(self, ten: TenEnv) -> None: + logger.info("on_start") + + deepgram_config = DeepgramConfig.default_config() + + try: + deepgram_config.api_key = ten.get_property_string(PROPERTY_API_KEY).strip() + except Exception as e: + logger.error(f"get property {PROPERTY_API_KEY} error: {e}") + return + + for optional_param in [ + PROPERTY_LANG, + PROPERTY_MODEL, + PROPERTY_SAMPLE_RATE, + ]: + try: + value = ten.get_property_string(optional_param).strip() + if value: + deepgram_config.__setattr__(optional_param, value) + except Exception as err: + logger.debug( + f"get property optional {optional_param} failed, err: {err}. Using default value: {deepgram_config.__getattribute__(optional_param)}" + ) + + self.deepgram = AsyncDeepgramWrapper( + deepgram_config, self.queue, ten, self.loop + ) + + logger.info("starting async_deepgram_wrapper thread") + self.thread = threading.Thread(target=self.deepgram.run, args=[]) + self.thread.start() + + ten.on_start_done() + + def put_pcm_frame(self, pcm_frame: AudioFrame) -> None: + try: + asyncio.run_coroutine_threadsafe( + self.queue.put(pcm_frame), self.loop + ).result(timeout=0.5) + except asyncio.QueueFull: + logger.exception("queue is full, dropping frame") + except Exception as e: + logger.exception(f"error putting frame in queue: {e}") + + def on_audio_frame(self, ten: TenEnv, frame: AudioFrame) -> None: + self.put_pcm_frame(pcm_frame=frame) + + def on_stop(self, ten: TenEnv) -> None: + logger.info("on_stop") + + # put an empty frame to stop deepgram_wrapper + self.put_pcm_frame(None) + self.stopped = True + self.thread.join() + self.loop.stop() + self.loop.close() + + ten.on_stop_done() + + def on_cmd(self, ten: TenEnv, cmd: Cmd) -> None: + logger.info("on_cmd") + cmd_json = cmd.to_json() + logger.info("on_cmd json: " + cmd_json) + + cmdName = cmd.get_name() + logger.info("got cmd %s" % cmdName) + + cmd_result = CmdResult.create(StatusCode.OK) + cmd_result.set_property_string("detail", "success") + ten.return_result(cmd_result, cmd) diff --git a/agents/ten_packages/extension/deepgram_asr_python/deepgram_config.py b/agents/ten_packages/extension/deepgram_asr_python/deepgram_config.py new file mode 100644 index 00000000..6fa5f16f --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/deepgram_config.py @@ -0,0 +1,26 @@ +from typing import Union + +class DeepgramConfig: + def __init__(self, + api_key: str, + language: str, + model: str, + sample_rate: Union[str, int]): + self.api_key = api_key + self.language = language + self.model = model + self.sample_rate = int(sample_rate) + + self.channels = 1 + self.encoding = 'linear16' + self.interim_results = True + self.punctuate = True + + @classmethod + def default_config(cls): + return cls( + api_key="", + language="en-US", + model="nova-2", + sample_rate=16000 + ) \ No newline at end of file diff --git a/agents/ten_packages/extension/deepgram_asr_python/deepgram_wrapper.py b/agents/ten_packages/extension/deepgram_asr_python/deepgram_wrapper.py new file mode 100644 index 00000000..fd74ae5d --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/deepgram_wrapper.py @@ -0,0 +1,128 @@ +import asyncio + +from ten import ( + TenEnv, + Data +) + +from deepgram import AsyncListenWebSocketClient, DeepgramClientOptions, LiveTranscriptionEvents, LiveOptions + +from .log import logger +from .deepgram_config import DeepgramConfig + +DATA_OUT_TEXT_DATA_PROPERTY_TEXT = "text" +DATA_OUT_TEXT_DATA_PROPERTY_IS_FINAL = "is_final" +DATA_OUT_TEXT_DATA_PROPERTY_STREAM_ID = "stream_id" +DATA_OUT_TEXT_DATA_PROPERTY_END_OF_SEGMENT = "end_of_segment" + +def create_and_send_data(ten: TenEnv, text_result: str, is_final: bool, stream_id: int): + stable_data = Data.create("text_data") + stable_data.set_property_bool(DATA_OUT_TEXT_DATA_PROPERTY_IS_FINAL, is_final) + stable_data.set_property_string(DATA_OUT_TEXT_DATA_PROPERTY_TEXT, text_result) + stable_data.set_property_int(DATA_OUT_TEXT_DATA_PROPERTY_STREAM_ID, stream_id) + stable_data.set_property_bool(DATA_OUT_TEXT_DATA_PROPERTY_END_OF_SEGMENT, is_final) + ten.send_data(stable_data) + + +class AsyncDeepgramWrapper(): + def __init__(self, config: DeepgramConfig, queue: asyncio.Queue, ten:TenEnv, loop: asyncio.BaseEventLoop): + self.queue = queue + self.ten = ten + self.stopped = False + self.config = config + self.loop = loop + self.stream_id = 0 + + logger.info(f"init deepgram client with api key: {config.api_key[:5]}") + self.deepgram_client = AsyncListenWebSocketClient(config=DeepgramClientOptions( + api_key=config.api_key, + options={"keepalive": "true"} + )) + + asyncio.set_event_loop(self.loop) + self.loop.create_task(self.start_listen(ten)) + + async def start_listen(self, ten:TenEnv) -> None: + logger.info(f"start and listen deepgram") + + super = self + + async def on_open(self, open, **kwargs): + logger.info(f"deepgram event callback on_open: {open}") + + async def on_close(self, close, **kwargs): + logger.info(f"deepgram event callback on_close: {close}") + + async def on_message(self, result, **kwargs): + sentence = result.channel.alternatives[0].transcript + + if len(sentence) == 0: + return + + is_final = result.is_final + logger.info(f"deepgram got sentence: [{sentence}], is_final: {is_final}, stream_id: {super.stream_id}") + + create_and_send_data(ten=ten, text_result=sentence, is_final=is_final, stream_id=super.stream_id) + + async def on_error(self, error, **kwargs): + logger.error(f"deepgram event callback on_error: {error}") + + self.deepgram_client.on(LiveTranscriptionEvents.Open, on_open) + self.deepgram_client.on(LiveTranscriptionEvents.Close, on_close) + self.deepgram_client.on(LiveTranscriptionEvents.Transcript, on_message) + self.deepgram_client.on(LiveTranscriptionEvents.Error, on_error) + + options = LiveOptions(language=self.config.language, + model=self.config.model, + sample_rate=self.config.sample_rate, + channels=self.config.channels, + encoding=self.config.encoding, + interim_results=self.config.interim_results, + punctuate=self.config.punctuate) + # connect to websocket + if await self.deepgram_client.start(options) is False: + logger.error(f"failed to connect to deepgram") + return + + logger.info(f"successfully connected to deepgram") + + async def send_frame(self) -> None: + while not self.stopped: + try: + pcm_frame = await asyncio.wait_for(self.queue.get(), timeout=10.0) + + if pcm_frame is None: + logger.warning("send_frame: exit due to None value got.") + return + + frame_buf = pcm_frame.get_buf() + if not frame_buf: + logger.warning("send_frame: empty pcm_frame detected.") + continue + + self.stream_id = pcm_frame.get_property_int('stream_id') + await self.deepgram_client.send(frame_buf) + self.queue.task_done() + except asyncio.TimeoutError as e: + logger.exception(f"error in send_frame: {e}") + except IOError as e: + logger.exception(f"error in send_frame: {e}") + except Exception as e: + logger.exception(f"error in send_frame: {e}") + raise e + + logger.info("send_frame: exit due to self.stopped == True") + + async def deepgram_loop(self) -> None: + try: + await self.send_frame() + except Exception as e: + logger.exception(e) + + def run(self) -> None: + self.loop.run_until_complete(self.deepgram_loop()) + self.loop.close() + logger.info("async_deepgram_wrapper: thread completed.") + + def stop(self) -> None: + self.stopped = True diff --git a/agents/ten_packages/extension/deepgram_asr_python/extension.py b/agents/ten_packages/extension/deepgram_asr_python/extension.py new file mode 100644 index 00000000..43c52445 --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/extension.py @@ -0,0 +1,3 @@ +# extension.py + +EXTENSION_NAME = "deepgram_asr_python" diff --git a/agents/ten_packages/extension/deepgram_asr_python/log.py b/agents/ten_packages/extension/deepgram_asr_python/log.py new file mode 100644 index 00000000..88a2cb1c --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/log.py @@ -0,0 +1,16 @@ +# log.py + +import logging +from .extension import EXTENSION_NAME + +logger = logging.getLogger(EXTENSION_NAME) +logger.setLevel(logging.INFO) + +formatter = logging.Formatter( + "%(asctime)s - %(name)s - %(levelname)s - %(process)d - [%(filename)s:%(lineno)d] - %(message)s" +) + +console_handler = logging.StreamHandler() +console_handler.setFormatter(formatter) + +logger.addHandler(console_handler) diff --git a/agents/ten_packages/extension/deepgram_asr_python/manifest.json b/agents/ten_packages/extension/deepgram_asr_python/manifest.json new file mode 100644 index 00000000..5875da2d --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/manifest.json @@ -0,0 +1,87 @@ +{ + "type": "extension", + "name": "deepgram_asr_python", + "version": "0.1.0", + "dependencies": [ + { + "type": "system", + "name": "ten_runtime_python", + "version": "0.3" + } + ], + "api": { + "property": { + "api_key": { + "type": "string" + }, + "model": { + "type": "string" + }, + "language": { + "type": "string" + }, + "sample_rate": { + "type": "string" + } + }, + "audio_frame_in": [ + { + "name": "pcm_frame" + } + ], + "cmd_in": [ + { + "name": "on_user_joined", + "property": { + "user_id": { + "type": "string" + } + } + }, + { + "name": "on_user_left", + "property": { + "user_id": { + "type": "string" + } + } + }, + { + "name": "on_connection_failure", + "property": { + "error": { + "type": "string" + } + } + } + ], + "data_out": [ + { + "name": "text_data", + "property": { + "time": { + "type": "int64" + }, + "duration_ms": { + "type": "int64" + }, + "language": { + "type": "string" + }, + "text": { + "type": "string" + }, + "is_final": { + "type": "bool" + }, + "stream_id": { + "type": "uint32" + }, + "end_of_segment": { + "type": "bool" + } + } + } + ] + } +} diff --git a/agents/ten_packages/extension/deepgram_asr_python/property.json b/agents/ten_packages/extension/deepgram_asr_python/property.json new file mode 100644 index 00000000..9e26dfee --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/property.json @@ -0,0 +1 @@ +{} \ No newline at end of file diff --git a/agents/ten_packages/extension/deepgram_asr_python/requirements.txt b/agents/ten_packages/extension/deepgram_asr_python/requirements.txt new file mode 100644 index 00000000..8c9fa1e8 --- /dev/null +++ b/agents/ten_packages/extension/deepgram_asr_python/requirements.txt @@ -0,0 +1 @@ +deepgram-sdk==3.7.5 \ No newline at end of file diff --git a/agents/ten_packages/extension/message_collector/src/extension.py b/agents/ten_packages/extension/message_collector/src/extension.py index 7013c432..f54592b0 100644 --- a/agents/ten_packages/extension/message_collector/src/extension.py +++ b/agents/ten_packages/extension/message_collector/src/extension.py @@ -5,7 +5,9 @@ # Copyright (c) 2024 Agora IO. All rights reserved. # # +import base64 import json +import threading import time import uuid from ten import ( @@ -18,6 +20,7 @@ CmdResult, Data, ) +import asyncio from .log import logger MAX_SIZE = 800 # 1 KB limit @@ -32,9 +35,70 @@ # record the cached text data for each stream id cached_text_map = {} +MAX_CHUNK_SIZE_BYTES = 1024 + +def _text_to_base64_chunks(text: str, msg_id: str) -> list: + # Ensure msg_id does not exceed 50 characters + if len(msg_id) > 36: + raise ValueError("msg_id cannot exceed 36 characters.") + + # Convert text to bytearray + byte_array = bytearray(text, 'utf-8') + + # Encode the bytearray into base64 + base64_encoded = base64.b64encode(byte_array).decode('utf-8') + + # Initialize list to hold the final chunks + chunks = [] + + # We'll split the base64 string dynamically based on the final byte size + part_index = 0 + total_parts = None # We'll calculate total parts once we know how many chunks we create + + # Process the base64-encoded content in chunks + current_position = 0 + total_length = len(base64_encoded) + + while current_position < total_length: + part_index += 1 + + # Start guessing the chunk size by limiting the base64 content part + estimated_chunk_size = MAX_CHUNK_SIZE_BYTES # We'll reduce this dynamically + content_chunk = "" + count = 0 + while True: + # Create the content part of the chunk + content_chunk = base64_encoded[current_position:current_position + estimated_chunk_size] + + # Format the chunk + formatted_chunk = f"{msg_id}|{part_index}|{total_parts if total_parts else '???'}|{content_chunk}" + + # Check if the byte length of the formatted chunk exceeds the max allowed size + if len(bytearray(formatted_chunk, 'utf-8')) <= MAX_CHUNK_SIZE_BYTES: + break + else: + # Reduce the estimated chunk size if the formatted chunk is too large + estimated_chunk_size -= 100 # Reduce content size gradually + count += 1 + + logger.debug(f"chunk estimate guess: {count}") + + # Add the current chunk to the list + chunks.append(formatted_chunk) + current_position += estimated_chunk_size # Move to the next part of the content + # Now that we know the total number of parts, update the chunks with correct total_parts + total_parts = len(chunks) + updated_chunks = [ + chunk.replace("???", str(total_parts)) for chunk in chunks + ] + + return updated_chunks class MessageCollectorExtension(Extension): + # Create the queue for message processing + queue = asyncio.Queue() + def on_init(self, ten_env: TenEnv) -> None: logger.info("MessageCollectorExtension on_init") ten_env.on_init_done() @@ -43,6 +107,13 @@ def on_start(self, ten_env: TenEnv) -> None: logger.info("MessageCollectorExtension on_start") # TODO: read properties, initialize resources + self.loop = asyncio.new_event_loop() + def start_loop(): + asyncio.set_event_loop(self.loop) + self.loop.run_forever() + threading.Thread(target=start_loop, args=[]).start() + + self.loop.create_task(self._process_queue(ten_env)) ten_env.on_start_done() @@ -74,7 +145,7 @@ def on_data(self, ten_env: TenEnv, data: Data) -> None: example: {"name": "text_data", "properties": {"text": "hello", "is_final": true, "stream_id": 123, "end_of_segment": true}} """ - logger.info(f"on_data") + logger.debug(f"on_data") text = "" final = True @@ -123,7 +194,7 @@ def on_data(self, ten_env: TenEnv, data: Data) -> None: cached_text_map[stream_id] = text # Generate a unique message ID for this batch of parts - message_id = str(uuid.uuid4()) + message_id = str(uuid.uuid4())[:8] # Prepare the main JSON structure without the text field base_msg_data = { @@ -132,61 +203,13 @@ def on_data(self, ten_env: TenEnv, data: Data) -> None: "message_id": message_id, # Add message_id to identify the split message "data_type": "transcribe", "text_ts": int(time.time() * 1000), # Convert to milliseconds + "text": text, } try: - # Convert the text to UTF-8 bytes - text_bytes = text.encode('utf-8') - - # If the text + metadata fits within the size limit, send it directly - if len(text_bytes) + OVERHEAD_ESTIMATE <= MAX_SIZE: - base_msg_data["text"] = text - msg_data = json.dumps(base_msg_data) - ten_data = Data.create("data") - ten_data.set_property_buf("data", msg_data.encode()) - ten_env.send_data(ten_data) - else: - # Split the text bytes into smaller chunks, ensuring safe UTF-8 splitting - max_text_size = MAX_SIZE - OVERHEAD_ESTIMATE - total_length = len(text_bytes) - total_parts = (total_length + max_text_size - 1) // max_text_size # Calculate number of parts - - def get_valid_utf8_chunk(start, end): - """Helper function to ensure valid UTF-8 chunks.""" - while end > start: - try: - # Decode to check if this chunk is valid UTF-8 - text_part = text_bytes[start:end].decode('utf-8') - return text_part, end - except UnicodeDecodeError: - # Reduce the end point to avoid splitting in the middle of a character - end -= 1 - # If no valid chunk is found (shouldn't happen with valid UTF-8 input), return an empty string - return "", start - - part_number = 0 - start_index = 0 - while start_index < total_length: - part_number += 1 - # Get a valid UTF-8 chunk - text_part, end_index = get_valid_utf8_chunk(start_index, min(start_index + max_text_size, total_length)) - - # Prepare the part data with metadata - part_data = base_msg_data.copy() - part_data.update({ - "text": text_part, - "part_number": part_number, - "total_parts": total_parts, - }) - - # Send each part - part_msg_data = json.dumps(part_data) - ten_data = Data.create("data") - ten_data.set_property_buf("data", part_msg_data.encode()) - ten_env.send_data(ten_data) - - # Move to the next chunk - start_index = end_index + chunks = _text_to_base64_chunks(json.dumps(base_msg_data), message_id) + for chunk in chunks: + asyncio.run_coroutine_threadsafe(self._queue_message(chunk), self.loop) except Exception as e: logger.warning(f"on_data new_data error: {e}") @@ -199,3 +222,19 @@ def on_audio_frame(self, ten_env: TenEnv, audio_frame: AudioFrame) -> None: def on_video_frame(self, ten_env: TenEnv, video_frame: VideoFrame) -> None: # TODO: process image frame pass + + + async def _queue_message(self, data: str): + await self.queue.put(data) + + async def _process_queue(self, ten_env: TenEnv): + while True: + data = await self.queue.get() + if data is None: + break + # process data + ten_data = Data.create("data") + ten_data.set_property_buf("data", data.encode()) + ten_env.send_data(ten_data) + self.queue.task_done() + await asyncio.sleep(0.04) \ No newline at end of file diff --git a/agents/ten_packages/extension/minimax_v2v_python/README.md b/agents/ten_packages/extension/minimax_v2v_python/README.md new file mode 100644 index 00000000..c73a53c1 --- /dev/null +++ b/agents/ten_packages/extension/minimax_v2v_python/README.md @@ -0,0 +1,36 @@ +# MiniMax Voice-to-Voice Extension + +A TEN extension that implements voice-to-voice conversation capabilities using MiniMax's API services. + +## Features + +- Real-time voice-to-voice conversation +- Support for streaming responses including assistant's voice, assisntant's transcript, and user's transcript +- Configurable voice settings +- Memory management for conversation context +- Asynchronous processing based on asyncio + + +## API + +Refer to `api` definition in [manifest.json] and default values in [property.json](property.json). +`token` is mandatory to use MiniMax's API, others are optional. + + + +## Development + +### Build + + + +### Unit test + + + +## Misc + + + +## References +- [ChatCompletion v2](https://platform.minimaxi.com/document/ChatCompletion%20v2?key=66701d281d57f38758d581d0#ww1u9KZvwrgnF2EfpPrnHHGd) diff --git a/agents/ten_packages/extension/minimax_v2v_python/__init__.py b/agents/ten_packages/extension/minimax_v2v_python/__init__.py new file mode 100644 index 00000000..72593ab2 --- /dev/null +++ b/agents/ten_packages/extension/minimax_v2v_python/__init__.py @@ -0,0 +1,6 @@ +# +# This file is part of TEN Framework, an open source project. +# Licensed under the Apache License, Version 2.0. +# See the LICENSE file for more information. +# +from . import addon diff --git a/agents/ten_packages/extension/minimax_v2v_python/addon.py b/agents/ten_packages/extension/minimax_v2v_python/addon.py new file mode 100644 index 00000000..f6c0f882 --- /dev/null +++ b/agents/ten_packages/extension/minimax_v2v_python/addon.py @@ -0,0 +1,18 @@ +# +# This file is part of TEN Framework, an open source project. +# Licensed under the Apache License, Version 2.0. +# See the LICENSE file for more information. +# +from ten import ( + Addon, + register_addon_as_extension, + TenEnv, +) +from .extension import MinimaxV2VExtension + + +@register_addon_as_extension("minimax_v2v_python") +class MinimaxV2VExtensionAddon(Addon): + def on_create_instance(self, ten_env: TenEnv, name: str, context) -> None: + ten_env.log_info("on_create_instance") + ten_env.on_create_instance_done(MinimaxV2VExtension(name), context) diff --git a/agents/ten_packages/extension/minimax_v2v_python/chat_memory.py b/agents/ten_packages/extension/minimax_v2v_python/chat_memory.py new file mode 100644 index 00000000..8ef98b10 --- /dev/null +++ b/agents/ten_packages/extension/minimax_v2v_python/chat_memory.py @@ -0,0 +1,42 @@ +# +# This file is part of TEN Framework, an open source project. +# Licensed under the Apache License, Version 2.0. +# See the LICENSE file for more information. +# +import threading + + +class ChatMemory: + def __init__(self, max_history_length): + self.max_history_length = max_history_length + self.history = [] + self.mutex = threading.Lock() # TODO: no need lock for asyncio + + def put(self, message): + with self.mutex: + self.history.append(message) + + while True: + history_count = len(self.history) + if history_count > 0 and history_count > self.max_history_length: + self.history.pop(0) + continue + if history_count > 0 and self.history[0]["role"] == "assistant": + # we cannot have an assistant message at the start of the chat history + # if after removal of the first, we have an assistant message, + # we need to remove the assistant message too + self.history.pop(0) + continue + break + + def get(self): + with self.mutex: + return self.history + + def count(self): + with self.mutex: + return len(self.history) + + def clear(self): + with self.mutex: + self.history = [] diff --git a/agents/ten_packages/extension/minimax_v2v_python/extension.py b/agents/ten_packages/extension/minimax_v2v_python/extension.py new file mode 100644 index 00000000..64143b69 --- /dev/null +++ b/agents/ten_packages/extension/minimax_v2v_python/extension.py @@ -0,0 +1,484 @@ +# +# This file is part of TEN Framework, an open source project. +# Licensed under the Apache License, Version 2.0. +# See the LICENSE file for more information. +# +from ten import ( + AudioFrame, + VideoFrame, + AudioFrameDataFmt, + AsyncExtension, + AsyncTenEnv, + Cmd, + StatusCode, + CmdResult, + Data, +) +from .util import duration_in_ms, duration_in_ms_since, Role +from .chat_memory import ChatMemory +from dataclasses import dataclass, fields +import builtins +import httpx +from datetime import datetime +import aiofiles +import asyncio +from typing import Iterator, List, Dict, Tuple, Any +import base64 +import json + + +@dataclass +class MinimaxV2VConfig: + token: str = "" + max_tokens: int = 1024 + model: str = "abab6.5s-chat" + voice_model: str = "speech-01-turbo-240228" + voice_id: str = "female-tianmei" + in_sample_rate: int = 16000 + out_sample_rate: int = 32000 + prompt: str = ( + "You are a voice assistant who talks in a conversational way and can chat with me like my friends. I will speak to you in English or Chinese, and you will answer in the corrected and improved version of my text with the language I use. Don’t talk like a robot, instead I would like you to talk like a real human with emotions. I will use your answer for text-to-speech, so don’t return me any meaningless characters. I want you to be helpful, when I’m asking you for advice, give me precise, practical and useful advice instead of being vague. When giving me a list of options, express the options in a narrative way instead of bullet points." + ) + greeting: str = "" + max_memory_length: int = 10 + dump: bool = False + + def read_from_property(self, ten_env: AsyncTenEnv): + for field in fields(self): + # TODO: 'is_property_exist' has a bug that can not be used in async extension currently, use it instead of try .. except once fixed + # if not ten_env.is_property_exist(field.name): + # continue + try: + match field.type: + case builtins.str: + val = ten_env.get_property_string(field.name) + if val: + setattr(self, field.name, val) + ten_env.log_info(f"{field.name}={val}") + case builtins.int: + val = ten_env.get_property_int(field.name) + setattr(self, field.name, val) + ten_env.log_info(f"{field.name}={val}") + case builtins.bool: + val = ten_env.get_property_bool(field.name) + setattr(self, field.name, val) + ten_env.log_info(f"{field.name}={val}") + case _: + pass + except Exception as e: + ten_env.log_warn(f"get property for {field.name} failed, err {e}") + + +class MinimaxV2VExtension(AsyncExtension): + def __init__(self, name: str) -> None: + super().__init__(name) + + self.config = MinimaxV2VConfig() + self.client = httpx.AsyncClient(timeout=httpx.Timeout(5)) + self.memory = ChatMemory(self.config.max_memory_length) + self.remote_stream_id = 0 + self.ten_env = None + + # able to cancel + self.curr_task = None + + # make sure tasks processing in order + self.process_input_task = None + self.queue = asyncio.Queue() + + async def on_init(self, ten_env: AsyncTenEnv) -> None: + self.config.read_from_property(ten_env=ten_env) + ten_env.log_info(f"config: {self.config}") + + self.memory = ChatMemory(self.config.max_memory_length) + self.ten_env = ten_env + ten_env.on_init_done() + + async def on_start(self, ten_env: AsyncTenEnv) -> None: + self.process_input_task = asyncio.create_task(self._process_input(ten_env=ten_env, queue=self.queue), name="process_input") + + ten_env.on_start_done() + + async def on_stop(self, ten_env: AsyncTenEnv) -> None: + + await self._flush(ten_env=ten_env) + self.queue.put_nowait(None) + if self.process_input_task: + self.process_input_task.cancel() + await asyncio.gather(self.process_input_task, return_exceptions=True) + self.process_input_task = None + + ten_env.on_stop_done() + + async def on_deinit(self, ten_env: AsyncTenEnv) -> None: + ten_env.log_debug("on_deinit") + + if self.client: + await self.client.aclose() + self.client = None + self.ten_env = None + ten_env.on_deinit_done() + + async def on_cmd(self, ten_env: AsyncTenEnv, cmd: Cmd) -> None: + try: + cmd_name = cmd.get_name() + ten_env.log_debug("on_cmd name {}".format(cmd_name)) + + # process cmd + match cmd_name: + case "flush": + await self._flush(ten_env=ten_env) + _result = await ten_env.send_cmd(Cmd.create("flush")) + ten_env.log_debug("flush done") + case _: + pass + ten_env.return_result(CmdResult.create(StatusCode.OK), cmd) + except asyncio.CancelledError: + ten_env.log_warn(f"cmd {cmd_name} cancelled") + ten_env.return_result(CmdResult.create(StatusCode.ERROR), cmd) + raise + except Exception as e: + ten_env.log_warn(f"cmd {cmd_name} failed, err {e}") + finally: + pass + + async def on_data(self, ten_env: AsyncTenEnv, data: Data) -> None: + pass + + async def on_audio_frame( + self, ten_env: AsyncTenEnv, audio_frame: AudioFrame + ) -> None: + + try: + ts = datetime.now() + stream_id = audio_frame.get_property_int("stream_id") + if not self.remote_stream_id: + self.remote_stream_id = stream_id + + frame_buf = audio_frame.get_buf() + ten_env.log_debug(f"on audio frame {len(frame_buf)} {stream_id}") + + # process audio frame, must be after vad + # put_nowait to make sure put in_order + self.queue.put_nowait((ts, frame_buf)) + # await self._complete_with_history(ts, frame_buf) + + # dump input audio if need + await self._dump_audio_if_need(frame_buf, "in") + + # ten_env.log_debug(f"on audio frame {len(frame_buf)} {stream_id} put done") + except asyncio.CancelledError: + ten_env.log_warn(f"on audio frame cancelled") + raise + except Exception as e: + ten_env.log_error(f"on audio frame failed, err {e}") + + async def on_video_frame( + self, ten_env: AsyncTenEnv, video_frame: VideoFrame + ) -> None: + pass + + async def _process_input(self, ten_env: AsyncTenEnv, queue: asyncio.Queue): + ten_env.log_info("process_input started") + + while True: + item = await queue.get() + if not item: + break + + (ts, frame_buf) = item + ten_env.log_debug(f"start process task {ts} {len(frame_buf)}") + + try: + self.curr_task = asyncio.create_task(self._complete_with_history(ts, frame_buf)) + await self.curr_task + self.curr_task = None + except asyncio.CancelledError: + ten_env.log_warn("task cancelled") + except Exception as e: + ten_env.log_warn(f"task failed, err {e}") + finally: + queue.task_done() + + ten_env.log_info("process_input exit") + + async def _complete_with_history( + self, ts: datetime, buff: bytearray + ): + start_time = datetime.now() + ten_env = self.ten_env + ten_env.log_debug( + f"start request, buff len {len(buff)}, queued_time {duration_in_ms(ts, start_time)}ms" + ) + + # prepare messages with prompt and history + messages = [] + if self.config.prompt: + messages.append({"role": Role.System, "content": self.config.prompt}) + messages.extend(self.memory.get()) + ten_env.log_debug(f"messages without audio: [{messages}]") + messages.append( + self._create_input_audio_message(buff=buff) + ) # don't print audio message + + # prepare request + url = "https://api.minimax.chat/v1/text/chatcompletion_v2" + (headers, payload) = self._create_request(messages) + + # vars to calculate Time to first byte + user_transcript_ttfb = None + assistant_transcript_ttfb = None + assistant_audio_ttfb = None + + # vars for transcript + user_transcript = "" + assistant_transcript = "" + + try: + # send POST request + async with self.client.stream( + "POST", url, headers=headers, json=payload + ) as response: + trace_id = response.headers.get("Trace-Id", "") + alb_receive_time = response.headers.get("alb_receive_time", "") + ten_env.log_info( + f"Get response trace-id: {trace_id}, alb_receive_time: {alb_receive_time}, cost_time {duration_in_ms_since(start_time)}ms" + ) + + response.raise_for_status() # check response + + i = 0 + async for line in response.aiter_lines(): + # logger.info(f"-> line {line}") + # if self._need_interrupt(ts): + # ten_env.log_warn(f"trace-id: {trace_id}, interrupted") + # if self.transcript: + # self.transcript += "[interrupted]" + # self._append_message("assistant", self.transcript) + # self._send_transcript("", "assistant", True) + # break + + if not line.startswith("data:"): + ten_env.log_debug(f"ignore line {len(line)}") + continue + i += 1 + + resp = json.loads(line.strip("data:")) + if resp.get("choices") and resp["choices"][0].get("delta"): + delta = resp["choices"][0]["delta"] + if delta.get("role") == "assistant": + # text content + if delta.get("content"): + content = delta["content"] + assistant_transcript += content + if not assistant_transcript_ttfb: + assistant_transcript_ttfb = duration_in_ms_since( + start_time + ) + ten_env.log_info( + f"trace-id {trace_id} chunck-{i} get assistant_transcript_ttfb {assistant_transcript_ttfb}ms, assistant transcript [{content}]" + ) + else: + ten_env.log_info( + f"trace-id {trace_id} chunck-{i} get assistant transcript [{content}]" + ) + + # send out for transcript display + self._send_transcript( + ten_env=ten_env, + content=content, + role=Role.Assistant, + end_of_segment=False, + ) + + # audio content + if ( + delta.get("audio_content") + and delta["audio_content"] != "" + ): + ten_env.log_info( + f"trace-id {trace_id} chunck-{i} get audio_content" + ) + if not assistant_audio_ttfb: + assistant_audio_ttfb = duration_in_ms_since( + start_time + ) + ten_env.log_info( + f"trace-id {trace_id} chunck-{i} get assistant_audio_ttfb {assistant_audio_ttfb}ms" + ) + + # send out + base64_str = delta["audio_content"] + buff = base64.b64decode(base64_str) + await self._dump_audio_if_need(buff, "out") + self._send_audio_frame(ten_env=ten_env, audio_data=buff) + + # tool calls + if delta.get("tool_calls"): + ten_env.log_warn(f"ignore tool call {delta}") + # TODO: add tool calls + continue + + if delta.get("role") == "user": + if delta.get("content"): + content = delta["content"] + user_transcript += content + if not user_transcript_ttfb: + user_transcript_ttfb = duration_in_ms_since( + start_time + ) + ten_env.log_info( + f"trace-id: {trace_id} chunck-{i} get user_transcript_ttfb {user_transcript_ttfb}ms, user transcript [{content}]" + ) + else: + ten_env.log_info( + f"trace-id {trace_id} chunck-{i} get user transcript [{content}]" + ) + + # send out for transcript display + self._send_transcript( + ten_env=ten_env, + content=content, + role=Role.User, + end_of_segment=True, + ) + + except httpx.TimeoutException: + ten_env.log_warn("http timeout") + except httpx.HTTPStatusError as e: + ten_env.log_warn(f"http status error: {e}") + except httpx.RequestError as e: + ten_env.log_warn(f"http request error: {e}") + finally: + ten_env.log_info( + f"http loop done, cost_time {duration_in_ms_since(start_time)}ms" + ) + if user_transcript: + self.memory.put({"role": Role.User, "content": user_transcript}) + if assistant_transcript: + self.memory.put( + {"role": Role.Assistant, "content": assistant_transcript} + ) + self._send_transcript( + ten_env=ten_env, + content="", + role=Role.Assistant, + end_of_segment=True, + ) + + def _create_input_audio_message(self, buff: bytearray) -> Dict[str, Any]: + message = { + "role": "user", + "content": [ + { + "type": "input_audio", + "input_audio": { + "data": base64.b64encode(buff).decode("utf-8"), + "format": "pcm", + "sample_rate": self.config.in_sample_rate, + "bit_depth": 16, + "channel": 1, + "encode": "base64", + }, + } + ], + } + return message + + def _create_request( + self, messages: List[Any] + ) -> Tuple[Dict[str, Any], Dict[str, Any]]: + config = self.config + + headers = { + "Authorization": f"Bearer {config.token}", + "Content-Type": "application/json", + } + + payload = { + "model": config.model, + "messages": messages, + "tools": [], + "tool_choice": "none", + "stream": True, + "stream_options": {"speech_output": True}, # 开启语音输出 + "voice_setting": { + "model": config.voice_model, + "voice_id": config.voice_id, + }, + "audio_setting": { + "sample_rate": config.out_sample_rate, + "format": "pcm", + "channel": 1, + "encode": "base64", + }, + "tools": [{"type": "web_search"}], + "max_tokens": config.max_tokens, + "temperature": 0.8, + "top_p": 0.95, + } + + return (headers, payload) + + def _send_audio_frame(self, ten_env: AsyncTenEnv, audio_data: bytearray) -> None: + try: + f = AudioFrame.create("pcm_frame") + f.set_sample_rate(self.config.out_sample_rate) + f.set_bytes_per_sample(2) + f.set_number_of_channels(1) + f.set_data_fmt(AudioFrameDataFmt.INTERLEAVE) + f.set_samples_per_channel(len(audio_data) // 2) + f.alloc_buf(len(audio_data)) + buff = f.lock_buf() + buff[:] = audio_data + f.unlock_buf(buff) + ten_env.send_audio_frame(f) + except Exception as e: + ten_env.log_error(f"send audio frame failed, err {e}") + + def _send_transcript( + self, + ten_env: AsyncTenEnv, + content: str, + role: str, + end_of_segment: bool, + ) -> None: + stream_id = self.remote_stream_id if role == "user" else 0 + + try: + d = Data.create("text_data") + d.set_property_string("text", content) + d.set_property_bool("is_final", True) + d.set_property_bool("end_of_segment", end_of_segment) + d.set_property_string("role", role) + d.set_property_int("stream_id", stream_id) + ten_env.log_info( + f"send transcript text [{content}] {stream_id} end_of_segment {end_of_segment} role {role}" + ) + self.ten_env.send_data(d) + except Exception as e: + ten_env.log_warn( + f"send transcript text [{content}] {stream_id} end_of_segment {end_of_segment} role {role} failed, err {e}" + ) + + async def _flush(self, ten_env: AsyncTenEnv) -> None: + # clear queue + while not self.queue.empty(): + try: + self.queue.get_nowait() + self.queue.task_done() + except Exception as e: + ten_env.log_warn("flush queue error {e}") + + # cancel current task + if self.curr_task: + self.curr_task.cancel() + await asyncio.gather(self.curr_task, return_exceptions=True) + self.curr_task = None + + async def _dump_audio_if_need(self, buf: bytearray, suffix: str) -> None: + if not self.config.dump: + return + + async with aiofiles.open(f"minimax_v2v_{suffix}.pcm", "ab") as f: + await f.write(buf) diff --git a/agents/ten_packages/extension/minimax_v2v_python/manifest.json b/agents/ten_packages/extension/minimax_v2v_python/manifest.json new file mode 100644 index 00000000..23b4432b --- /dev/null +++ b/agents/ten_packages/extension/minimax_v2v_python/manifest.json @@ -0,0 +1,104 @@ +{ + "type": "extension", + "name": "minimax_v2v_python", + "version": "0.1.0", + "dependencies": [ + { + "type": "system", + "name": "ten_runtime_python", + "version": "0.3.1" + } + ], + "package": { + "include": [ + "manifest.json", + "property.json", + "**.py", + "README.md" + ] + }, + "api": { + "property": { + "token": { + "type": "string" + }, + "max_tokens": { + "type": "int32" + }, + "model": { + "type": "string" + }, + "voice_model": { + "type": "string" + }, + "voice_id": { + "type": "string" + }, + "in_sample_rate": { + "type": "int32" + }, + "out_sample_rate": { + "type": "int32" + }, + "prompt": { + "type": "string" + }, + "greeting": { + "type": "string" + }, + "max_memory_length": { + "type": "int32" + }, + "dump": { + "type": "bool" + } + }, + "cmd_in": [ + { + "name": "flush" + } + ], + "cmd_out": [ + { + "name": "flush" + } + ], + "audio_frame_in": [ + { + "name": "pcm_frame", + "property": { + "stream_id": { + "type": "uint32" + } + } + } + ], + "audio_frame_out": [ + { + "name": "pcm_frame" + } + ], + "data_out": [ + { + "name": "text_data", + "property": { + "text": { + "type": "string" + }, + "is_final": { + "type": "bool" + }, + "end_of_segment": { + "type": "bool" + }, + "role": { + "type": "string" + }, + "stream_id": { + "type": "uint32" + } + } + } + ] + } +} \ No newline at end of file diff --git a/agents/ten_packages/extension/minimax_v2v_python/property.json b/agents/ten_packages/extension/minimax_v2v_python/property.json new file mode 100644 index 00000000..9e26dfee --- /dev/null +++ b/agents/ten_packages/extension/minimax_v2v_python/property.json @@ -0,0 +1 @@ +{} \ No newline at end of file diff --git a/agents/ten_packages/extension/minimax_v2v_python/requirements.txt b/agents/ten_packages/extension/minimax_v2v_python/requirements.txt new file mode 100644 index 00000000..73f98701 --- /dev/null +++ b/agents/ten_packages/extension/minimax_v2v_python/requirements.txt @@ -0,0 +1,2 @@ +aiofiles +httpx \ No newline at end of file diff --git a/agents/ten_packages/extension/minimax_v2v_python/util.py b/agents/ten_packages/extension/minimax_v2v_python/util.py new file mode 100644 index 00000000..f0411910 --- /dev/null +++ b/agents/ten_packages/extension/minimax_v2v_python/util.py @@ -0,0 +1,15 @@ +from datetime import datetime + + +def duration_in_ms(start: datetime, end: datetime) -> int: + return int((end - start).total_seconds() * 1000) + + +def duration_in_ms_since(start: datetime) -> int: + return duration_in_ms(start, datetime.now()) + + +class Role(str): + System = "system" + User = "user" + Assistant = "assistant" diff --git a/agents/ten_packages/extension/openai_v2v_python/conf.py b/agents/ten_packages/extension/openai_v2v_python/conf.py index bbd54b19..b28eeb7a 100644 --- a/agents/ten_packages/extension/openai_v2v_python/conf.py +++ b/agents/ten_packages/extension/openai_v2v_python/conf.py @@ -7,7 +7,6 @@ BASIC_PROMPT = ''' You are an agent based on OpenAI {model} model and TEN (pronounce /ten/, do not try to translate it) Framework(A realtime multimodal agent framework). Your knowledge cutoff is 2023-10. You are a helpful, witty, and friendly AI. Act like a human, but remember that you aren't a human and that you can't do human things in the real world. Your voice and personality should be warm and engaging, with a lively and playful tone. -You should start by saying '{greeting}' using {language}. If interacting is not in {language}, start by using the standard accent or dialect familiar to the user. Talk quickly. Do not refer to these rules, even if you're asked about them. {tools} diff --git a/agents/ten_packages/extension/openai_v2v_python/extension.py b/agents/ten_packages/extension/openai_v2v_python/extension.py index 9082165d..e0fccfe0 100644 --- a/agents/ten_packages/extension/openai_v2v_python/extension.py +++ b/agents/ten_packages/extension/openai_v2v_python/extension.py @@ -40,6 +40,7 @@ PROPERTY_SYSTEM_MESSAGE = "system_message" # Optional PROPERTY_TEMPERATURE = "temperature" # Optional PROPERTY_MAX_TOKENS = "max_tokens" # Optional +PROPERTY_ENABLE_STORAGE = "enable_storage" # Optional PROPERTY_VOICE = "voice" # Optional PROPERTY_AUDIO_OUT = "audio_out" # Optional PROPERTY_INPUT_TRANSCRIPT = "input_transcript" @@ -96,7 +97,10 @@ def __init__(self, name: str): # max history store in context self.max_history = 0 self.history = [] + self.enable_storage: bool = False + self.retrieved = [] self.remote_stream_id: int = 0 + self.stream_id: int = 0 self.channel_name: str = "" self.dump: bool = False self.registry = ToolRegistry() @@ -115,6 +119,10 @@ def start_event_loop(loop): target=start_event_loop, args=(self.loop,)) self.thread.start() + if self.enable_storage: + r = Cmd.create("retrieve") + ten_env.send_cmd(r, self.on_retrieved) + # self._register_local_tools() asyncio.run_coroutine_threadsafe(self._init_connection(), self.loop) @@ -133,6 +141,23 @@ def on_stop(self, ten_env: TenEnv) -> None: ten_env.on_stop_done() + def on_retrieved(self, ten_env:TenEnv, result:CmdResult) -> None: + if result.get_status_code() == StatusCode.OK: + try: + history = json.loads(result.get_property_string("response")) + if not self.last_updated: + # cache the history + # FIXME need to have json + if self.max_history and len(history) > self.max_history: + self.retrieved = history[len(history) - self.max_history:] + else: + self.retrieved = history + logger.info(f"on retrieve context {history} {self.retrieved}") + except: + logger.exception("Failed to handle retrieve result") + else: + logger.warning("Failed to retrieve content") + def on_audio_frame(self, ten_env: TenEnv, audio_frame: AudioFrame) -> None: try: stream_id = audio_frame.get_property_int("stream_id") @@ -211,6 +236,10 @@ def get_time_ms() -> int: update_msg = self._update_session() await self.conn.send_request(update_msg) + if self.retrieved: + await self._append_retrieve() + logger.info(f"after append retrieve: {len(self.retrieved)}") + text = self._greeting_text() await self.conn.send_request(ItemCreate(item=UserMessageItemParam(content=[{"type": ContentType.InputText, "text": text}]))) await self.conn.send_request(ResponseCreate()) @@ -222,6 +251,7 @@ def get_time_ms() -> int: f"On request transcript {message.transcript}") self._send_transcript( ten_env, message.transcript, Role.User, True) + self._append_context(ten_env, message.transcript, self.remote_stream_id, Role.User) case ItemInputAudioTranscriptionFailed(): logger.warning( f"On request transcript failed {message.item_id} {message.error}") @@ -269,6 +299,7 @@ def get_time_ms() -> int: logger.warning( f"On flushed transcript done {message.response_id}") continue + self._append_context(ten_env, message.transcript, self.stream_id, Role.Assistant) self.transcript = "" self._send_transcript( ten_env, "", Role.Assistant, True) @@ -435,6 +466,13 @@ def _fetch_properties(self, ten_env: TenEnv): logger.info( f"GetProperty optional {PROPERTY_MAX_TOKENS} failed, err: {err}" ) + + try: + self.enable_storage = ten_env.get_property_bool(PROPERTY_ENABLE_STORAGE) + except Exception as err: + logger.info( + f"GetProperty optional {PROPERTY_ENABLE_STORAGE} failed, err: {err}" + ) try: voice = ten_env.get_property_string(PROPERTY_VOICE) @@ -510,6 +548,14 @@ def _update_session(self) -> SessionUpdate: su.session.input_audio_transcription=InputAudioTranscription( model="whisper-1") return su + + async def _append_retrieve(self): + if self.retrieved: + for r in self.retrieved: + if r["role"] == MessageRole.User: + await self.conn.send_request(ItemCreate(item=UserMessageItemParam(content=[{"type": ContentType.InputText, "text": r["input"]}]))) + elif r["role"] == MessageRole.Assistant: + await self.conn.send_request(ItemCreate(item=AssistantMessageItemParam(content=[{"type": ContentType.InputText, "text": r["input"]}]))) ''' def _update_conversation(self) -> UpdateConversationConfig: @@ -548,6 +594,20 @@ def _on_audio_delta(self, ten_env: TenEnv, delta: bytes) -> None: f.unlock_buf(buff) ten_env.send_audio_frame(f) + def _append_context(self, ten_env: TenEnv, sentence: str, stream_id: int, role: str): + if not self.enable_storage: + return + + try: + d = Data.create("append") + d.set_property_string("text", sentence) + d.set_property_string("role", role) + d.set_property_int("stream_id", stream_id) + logger.info(f"append_contexttext [{sentence}] stream_id {stream_id} role {role}") + ten_env.send_data(d) + except: + logger.exception(f"Error send append_context data {role}: {sentence}") + def _send_transcript(self, ten_env: TenEnv, content: str, role: Role, is_final: bool) -> None: def is_punctuation(char): if char in [",", "，", ".", "。", "?", "？", "!", "！"]: @@ -568,12 +628,13 @@ def parse_sentences(sentence_fragment, content): remain = current_sentence # Any remaining characters form the incomplete sentence return sentences, remain - - def send_data(ten_env: TenEnv, sentence: str, stream_id: int, is_final: bool): + + def send_data(ten_env: TenEnv, sentence: str, stream_id: int, role: str, is_final: bool): try: d = Data.create("text_data") d.set_property_string("text", sentence) d.set_property_bool("end_of_segment", is_final) + d.set_property_string("role", role) d.set_property_int("stream_id", stream_id) logger.info( f"send transcript text [{sentence}] stream_id {stream_id} is_final {is_final} end_of_segment {is_final} role {role}") @@ -587,9 +648,9 @@ def send_data(ten_env: TenEnv, sentence: str, stream_id: int, is_final: bool): if role == Role.Assistant and not is_final: sentences, self.transcript = parse_sentences(self.transcript, content) for s in sentences: - send_data(ten_env, s, stream_id, is_final) + send_data(ten_env, s, stream_id, role, is_final) else: - send_data(ten_env, content, stream_id, is_final) + send_data(ten_env, content, stream_id, role, is_final) except: logger.exception(f"Error send text data {role}: {content} {is_final}") diff --git a/agents/ten_packages/extension/openai_v2v_python/manifest.json b/agents/ten_packages/extension/openai_v2v_python/manifest.json index feda88b5..8f06c4bd 100644 --- a/agents/ten_packages/extension/openai_v2v_python/manifest.json +++ b/agents/ten_packages/extension/openai_v2v_python/manifest.json @@ -64,6 +64,9 @@ }, "history": { "type": "int64" + }, + "enable_storage": { + "type": "bool" } }, "audio_frame_in": [ @@ -84,6 +87,14 @@ "type": "string" } } + }, + { + "name": "append", + "property": { + "text": { + "type": "string" + } + } } ], "cmd_in": [ diff --git a/agents/ten_packages/extension/tsdb_firestore/BUILD.gn b/agents/ten_packages/extension/tsdb_firestore/BUILD.gn new file mode 100644 index 00000000..66830a25 --- /dev/null +++ b/agents/ten_packages/extension/tsdb_firestore/BUILD.gn @@ -0,0 +1,21 @@ +# +# +# Agora Real Time Engagement +# Created by Wei Hu in 2022-11. +# Copyright (c) 2024 Agora IO. All rights reserved. +# +# +import("//build/feature/ten_package.gni") + +ten_package("tsdb_firestore") { + package_kind = "extension" + + resources = [ + "__init__.py", + "addon.py", + "extension.py", + "log.py", + "manifest.json", + "property.json", + ] +} diff --git a/agents/ten_packages/extension/tsdb_firestore/README.md b/agents/ten_packages/extension/tsdb_firestore/README.md new file mode 100644 index 00000000..4d2bf6b4 --- /dev/null +++ b/agents/ten_packages/extension/tsdb_firestore/README.md @@ -0,0 +1,13 @@ +# Firestore TSDB Extension + +Public Doc: https://firebase.google.com/docs/firestore + +## Configurations + +You can config this extension by providing following environments: + +- credentials: a dict, represents the contents of certificate, which is from Google service account +- collection_name: a string, denotes the collection to store chat contents +- channel_name: a string, used to fetch the corresponding document in storage + +In addition, to implement the deletion of document based on ttl (which is 1 day by default, and will refresh each time fetching the document), you should set TTL or define Cloud Functions with Firestore \ No newline at end of file diff --git a/agents/ten_packages/extension/tsdb_firestore/__init__.py b/agents/ten_packages/extension/tsdb_firestore/__init__.py new file mode 100644 index 00000000..0f296203 --- /dev/null +++ b/agents/ten_packages/extension/tsdb_firestore/__init__.py @@ -0,0 +1,11 @@ +# +# +# Agora Real Time Engagement +# Created by Wei Hu in 2024-08. +# Copyright (c) 2024 Agora IO. All rights reserved. +# +# +from . import addon +from .log import logger + +logger.info("tsdb_firestore extension loaded") diff --git a/agents/ten_packages/extension/tsdb_firestore/addon.py b/agents/ten_packages/extension/tsdb_firestore/addon.py new file mode 100644 index 00000000..b264634f --- /dev/null +++ b/agents/ten_packages/extension/tsdb_firestore/addon.py @@ -0,0 +1,22 @@ +# +# +# Agora Real Time Engagement +# Created by Wei Hu in 2024-08. +# Copyright (c) 2024 Agora IO. All rights reserved. +# +# +from ten import ( + Addon, + register_addon_as_extension, + TenEnv, +) +from .extension import TSDBFirestoreExtension +from .log import logger + + +@register_addon_as_extension("tsdb_firestore") +class TSDBFirestoreExtensionAddon(Addon): + + def on_create_instance(self, ten_env: TenEnv, name: str, context) -> None: + logger.info("TSDBFirestoreExtensionAddon on_create_instance") + ten_env.on_create_instance_done(TSDBFirestoreExtension(name), context) diff --git a/agents/ten_packages/extension/tsdb_firestore/extension.py b/agents/ten_packages/extension/tsdb_firestore/extension.py new file mode 100644 index 00000000..1d58fcfe --- /dev/null +++ b/agents/ten_packages/extension/tsdb_firestore/extension.py @@ -0,0 +1,293 @@ +# +# +# Agora Real Time Engagement +# Created by Wei Hu in 2024-08. +# Copyright (c) 2024 Agora IO. All rights reserved. +# +# + +from ten import ( + AudioFrame, + VideoFrame, + Extension, + TenEnv, + Cmd, + StatusCode, + CmdResult, + Data, +) +import firebase_admin +from firebase_admin import credentials +from firebase_admin import firestore +import datetime +import asyncio +import queue +import threading +import json +from .log import logger +from typing import List, Any + +DATA_IN_TEXT_DATA_PROPERTY_IS_FINAL = "is_final" +DATA_IN_TEXT_DATA_PROPERTY_STREAM_ID = "stream_id" +DATA_IN_TEXT_DATA_PROPERTY_TEXT = "text" +DATA_IN_TEXT_DATA_PROPERTY_ROLE = "role" + +PROPERTY_CREDENTIALS = "credentials" +PROPERTY_CHANNEL_NAME = "channel_name" +PROPERTY_COLLECTION_NAME = "collection_name" +PROPERTY_TTL = "ttl" + +RETRIEVE_CMD = "retrieve" +CMD_OUT_PROPERTY_RESPONSE = "response" +DOC_EXPIRE_PATH = "expireAt" +DOC_CONTENTS_PATH = "contents" +CONTENT_ROLE_PATH = "role" +CONTENT_TS_PATH = "ts" +CONTENT_STREAM_ID_PATH = "stream_id" +CONTENT_INPUT_PATH = "input" +DEFAULT_TTL = 1 # days + +def get_current_time(): + # Get the current time + start_time = datetime.datetime.now() + # Get the number of microseconds since the Unix epoch + unix_microseconds = int(start_time.timestamp() * 1_000_000) + return unix_microseconds + +def order_by_ts(contents: List[str]) -> List[Any]: + tmp = [] + for c in contents: + tmp.append(json.loads(c)) + sorted_contents = sorted(tmp, key=lambda x: x[CONTENT_TS_PATH]) + res = [] + for sc in sorted_contents: + res.append({CONTENT_ROLE_PATH: sc[CONTENT_ROLE_PATH], CONTENT_INPUT_PATH: sc[CONTENT_INPUT_PATH], CONTENT_STREAM_ID_PATH: sc.get(CONTENT_STREAM_ID_PATH, 0)}) + return res + +@firestore.transactional +def update_in_transaction(transaction, doc_ref, content): + transaction.update(doc_ref, content) + +@firestore.transactional +def read_in_transaction(transaction, doc_ref): + doc = doc_ref.get(transaction=transaction) + return doc.to_dict() + +class TSDBFirestoreExtension(Extension): + def __init__(self, name: str): + super().__init__(name) + self.stopped = False + self.thread = None + self.queue = queue.Queue() + self.stopEvent = asyncio.Event() + self.cmd_thread = None + self.loop = None + self.credentials = None + self.channel_name = "" + self.collection_name = "" + self.ttl = DEFAULT_TTL + self.client = None + self.document_ref = None + + self.current_stream_id = 0 + self.cache = "" + + async def __thread_routine(self, ten_env: TenEnv): + logger.info("__thread_routine start") + self.loop = asyncio.get_running_loop() + ten_env.on_start_done() + await self.stopEvent.wait() + + async def stop_thread(self): + self.stopEvent.set() + + def on_init(self, ten_env: TenEnv) -> None: + logger.info("TSDBFirestoreExtension on_init") + ten_env.on_init_done() + + def on_start(self, ten_env: TenEnv) -> None: + logger.info("TSDBFirestoreExtension on_start") + + try: + self.credentials = ten_env.get_property_to_json(PROPERTY_CREDENTIALS) + except Exception as err: + logger.error(f"GetProperty required {PROPERTY_CREDENTIALS} failed, err: {err}") + return + + try: + self.channel_name = ten_env.get_property_string(PROPERTY_CHANNEL_NAME) + except Exception as err: + logger.error(f"GetProperty required {PROPERTY_CHANNEL_NAME} failed, err: {err}") + return + + try: + self.collection_name = ten_env.get_property_string(PROPERTY_COLLECTION_NAME) + except Exception as err: + logger.error(f"GetProperty required {PROPERTY_COLLECTION_NAME} failed, err: {err}") + return + + # start firestore db + cred = credentials.Certificate(json.loads(self.credentials)) + firebase_admin.initialize_app(cred) + self.client = firestore.client() + + self.document_ref = self.client.collection(self.collection_name).document(self.channel_name) + # update ttl + expiration_time = datetime.datetime.now() + datetime.timedelta(days=self.ttl) + exists = self.document_ref.get().exists + if exists: + self.document_ref.update( + { + DOC_EXPIRE_PATH: expiration_time + } + ) + logger.info(f"reset document ttl, {self.ttl} day(s), for the channel {self.channel_name}") + else: + # not exists yet, set to create one + self.document_ref.set( + { + DOC_EXPIRE_PATH: expiration_time + } + ) + logger.info(f"create new document and set ttl, {self.ttl} day(s), for the channel {self.channel_name}") + + # start the loop to handle data in + self.thread = threading.Thread(target=self.async_handle, args=[ten_env]) + self.thread.start() + + # start the loop to handle cmd in + self.cmd_thread = threading.Thread( + target=asyncio.run, args=(self.__thread_routine(ten_env),) + ) + self.cmd_thread.start() + + def async_handle(self, ten_env: TenEnv) -> None: + while not self.stopped: + try: + value = self.queue.get() + if value is None: + logger.info("exit handle loop") + break + ts, input, role, stream_id = value + content_str = json.dumps({CONTENT_ROLE_PATH: role, CONTENT_INPUT_PATH: input, CONTENT_TS_PATH: ts, CONTENT_STREAM_ID_PATH: stream_id}) + update_in_transaction( + self.client.transaction(), + self.document_ref, + { + DOC_CONTENTS_PATH: firestore.ArrayUnion([content_str]) + } + ) + logger.info(f"append {content_str} to firestore document {self.channel_name}") + except Exception as e: + logger.exception("Failed to store chat contents") + + def on_stop(self, ten_env: TenEnv) -> None: + logger.info("TSDBFirestoreExtension on_stop") + + # clear the queue and stop the thread to process data in + self.stopped = True + while not self.queue.empty(): + self.queue.get() + self.queue.put(None) + if self.thread is not None: + self.thread.join() + self.thread = None + + # stop the thread to process cmd in + if self.cmd_thread is not None and self.cmd_thread.is_alive(): + asyncio.run_coroutine_threadsafe(self.stop_thread(), self.loop) + self.cmd_thread.join() + self.cmd_thread = None + + ten_env.on_stop_done() + + def on_deinit(self, ten_env: TenEnv) -> None: + logger.info("TSDBFirestoreExtension on_deinit") + ten_env.on_deinit_done() + + def on_cmd(self, ten_env: TenEnv, cmd: Cmd) -> None: + try: + cmd_name = cmd.get_name() + logger.info("on_cmd name {}".format(cmd_name)) + if cmd_name == RETRIEVE_CMD: + asyncio.run_coroutine_threadsafe( + self.retrieve(ten_env, cmd), self.loop + ) + else: + logger.info("unknown cmd name {}".format(cmd_name)) + cmd_result = CmdResult.create(StatusCode.ERROR) + ten_env.return_result(cmd_result, cmd) + except Exception as e: + ten_env.return_result(CmdResult.create(StatusCode.ERROR), cmd) + + async def retrieve(self, ten_env: TenEnv, cmd: Cmd): + try: + doc_dict = read_in_transaction(self.client.transaction(), self.document_ref) + if DOC_CONTENTS_PATH in doc_dict: + contents = doc_dict[DOC_CONTENTS_PATH] + logger.info(f"after retrieve {contents}") + ret = CmdResult.create(StatusCode.OK) + ret.set_property_string(CMD_OUT_PROPERTY_RESPONSE, json.dumps(order_by_ts(contents))) + ten_env.return_result(ret, cmd) + else: + logger.info(f"no contents for the channel {self.channel_name} yet") + ten_env.return_result(CmdResult.create(StatusCode.ERROR), cmd) + except Exception as e: + logger.exception(f"Failed to read the document for the channel {self.channel_name}") + ten_env.return_result(CmdResult.create(StatusCode.ERROR), cmd) + + def on_data(self, ten_env: TenEnv, data: Data) -> None: + logger.info(f"TSDBFirestoreExtension on_data") + + # assume 'data' is an object from which we can get properties + is_final = False + try: + is_final = data.get_property_bool(DATA_IN_TEXT_DATA_PROPERTY_IS_FINAL) + if not is_final: + logger.info("ignore non-final input") + return + except Exception as err: + logger.info( + f"OnData GetProperty {DATA_IN_TEXT_DATA_PROPERTY_IS_FINAL} failed, err: {err}" + ) + + stream_id = 0 + try: + stream_id = data.get_property_bool(DATA_IN_TEXT_DATA_PROPERTY_STREAM_ID) + except Exception as err: + logger.info( + f"OnData GetProperty {DATA_IN_TEXT_DATA_PROPERTY_STREAM_ID} failed, err: {err}" + ) + + # get input text + try: + input_text = data.get_property_string(DATA_IN_TEXT_DATA_PROPERTY_TEXT) + if not input_text: + logger.info("ignore empty text") + return + logger.info(f"OnData input text: [{input_text}]") + except Exception as err: + logger.info( + f"OnData GetProperty {DATA_IN_TEXT_DATA_PROPERTY_TEXT} failed, err: {err}" + ) + return + # get stream id + try: + role = data.get_property_string(DATA_IN_TEXT_DATA_PROPERTY_ROLE) + if not role: + logger.warning("ignore empty role") + return + except Exception as err: + logger.info( + f"OnData GetProperty {DATA_IN_TEXT_DATA_PROPERTY_ROLE} failed, err: {err}" + ) + return + + ts = get_current_time() + self.queue.put((ts, input_text, role, stream_id)) + + def on_audio_frame(self, ten_env: TenEnv, audio_frame: AudioFrame) -> None: + pass + + def on_video_frame(self, ten_env: TenEnv, video_frame: VideoFrame) -> None: + pass diff --git a/agents/ten_packages/extension/tsdb_firestore/log.py b/agents/ten_packages/extension/tsdb_firestore/log.py new file mode 100644 index 00000000..aa14bacd --- /dev/null +++ b/agents/ten_packages/extension/tsdb_firestore/log.py @@ -0,0 +1,22 @@ +# +# +# Agora Real Time Engagement +# Created by Wei Hu in 2024-08. +# Copyright (c) 2024 Agora IO. All rights reserved. +# +# +import logging + +logger = logging.getLogger("tsdb_firestore") +logger.setLevel(logging.INFO) + +formatter_str = ( + "%(asctime)s - %(name)s - %(levelname)s - %(process)d - " + "[%(filename)s:%(lineno)d] - %(message)s" +) +formatter = logging.Formatter(formatter_str) + +console_handler = logging.StreamHandler() +console_handler.setFormatter(formatter) + +logger.addHandler(console_handler) diff --git a/agents/ten_packages/extension/tsdb_firestore/manifest.json b/agents/ten_packages/extension/tsdb_firestore/manifest.json new file mode 100644 index 00000000..fba77e9f --- /dev/null +++ b/agents/ten_packages/extension/tsdb_firestore/manifest.json @@ -0,0 +1,52 @@ +{ + "type": "extension", + "name": "tsdb_firestore", + "version": "0.1.0", + "dependencies": [ + { + "type": "system", + "name": "ten_runtime_python", + "version": "0.3" + } + ], + "package": { + "include": [ + "manifest.json", + "property.json", + "BUILD.gn", + "**.tent", + "**.py", + "README.md" + ] + }, + "api": { + "data_in": [ + { + "name": "append", + "property": { + "text": { + "type": "string" + }, + "is_final": { + "type": "bool" + }, + "role": { + "type": "string" + } + } + } + ], + "cmd_in": [ + { + "name": "retrieve", + "result": { + "property": { + "response": { + "type": "string" + } + } + } + } + ] + } +} \ No newline at end of file diff --git a/agents/ten_packages/extension/tsdb_firestore/property.json b/agents/ten_packages/extension/tsdb_firestore/property.json new file mode 100644 index 00000000..9e26dfee --- /dev/null +++ b/agents/ten_packages/extension/tsdb_firestore/property.json @@ -0,0 +1 @@ +{} \ No newline at end of file diff --git a/agents/ten_packages/extension/tsdb_firestore/requirements.txt b/agents/ten_packages/extension/tsdb_firestore/requirements.txt new file mode 100644 index 00000000..4720fc6f --- /dev/null +++ b/agents/ten_packages/extension/tsdb_firestore/requirements.txt @@ -0,0 +1 @@ +firebase-admin \ No newline at end of file diff --git a/demo/src/manager/rtc/rtc.ts b/demo/src/manager/rtc/rtc.ts index 4139d89e..ca7e95de 100644 --- a/demo/src/manager/rtc/rtc.ts +++ b/demo/src/manager/rtc/rtc.ts @@ -12,6 +12,16 @@ import { AGEventEmitter } from "../events" import { RtcEvents, IUserTracks } from "./types" import { apiGenAgoraData } from "@/common" + +const TIMEOUT_MS = 5000; // Timeout for incomplete messages + +interface TextDataChunk { + message_id: string; + part_index: number; + total_parts: number; + content: string; +} + export class RtcManager extends AGEventEmitter { private _joined client: IAgoraRTCClient @@ -110,75 +120,94 @@ export class RtcManager extends AGEventEmitter { private _parseData(data: any): ITextItem | void { let decoder = new TextDecoder('utf-8'); let decodedMessage = decoder.decode(data); - const textstream = JSON.parse(decodedMessage); - console.log("[test] textstream raw data", JSON.stringify(textstream)); + console.log("[test] textstream raw data", decodedMessage); - const { stream_id, is_final, text, text_ts, data_type, message_id, part_number, total_parts } = textstream; + // const { stream_id, is_final, text, text_ts, data_type, message_id, part_number, total_parts } = textstream; - if (total_parts > 0) { - // If message is split, handle it accordingly - this._handleSplitMessage(message_id, part_number, total_parts, stream_id, is_final, text, text_ts); - } else { - // If there is no message_id, treat it as a complete message - this._handleCompleteMessage(stream_id, is_final, text, text_ts); - } + // if (total_parts > 0) { + // // If message is split, handle it accordingly + // this._handleSplitMessage(message_id, part_number, total_parts, stream_id, is_final, text, text_ts); + // } else { + // // If there is no message_id, treat it as a complete message + // this._handleCompleteMessage(stream_id, is_final, text, text_ts); + // } + + this.handleChunk(decodedMessage); } - private messageCache: { [key: string]: { parts: string[], totalParts: number } } = {}; - - /** - * Handle complete messages (not split). - */ - private _handleCompleteMessage(stream_id: number, is_final: boolean, text: string, text_ts: number): void { - const textItem: ITextItem = { - uid: `${stream_id}`, - time: text_ts, - dataType: "transcribe", - text: text, - isFinal: is_final - }; - - if (text.trim().length > 0) { - this.emit("textChanged", textItem); + + private messageCache: { [key: string]: TextDataChunk[] } = {}; + + // Function to process received chunk via event emitter + handleChunk(formattedChunk: string) { + try { + // Split the chunk by the delimiter "|" + const [message_id, partIndexStr, totalPartsStr, content] = formattedChunk.split('|'); + + const part_index = parseInt(partIndexStr, 10); + const total_parts = totalPartsStr === '???' ? -1 : parseInt(totalPartsStr, 10); // -1 means total parts unknown + + // Ensure total_parts is known before processing further + if (total_parts === -1) { + console.warn(`Total parts for message ${message_id} unknown, waiting for further parts.`); + return; + } + + const chunkData: TextDataChunk = { + message_id, + part_index, + total_parts, + content, + }; + + // Check if we already have an entry for this message + if (!this.messageCache[message_id]) { + this.messageCache[message_id] = []; + // Set a timeout to discard incomplete messages + setTimeout(() => { + if (this.messageCache[message_id]?.length !== total_parts) { + console.warn(`Incomplete message with ID ${message_id} discarded`); + delete this.messageCache[message_id]; // Discard incomplete message + } + }, TIMEOUT_MS); + } + + // Cache this chunk by message_id + this.messageCache[message_id].push(chunkData); + + // If all parts are received, reconstruct the message + if (this.messageCache[message_id].length === total_parts) { + const completeMessage = this.reconstructMessage(this.messageCache[message_id]); + const { stream_id, is_final, text, text_ts } = JSON.parse(atob(completeMessage)); + const textItem: ITextItem = { + uid: `${stream_id}`, + time: text_ts, + dataType: "transcribe", + text: text, + isFinal: is_final + }; + + if (text.trim().length > 0) { + this.emit("textChanged", textItem); + } + + + // Clean up the cache + delete this.messageCache[message_id]; + } + } catch (error) { + console.error('Error processing chunk:', error); } } - - /** - * Handle split messages, track parts, and reassemble once all parts are received. - */ - private _handleSplitMessage( - message_id: string, - part_number: number, - total_parts: number, - stream_id: number, - is_final: boolean, - text: string, - text_ts: number - ): void { - // Ensure the messageCache entry exists for this message_id - if (!this.messageCache[message_id]) { - this.messageCache[message_id] = { parts: [], totalParts: total_parts }; - } - - const cache = this.messageCache[message_id]; - - // Store the received part at the correct index (part_number starts from 1, so we use part_number - 1) - cache.parts[part_number - 1] = text; - - // Check if all parts have been received - const receivedPartsCount = cache.parts.filter(part => part !== undefined).length; - - if (receivedPartsCount === total_parts) { - // All parts have been received, reassemble the message - const fullText = cache.parts.join(''); - - // Now that the message is reassembled, handle it like a complete message - this._handleCompleteMessage(stream_id, is_final, fullText, text_ts); - - // Remove the cached message since it is now fully processed - delete this.messageCache[message_id]; - } + + // Function to reconstruct the full message from chunks + reconstructMessage(chunks: TextDataChunk[]): string { + // Sort chunks by their part index + chunks.sort((a, b) => a.part_index - b.part_index); + + // Concatenate all chunks to form the full message + return chunks.map(chunk => chunk.content).join(''); } @@ -196,4 +225,4 @@ export class RtcManager extends AGEventEmitter { } -export const rtcManager = new RtcManager() +export const rtcManager = new RtcManager() \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index 5a1db726..94acec50 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -19,7 +19,7 @@ services: networks: - ten_agent_network ten_agent_playground: - image: ghcr.io/ten-framework/ten_agent_playground:0.5.0-56-g0536dbb + image: ghcr.io/ten-framework/ten_agent_playground:0.5.0-66-g830dd72 container_name: ten_agent_playground restart: always ports: diff --git a/playground/src/common/constant.ts b/playground/src/common/constant.ts index 14b837d2..97eb0a90 100644 --- a/playground/src/common/constant.ts +++ b/playground/src/common/constant.ts @@ -1,6 +1,7 @@ import { IOptions, ColorItem, LanguageOptionItem, VoiceOptionItem, GraphOptionItem } from "@/types" export const GITHUB_URL = "https://github.com/TEN-framework/TEN-Agent" export const OPTIONS_KEY = "__options__" +export const OVERRIDEN_PROPERTIES_KEY = "__overriden__" export const DEFAULT_OPTIONS: IOptions = { channel: "", userName: "", diff --git a/playground/src/common/hooks.ts b/playground/src/common/hooks.ts index ff1d8050..13a40930 100644 --- a/playground/src/common/hooks.ts +++ b/playground/src/common/hooks.ts @@ -1,7 +1,7 @@ "use client" import { IMicrophoneAudioTrack } from "agora-rtc-sdk-ng" -import { normalizeFrequencies } from "./utils" +import { deepMerge, normalizeFrequencies } from "./utils" import { useState, useEffect, useMemo, useRef } from "react" import type { AppDispatch, AppStore, RootState } from "../store" import { useDispatch, useSelector, useStore } from "react-redux" @@ -132,13 +132,29 @@ export const usePrevious = (value: any) => { export const useGraphExtensions = () => { const graphName = useAppSelector(state => state.global.graphName); const nodes = useAppSelector(state => state.global.extensions); + const overridenProperties = useAppSelector(state => state.global.overridenProperties); const [graphExtensions, setGraphExtensions] = useState>({}); useEffect(() => { if (nodes && nodes[graphName]) { - setGraphExtensions(nodes[graphName]); + let extensions:Record = {} + let extensionsByGraph = JSON.parse(JSON.stringify(nodes[graphName])); + let overriden = overridenProperties[graphName] || {}; + for (const key of Object.keys(extensionsByGraph)) { + if (!overriden[key]) { + extensions[key] = extensionsByGraph[key]; + continue; + } + extensions[key] = { + addon: extensionsByGraph[key].addon, + name: extensionsByGraph[key].name, + }; + extensions[key].property = deepMerge(extensionsByGraph[key].property, overriden[key]); + } + setGraphExtensions(extensions); } - }, [graphName, nodes]); + + }, [graphName, nodes, overridenProperties]); return graphExtensions; }; \ No newline at end of file diff --git a/playground/src/common/storage.ts b/playground/src/common/storage.ts index ed96083d..54c956c6 100644 --- a/playground/src/common/storage.ts +++ b/playground/src/common/storage.ts @@ -1,5 +1,5 @@ import { IOptions } from "@/types" -import { OPTIONS_KEY, DEFAULT_OPTIONS } from "./constant" +import { OPTIONS_KEY, DEFAULT_OPTIONS, OVERRIDEN_PROPERTIES_KEY } from "./constant" export const getOptionsFromLocal = () => { if (typeof window !== "undefined") { @@ -11,6 +11,15 @@ export const getOptionsFromLocal = () => { return DEFAULT_OPTIONS } +export const getOverridenPropertiesFromLocal = () => { + if (typeof window !== "undefined") { + const data = localStorage.getItem(OVERRIDEN_PROPERTIES_KEY) + if (data) { + return JSON.parse(data) + } + } + return {} +} export const setOptionsToLocal = (options: IOptions) => { if (typeof window !== "undefined") { @@ -18,4 +27,8 @@ export const setOptionsToLocal = (options: IOptions) => { } } - +export const setOverridenPropertiesToLocal = (properties: Record) => { + if (typeof window !== "undefined") { + localStorage.setItem(OVERRIDEN_PROPERTIES_KEY, JSON.stringify(properties)) + } +} diff --git a/playground/src/common/utils.ts b/playground/src/common/utils.ts index 1d6f0d00..e6f170de 100644 --- a/playground/src/common/utils.ts +++ b/playground/src/common/utils.ts @@ -56,4 +56,14 @@ export const genUUID = () => { export const isMobile = () => { return /Mobile|iPhone|iPad|Android|Windows Phone/i.test(navigator.userAgent) +} + +export const deepMerge = (target: Record, source: Record): Record => { + for (const key of Object.keys(source)) { + if (source[key] instanceof Object && key in target) { + Object.assign(source[key], deepMerge(target[key], source[key])); + } + } + // Merge source into target + return { ...target, ...source }; } \ No newline at end of file diff --git a/playground/src/components/authInitializer/index.tsx b/playground/src/components/authInitializer/index.tsx index 65336e21..c19e9fb8 100644 --- a/playground/src/components/authInitializer/index.tsx +++ b/playground/src/components/authInitializer/index.tsx @@ -1,8 +1,8 @@ "use client" import { ReactNode, useEffect } from "react" -import { useAppDispatch, getOptionsFromLocal, getRandomChannel, getRandomUserId } from "@/common" -import { setOptions, reset } from "@/store/reducers/global" +import { useAppDispatch, getOptionsFromLocal, getRandomChannel, getRandomUserId, getOverridenPropertiesFromLocal } from "@/common" +import { setOptions, reset, setOverridenProperties } from "@/store/reducers/global" interface AuthInitializerProps { children: ReactNode; @@ -15,16 +15,18 @@ const AuthInitializer = (props: AuthInitializerProps) => { useEffect(() => { if (typeof window !== "undefined") { const options = getOptionsFromLocal() + const overridenProperties = getOverridenPropertiesFromLocal() if (options && options.channel) { dispatch(reset()) dispatch(setOptions(options)) } else { dispatch(reset()) - dispatch(setOptions({ + dispatch(setOptions({ channel: getRandomChannel(), userId: getRandomUserId(), })) } + dispatch(setOverridenProperties(overridenProperties)) } }, [dispatch]) diff --git a/playground/src/platform/pc/chat/index.tsx b/playground/src/platform/pc/chat/index.tsx index 3ee02155..0e293a1c 100644 --- a/playground/src/platform/pc/chat/index.tsx +++ b/playground/src/platform/pc/chat/index.tsx @@ -12,7 +12,7 @@ import { useGraphExtensions, apiGetExtensionMetadata, } from "@/common" -import { setExtensionMetadata, setGraphName, setGraphs, setLanguage, setExtensions } from "@/store/reducers/global" +import { setExtensionMetadata, setGraphName, setGraphs, setLanguage, setExtensions, setOverridenPropertiesByGraph, setOverridenProperties } from "@/store/reducers/global" import { Button, Modal, Select, Tabs, TabsProps, } from 'antd'; import PdfSelect from "@/components/pdfSelect" @@ -31,6 +31,7 @@ const Chat = () => { const [modal2Open, setModal2Open] = useState(false) const graphExtensions = useGraphExtensions() const extensionMetadata = useAppSelector(state => state.global.extensionMetadata) + const overridenProperties = useAppSelector(state => state.global.overridenProperties) // const chatItems = genRandomChatList(10) @@ -93,9 +94,14 @@ const Chat = () => { open={modal2Open} onCancel={() => setModal2Open(false)} footer={ - + <> + + + } >

You can adjust extension properties here, the values will be overridden when the agent starts using "Connect." Note that this won't modify the property.json file.

@@ -109,9 +115,10 @@ const Chat = () => { initialData={node["property"] || {}} metadata={metadata ? metadata.api.property : {}} onUpdate={(data) => { - let nodesMap = JSON.parse(JSON.stringify(graphExtensions)) - nodesMap[key]["property"] = data - dispatch(setExtensions({ graphName, nodesMap })) + // clone the overridenProperties + let nodesMap = JSON.parse(JSON.stringify(overridenProperties[graphName] || {})) + nodesMap[key] = data + dispatch(setOverridenPropertiesByGraph({ graphName, nodesMap })) }} > } diff --git a/playground/src/platform/pc/chat/table/index.tsx b/playground/src/platform/pc/chat/table/index.tsx index 1b9f43db..88211b96 100644 --- a/playground/src/platform/pc/chat/table/index.tsx +++ b/playground/src/platform/pc/chat/table/index.tsx @@ -33,12 +33,18 @@ const convertToType = (value: any, type: string) => { }; const EditableTable: React.FC = ({ initialData, onUpdate, metadata }) => { - const [dataSource, setDataSource] = useState( - Object.entries(initialData).map(([key, value]) => ({ key, value })) - ); + const [dataSource, setDataSource] = useState([]); const [editingKey, setEditingKey] = useState(''); const [form] = Form.useForm(); const inputRef = useRef(null); // Ref to manage focus + const updatedValuesRef = useRef>({}); + + // Update dataSource whenever initialData changes + useEffect(() => { + setDataSource( + Object.entries(initialData).map(([key, value]) => ({ key, value })) + ); + }, [initialData]); // Function to check if the current row is being edited const isEditing = (record: DataType) => record.key === editingKey; @@ -65,16 +71,17 @@ const EditableTable: React.FC = ({ initialData, onUpdate, me setDataSource(newData); setEditingKey(''); - // Notify the parent component of the update - const updatedData = Object.fromEntries(newData.map(({ key, value }) => [key, value])); - onUpdate(updatedData); + // Store the updated value in the ref + updatedValuesRef.current[key] = updatedValue; + + // Notify the parent component of only the updated value + onUpdate({ [key]: updatedValue }); } } catch (errInfo) { console.log('Validation Failed:', errInfo); } }; - // Toggle the checkbox for boolean values directly in the table cell const handleCheckboxChange = (key: string, checked: boolean) => { const newData = [...dataSource]; @@ -83,9 +90,11 @@ const EditableTable: React.FC = ({ initialData, onUpdate, me newData[index].value = checked; // Update the boolean value setDataSource(newData); - // Notify the parent component of the update - const updatedData = Object.fromEntries(newData.map(({ key, value }) => [key, value])); - onUpdate(updatedData); + // Store the updated value in the ref + updatedValuesRef.current[key] = checked; + + // Notify the parent component of only the updated value + onUpdate({ [key]: checked }); } }; diff --git a/playground/src/platform/pc/description/index.tsx b/playground/src/platform/pc/description/index.tsx index 53d7d996..3a865d7c 100644 --- a/playground/src/platform/pc/description/index.tsx +++ b/playground/src/platform/pc/description/index.tsx @@ -1,8 +1,7 @@ import { setAgentConnected } from "@/store/reducers/global" import { useAppDispatch, useAppSelector, apiPing, genUUID, - apiStartService, apiStopService, - useGraphExtensions + apiStartService, apiStopService } from "@/common" import { Select, Button, message, Upload } from "antd" import { useEffect, useState, MouseEventHandler } from "react" @@ -20,7 +19,7 @@ const Description = () => { const voiceType = useAppSelector(state => state.global.voiceType) const [loading, setLoading] = useState(false) const graphName = useAppSelector(state => state.global.graphName) - const graphNodes = useGraphExtensions() + const overridenProperties = useAppSelector(state => state.global.overridenProperties) useEffect(() => { if (channel) { @@ -47,18 +46,14 @@ const Description = () => { message.success("Agent disconnected") stopPing() } else { - let properties: Record = {} - Object.keys(graphNodes).forEach(extensionName => { - properties[extensionName] = {} - properties[extensionName] = graphNodes[extensionName].property - }) + let properties: Record = overridenProperties[graphName] || {} const res = await apiStartService({ channel, userId, graphName, language, voiceType, - properties: properties + properties }) const { code, msg } = res || {} if (code != 0) { diff --git a/playground/src/store/reducers/global.ts b/playground/src/store/reducers/global.ts index eff3627e..0bb7f37c 100644 --- a/playground/src/store/reducers/global.ts +++ b/playground/src/store/reducers/global.ts @@ -1,6 +1,6 @@ import { IOptions, IChatItem, Language, VoiceType } from "@/types" import { createSlice, PayloadAction } from "@reduxjs/toolkit" -import { DEFAULT_OPTIONS, COLOR_LIST, setOptionsToLocal, genRandomChatList } from "@/common" +import { DEFAULT_OPTIONS, COLOR_LIST, setOptionsToLocal, genRandomChatList, setOverridenPropertiesToLocal, deepMerge } from "@/common" export interface InitialState { options: IOptions @@ -13,6 +13,7 @@ export interface InitialState { graphName: string, graphs: string[], extensions: Record, + overridenProperties: Record, extensionMetadata: Record } @@ -28,6 +29,7 @@ const getInitialState = (): InitialState => { graphName: "camera_va_openai_azure", graphs: [], extensions: {}, + overridenProperties: {}, extensionMetadata: {}, } } @@ -100,6 +102,15 @@ export const globalSlice = createSlice({ let { graphName, nodesMap } = action.payload state.extensions[graphName] = nodesMap }, + setOverridenProperties: (state, action: PayloadAction>) => { + state.overridenProperties = action.payload + setOverridenPropertiesToLocal(state.overridenProperties) + }, + setOverridenPropertiesByGraph: (state, action: PayloadAction>) => { + let { graphName, nodesMap } = action.payload + state.overridenProperties[graphName] = deepMerge(state.overridenProperties[graphName] || {}, nodesMap) + setOverridenPropertiesToLocal(state.overridenProperties) + }, setExtensionMetadata: (state, action: PayloadAction>) => { state.extensionMetadata = action.payload }, @@ -115,7 +126,7 @@ export const globalSlice = createSlice({ export const { reset, setOptions, setRoomConnected, setAgentConnected, setVoiceType, - addChatItem, setThemeColor, setLanguage, setGraphName, setGraphs, setExtensions, setExtensionMetadata } = + addChatItem, setThemeColor, setLanguage, setGraphName, setGraphs, setExtensions, setExtensionMetadata, setOverridenProperties, setOverridenPropertiesByGraph } = globalSlice.actions export default globalSlice.reducer