Alibaba Cloud Model Studio:Function calling

## Step 1: Define tool functions

# Add the import for the random module
import random
from datetime import datetime

# Simulate a weather query tool. Example output: "The weather in Beijing today is rainy."
def get_current_weather(arguments):
    # Define a list of alternative weather conditions
    weather_conditions = ["Sunny", "Cloudy", "Rainy"]
    # Randomly select a weather condition
    random_weather = random.choice(weather_conditions)
    # Extract location information from JSON
    location = arguments["location"]
    # Return the formatted weather information
    return f"The weather in {location} today is {random_weather}."

# A tool to query the current time. Example output: "Current time: 2024-04-15 17:15:18."
def get_current_time():
    # Get the current date and time
    current_datetime = datetime.now()
    # Format the current date and time
    formatted_time = current_datetime.strftime('%Y-%m-%d %H:%M:%S')
    # Return the formatted current time
    return f"Current time: {formatted_time}."

# Test the tool functions and print the results. You can remove the following four lines of test code when running the subsequent steps.
print("Testing tool output:")
print(get_current_weather({"location": "Shanghai"}))
print(get_current_time())
print("\n")

// Step 1: Define tool functions

// Import the time query tool
import { format } from 'date-fns';

function getCurrentWeather(args) {
    // Define a list of alternative weather conditions
    const weatherConditions = ["Sunny", "Cloudy", "Rainy"];
    // Randomly select a weather condition
    const randomWeather = weatherConditions[Math.floor(Math.random() * weatherConditions.length)];
    // Extract location information from JSON
    const location = args.location;
    // Return the formatted weather information
    return `The weather in ${location} today is ${randomWeather}.`;
}

function getCurrentTime() {
    // Get the current date and time
    const currentDatetime = new Date();
    // Format the current date and time
    const formattedTime = format(currentDatetime, 'yyyy-MM-dd HH:mm:ss');
    // Return the formatted current time
    return `Current time: ${formattedTime}.`;
}

// Test the tool functions and print the results. You can remove the following four lines of test code when running the subsequent steps.
console.log("Testing tool output:")
console.log(getCurrentWeather({location:"Shanghai"}));
console.log(getCurrentTime());
console.log("\n")

# Paste the following code after the Step 1 code

## Step 2: Create the tools array

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "Useful for when you want to know the current time.",
            "parameters": {}
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Useful for when you want to query the weather in a specific city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "A city or district, such as Beijing, Hangzhou, or Yuhang.",
                    }
                },
                "required": ["location"]
            }
        }
    }
]
tool_name = [tool["function"]["name"] for tool in tools]
print(f"Created {len(tools)} tools: {tool_name}\n")

// Paste the following code after the Step 1 code

// Step 2: Create the tools array

const tools = [
    {
      type: "function",
      function: {
        name: "get_current_time",
        description: "Useful for when you want to know the current time.",
        parameters: {}
      }
    },
    {
      type: "function",
      function: {
        name: "get_current_weather",
        description: "Useful for when you want to query the weather in a specific city.",
        parameters: {
          type: "object",
          properties: {
            location: {
              type: "string",
              description: "A city or district, such as Beijing, Hangzhou, or Yuhang.",
            }
          },
          required: ["location"]
        }
      }
    }
  ];

const toolNames = tools.map(tool => tool.function.name);
console.log(`Created ${tools.length} tools: ${toolNames.join(', ')}\n`);

# Step 3: Create the messages array
# Paste the following code after the Step 2 code
# Example User Message for a text generation model
messages = [
    {
        "role": "system",
        "content": """You are a helpful assistant. If the user asks about the weather, call the 'get_current_weather' function;
     if the user asks about the time, call the 'get_current_time' function.
     Please answer the questions in a friendly tone.""",
    },
    {
        "role": "user",
        "content": "Weather in Shanghai"
    }
]

# Example User Message for a multimodal model
# messages=[
#  {
#         "role": "system",
#         "content": """You are a helpful assistant. If the user asks about the weather, call the 'get_current_weather' function;
#      if the user asks about the time, call the 'get_current_time' function.
#      Please answer the questions in a friendly tone.""",
#     },
#     {"role": "user",
#      "content": [{"type": "image_url","image_url": {"url": "https://img.alicdn.com/imgextra/i2/O1CN01FbTJon1ErXVGMRdsN_!!6000000000405-0-tps-1024-683.jpg"}},
#                  {"type": "text", "text": "Query the current weather for the location in the image"}]},
# ]

print("messages array created\n")

// Step 3: Create the messages array
// Paste the following code after the Step 2 code
const messages = [
    {
        role: "system",
        content: "You are a helpful assistant. If the user asks about the weather, call the 'get_current_weather' function; if the user asks about the time, call the 'get_current_time' function. Please answer the questions in a friendly tone.",
    },
    {
        role: "user",
        content: "Weather in Shanghai"
    }
];
// Example User Message for a multimodal model,
// const messages: [{
//     role: "user",
//     content: [{type: "image_url", image_url: {"url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"}},
//               {type: "text", text: "What is depicted in the image?"}]
//   }];

console.log("messages array created\n");

# Step 4: Make a function calling
# Paste the following code after the Step 3 code
from openai import OpenAI
import os

client = OpenAI(
    # API keys vary by region. To get an API key, see https://www.alibabacloud.com/help/en/model-studio/get-api-key
    # If you have not configured the environment variable, replace the following line with your Model Studio API key: api_key="sk-xxx",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # Singapore region. Replace {WorkspaceId} with your actual Workspace ID. URLs vary by region.
    base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1",
)

def function_calling():
    completion = client.chat.completions.create(
        # This example uses qwen3.8-max. You can change the model name as needed. For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models
        model="qwen3.8-max",
        extra_body={"enable_thinking": False},
        messages=messages,
        tools=tools
    )
    print("Returned object:")
    print(completion.choices[0].message.model_dump_json())
    print("\n")
    return completion

print("Making a function calling...")
completion = function_calling()

