Força pelo menos uma ferramenta tool_choice para "required". O modelo sempre retorna uma chamada de ferramenta, então tool_calls nunca está vazio. Use isso quando cada pergunta no cenário requer uma ferramenta. Confirme que as perguntas são relevantes para as ferramentas disponíveis primeiro, ou os resultados podem ser inesperados.

Tradução de apoio para consulta. Código e identificadores técnicos preservados. Consultar referência original ↗

Conecte modelos a ferramentas externas. Modelos de linguagem grandes (LMLs) não podem acessar dados em tempo real ou sistemas externos. A Chamada de Funções permite que modelos chamem ferramentas externas -- como APIs, bancos de dados e funções definidas pelo usuário -- para que um modelo possa recuperar informações ou executar ações além de suas capacidades incorporadas. Você define ferramentas, o modelo decide quando chamá-las, e sua aplicação executa as chamadas.

Como funciona

A função Chamada funciona através de uma interação multi-passo entre sua aplicação e o LLM.

Faça a primeira chamada de modelo A aplicação envia a pergunta do usuário e uma lista de ferramentas disponíveis para a LLM.
Recebeu instruções do modelo (nome da ferramenta e parâmetros de entrada) Se o modelo decidir chamar uma ferramenta externa, ela retorna uma instrução JSON que especifica o nome da função e parâmetros de entrada para passar.

Se o modelo decidir não chamar uma ferramenta, ela retorna uma resposta em linguagem natural.

