Pemanggilan fungsi

import os
import uuid
import threading
import traceback
import json
import base64
import signal
import sys
import time
from typing import Dict, Any, Optional, List
import pyaudio
import queue
import contextlib
import dashscope
from dashscope.audio.qwen_omni import *

# ==================== Definisi Konstanta ====================
VOICE = 'Tina'
MODEL = "qwen3.5-omni-plus-realtime"
# Untuk mengakses wilayah Beijing, ganti WS_URL dengan: wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime
WS_URL = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime"
# Konfigurasikan kunci API. Jika Anda belum mengatur variabel lingkungan, ganti baris berikut dengan kunci API Anda: dashscope.api_key = "sk-xxx"
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
AUDIO_SAMPLE_RATE = 16000
AUDIO_CHUNK_SIZE = 3200
OUTPUT_AUDIO_SAMPLE_RATE = 24000

# ==================== Definisi Alat (Tool) ====================
def get_train_price(src: str, dst: str) -> str:
    """Kueri harga tiket kereta"""
    return f"The train ticket price from {src} to {dst} is 100-200 CNY."

def get_flight_price(src: str, dst: str) -> str:
    """Kueri harga tiket pesawat"""
    return f"The flight ticket price from {src} to {dst} is 200-300 USD."

def get_current_weather(location: str) -> str:
    """Kueri cuaca di kota tertentu"""
    return f"The weather in {location} today is changing from haze to sunny, with a temperature of 4/-4°C and a light breeze."

# Definisi alat (tool) format OpenAI terpadu
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Berguna saat Anda ingin menanyakan cuaca di kota tertentu.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "Sebuah kota atau distrik, seperti Beijing, Hangzhou, atau Yuhang.",
                    }
                },
                "required": ["location"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_flight_price",
            "description": "Berguna saat Anda ingin menanyakan harga tiket pesawat.",
            "parameters": {
                "type": "object",
                "properties": {
                    "src": {
                        "type": "string",
                        "description": "Kota keberangkatan penerbangan, seperti Beijing atau Hangzhou.",
                    },
                    "dst": {
                        "type": "string",
                        "description": "Kota kedatangan penerbangan, seperti Beijing atau Hangzhou.",
                    },
                },
                "required": ["src", "dst"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_train_price",
            "description": "Berguna saat Anda ingin menanyakan harga tiket kereta.",
            "parameters": {
                "type": "object",
                "properties": {
                    "src": {
                        "type": "string",
                        "description": "Kota keberangkatan kereta, seperti Beijing atau Hangzhou.",
                    },
                    "dst": {
                        "type": "string",
                        "description": "Kota kedatangan kereta, seperti Beijing atau Hangzhou.",
                    },
                },
                "required": ["src", "dst"],
            },
        },
    },
]

# Pemetaan nama alat (tool) ke fungsi
TOOL_FUNCTIONS = {
    "get_current_weather": get_current_weather,
    "get_flight_price": get_flight_price,
    "get_train_price": get_train_price,
}

# ==================== Penanganan Panggilan Alat (Tool Call) ====================
def handle_tool_call(tool_call_response: Dict[str, Any]) -> Dict[str, Any]:
    """
    Menangani permintaan panggilan alat (tool call)

    Args:
        tool_call_response: Informasi panggilan alat yang berisi nama, argumen, dan call_id

    Returns:
        Respons panggilan alat yang diperbarui yang berisi bidang output
    """
    try:
        function_name = tool_call_response['name']
        tool_call_arguments = json.loads(tool_call_response['arguments'])

        print(f'[Tool Call] Start processing: name={function_name}, args={tool_call_arguments}')

        # Temukan fungsi yang sesuai
        if function_name not in TOOL_FUNCTIONS:
            tool_call_response['output'] = f"Client did not find the tool: {function_name}"
            print(f'[Tool Call] Error: Tool not found {function_name}')
            return tool_call_response

        # Panggil fungsi
        func = TOOL_FUNCTIONS[function_name]
        result = func(**tool_call_arguments)
        tool_call_response['output'] = result

        print(f'[Tool Call] Completed: {result}')
        return tool_call_response

    except Exception as e:
        error_msg = f"Tool call failed: {str(e)}"
        tool_call_response['output'] = error_msg
        print(f'[Tool Call] Exception: {error_msg}')
        traceback.print_exc()
        return tool_call_response