// Step 4: Make a function calling
// Paste the following code after the Step 3 code
import OpenAI from "openai";
const openai = new OpenAI(
    {
        // API keys vary by region. To get an API key, see https://www.alibabacloud.com/help/en/model-studio/get-api-key
        // If you have not configured the environment variable, replace the following line with your Model Studio API key: apiKey: "sk-xxx",
        apiKey: process.env.DASHSCOPE_API_KEY,
        // Singapore region. Replace {WorkspaceId} with your actual Workspace ID. URLs vary by region.
        baseURL: "https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1"
    }
);

async function functionCalling() {
    const completion = await openai.chat.completions.create({
        model: "qwen3.8-max",  // This example uses qwen3.8-max. You can change the model name as needed. For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models
        enable_thinking: false,
        messages: messages,
        tools: tools
    });
    console.log("Returned object:");
    console.log(JSON.stringify(completion.choices[0].message));
    console.log("\n");
    return completion;
}

const completion = await functionCalling();

# Step 5: Run the tool function
# Paste the following code after the Step 4 code
import json

print("Running the tool function...")
# Get the function name and input parameters from the returned result
function_name = completion.choices[0].message.tool_calls[0].function.name
arguments_string = completion.choices[0].message.tool_calls[0].function.arguments

# Use the json module to parse the parameter string
arguments = json.loads(arguments_string)
# Create a function mapping table
function_mapper = {
    "get_current_weather": get_current_weather,
    "get_current_time": get_current_time
}
# Get the function entity
function = function_mapper[function_name]
# If the input parameter is empty, call the function directly
if arguments == {}:
    function_output = function()
# Otherwise, pass the parameters and then call the function
else:
    function_output = function(arguments)
# Print the tool's output
print(f"Tool function output: {function_output}\n")

// Step 5: Run the tool function
// Paste the following code after the Step 4 code

console.log("Running the tool function...");
const function_name = completion.choices[0].message.tool_calls[0].function.name;
const arguments_string = completion.choices[0].message.tool_calls[0].function.arguments;

// Use the JSON module to parse the parameter string
const args = JSON.parse(arguments_string);

// Create a function mapping table
const functionMapper = {
    "get_current_weather": getCurrentWeather,
    "get_current_time": getCurrentTime
};

// Get the function entity
const func = functionMapper[function_name];

// If the input parameter is empty, call the function directly
let functionOutput;
if (Object.keys(args).length === 0) {
    functionOutput = func();
} else {
    // Otherwise, pass the parameters and then call the function
    functionOutput = func(args);
}

// Print the tool's output
console.log(`Tool function output: ${functionOutput}\n`);

# Step 6: Submit the tool output to the LLM
# Paste the following code after the Step 5 code

messages.append(completion.choices[0].message)
print("Assistant message added")
messages.append({"role": "tool", "content": function_output, "tool_call_id": completion.choices[0].message.tool_calls[0].id})
print("Tool message added\n")

// Step 6: Submit the tool output to the LLM
// Paste the following code after the Step 5 code

messages.push(completion.choices[0].message);
console.log("Assistant message added")
messages.push({
    "role": "tool",
    "content": functionOutput,
    "tool_call_id": completion.choices[0].message.tool_calls[0].id
});
console.log("Tool message added\n");

# Step 7: Let the LLM summarize the tool output
# Paste the following code after the Step 6 code
print("Summarizing the tool output...")
completion = function_calling()

// Step 7: Let the LLM summarize the tool output
// Paste the following code after the Step 6 code

console.log("Summarizing the tool output...");
const completion_1 = await functionCalling();

def function_calling():
    completion = client.chat.completions.create(
        model="qwen3.8-max",  # This example uses qwen3.8-max. You can change the model name as needed.
        extra_body={"enable_thinking": False},
        messages=messages,
        tools=tools,
        # New parameter
        parallel_tool_calls=True
    )
    print("Returned object:")
    print(completion.choices[0].message.model_dump_json())
    print("\n")
    return completion

print("Making a function calling...")
completion = function_calling()

async function functionCalling() {
    const completion = await openai.chat.completions.create({
        model: "qwen3.8-max",  // This example uses qwen3.8-max. You can change the model name as needed.
        enable_thinking: false,
        messages: messages,
        tools: tools,
        parallel_tool_calls: true
    });
    console.log("Returned object:");
    console.log(JSON.stringify(completion.choices[0].message));
    console.log("\n");
    return completion;
}

const completion = await functionCalling();

def function_calling():
    completion = client.chat.completions.create(
        model="qwen3.8-max",
        extra_body={"enable_thinking": False},
        messages=messages,
        tools=tools,
        tool_choice="required"
    )
    print(completion.model_dump_json())

function_calling()

async function functionCalling() {
    const completion = await openai.chat.completions.create({
        model: "qwen3.8-max",
        enable_thinking: false,
        messages: messages,
        tools: tools,
        tool_choice: "required"
    });
    console.log("Returned object:");
    console.log(JSON.stringify(completion.choices[0].message));
    console.log("\n");
    return completion;
}

const completion = await functionCalling();

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # Singapore region. Replace {WorkspaceId} with your actual Workspace ID. URLs vary by region.
    base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1",
)
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Useful for when you want to query the weather in a specific city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "A city or district, such as Beijing, Hangzhou, or Yuhang.",
                    }
                },
                "required": ["location"],
            },
        },
    },
]

stream = client.chat.completions.create(
    model="qwen3.8-max",
    extra_body={"enable_thinking": False},
    messages=[{"role": "user", "content": "Weather in Hangzhou?"}],
    tools=tools,
    stream=True
)

for chunk in stream:
    delta = chunk.choices[0].delta
    print(delta.tool_calls)

import { OpenAI } from "openai";