** Execute a ferramenta em sua aplicação** Sua aplicação executa a ferramenta especificada e obtém sua saída.
Faça a segunda chamada de modelo Adicione a saída da ferramenta ao contexto do modelo (o array de mensagens), e então ligue para o modelo novamente.
**Receber a resposta do modelo final ** O modelo combina a saída da ferramenta com a pergunta do usuário para gerar uma resposta em linguagem natural. O seguinte diagrama ilustra o fluxo de trabalho: [Imagem] (https://help-static-aliyun-doc.aliyuncs.com/assets/img/en-US/8226342771/CAEQUxiBgMCR3fao4RkiIDY0YmY0NWU2NzM0MzQ3NWU5OWY1ZWRjM2Y2MWI4NTMx4358988_20240409160230.951.svg)

Modelos apoiados.

Todos os modelos de geração de texto de propósito geral suportam a chamada de função, incluindo modelos de terceiros (DeepSeek, Kimi, GLM, MiniMax) e o Qwen3.8 Série de código aberto (qwen3.8-2.4t-a95b). Para a visão, Qwen3-VL, qwen3.8-omni-flash, qwen3.5-omni-plus, qwen3.5-omni-flash, e qwen3-omni-flash Também apoie. Veja [Modelos] (https://www.qwencloud.com/developer-guides/getting-started/text-generation-models) para a lista completa. Quando você usa a função Chamando com GLM Modelos de série, você deve incluir extra_body={"tool_stream": True} em seu pedido. Caso contrário, o modelo não voltará. tool_callsE a chamada de ferramentas não funcionará. Por glm-5.3, tool_stream predefinições para true Então não precisa passar isso explicitamente. Para modelos de discurso em tempo real, qwen3.8-omni-flash-realtime, qwen3.5-omni-plus-realtime, e qwen3.5-omni-flash-realtime A função de suporte é chamada via WebSocket. Veja em tempo real para detalhes de implementação. Para modelos de chat de voz, o Qwen-Audio-Realtime Série (Qwen-Audio-3.1-Realtime-Plus série, Qwen-Audio-3.0-Realtime-Plus série, e Qwen-Audio-3.0-Realtime-Flash - Também apoio a chamada.

Começando.

Esta seção mostra como usar a função chamada com um cenário de busca do tempo.

OpenAI compatível

DashScope Python Node.js Entendido.

from openai import OpenAI
from datetime import datetime
import json
import os
import random

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qwencloudapi.com/compatible-mode/v1",
)
# Simulate user question
USER_QUESTION = "What's the weather in Singapore?"
# Define tool list
tools = [
  {
    "type": "function",
    "function": {
      "name": "get_current_weather",
      "description": "Useful when you want to check the weather for a specific city.",
      "parameters": {
        "type": "object",
        "properties": {
          "location": {
            "type": "string",
            "description": "City or county, such as Singapore or New York.",
          }
        },
        "required": ["location"],
      },
    },
  },
]

# Simulate weather lookup tool
def get_current_weather (arguments):
  weather_conditions = ["sunny", "cloudy", "rainy"]
  random_weather = random.choice(weather_conditions)
  location = arguments["location"]
  return f"{location} is {random_weather} today."

# Wrap model response function
def get_response (messages):
  completion = client.chat.completions.create(
    model="qwen3.8-max",
    messages=messages,
    tools=tools,
  )
  return completion

messages = [{"role": "user", "content": USER_QUESTION}]
response = get_response(messages)
assistant_output = response.choices[0].message
if assistant_output.content is None:
  assistant_output.content = ""
messages.append(assistant_output)
# If no tool is needed, output content directly
if assistant_output.tool_calls is None:
  print (f"No weather tool call needed. Direct reply: {assistant_output.content}")
else:
  # Enter tool calling loop
  while assistant_output.tool_calls is not None:
    tool_call = assistant_output.tool_calls[0]
    tool_call_id = tool_call.id
    func_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    print (f"Calling tool [{func_name}], parameters: {arguments}")
    # Execute tool
    tool_result = get_current_weather(arguments)
    # Build tool response message
    tool_message = {
      "role": "tool",
      "tool_call_id": tool_call_id,
      "content": tool_result,
    }
    print (f"Tool returned: {tool_message['content']}")
    messages.append(tool_message)
    # Call model again to get summarized natural-language reply
    response = get_response(messages)
    assistant_output = response.choices[0].message
    if assistant_output.content is None:
      assistant_output.content = ""
    messages.append(assistant_output)
  print (f"Final assistant reply: {assistant_output.content}")

qwen3.8-max usa o MultiModalConversation Interface. Para modelos como qwen-plus e qwen3.7-max, use o Generation Interface -- veja [Geração de texto rápido start] (https://www.qwencloud.com/developer-guides/text-generation/quickstart) para exemplos. Python Java Entendido.

import os
from dashscope import MultiModalConversation
import dashscope
import json
import random
dashscope.base_http_api_url = 'https://maas.qwencloudapi.com/api/v1'

# 1. Define tool list
tools = [
  {
    "type": "function",
    "function": {
      "name": "get_current_weather",
      "description": "Useful when you want to check the weather for a specific city.",
      "parameters": {
        "type": "object",
        "properties": {
          "location": {
            "type": "string",
            "description": "City or county, such as Singapore or New York.",
          }
        },
        "required": ["location"],
      },
    },
  }
]

# 2. Simulate weather lookup tool
def get_current_weather (arguments):
  weather_conditions = ["sunny", "cloudy", "rainy"]
  random_weather = random.choice(weather_conditions)
  location = arguments["location"]
  return f"{location} is {random_weather} today."

# 3. Wrap model response function
def get_response (messages):
  response = MultiModalConversation.call(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    model="qwen3.8-max",
    messages=messages,
    tools=tools,
  )
  return response

# 4. Initialize conversation history
messages = [
  {
    "role": "user",
    "content": "What's the weather in Singapore?"
  }
]

# 5. First model call
response = get_response(messages)
assistant_output = response.output.choices[0].message
messages.append(assistant_output)

# 6. Determine whether to call a tool
if "tool_calls" not in assistant_output or not assistant_output["tool_calls"]:
  print (f"No tool call needed. Direct reply: {assistant_output['content'][0]['text']}")
else:
  # 7. Enter tool calling loop
  while "tool_calls" in assistant_output and assistant_output["tool_calls"]:
    tool_call = assistant_output["tool_calls"][0]
    func_name = tool_call["function"]["name"]
    arguments = json.loads(tool_call["function"]["arguments"])
    tool_call_id = tool_call.get("id")
    print (f"Calling tool [{func_name}], parameters: {arguments}")
    tool_result = get_current_weather(arguments)
    tool_message = {
      "role": "tool",
      "content": tool_result,
      "tool_call_id": tool_call_id
    }
    print (f"Tool returned: {tool_message['content']}")
    messages.append(tool_message)
    response = get_response(messages)
    assistant_output = response.output.choices[0].message
    messages.append(assistant_output)
  # 8. Output final natural-language reply
  print (f"Final assistant reply: {assistant_output['content'][0]['text']}")

Executar o código produz a seguinte saída: Saída Entendido.

Calling tool [get_current_weather], parameters: {'location': 'Singapore'}
Tool returned: Singapore is cloudy today.
Final assistant reply: Today's weather in Singapore is cloudy.

Referência do esquema de ferramentas

Cada ferramenta é um objeto JSON com a seguinte estrutura: Entendido.

{
  "type": "function",
  "function": {
    "name": "get_current_weather",
    "description": "Useful when you want to check the weather for a specific city.",
    "parameters": {
      "type": "object",
      "properties": {
        "location": {
          "type": "string",
          "description": "City or county, such as Singapore or New York."
        }
      },
      "required": ["location"]
    }
  }
}

Campo	Descrição
`type`	Sempre.`"function"`.
`function.name`	O nome da função. Deve corresponder à função que sua aplicação executa.
`function.description`	Descreve o propósito da ferramenta. O modelo usa isso para decidir quando ligar para a ferramenta.
`function.parameters`	Um objeto JSON Schema descrevendo os parâmetros de entrada da função. Omita se a função não receber nenhuma entrada.
`function.parameters.properties`	Cada chave é um nome de parâmetro. Valores descrevem o tipo e o propósito do parâmetro.
`function.parameters.required`	Arranjo os nomes dos parâmetros necessários.

Escreva descrições claras e específicas. O modelo depende de description campos para selecionar a ferramenta certa e extrair os parâmetros certos.

Especifique o comportamento de chamada de ferramentas.

Ferramenta paralela chamando

Por padrão, o modelo retorna uma chamada por resposta. Se o pedido do usuário requer várias chamadas de ferramentas independentes, como "Qual é o clima em Pequim e Xangai?" parallel_tool_calls para true: Python Node.js Entendido.

completion = client.chat.completions.create(
  model="qwen3.8-max",
  messages=messages,
  tools=tools,
  parallel_tool_calls=True
)

O tool_calls Array então contém várias entradas: Entendido.

{
  "role": "assistant",
  "tool_calls": [
    {
      "function": { "name": "get_current_weather", "arguments": "{\"location\": \"Beijing\"}" },
      "index": 0, "id": "call_c2d8a3a2...", "type": "function"
    },
    {
      "function": { "name": "get_current_weather", "arguments": "{\"location\": \"Shanghai\"}" },
      "index": 1, "id": "call_dc7f2f67...", "type": "function"
    }
  ]
}

Use a chamada de ferramentas paralelas apenas quando as tarefas são independentes. Se as tarefas dependem umas das outras (como quando a entrada da ferramenta A depende da saída da ferramenta B), use o loop de "Getting" que começou a chamar as ferramentas em série.

Chamada de ferramenta forçada (tool_ choice)

O tool_choice O parâmetro controla se o modelo chama uma ferramenta. O padrão é "auto" (modelo decide).

** Força uma ferramenta específica**: conjunto tool_choice para {"type": "function", "function": {"name": "get_current_weather"}}. O modelo ignora a seleção de ferramentas e sempre chama a função especificada.
Força pelo menos uma ferramenta tool_choice para "required". O modelo sempre retorna uma chamada de ferramenta, então tool_calls nunca está vazio. Use isso quando cada pergunta no cenário requer uma ferramenta. Confirme que as perguntas são relevantes para as ferramentas disponíveis primeiro, ou os resultados podem ser inesperados.

O Qwen Modelos de série não suportam configurações. tool_choice para "required"Em modo não pensante required Não posso garantir que uma ferramenta é sempre chamada, e no modo de pensar "required" Não é suportado.

Bloqueie todas as ferramentas tool_choice para "none" (ou omitir o tools parâmetro). O modelo responde diretamente sem chamar nenhuma ferramenta. Python Node.js Entendido.

# Force a specific tool
completion = client.chat.completions.create(
  model="qwen3.8-max",
  messages=messages,
  tools=tools,
  tool_choice={"type": "function", "function": {"name": "get_current_weather"}},
  extra_body={"enable_thinking": False}
)

# Force at least one tool
completion = client.chat.completions.create(
  model="kimi/kimi-k3",
  messages=messages,
  tools=tools,
  tool_choice="required"
)

# Block all tools
completion = client.chat.completions.create(
  model="qwen3.8-max",
  messages=messages,
  tools=tools,
  tool_choice="none"
)

Modelos como qwen3.8-max Ativar modo de pensamento por padrão. O modo de pensar só suporta tool_choice - Pronto para... "auto" Ou "none". Para forçar uma ferramenta específica, desativar o modo de pensar primeiro, definindo enable_thinking para false.

Remova o tool_choice parâmetro ao resumir as saídas da ferramenta. Caso contrário, a API ainda retorna informações de chamada de ferramenta em vez de uma resposta em linguagem natural.

Conversas multi-voltas

Um usuário pode perguntar "Qual é o tempo em Pequim?" no primeiro round e "E Xangai?" no segundo round. Sem o contexto de primeira rodada, o modelo não pode determinar qual ferramenta chamar. Em conversas multi-turnos, mantenha as mensagens depois de cada rodada. Então adicione a nova mensagem de usuário e invoque a função chamando novamente. A estrutura das mensagens é assim: Entendido.

[
  "System message -- Strategy guiding the model to call tools",
  "User message -- User's question",
  "Assistant message -- Tool call information returned by the model",
  "Tool message -- Tool output",
  "Assistant message -- Model's summary of tool call information",
  "User message -- User's second question"
]

Fluxo

Para conceitos de streaming gerais (protocolo SSE, como permitir streaming, faturamento e uso de fichas), veja Streaming output. Quando se transmite com a função chamando, a informação da chamada de ferramenta chega em pedaços:

** Nome da função da ferramenta**: retornado no primeiro pedaço.
** Argumentos de chamada de ferramentas**: retornados incrementalmente em pedaços subsequentes. Você deve agregar os deltas de argumento antes de analisar JSON. Adicionar stream=True ao seu pedido, junte-se aos pedaços: Python Node.js Entendido.

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())

Ao construir a mensagem assistente para a segunda chamada modelo, substitua o tool_calls campo com o conteúdo agregado.

API de respostas

A API [Responsáveis] (https://www.qwencloud.com/api-reference/chat/openai-responses) usa um formato de resposta diferente para chamadas de ferramentas. Em vez de tool_calls na mensagem assistente, as chamadas de ferramentas aparecem como function_call - Não. output Envie ferramentas com seu pedido. Python Entendido.

from openai import OpenAI
import os

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qwencloudapi.com/compatible-mode/v1",
)

tools = [
  {
    "type": "function",
    "name": "get_current_weather",
    "description": "Useful when you want to check the weather for a specific city.",
    "parameters": {
      "type": "object",
      "properties": {
        "location": {
          "type": "string",
          "description": "City or county, such as Singapore or New York.",
        }
      },
      "required": ["location"],
    },
  },
]

response = client.responses.create(
  model="qwen3.8-max",
  tools=tools,
  input="What's the weather in Singapore?",
)

Passo 2 -- Processar a função_chamada item de saída: output Array contém um function_call item: Entendido.

{
  "type": "function_call",
  "id": "fc_12345",
  "call_id": "call_xxx",
  "name": "get_current_weather",
  "arguments": "{\"location\": \"Singapore\"}"
}

** Passo 3 -- Devolva o resultado da ferramenta:** Após executar a função, passe o resultado de volta usando um function_call_output item: Python Entendido.

tool_result = get_current_weather({"location": "Singapore"})
response = client.responses.create(
  model="qwen3.8-max",
  tools=tools,
  input=[
    {"type": "message", "role": "user", "content": "What's the weather in Singapore?"},
    response.output[0],  # The function_call item
    {
      "type": "function_call_output",
      "call_id": response.output[0].call_id,
      "output": tool_result,
    },
  ],
)
print (response.output_text)

A saída da ferramenta geralmente é uma corda. No entanto, quando o modelo precisa entender uma imagem produzida pela ferramenta (como uma ferramenta de geração de imagens), você pode passar um array de conteúdo com input_text e input_image Tipos. Uma URL de imagem passada como uma string é tratada como texto simples e não é analisada como uma imagem. Na API Responses, as definições de ferramentas usam uma estrutura plana (name e parameters no nível superior) em vez de o aninhado function Embrulho usado em Completações de Chat. Veja [Responsáveis API] (https://www.qwencloud.com/api-reference/chat/openai-responses) para detalhes. Tanto a API de Completações de Chat quanto a API de Respostas suportam a saída de ferramentas como uma string ou um array de conteúdo. Ao usar um array de conteúdo, cada item tem um type Campo (input_text Ou input_image), a text campo (para conteúdo de texto), e um image_url campo (para URLs de imagem ou dados codificados pela Base64).

Qwen-Omni Modelos

Não em tempo real.

Durante a fase de recuperação de informações da ferramenta, Qwen3.5-Omni e Qwen3-Omni-Flash Difere de outros modelos de duas maneiras:

**Streaming saída é necessária:**Set stream=True Quando recuperar informações de ferramentas.
Texto de saída apenas (recomendado): modalities=["text"] para evitar saída de áudio desnecessária durante a seleção de ferramentas.

Veja [Audio e vídeo de compreensão] (https://www.qwencloud.com/developer-guides/speech/multimodal-speech) para detalhes sobre Qwen-Omni Modelos. Python Node.js Entendido.

from openai import OpenAI
import os

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qwencloudapi.com/compatible-mode/v1",
)
tools = [
  {
    "type": "function",
    "function": {
      "name": "get_current_weather",
      "description": "Useful when you want to check the weather for a specific city.",
      "parameters": {
        "type": "object",
        "properties": {
          "location": {
            "type": "string",
            "description": "City or county, such as Singapore or New York.",
          }
        },
        "required": ["location"],
      },
    },
  },
]

completion = client.chat.completions.create(
  model="qwen3-omni-flash",
  messages=[{"role": "user", "content": "What's the weather in Singapore?"}],
  modalities=["text"],
  stream=True,
  tools=tools
)

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

Executar o código dá esta saída: Saída Entendido.

[ChoiceDeltaToolCall(index=0, id='call_391c8e5787bc4972a388aa', function=ChoiceDeltaToolCallFunction(arguments=None, name='get_current_weather'), type='function')]
[ChoiceDeltaToolCall(index=0, id='call_391c8e5787bc4972a388aa', function=ChoiceDeltaToolCallFunction(arguments=' {"location": "Singapore"}', name=None), type='function')]
None

Veja a seção de Streaming acima para o código para agregar pedaços de argumentos.

Em tempo real.

Qwen3.8-Omni-Flash-Realtime, Qwen3.5-Omni-Plus-Realtime, e Qwen3.5-Omni-Flash-Realtime Apoio personalizado Função Chamada para conversas de voz. O fluxo de trabalho seguinte usa o protocolo DashScope SDK ou WebSocket nativo. Para conexão e configuração SDK, veja o [guia em tempo real] (https://www.qwencloud.com/developer-guides/speech/realtime-multimodal-speech). Para ferramentas MCP executadas por servidor e o fluxo de aprovação Qwen3.8-Omni-Flash-Realtime, veja o [fluxo de interação MPC] (https://www.qwencloud.com/api-reference/real-time-multimodal/interaction-process#qwen38-mcp-flow Depois de estabelecer uma conexão WebSocket, você pode passar as definições de ferramentas através de session.update Para iniciar o seguinte fluxo de interação:

O usuário faz uma pergunta com a voz. O cliente captura o áudio e o envia para o servidor (correspondente ao append_audio() método).
Após a detecção de atividade de voz (VAD) do servidor detectar o fim da fala, ele realiza inferência de modelo e determina que uma ferramenta precisa ser chamada.
O servidor devolve a informação de chamada de ferramenta ao cliente (correspondente ao response.function_call_arguments.done evento), incluindo o nome da função (name), parâmetros de função (arguments- E chame a identidade.call_id). Exemplo: Entendido.

{
  "type": "response.function_call_arguments.done",
  "response_id": "resp_JnTOsWXlFhKcFohZbtfz6",
  "item_id": "item_Rhcms7CauTNsQprV5S4Hr",
  "output_index": 0,
  "name": "get_current_weather",
  "call_id": "call_2be200f4cafe419b9530dd",
  "arguments": "{\"location\": \"Hangzhou\"}"
}

O cliente executa a função de ferramenta correspondente localmente com base no nome da função e parâmetros, e obtém o resultado. O cliente envia os resultados da ferramenta e desencadeia a resposta final.
O cliente envia o resultado da execução da ferramenta de volta para o servidor (correspondente ao conversation.item.create evento), incluindo a identificação de chamada (call_id) e resultado de execução (output). Exemplo: Entendido.

{
  "type": "conversation.item.create",
  "item": {
    "type": "function_call_output",
    "call_id": "call_2be200f4cafe419b9530dd",
    "output": "The weather in Hangzhou today is sunny, with a temperature of 25°C and a light breeze."
  }
}

O cliente continua enviando um response.create Evento para ativar o servidor para gerar a resposta final de voz baseada no resultado da execução da ferramenta.
O cliente recebe a voz e o texto retornado pelo servidor (correspondente ao response.audio.delta e response.audio_transcript.delta E toca a resposta de voz para o usuário. O Qwen-Omni-Realtime série não suporta o tool_choice e parallel_tool_calls Parâmetros. Para mais detalhes sobre Qwen-Omni Modelos em tempo real e protocolo WebSocket, veja [Fala multimodal em tempo real] (https://www.qwencloud.com/developer-guides/speech/realtime-multimodal-speech), [Acontecimentos do cliente] (https://www.qwencloud.com/api-reference/real-time-multimodal/client-events), e [Eventos de servidor] (https://www.qwencloud.com/api-reference/real-time-multimodal/server-events). DashScope Python SDK DashScope Java SDK WebSocket (Python) Entendido.

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"
WS_URL = "wss://maas.qwencloudapi.com/api-ws/v1/realtime"
# Configure the API key. If the environment variable is not set, replace the line below with 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 weather for a specified city"""
  return f"The weather in {location} today is hazy turning 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 when you want to check the weather in a specific city.",
      "parameters": {
        "type": "object",
        "properties": {
          "location": {
            "type": "string",
            "description": "City or county, such as Beijing, Hangzhou, or Yuhang District.",
          }
        },
        "required": ["location"],
      },
    },
  },
  {
    "type": "function",
    "function": {
      "name": "get_flight_price",
      "description": "Useful 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 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 from 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]:
  """
  Handle tool call requests

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

  Returns:
    Updated tool call response including 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] Complete: {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:
  """Send tool call response 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:
  - Decoder thread: Decodes base64-encoded audio data into raw PCM data
  - Player thread: Writes PCM data to the audio output device

  Supports dynamic addition of audio data, cancellation of playback, saving audio files, etc.
  """

  def __init__(self, pya: pyaudio.PyAudio, sample_rate=24000, chunk_size_ms=100, save_file=False):
    """
    Initialize 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):
    """Decoder 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):
    """Player 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 chunk (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 done
      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 done
      elif event_type == 'response.done':
        print ('====== Response done ======')
        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 chunk: {audio_delay:.0f}ms')
          else:
            print ('[Metric] Metric info not available (might be a response after a tool call)')

        self.all_response_text = ''

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

  def process_pending_tool_calls (self) -> bool:
    """
    Process pending tool calls

    Returns:
      Whether there are new tool calls to respond to
    """
    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 audio manager
  audio_manager = AudioManager()

  # Create callback handler
  callback = OmniCallback(audio_manager)

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

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

  # Establish 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 signal handler
  def signal_handler (sig, frame):
    print ('\n Ctrl+C received, 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 complete, creating new response ***")
        conversation.create_response(
          instructions=None,
          output_modalities=omni_output_modalities
        )
        print ('====== Tool call processing complete ======\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()

Modo de pensar

Modelos profundos raciocinam antes de gerar chamadas de ferramentas. Preparar. enable_thinking=True para ver o processo de raciocínio do modelo. A resposta inclui: reasoning_content Antes que a ferramenta chame:

O modelo de saída reasoning_content mostrando sua análise de intenção do usuário, seleção de ferramentas e planejamento de parâmetros.
O modelo então sai do tool_calls Como sempre. Ao passar a mensagem assistente de volta em pedidos subsequentes, você deve incluir o reasoning_content Campo. Com o modo de pensar ativado, o tool_choice parâmetro só suporta "auto" (padrão) ou "none". Python Entendido.

completion = client.chat.completions.create(
  model="qwen3.8-max",  # Use a thinking-capable model
  messages=messages,
  tools=tools,
  extra_body={"enable_thinking": True},
  stream=True,
)

Para transmissão completa código parse com reasoning_content Manuseando, veja [Modo de pensar] (https://www.qwencloud.com/developer-guides/text-generation/thinking).

Boas práticas.

Exatidão de seleção de ferramentas de teste: Construir um conjunto de dados de avaliação espelhando cenários reais. Precisão de seleção de ferramentas, precisão de extração de parâmetros e taxa de sucesso de ponta a ponta.
Otimizar descrições de ferramentas: Quando o modelo seleciona a ferramenta errada ou extrai parâmetros errados, refine descrições e prompts de sistema antes de atualizar modelos.
** Mantenha o candidato conjunto pequeno**: Limitar ferramentas passadas para o modelo para não mais de 20. Use uma camada de roteamento (pesquisa semântica, filtragem de palavras-chave ou um roteador LLM leve) para pré-filtrar grandes bibliotecas de ferramentas.
Princípio do menos privilégio: padrão para ferramentas somente de leitura. Nunca dê acesso direto ao modelo a operações perigosas (execução de código, exclusão de arquivos, transferências financeiras).
** Adicione confirmação humana para operações de escrita**: O modelo pode gerar pedidos de ação, mas ações irreversíveis (enviar e-mail, modificar dados) devem exigir confirmação do usuário.
Atribua tempo a cada passo. Quando falhar, devolva uma mensagem clara como "Desculpe, não posso recuperar essa informação agora. Por favor, tente novamente mais tarde."
Mostrar progresso na UI, mostrar mensagens como "Procurando o tempo para você..." quando iniciar a execução da ferramenta.

Passe informações da ferramenta através da mensagem do sistema

Alternativa: incorporar informações de ferramentas no sistema. Nós recomendamos usar o tools parâmetro (mostrado ao longo deste guia). Se precisar de controle direto sobre o prompt de ferramentas, incorpore definições de ferramentas na mensagem do sistema usando este modelo: Python Entendido.

import os
from openai import OpenAI
import json

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qwencloudapi.com/compatible-mode/v1",
)

tools = [
  {
    "type": "function",
    "function": {
      "name": "get_current_time",
      "description": "Useful when you want to know the current time.",
      "parameters": {}
    }
  },
  {
    "type": "function",
    "function": {
      "name": "get_current_weather",
      "description": "Useful when you want to check the weather for a specific city.",
      "parameters": {
        "type": "object",
        "properties": {
          "location": {
            "type": "string",
            "description": "City or county, such as Singapore or New York."
          }
        },
        "required": ["location"]
      }
    }
  }
]

custom_prompt = "You are a helpful assistant."
tools_content = "\n".join(json.dumps(tool, ensure_ascii=False) for tool in tools)

system_prompt = f"""{custom_prompt}

# Tools

You may call one or more functions to assist with the user query.

You are provided with function signatures within <tools></tools> XML tags:
<tools>
{tools_content}
</tools>

For each function call, return a json object with function name and arguments within <tool_call></tool_call> XML tags:
<tool_call>
{{"name": <function-name>, "arguments": <args-json-object>}}
</tool_call>"""

messages = [
  {"role": "system", "content": system_prompt},
  {"role": "user", "content": "What time is it?"}
]

completion = client.chat.completions.create(
  model="qwen3.8-max",
  messages=messages,
)
print (completion.choices[0].message.content)

Depois de executar o código, use um analisador XML para extrair informações de chamada de ferramenta entre o <tool_call> e </tool_call> Marcas.

Contagem

Além de fichas no array de mensagens, descrições de ferramentas também contam como tokens de entrada e são faturadas como parte do prompt.

Códigos de erro.

Se uma ligação falhar, veja [Mensagens de erro] (https://www.qwencloud.com/api-reference/preparation/error-messages). [Anterior] (https://www.qwencloud.com/developer-guides/embeddings/reranking[Busca na Web Respostas do modelo Ground em dados da web em tempo real Próximo] (https://www.qwencloud.com/developer-guides/tool-calling/web-search)