def send_tool_call_response(conversation: OmniRealtimeConversation, response: Dict[str, Any]) -> None:
    """Mengirimkan hasil panggilan alat (tool call) ke server"""
    conversation.create_item({
        "id": 'item_' + uuid.uuid4().hex,
        "type": "function_call_output",
        "call_id": response['call_id'],
        "output": response["output"],
    })

# ==================== Pemutar Audio PCM ====================
class PCMPlayer:
    """
    Pemutar Audio PCM

    Menggunakan arsitektur dual-thread untuk pemutaran audio real-time:
    - Thread decoding: Mendekode data audio yang dienkode base64 menjadi data PCM mentah
    - Thread pemutaran: Menulis data PCM ke perangkat output audio

    Mendukung penambahan data audio secara dinamis, pembatalan pemutaran, penyimpanan file audio, dll.
    """

    def __init__(self, pya: pyaudio.PyAudio, sample_rate=24000, chunk_size_ms=100, save_file=False):
        """
        Menginisialisasi pemutar PCM

        Args:
            pya: Instans pyaudio.PyAudio
            sample_rate: Laju pengambilan sampel audio (Hz), default 24000
            chunk_size_ms: Ukuran chunk audio (milidetik), memengaruhi latensi pembatalan pemutaran, default 100ms
            save_file: Apakah akan menyimpan audio yang diputar ke file (result.pcm), default False
        """

        self.pya = pya
        self.sample_rate = sample_rate
        self.chunk_size_bytes = chunk_size_ms * sample_rate * 2 // 1000
        self.player_stream = pya.open(format=pyaudio.paInt16,
                                       channels=1,
                                       rate=sample_rate,
                                       output=True)

        self.raw_audio_buffer: queue.Queue = queue.Queue()
        self.b64_audio_buffer: queue.Queue = queue.Queue()
        self.status_lock = threading.Lock()
        self.status = 'playing'
        self.decoder_thread = threading.Thread(target=self.decoder_loop)
        self.player_thread = threading.Thread(target=self.player_loop)
        self.decoder_thread.start()
        self.player_thread.start()
        self.complete_event: threading.Event = None
        self.save_file = save_file
        if self.save_file:
            self.out_file = open('result.pcm', 'wb')

    def decoder_loop(self):
        """Thread decoding: Mendekode data audio base64 menjadi data PCM mentah"""
        while self.status != 'stop':
            recv_audio_b64 = None
            with contextlib.suppress(queue.Empty):
                recv_audio_b64 = self.b64_audio_buffer.get(timeout=0.1)
            if recv_audio_b64 is None:
                continue
            recv_audio_raw = base64.b64decode(recv_audio_b64)
            # dorong data audio mentah ke antrian per chunk
            for i in range(0, len(recv_audio_raw), self.chunk_size_bytes):
                chunk = recv_audio_raw[i:i + self.chunk_size_bytes]
                self.raw_audio_buffer.put(chunk)
                if self.save_file:
                    self.out_file.write(chunk)

    def player_loop(self):
        """Thread pemutaran: Menulis data PCM ke perangkat output audio"""
        while self.status != 'stop':
            recv_audio_raw = None
            with contextlib.suppress(queue.Empty):
                recv_audio_raw = self.raw_audio_buffer.get(timeout=0.1)
            if recv_audio_raw is None:
                if self.complete_event:
                    self.complete_event.set()
                continue
            # tulis chunk ke pemutar audio pyaudio, tunggu hingga selesai memutar chunk ini.
            self.player_stream.write(recv_audio_raw)

    def cancel_playing(self):
        """Batalkan pemutaran: Kosongkan semua antrian buffer"""
        self.b64_audio_buffer.queue.clear()
        self.raw_audio_buffer.queue.clear()

    def add_data(self, data):
        """Tambahkan data audio yang dienkode base64 ke antrian pemutaran"""
        self.b64_audio_buffer.put(data)

    def wait_for_complete(self):
        """Tunggu hingga pemutaran selesai"""
        self.complete_event = threading.Event()
        self.complete_event.wait()
        self.complete_event = None

    def shutdown(self):
        """Matikan pemutar dan lepaskan sumber daya"""
        self.status = 'stop'
        self.decoder_thread.join()
        self.player_thread.join()
        self.player_stream.close()
        if self.save_file:
            self.out_file.close()