const openai = new OpenAI(
    {
        // API keys vary by region. To get an API key, see https://www.alibabacloud.com/help/en/model-studio/get-api-key
        // If you have not configured the environment variable, replace the following line with your Model Studio API key: apiKey: "sk-xxx",
        apiKey: process.env.DASHSCOPE_API_KEY,
        // China (Beijing) region. Replace {WorkspaceId} with your actual Workspace ID. URLs vary by region.
        baseURL: "https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
    }
);
const tools = [
    {
        "type": "function",
        "function": {
            "name": "getCurrentWeather",
            "description": "Useful for when you want to query the weather in a specific city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "A city or district, such as Beijing, Hangzhou, or Yuhang."
                    }
                },
                "required": ["location"]
            }
        }
    }
];

const stream = await openai.chat.completions.create({
    model: "qwen3.8-max",
    enable_thinking: false,
    messages: [{ role: "user", content: "Weather in Beijing" }],
    tools: tools,
    stream: true,
});

for await (const chunk of stream) {
    const delta = chunk.choices[0].delta;
    console.log(delta.tool_calls);
}

tool_calls = {}
for response_chunk in stream:
    delta_tool_calls = response_chunk.choices[0].delta.tool_calls
    if delta_tool_calls:
        for tool_call_chunk in delta_tool_calls:
            call_index = tool_call_chunk.index
            tool_call_chunk.function.arguments = tool_call_chunk.function.arguments or ""
            if call_index not in tool_calls:
                tool_calls[call_index] = tool_call_chunk
            else:
                tool_calls[call_index].function.arguments += tool_call_chunk.function.arguments
print(tool_calls[0].model_dump_json())

const toolCalls = {};
for await (const responseChunk of stream) {
  const deltaToolCalls = responseChunk.choices[0]?.delta?.tool_calls;
  if (deltaToolCalls) {
    for (const toolCallChunk of deltaToolCalls) {
      const index = toolCallChunk.index;
      toolCallChunk.function.arguments = toolCallChunk.function.arguments || "";
      if (!toolCalls[index]) {
        toolCalls[index] = { ...toolCallChunk };
        if (!toolCalls[index].function) {
            toolCalls[index].function = { name: '', arguments: '' };
        }
      }
      else if (toolCallChunk.function?.arguments) {
        toolCalls[index].function.arguments += toolCallChunk.function.arguments;
      }
    }
  }
}
console.log(JSON.stringify(toolCalls[0]));

from openai import OpenAI
import json
import os
import random

# Initialize the client
client = OpenAI(
    # If you have not configured the environment variable, replace the following line with your Model Studio API key: api_key="sk-xxx",
    # API keys vary by region. To get an API key, see https://www.alibabacloud.com/help/en/model-studio/get-api-key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1",
)
# Simulate a user question
USER_QUESTION = "What's the weather like in Singapore?"
# Define the tool list
tools = [
    {
        "type": "function",
        "name": "get_current_weather",
        "description": "Useful for when you want to query the weather in a specific city.",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "A city or district, such as Singapore or London.",
                }
            },
            "required": ["location"],
        },
    }
]

# Simulate a weather query tool
def get_current_weather(arguments):
    weather_conditions = ["Sunny", "Cloudy", "Rainy"]
    random_weather = random.choice(weather_conditions)
    location = arguments["location"]
    return f"The weather in {location} today is {random_weather}."

# Encapsulate the model response function
def get_response(input_data):
    response = client.responses.create(
        model="qwen3.8-max",
        extra_body={"enable_thinking": False},
        input=input_data,
        tools=tools,
    )
    return response

# Maintain the conversation context
conversation = [{"role": "user", "content": USER_QUESTION}]

response = get_response(conversation)
function_calls = [item for item in response.output if item.type == "function_call"]
# If no tool call is needed, print the content directly
if not function_calls:
    print(f"Assistant's final response: {response.output_text}")
else:
    # Enter the tool calling loop
    while function_calls:
        for fc in function_calls:
            func_name = fc.name
            arguments = json.loads(fc.arguments)
            print(f"Calling tool [{func_name}], arguments: {arguments}")
            # Run the tool
            tool_result = get_current_weather(arguments)
            print(f"Tool returns: {tool_result}")
            # Append the tool call and result as a pair to the context
            conversation.append(
                {
                    "type": "function_call",
                    "name": fc.name,
                    "arguments": fc.arguments,
                    "call_id": fc.call_id,
                }
            )
            conversation.append(
                {
                    "type": "function_call_output",
                    "call_id": fc.call_id,
                    "output": tool_result,
                }
            )
        # Call the model again with the full context
        response = get_response(conversation)
        function_calls = [
            item for item in response.output if item.type == "function_call"
        ]
    print(f"Assistant's final response: {response.output_text}")

import OpenAI from "openai";

// Initialize the client
const openai = new OpenAI({
  // API keys vary by region. To get an API key, see https://www.alibabacloud.com/help/en/model-studio/get-api-key
  // If you have not configured the environment variable, replace the following line with your Model Studio API key: apiKey: "sk-xxx",
  apiKey: process.env.DASHSCOPE_API_KEY,
  baseURL:
    "https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1",
});

// Define the tool list
const tools = [
  {
    type: "function",
    name: "get_current_weather",
    description: "Useful for when you want to query the weather in a specific city.",
    parameters: {
      type: "object",
      properties: {
        location: {
          type: "string",
          description: "A city or district, such as Singapore or London.",
        },
      },
      required: ["location"],
    },
  },
];

// Simulate a weather query tool
const getCurrentWeather = (args) => {
  const weatherConditions = ["Sunny", "Cloudy", "Rainy"];
  const randomWeather =
    weatherConditions[Math.floor(Math.random() * weatherConditions.length)];
  const location = args.location;
  return `The weather in ${location} today is ${randomWeather}.`;
};

// Encapsulate the model response function
const getResponse = async (inputData) => {
  const response = await openai.responses.create({
    model: "qwen3.8-max",
    enable_thinking: false,
    input: inputData,
    tools: tools,
  });
  return response;
};

const main = async () => {
  const userQuestion = "Weather in Singapore";

  // Maintain the conversation context
  const conversation = [{ role: "user", content: userQuestion }];

  let response = await getResponse(conversation);
  let functionCalls = response.output.filter(
    (item) => item.type === "function_call"
  );
  // If no tool call is needed, print the content directly
  if (functionCalls.length === 0) {
    console.log(`Assistant's final response: ${response.output_text}`);
  } else {
    // Enter the tool calling loop
    while (functionCalls.length > 0) {
      for (const fc of functionCalls) {
        const funcName = fc.name;
        const args = JSON.parse(fc.arguments);
        console.log(`Calling tool [${funcName}], arguments:`, args);
        // Run the tool
        const toolResult = getCurrentWeather(args);
        console.log(`Tool returns: ${toolResult}`);
        // Append the tool call and result as a pair to the context
        conversation.push({
          type: "function_call",
          name: fc.name,
          arguments: fc.arguments,
          call_id: fc.call_id,
        });
        conversation.push({
          type: "function_call_output",
          call_id: fc.call_id,
          output: toolResult,
        });
      }
      // Call the model again with the full context
      response = await getResponse(conversation);
      functionCalls = response.output.filter(
        (item) => item.type === "function_call"
      );
    }
    console.log(`Assistant's final response: ${response.output_text}`);
  }
};

// Start the program
main().catch(console.error);

from openai import OpenAI
import os

client = OpenAI(
    # API keys vary by region. To get an API key, see https://www.alibabacloud.com/help/en/model-studio/get-api-key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # Singapore region. Replace {WorkspaceId} with your actual Workspace ID. URLs vary by region.
    base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1",
)
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Useful for when you want to query the weather in a specific city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "A city or district, such as Beijing, Hangzhou, or Yuhang.",
                    }
                },
                "required": ["location"],
            },
        },
    },
]

completion = client.chat.completions.create(
    model="qwen3.5-omni-plus",
    messages=[{"role": "user", "content": "Weather in Hangzhou?"}],

    # Set the modality of the output data. Valid values: ["text"], ["text","audio"]. We recommend setting it to ["text"].
    modalities=["text"],

    # stream must be set to True, otherwise an error will occur.
    stream=True,
    tools=tools
)

for chunk in completion:
    # If the output includes the audio modality, change the following condition to: if chunk.choices and not hasattr(chunk.choices[0].delta, "audio"):
    if chunk.choices:
        delta = chunk.choices[0].delta
        print(delta.tool_calls)

import { OpenAI } from "openai";

const openai = new OpenAI(
    {
        // API keys vary by region. To get an API key, see https://www.alibabacloud.com/help/en/model-studio/get-api-key
        // If you have not configured the environment variable, replace the following line with your Model Studio API key: apiKey: "sk-xxx",
        apiKey: process.env.DASHSCOPE_API_KEY,
        // Singapore region. Replace {WorkspaceId} with your actual Workspace ID. URLs vary by region.
        baseURL: "https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1"
    }
);
const tools = [
    {
        "type": "function",
        "function": {
            "name": "getCurrentWeather",
            "description": "Useful for when you want to query the weather in a specific city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "A city or district, such as Beijing, Hangzhou, or Yuhang."
                    }
                },
                "required": ["location"]
            }
        }
    }
];

const stream = await openai.chat.completions.create({
    model: "qwen3-omni-flash",
    messages: [
        {
            "role": "user",
            "content": "Weather in Hangzhou"
        }],
    stream: true,
    // Set the modality of the output data. Valid values: ["text"], ["text","audio"]. We recommend setting it to ["text"].
    modalities: ["text"],
    tools:tools
});

for await (const chunk of stream) {
    // If the output includes audio, replace the conditional statement with: if (chunk.choices?.length && chunk.choices[0].delta && !('audio' in chunk.choices[0].delta))
    if (chunk.choices?.length){
    const delta = chunk.choices[0].delta;
    console.log(delta.tool_calls);
}}

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 *

# ==================== Constant Definitions ====================
VOICE = 'Tina'
MODEL = "qwen3.5-omni-plus-realtime"
# To access the Beijing region, replace WS_URL with: 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"
# Configure the API key. If you have not set the environment variable, replace the following line with your API key: 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

# ==================== Tool Definitions ====================
def get_train_price(src: str, dst: str) -> str:
    """Query train ticket prices"""
    return f"The train ticket price from {src} to {dst} is 100-200 CNY."

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

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

# Unified OpenAI format tool definitions
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Useful for when you want to query the weather in a specific city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "A city or district, such as Beijing, Hangzhou, or Yuhang.",
                    }
                },
                "required": ["location"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_flight_price",
            "description": "Useful for when you want to query flight ticket prices.",
            "parameters": {
                "type": "object",
                "properties": {
                    "src": {
                        "type": "string",
                        "description": "The departure city of the flight, such as Beijing or Hangzhou.",
                    },
                    "dst": {
                        "type": "string",
                        "description": "The arrival city of the flight, such as Beijing or Hangzhou.",
                    },
                },
                "required": ["src", "dst"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_train_price",
            "description": "Useful for when you want to query train ticket prices.",
            "parameters": {
                "type": "object",
                "properties": {
                    "src": {
                        "type": "string",
                        "description": "The departure city of the train, such as Beijing or Hangzhou.",
                    },
                    "dst": {
                        "type": "string",
                        "description": "The arrival city of the train, such as Beijing or Hangzhou.",
                    },
                },
                "required": ["src", "dst"],
            },
        },
    },
]

# Mapping of tool names to functions
TOOL_FUNCTIONS = {
    "get_current_weather": get_current_weather,
    "get_flight_price": get_flight_price,
    "get_train_price": get_train_price,
}