# ==================== Manajer Audio ====================
class AudioManager:
    """Mengelola sumber daya input dan output audio"""

    def __init__(self):
        self.pya: Optional[pyaudio.PyAudio] = None
        self.mic_stream: Optional[pyaudio.Stream] = None
        self.player: Optional[PCMPlayer] = None

    def initialize(self) -> None:
        """Inisialisasi perangkat audio"""
        print('Initializing audio devices...')
        self.pya = pyaudio.PyAudio()
        self.mic_stream = self.pya.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=AUDIO_SAMPLE_RATE,
            input=True
        )
        self.player = PCMPlayer(self.pya, sample_rate=OUTPUT_AUDIO_SAMPLE_RATE)
        print('Audio devices initialized')

    def read_audio_chunk(self) -> Optional[bytes]:
        """Baca sebuah chunk data audio"""
        if not self.mic_stream:
            return None
        try:
            return self.mic_stream.read(AUDIO_CHUNK_SIZE, exception_on_overflow=False)
        except Exception as e:
            print(f'[Error] Failed to read audio data: {e}')
            return None

    def cleanup(self) -> None:
        """Bersihkan sumber daya audio"""
        print('Cleaning up audio resources...')
        if self.player:
            self.player.shutdown()
        if self.mic_stream:
            self.mic_stream.close()
        if self.pya:
            self.pya.terminate()
        print('Audio resources cleaned up')

# ==================== Penangan Callback ====================
class OmniCallback(OmniRealtimeCallback):
    """Penangan callback percakapan real-time Omni"""

    def __init__(self, audio_manager: AudioManager):
        self.audio_manager = audio_manager
        self.tool_calls: Dict[str, Dict[str, Any]] = {}
        self.all_response_text: str = ''
        self.last_package_time: float = 0
        self.is_first_text: bool = True
        self.is_first_audio: bool = True
        self.conversation: Optional[OmniRealtimeConversation] = None

    def set_conversation(self, conversation: OmniRealtimeConversation) -> None:
        """Atur referensi instans percakapan"""
        self.conversation = conversation

    def on_open(self) -> None:
        """Callback saat koneksi terjalin"""
        print('Connection established')
        self.audio_manager.initialize()
        self.last_package_time = time.time() * 1000
        self.is_first_text = True
        self.is_first_audio = True
        self.tool_calls = {}
        self.all_response_text = ''

    def on_close(self, close_status_code: int, close_msg: str) -> None:
        """Callback saat koneksi ditutup"""
        print(f'Connection closed: code={close_status_code}, msg={close_msg}')
        self.audio_manager.cleanup()
        sys.exit(0)

    def on_event(self, response: Dict[str, Any]) -> None:
        """Tangani callback event"""
        try:
            event_type = response.get('type', '')

            # Sesi dibuat
            if event_type == 'session.created':
                print(f'Session started: {response["session"]["id"]}')

            # Speech-to-text selesai
            elif event_type == 'conversation.item.input_audio_transcription.completed':
                print(f'User question: {response.get("transcript", "")}')

            # Respons teks inkremental
            elif event_type in ('response.audio_transcript.delta', 'response.text.delta'):
                if self.is_first_text:
                    self.is_first_text = False
                    latency = time.time() * 1000 - self.last_package_time
                    print(f'Time to first token (VAD end): {latency:.0f} ms')

                text = response.get('delta', '')
                self.all_response_text += text

            # Respons audio inkremental
            elif event_type == 'response.audio.delta':
                if self.is_first_audio:
                    self.is_first_audio = False
                    latency = time.time() * 1000 - self.last_package_time
                    print(f'Time to first audio (VAD end): {latency:.0f} ms')

                audio_interval = time.time() * 1000 - self.last_package_time
                print(f'Audio interval: {audio_interval:.0f} ms')
                self.last_package_time = time.time() * 1000

                recv_audio_b64 = response.get('delta', '')
                if self.audio_manager.player:
                    self.audio_manager.player.add_data(recv_audio_b64)

            # VAD mendeteksi awal ucapan
            elif event_type == 'input_audio_buffer.speech_started':
                print('====== VAD detected speech start ======')
                if self.audio_manager.player:
                    self.audio_manager.player.cancel_playing()

            # VAD mendeteksi akhir ucapan
            elif event_type == 'input_audio_buffer.speech_stopped':
                print('====== VAD detected speech end ======')
                self.last_package_time = time.time() * 1000
                self.is_first_text = True
                self.is_first_audio = True
                self.tool_calls = {}

            # Argumen panggilan fungsi selesai
            elif event_type == 'response.function_call_arguments.done':
                print('====== Received tool call request ======')
                call_id = response.get('call_id', '')
                self.tool_calls[call_id] = response.copy()
                self.tool_calls[call_id]['processed'] = False

            # Respons selesai
            elif event_type == 'response.done':
                print('====== Response completed ======')
                print(f'Full response: {self.all_response_text}')

                if self.conversation:
                    response_id = self.conversation.get_last_response_id()
                    text_delay = self.conversation.get_last_first_text_delay()
                    audio_delay = self.conversation.get_last_first_audio_delay()

                    # Cetak metrik terperinci hanya jika semua tersedia
                    if response_id is not None and text_delay is not None and audio_delay is not None:
                        print(f'[Metric] Response ID: {response_id}, '
                              f'Time to first token: {text_delay:.0f}ms, '
                              f'Time to first audio: {audio_delay:.0f}ms')
                    else:
                        print('[Metrik] Informasi metrik untuk sementara tidak tersedia (kemungkinan respons setelah panggilan alat)')

                self.all_response_text = ''

        except Exception as e:
            print(f'[Error] Exception handling event: {e}')
            traceback.print_exc()

    def process_pending_tool_calls(self) -> bool:
        """
        Memproses panggilan alat (tool call) yang tertunda

        Returns:
            Apakah ada panggilan alat baru yang memerlukan respons
        """
        has_pending = False

        for call_id, tool_call in self.tool_calls.items():
            if not tool_call.get('processed', False):
                has_pending = True
                tool_call['processed'] = True

                # Tangani panggilan alat (tool call)
                result = handle_tool_call(tool_call)

                # Kirim hasilnya ke server
                if self.conversation:
                    send_tool_call_response(self.conversation, result)

        return has_pending