# ==================== Tool Call Handling ====================
def handle_tool_call(tool_call_response: Dict[str, Any]) -> Dict[str, Any]:
    """
    Handles a tool call request

    Args:
        tool_call_response: Tool call information containing name, arguments, and call_id

    Returns:
        Updated tool call response containing the output field
    """
    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}')

        # Find the corresponding function
        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

        # Call the function
        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:
    """Sends the tool call result to the server"""
    conversation.create_item({
        "id": 'item_' + uuid.uuid4().hex,
        "type": "function_call_output",
        "call_id": response['call_id'],
        "output": response["output"],
    })

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

    Uses a dual-thread architecture for real-time audio playback:
    - Decoding thread: Decodes base64-encoded audio data into raw PCM data
    - Playback thread: Writes PCM data to the audio output device

    Supports dynamically adding audio data, canceling playback, saving audio files, etc.
    """

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

        Args:
            pya: pyaudio.PyAudio instance
            sample_rate: Audio sampling rate (Hz), default 24000
            chunk_size_ms: Audio chunk size (milliseconds), affects playback cancellation latency, default 100ms
            save_file: Whether to save the played audio to a 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):
        """Decoding thread: Decodes base64 audio data into raw PCM data"""
        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)
            # push raw audio data into queue by 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):
        """Playback thread: Writes PCM data to the audio output device"""
        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
            # write chunk to pyaudio audio player, wait until finish playing this chunk.
            self.player_stream.write(recv_audio_raw)

    def cancel_playing(self):
        """Cancel playback: Clear all buffer queues"""
        self.b64_audio_buffer.queue.clear()
        self.raw_audio_buffer.queue.clear()

    def add_data(self, data):
        """Add base64-encoded audio data to the playback queue"""
        self.b64_audio_buffer.put(data)

    def wait_for_complete(self):
        """Wait for playback to complete"""
        self.complete_event = threading.Event()
        self.complete_event.wait()
        self.complete_event = None

    def shutdown(self):
        """Shut down the player and release resources"""
        self.status = 'stop'
        self.decoder_thread.join()
        self.player_thread.join()
        self.player_stream.close()
        if self.save_file:
            self.out_file.close()

# ==================== Audio Manager ====================
class AudioManager:
    """Manages audio input and output resources"""

    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:
        """Initialize audio devices"""
        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]:
        """Read an audio data chunk"""
        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:
        """Clean up audio resources"""
        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')

# ==================== Callback Handler ====================
class OmniCallback(OmniRealtimeCallback):
    """Omni real-time conversation callback handler"""

    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:
        """Set the conversation instance reference"""
        self.conversation = conversation

    def on_open(self) -> None:
        """Callback on connection establishment"""
        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 on connection closure"""
        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:
        """Handle event callbacks"""
        try:
            event_type = response.get('type', '')

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

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

            # Incremental text response
            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

            # Incremental audio response
            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 detected speech start
            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 detected speech end
            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 = {}

            # Function call arguments completed
            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

            # Response completed
            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()

                    # Print detailed metrics only when all are available
                    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('[Metric] Metric information is temporarily unavailable (possibly a response after a tool call)')

                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:
        """
        Processes pending tool calls

        Returns:
            Whether there are new tool calls that need a response
        """
        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

                # Handle the tool call
                result = handle_tool_call(tool_call)

                # Send the result to the server
                if self.conversation:
                    send_tool_call_response(self.conversation, result)

        return has_pending

# ==================== Main Program ====================
def main():
    """Main function"""
    print('Initializing Omni real-time conversation...')

    # Create an audio manager
    audio_manager = AudioManager()

    # Create a callback handler
    callback = OmniCallback(audio_manager)

    # Create a conversation instance
    conversation = OmniRealtimeConversation(
        api_key=dashscope.api_key,
        url=WS_URL,
        model=MODEL,
        callback=callback,
    )

    # Set the conversation reference in the callback
    callback.set_conversation(conversation)

    # Establish the connection
    conversation.connect()

    # Configure session parameters
    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,
    )

    # Set up signal handling
    def signal_handler(sig, frame):
        print('\nReceived Ctrl+C, stopping...')
        conversation.close()
        audio_manager.cleanup()
        print('Omni real-time conversation stopped')
        sys.exit(0)

    signal.signal(signal.SIGINT, signal_handler)
    print("Press Ctrl+C to stop the conversation...\n")

    # Main loop: continuously send audio and check for tool calls
    try:
        while True:
            # Process pending tool calls
            has_tool_calls = callback.process_pending_tool_calls()

            if has_tool_calls:
                print("*** Tool call completed, creating new response ***")
                conversation.create_response(
                    instructions=None,
                    output_modalities=omni_output_modalities
                )
                print('====== Tool call processing completed ======\n')

            # Read and send audio data
            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()

import com.alibaba.dashscope.audio.omni.*;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import javax.sound.sampled.*;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;

public class Main {

    public static void main(String[] args) {
        try {
            // Initialize components
            AudioPlayer audioPlayer = new AudioPlayer();
            ToolRegistry toolRegistry = new ToolRegistry();
            ConversationHandler handler = new ConversationHandler(audioPlayer, toolRegistry);

            // Create and configure the session
            OmniRealtimeParam param = OmniRealtimeParam.builder()
                    .model("qwen3.5-omni-plus-realtime")
                    .apikey(System.getenv("DASHSCOPE_API_KEY"))
                    // To access the Beijing region, replace the url with: wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime
                    .url("wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime")
                    .build();

            OmniRealtimeConversation conversation = new OmniRealtimeConversation(param, handler);
            conversation.connect();

            // Configure session parameters
            configureSession(conversation, toolRegistry);

            // Start audio capture
            startAudioCapture(conversation, handler);

            // Clean up resources
            cleanup(conversation, audioPlayer);

        } catch (NoApiKeyException e) {
            System.err.println("API KEY not found: Please set the DASHSCOPE_API_KEY environment variable");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void configureSession(OmniRealtimeConversation conversation, ToolRegistry toolRegistry) {
        HashMap<String, Object> additionalConfig = new HashMap<>();
        additionalConfig.put("tools", toolRegistry.buildToolsDefinition());

        conversation.updateSession(OmniRealtimeConfig.builder()
                .modalities(Arrays.asList(OmniRealtimeModality.AUDIO, OmniRealtimeModality.TEXT))
                .voice("Tina")
                .enableTurnDetection(true)
                .enableInputAudioTranscription(true)
                .parameters(additionalConfig)
                .build());

        System.out.println("Tool calling is enabled. Please start speaking (Press Ctrl+C to exit)...");
    }

    private static void startAudioCapture(OmniRealtimeConversation conversation, ConversationHandler handler)
            throws LineUnavailableException {
        AudioFormat format = new AudioFormat(16000, 16, 1, true, false);
        TargetDataLine mic = AudioSystem.getTargetDataLine(format);
        mic.open(format);
        mic.start();

        ByteBuffer buffer = ByteBuffer.allocate(3200);
        while (!handler.getShouldStop().get()) {
            int bytesRead = mic.read(buffer.array(), 0, buffer.capacity());
            if (bytesRead > 0) {
                conversation.appendAudio(Base64.getEncoder().encodeToString(buffer.array()));

                // Check and process pending tool calls
                if (handler.hasPendingToolCalls()) {
                    System.out.println("*** create response after call tools");
                    handler.processPendingToolCalls(conversation);
                    conversation.createResponse(null, Arrays.asList(OmniRealtimeModality.AUDIO, OmniRealtimeModality.TEXT));
                    System.out.println("======TOOL CALL END======");
                }
            }
            try {
                Thread.sleep(20);
            } catch (InterruptedException ignored) {}
        }

        mic.close();
    }

    private static void cleanup(OmniRealtimeConversation conversation, AudioPlayer audioPlayer) {
        try {
            conversation.close(1000, "Normal exit");
            audioPlayer.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * Audio Player - Responsible for sequential playback of audio data
     */
    static class AudioPlayer {
        private final SourceDataLine line;
        private final Queue<byte[]> audioQueue = new ConcurrentLinkedQueue<>();
        private final Thread playerThread;
        private final AtomicBoolean shouldStop = new AtomicBoolean(false);

        public AudioPlayer() throws LineUnavailableException {
            AudioFormat format = new AudioFormat(24000, 16, 1, true, false);
            line = AudioSystem.getSourceDataLine(format);
            line.open(format);
            line.start();

            playerThread = new Thread(this::playLoop, "AudioPlayer");
            playerThread.start();
        }

        private void playLoop() {
            while (!shouldStop.get()) {
                byte[] audio = audioQueue.poll();
                if (audio != null) {
                    line.write(audio, 0, audio.length);
                } else {
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException ignored) {}
                }
            }
        }

        public void play(String base64Audio) {
            audioQueue.add(Base64.getDecoder().decode(base64Audio));
        }

        public void close() {
            shouldStop.set(true);
            try {
                playerThread.join(1000);
            } catch (InterruptedException ignored) {}
            line.drain();
            line.close();
        }
    }

    /**
     * Tool Registry - Manages available tools and their implementations
     */
    static class ToolRegistry {
        private final Map<String, Function<JsonObject, String>> tools = new ConcurrentHashMap<>();
        private final Map<String, JsonObject> pendingToolCalls = new ConcurrentHashMap<>();

        public ToolRegistry() {
            registerDefaultTools();
        }

        private void registerDefaultTools() {
            registerTool("get_current_weather", this::getCurrentWeather);
            registerTool("get_flight_price", this::getFlightPrice);
            registerTool("get_train_price", this::getTrainPrice);
        }

        public void registerTool(String name, Function<JsonObject, String> handler) {
            tools.put(name, handler);
        }

        /**
         * Build tool definitions (OpenAI format)
         */
        public List<Map<String, Object>> buildToolsDefinition() {
            List<Map<String, Object>> definitions = new ArrayList<>();

            definitions.add(createFunctionDefinition(
                    "get_current_weather",
                    "Useful for when you want to query the weather in a specific city.",
                    createParamsSchema(
                            Collections.singletonMap("location",
                                    createProperty("string", "A city or district, such as Beijing, Hangzhou, or Yuhang.")),
                            Collections.singletonList("location")
                    )
            ));

            Map<String, Object> flightProps = new HashMap<>();
            flightProps.put("src", createProperty("string", "The departure city of the flight, such as Beijing or Hangzhou."));
            flightProps.put("dst", createProperty("string", "The arrival city of the flight, such as Beijing or Hangzhou."));
            definitions.add(createFunctionDefinition(
                    "get_flight_price",
                    "Useful for when you want to query flight ticket prices.",
                    createParamsSchema(flightProps, Arrays.asList("src", "dst"))
            ));

            Map<String, Object> trainProps = new HashMap<>();
            trainProps.put("src", createProperty("string", "The departure city of the train, such as Beijing or Hangzhou."));
            trainProps.put("dst", createProperty("string", "The arrival city of the train, such as Beijing or Hangzhou."));
            definitions.add(createFunctionDefinition(
                    "get_train_price",
                    "Useful for when you want to query train ticket prices.",
                    createParamsSchema(trainProps, Arrays.asList("src", "dst"))
            ));

            return definitions;
        }

        private Map<String, Object> createFunctionDefinition(String name, String description, Map<String, Object> parameters) {
            Map<String, Object> function = new HashMap<>();
            function.put("name", name);
            function.put("description", description);
            function.put("parameters", parameters);

            Map<String, Object> tool = new HashMap<>();
            tool.put("type", "function");
            tool.put("function", function);
            return tool;
        }

        private Map<String, Object> createParamsSchema(Map<String, Object> properties, List<String> required) {
            Map<String, Object> schema = new HashMap<>();
            schema.put("type", "object");
            schema.put("properties", properties);
            schema.put("required", required);
            return schema;
        }

        private Map<String, Object> createProperty(String type, String description) {
            Map<String, Object> prop = new HashMap<>();
            prop.put("type", type);
            prop.put("description", description);
            return prop;
        }

        /**
         * Add a tool call to the pending queue
         */
        public void addPendingToolCall(String callId, JsonObject toolCall) {
            pendingToolCalls.put(callId, toolCall);
        }

        /**
         * Check if there are pending tool calls
         */
        public boolean hasPendingToolCalls() {
            return !pendingToolCalls.isEmpty();
        }

        /**
         * Process all pending tool calls
         */
        public void processPendingToolCalls(OmniRealtimeConversation conversation) {
            if (pendingToolCalls.isEmpty()) {
                return;
            }

            for (Map.Entry<String, JsonObject> entry : pendingToolCalls.entrySet()) {
                String callId = entry.getKey();
                JsonObject toolCall = entry.getValue();

                String result = executeTool(toolCall);
                sendToolResult(conversation, callId, result);
            }

            pendingToolCalls.clear();
        }

        private String executeTool(JsonObject toolCall) {
            String functionName = toolCall.get("name").getAsString();
            JsonObject arguments = new Gson().fromJson(
                    toolCall.get("arguments").getAsString(),
                    JsonObject.class
            );

            System.out.println("[Tool Call] start handling: " + functionName + ", args: " + arguments);

            Function<JsonObject, String> handler = tools.get(functionName);
            if (handler == null) {
                return "Client did not find this tool. Call failed.";
            }

            String result = handler.apply(arguments);
            System.out.println("[Tool Call] response: " + result);
            return result;
        }

        private void sendToolResult(OmniRealtimeConversation conversation, String callId, String output) {
            JsonObject item = new JsonObject();
            item.addProperty("id", "item_" + UUID.randomUUID().toString().replace("-", ""));
            item.addProperty("type", "function_call_output");
            item.addProperty("call_id", callId);
            item.addProperty("output", output);

            conversation.createItem(item);
        }

        // ===== Tool Implementations =====

        private String getCurrentWeather(JsonObject args) {
            String location = args.get("location").getAsString();
            return "The weather in " + location + " today is changing from haze to sunny, with a temperature of 4/-4°C and a light breeze.";
        }

        private String getFlightPrice(JsonObject args) {
            String src = args.get("src").getAsString();
            String dst = args.get("dst").getAsString();
            return "The flight ticket price from " + src + " to " + dst + " is 200-300 USD.";
        }

        private String getTrainPrice(JsonObject args) {
            String src = args.get("src").getAsString();
            String dst = args.get("dst").getAsString();
            return "invalid apikey error";
        }
    }

    /**
     * Conversation Handler - Handles WebSocket events
     */
    static class ConversationHandler extends OmniRealtimeCallback {
        private final AudioPlayer audioPlayer;
        private final ToolRegistry toolRegistry;
        private final AtomicBoolean shouldStop = new AtomicBoolean(false);
        private final AtomicReference<StringBuilder> responseTextRef = new AtomicReference<>(new StringBuilder());

        private long lastPackageTime = 0;
        private boolean isFirstText = true;
        private boolean isFirstAudio = true;

        public ConversationHandler(AudioPlayer audioPlayer, ToolRegistry toolRegistry) {
            this.audioPlayer = audioPlayer;
            this.toolRegistry = toolRegistry;
        }

        public AtomicBoolean getShouldStop() {
            return shouldStop;
        }

        @Override
        public void onOpen() {
            System.out.println("Connection established");
        }

        @Override
        public void onClose(int code, String reason) {
            System.out.println("Connection closed");
            shouldStop.set(true);
        }

        @Override
        public void onEvent(JsonObject message) {
            String type = message.get("type").getAsString();

            switch (type) {
                case "session.created":
                    handleSessionCreated(message);
                    break;
                case "conversation.item.input_audio_transcription.completed":
                    handleTranscriptionCompleted(message);
                    break;
                case "response.audio_transcript.delta":
                case "response.text.delta":
                    handleTextDelta(message);
                    break;
                case "response.audio.delta":
                    handleAudioDelta(message);
                    break;
                case "input_audio_buffer.speech_started":
                    handleSpeechStarted();
                    break;
                case "input_audio_buffer.speech_stopped":
                    handleSpeechStopped();
                    break;
                case "response.function_call_arguments.done":
                    handleFunctionCall(message);
                    break;
                case "response.done":
                    handleResponseDone();
                    break;
                default:
                    break;
            }
        }

        private void handleSessionCreated(JsonObject message) {
            String sessionId = message.get("session").getAsJsonObject().get("id").getAsString();
            System.out.println("start session: " + sessionId);
        }

        private void handleTranscriptionCompleted(JsonObject message) {
            System.out.println("question: " + message.get("transcript").getAsString());
        }

        private void handleTextDelta(JsonObject message) {
            if (isFirstText) {
                isFirstText = false;
                System.out.println("first text latency from vad end: " +
                        (System.currentTimeMillis() - lastPackageTime) + " ms");
            }
            String text = message.get("delta").getAsString();
            responseTextRef.get().append(text);
        }

        private void handleAudioDelta(JsonObject message) {
            if (isFirstAudio) {
                isFirstAudio = false;
                System.out.println("first audio latency from vad end: " +
                        (System.currentTimeMillis() - lastPackageTime) + " ms");
            }
            System.out.println("audio interval: " + (System.currentTimeMillis() - lastPackageTime) + " ms");
            lastPackageTime = System.currentTimeMillis();
            audioPlayer.play(message.get("delta").getAsString());
        }

        private void handleSpeechStarted() {
            System.out.println("======VAD Speech Start======");
        }

        private void handleSpeechStopped() {
            System.out.println("======VAD Speech End======");
            lastPackageTime = System.currentTimeMillis();
            isFirstText = true;
            isFirstAudio = true;
        }

        private void handleFunctionCall(JsonObject message) {
            System.out.println("======TOOL CALL======");
            String callId = message.get("call_id").getAsString();
            toolRegistry.addPendingToolCall(callId, message);
        }

        private void handleResponseDone() {
            System.out.println("======RESPONSE DONE======");
            System.out.println("all response text: " + responseTextRef.get());
            responseTextRef.set(new StringBuilder());
        }

        /**
         * Check if there are pending tool calls
         */
        public boolean hasPendingToolCalls() {
            return toolRegistry.hasPendingToolCalls();
        }

        /**
         * Process all pending tool calls
         */
        public void processPendingToolCalls(OmniRealtimeConversation conversation) {
            toolRegistry.processPendingToolCalls(conversation);
        }
    }
}

import asyncio
import json
import base64
import os
import pyaudio
import websockets

# ==================== Constant Definitions ====================
API_KEY = os.getenv("DASHSCOPE_API_KEY")
# To access the Beijing region, replace with:
# wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime
URL = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime"
MODEL = "qwen3.5-omni-plus-realtime"
VOICE = "Ethan"

# ==================== Tool Definitions ====================
def get_current_weather(location):
    """Query the weather in a specific city"""
    return f"The weather in {location} today is changing from haze to sunny, with a temperature of 4/-4°C and a light breeze."

def get_flight_price(src, dst):
    """Query flight ticket prices"""
    return f"The flight ticket price from {src} to {dst} is 200-300 USD."

def get_train_price(src, dst):
    """Query train ticket prices"""
    return f"The train ticket price from {src} to {dst} is 100-200 CNY."

# Mapping of tool names to functions
TOOL_FUNCTIONS = {
    "get_current_weather": get_current_weather,
    "get_flight_price": get_flight_price,
    "get_train_price": get_train_price,
}

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Useful for when you want to query the weather in a specific city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "A city or district, such as Beijing, Hangzhou, or Yuhang.",
                    }
                },
                "required": ["location"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_flight_price",
            "description": "Useful for when you want to query flight ticket prices.",
            "parameters": {
                "type": "object",
                "properties": {
                    "src": {
                        "type": "string",
                        "description": "The departure city of the flight, such as Beijing or Hangzhou.",
                    },
                    "dst": {
                        "type": "string",
                        "description": "The arrival city of the flight, such as Beijing or Hangzhou.",
                    },
                },
                "required": ["src", "dst"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_train_price",
            "description": "Useful for when you want to query train ticket prices.",
            "parameters": {
                "type": "object",
                "properties": {
                    "src": {
                        "type": "string",
                        "description": "The departure city of the train, such as Beijing or Hangzhou.",
                    },
                    "dst": {
                        "type": "string",
                        "description": "The arrival city of the train, such as Beijing or Hangzhou.",
                    },
                },
                "required": ["src", "dst"],
            },
        },
    },
]

# ==================== Tool Call Handling ====================
def handle_tool_call(name, arguments_str):
    """
    Handles a tool call request

    Args:
        name: Tool function name
        arguments_str: JSON formatted input parameter string

    Returns:
        Tool execution result string
    """
    try:
        arguments = json.loads(arguments_str)
        print(f'[Tool Call] Start processing: name={name}, args={arguments}')