# ==================== Program Utama ====================
def main():
    """Fungsi utama"""
    print('Initializing Omni real-time conversation...')

    # Buat manajer audio
    audio_manager = AudioManager()

    # Buat penangan callback
    callback = OmniCallback(audio_manager)

    # Buat instans percakapan
    conversation = OmniRealtimeConversation(
        api_key=dashscope.api_key,
        url=WS_URL,
        model=MODEL,
        callback=callback,
    )

    # Atur referensi percakapan di callback
    callback.set_conversation(conversation)

    # Buat koneksi
    conversation.connect()

    # Konfigurasikan parameter sesi
    omni_output_modalities = [MultiModality.AUDIO, MultiModality.TEXT]

    conversation.update_session(
        output_modalities=omni_output_modalities,
        voice=VOICE,
        input_audio_format=AudioFormat.PCM_16000HZ_MONO_16BIT,
        output_audio_format=AudioFormat.PCM_24000HZ_MONO_16BIT,
        enable_input_audio_transcription=True,
        enable_turn_detection=True,
        turn_detection_type='server_vad',
        tools=TOOLS,
    )

    # Siapkan penanganan sinyal
    def signal_handler(sig, frame):
        print('\nMenerima Ctrl+C, menghentikan...')
        conversation.close()
        audio_manager.cleanup()
        print('Percakapan real-time Omni dihentikan')
        sys.exit(0)

    signal.signal(signal.SIGINT, signal_handler)
    print("Tekan Ctrl+C untuk menghentikan percakapan...\n")

    # Loop utama: terus-menerus mengirim audio dan memeriksa panggilan alat (tool call)
    try:
        while True:
            # Proses panggilan alat (tool call) yang tertunda
            has_tool_calls = callback.process_pending_tool_calls()

            if has_tool_calls:
                print("*** Panggilan alat selesai, membuat respons baru ***")
                conversation.create_response(
                    instructions=None,
                    output_modalities=omni_output_modalities
                )
                print('====== Pemrosesan panggilan alat selesai ======\n')

            # Baca dan kirim data audio
            audio_data = audio_manager.read_audio_chunk()
            if audio_data:
                audio_b64 = base64.b64encode(audio_data).decode('ascii')
                conversation.append_audio(audio_b64)
            else:
                break

    except KeyboardInterrupt:
        signal_handler(signal.SIGINT, None)
    except Exception as e:
        print(f'[Error] Main loop exception: {e}')
        traceback.print_exc()
    finally:
        conversation.close()
        audio_manager.cleanup()

if __name__ == '__main__':
    main()