        func = TOOL_FUNCTIONS.get(name)
        if func is None:
            result = f"Client did not find the tool: {name}"
            print(f'[Tool Call] Error: {result}')
            return result

        result = func(**arguments)
        print(f'[Tool Call] Completed: {result}')
        return result

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

# ==================== Main Program ====================
async def main():
    """Main function: Establishes a WebSocket connection and conducts a voice conversation"""
    pya = pyaudio.PyAudio()
    speaker = pya.open(format=pyaudio.paInt16, channels=1, rate=24000, output=True)

    # Establish WebSocket connection
    headers = {
        "Authorization": f"bearer {API_KEY}",
        "X-DashScope-OmniRealtime": "true",
    }
    async with websockets.connect(
        f"{URL}?model={MODEL}", additional_headers=headers,
    ) as ws:
        await ws.recv()

        # Configure session parameters
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "voice": VOICE,
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "instructions": "You are a personal assistant named Xiaoyun",
                "turn_detection": {"type": "server_vad"},
                "input_audio_transcription": {"model": "qwen3-asr-flash-realtime"},
                "tools": TOOLS,
            },
        }))
        await ws.recv()

        # Audio capture coroutine
        async def send_audio():
            mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
            try:
                while True:
                    data = mic.read(3200, exception_on_overflow=False)
                    await ws.send(json.dumps({
                        "type": "input_audio_buffer.append",
                        "audio": base64.b64encode(data).decode(),
                    }))
                    await asyncio.sleep(0.01)
            except asyncio.CancelledError:
                mic.close()

        pending = {}
        all_response_text = ""
        send_task = asyncio.create_task(send_audio())
        print("Tool calling is enabled. Speak into the microphone (Ctrl+C to exit)...")

        # Event handling loop
        async for raw in ws:
            msg = json.loads(raw)
            t = msg["type"]

            # Session created
            if t == "session.created":
                print(f"Session started: {msg['session']['id']}")

            # Play audio
            elif t == "response.audio.delta":
                speaker.write(base64.b64decode(msg["delta"]))

            # Incremental text response
            elif t in ("response.audio_transcript.delta", "response.text.delta"):
                all_response_text += msg.get("delta", "")

            # User speech-to-text
            elif t == "conversation.item.input_audio_transcription.completed":
                print(f"[User] {msg['transcript']}")

            # VAD detected speech start
            elif t == "input_audio_buffer.speech_started":
                print("====== VAD detected speech start ======")

            # VAD detected speech end
            elif t == "input_audio_buffer.speech_stopped":
                print("====== VAD detected speech end ======")

            # Received tool call request
            elif t == "response.function_call_arguments.done":
                print("====== Received tool call request ======")
                pending[msg["call_id"]] = {
                    "name": msg["name"],
                    "arguments": msg["arguments"],
                }

            # Response completed
            elif t == "response.done":
                if pending:
                    # Run pending tool calls
                    for cid, info in pending.items():
                        result = handle_tool_call(info["name"], info["arguments"])
                        # Send tool execution result
                        await ws.send(json.dumps({
                            "type": "conversation.item.create",
                            "item": {
                                "type": "function_call_output",
                                "call_id": cid,
                                "output": result,
                            },
                        }))
                    pending.clear()
                    # Trigger the server to continue generating a response
                    await ws.send(json.dumps({
                        "type": "response.create",
                        "response": {"modalities": ["text", "audio"]},
                    }))
                    print("====== Tool call processing completed ======")
                else:
                    # Normal response completed, print the full response
                    if all_response_text:
                        print(f"[Model] {all_response_text}")
                    all_response_text = ""

        send_task.cancel()
        speaker.close()
        pya.terminate()

asyncio.run(main())