全双工语音大模型(算法编码：8000200008)

协议类型：WebSocket

---

一、接入点

环境	地址
公网	wss://api-maas.teleai.com.cn/aipaas/voice/v1/full/duplex

• API Path ：/aipaas/voice/v1/full/duplex
• 请求方法 ：WebSocket
• Content-Type ：application/json

---

二、鉴权

采用请求头携带鉴权信息，在建立 WebSocket 连接时通过 Header 传递 ：

2.1 标准鉴权

Authorization: <鉴权信息>

• 鉴权规则请参见：开发指南 - 签名认证方式。
• 鉴权在线生成工具：API在线签名计算工具。

请求头	必填	说明
Content-Type	是	内容类型，固定 application/json
X-APP-ID	是	控制台 → 应用管理 → AppID
Device-Uuid	是	设备uuid
Authorization	是	鉴权信息

2.2 Bearer Token 鉴权

Authorization: Bearer <用户申请的API-KEY>

• API Key 鉴权，请在 API Key 管理 页面，获取长效 API Key。

请求头	必填	说明
Content-Type	是	内容类型，固定 application/json
Device-Uuid	是	设备uuid
Authorization	是	Bearer

---

三、能力简介

语音对话机器人可以与用户开展实时语音对话。

• 输入格式：PCM 编码
• 通道：支持单声道输入
• 输入采样率：支持 8000/16000 Hz
• 合成音频采样率：22050 Hz

---

四、请求与响应

4.1 请求参数（Json）

参数名	说明	必填	类型
req_id	请求 id，记录该值便于排查问题	是	string
req_status	当前的请求状态。0：开始对话；1：发送音频数据；2：结束发送音频数据；3：手动控制打断	是	int
sampling_rate	音频采样率，默认值 16000Hz	否	int
num_channels	输入音频数据的通道数，默认值是 1	否	int
interrupt_sensitivity	打断敏感度，取值范围暂时没有限制，默认值是 0.5	否	float
enable_dialog_output	是否开启对话文本输出，默认值是 true	否	boolean
enable_button_interrupt	是否开启按键打断，默认值是 true	否	boolean
audio_stream	音频数据，经过 base64 编码	否	string

4.2 响应体（Json Object）

参数名	说明	必填	类型
code	返回码	是	int
message	返回码说明	是	string
sid	会话id，用于记录本次会话	是	string
res_status	当前服务端状态。0：服务端对话功能就绪；1：返回服务端对话结果；2：音频数据发送完成，服务端对话结束	是	int
result	识别或合成结果	是	object

结果 result

参数名	说明	必填	类型
result>>type	本次结果的类型：0：ASR；1：TTS；2：NLP 回答；3：Control Signal	是	int
result>>recog_status	当前识别状态。1：服务端检测到有效语音开始时间点；2：返回中间转写结果；3：服务端检测到一句话结束，返回该句的转写结果；4：服务端收到音频流结束的信号，返回识别结束信号	是	int
result>>text	type 为 0 时为语音识别结果，type 为 2 时，为 NLP 结果	是	string
result>>audio	合成音频数据，经过 base64 编码	是	string
result>>is_end	标志位。true：最后一个合成片段；false：中间合成片段	是	boolean
result>>command	当前控制信号表示的命令。Interrupt：打断信号	是	string

---

五、状态码说明

5.1 算法状态码

返回码	解释	说明	解决方法
10301	Required parameter miss	参数错误	检查请求体是否符合接口协议
10304	Parse request body fail	请求格式错误	查看请求的 URL body 格式是否正确，参考接口文档
10503	Server connection time out	服务连接超时	联系技术人员
10903	Process failed	处理失败	联系技术人员
10000	Success	成功	执行下一步操作

通用状态码请参考【状态码】中的【网关认证】。

5.2 公共状态码

• 接口响应的公共状态码由网关统一定义，一般用于标识鉴权失败、限流或服务异常，具体业务错误码、错误信息及排查建议请参考：开发指南-公共信息及状态码。

---

六、调用示例

6.1 发送开始对话请求

名称	类型	必需	说明
req_id	string	是	请求全局唯一
req_status	int	是	当前的请求状态。0：开始对话送音频数据
sampling_rate	int	否	音频采样率，默认值 16000Hz
num_channels	int	否	输入音频数据的通道数，默认值是 1
interrupt_sensitivity	float	否	打断敏感度，默认值是 0.5
enable_dialog_output	bool	否	是否开启对话文本输出，默认值是 true
enable_button_interrupt	bool	否	是否开启按键打断，默认值是 true

{
    "req_id": "test001",
    "req_status": 0,
    "sampling_rate": 16000,
    "num_channels": 1
}

6.2 发送语音数据

名称	类型	必需	说明
req_status	int	是	当前的请求状态。1：发送音频数据
audio_stream	string	是	音频数据，经过 base64 编码

{
    "req_status": 1,
    "audio_stream": "yourBase64"
}

6.3 结束发送音频数据

名称	类型	必需	说明
req_status	int	是	当前的请求状态。2：结束发送语音数据

{
    "req_status": 2
}

6.4 发送语音数据/控制信号

客户端接收到服务端发送的确认响应后，开始持续发送音频数据。

名称	类型	必需	说明
req_status	int	是	当前的请求状态。1：发送音频数据
audio_stream	string	是	音频数据，经过 base64 编码

{
  "req_status": 1,
  "audio_stream": "yourBase64"
}

客户端接收到服务端发送的确认响应后，可以发送控制信号。

名称	类型	必需	说明
req_status	int	是	当前的请求状态。3：发送控制信号
command	string	是	当前控制信号表示的命令- Interrupt 用于按键打断；- Playback Finished 音频播放完成

NOTE：每一轮对话，客户端播放完机器人返回的所有音频后，客户端必须准确向服务发送 Playback Finished 命令

6.5 Java 版本

import com.alibaba.fastjson.JSONObject;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;

/**
 * 为保证示例可以正常运行，请确认maven中是否存在以下依赖：
 *
 *     org.java-websocket
 *     Java-WebSocket
 *     1.5.3
 *
 *
 *     com.alibaba
 *     fastjson
 *     2.0.49
 *
 */
public class WebSocketExample extends WebSocketClient {

    public WebSocketExample(URI serverUri, Map headers) {
        super(serverUri, headers);
    }

    public static void main(String[] args) {
        try {
            // 创建WebSocket客户端
            URI uri = new URI("算法调用地址");
            Map headers = new HashMap();
            //调用鉴权
            headers.put("Content-Type", "application/json");
            headers.put("X-APP-ID", "yourAppId");
            headers.put("Authorization", "yourAuthorization");

            WebSocketExample client = new WebSocketExample(uri, headers);
            client.connect();

            // 等待一段时间后关闭连接
            Thread.sleep(5000);
            client.close();
        } catch (URISyntaxException | InterruptedException e) {
            e.printStackTrace();
            System.out.println("WebSocket error: " + e.getMessage());
        }
    }

    @Override
    public void onOpen(ServerHandshake handshakedata) {
        System.out.println("WebSocket connection opened");
        // 发送消息
        try {
            // 初始化连接消息
            JSONObject initSend = new JSONObject();
            initSend.put("req_id", "test001");
            initSend.put("req_status", 0);
            initSend.put("sampling_rate", 16000);
            initSend.put("num_channels", 1);
            send(initSend.toJSONString());
            // 发送音频文件base64
            JSONObject audioSend = new JSONObject();
            String audioFilePath = "音频文件路径";
            // 读取音频文件的字节
            byte[] audioBytes = Files.readAllBytes(Paths.get(audioFilePath));
            List chunkedList = chunkByteArray(audioBytes, 400);
            // 将字节转换为Base64编码的字符串
            for (byte[] chunked : chunkedList) {
                String base64 = Base64.getEncoder().encodeToString(chunked);
                audioSend.put("req_status", 1);
                audioSend.put("audio_stream", base64);
                send(audioSend.toJSONString());
            }
            // 发送结束消息
            JSONObject endSend = new JSONObject();
            endSend.put("req_status", 2);
            send(endSend.toJSONString());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("WebSocket connection closed: " + reason);
    }

    @Override
    public void onError(Exception ex) {
        System.err.println("WebSocket error: " + ex.getMessage());
    }

    /**
     * 将byte数组按照指定长度截断并存放在list中
     *
     * @param originalArray 原始byte数组
     * @param chunkSize     每个截断部分的长度
     * @return 包含截断后byte数组的list
     */
    public static List chunkByteArray(byte[] originalArray, int chunkSize) {
        List chunkedList = new ArrayList();

        if (originalArray == null || chunkSize <= 0) {
            throw new IllegalArgumentException("Invalid input parameters");
        }

        for (int i = 0; i < originalArray.length; i += chunkSize) {
            int remainingLength = originalArray.length - i;
            int lengthToCopy = Math.min(chunkSize, remainingLength);

            byte[] chunk = new byte[lengthToCopy];
            System.arraycopy(originalArray, i, chunk, 0, lengthToCopy);

            chunkedList.add(chunk);
        }

        return chunkedList;
    }

}

6.6 Python 版本

import logging
import os
import json
import time
import base64
import asyncio
import argparse
import websockets
import multiprocessing as mp
from scipy.io import wavfile

# 配置日志记录
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

# 定义音频流的最大切分长度
CHUNK_SIZE = 6400

async def uws_loop_send(args):
    """
    发送音频流的异步函数
    :param args: 包含 websocket, idx, wav_id, audio_file, task_id, t_slp 的元组
    """
    websocket, idx, wav_id, audio_file, task_id, t_slp = args
    try:
        logger.info(f"pid: {os.getpid()} idx: {idx:3}, task_id: {task_id:36} send_info: start loop send 1")
        while True:
            is_end = False
            # 读取音频文件的一段数据
            audio_stream = audio_file.read(CHUNK_SIZE)
            # 将音频数据编码为 base64 格式
            audio_stream = base64.b64encode(audio_stream).decode("utf-8")
            # 检查是否到达文件末尾
            try_stream = audio_file.read(1)
            if not try_stream:
                is_end = True
            else:
                audio_file.seek(-1, 1)
            # 构建发送的消息
            mid_json = {
                "audio_stream": audio_stream,
                "req_status": 1
            }
            # 发送消息
            await websocket.send(json.dumps(mid_json))
            await asyncio.sleep(t_slp)
            if is_end:
                logger.info(f"pid: {os.getpid()} idx: {idx:3} task_id: {task_id:36} send_info: end loop send 1")
                break

    except Exception as e:
        logger.error(f"pid: {os.getpid()} idx: {idx:3} task_id: {task_id:36} send_info: send 1 fail: {e}")
        return 1

    return 0

async def uws_recv(args):
    """
    接收服务器响应的异步函数
    :param args: 包含 websocket, idx, wav_id, task_id 的元组
    """
    websocket, idx, wav_id, task_id = args
    logger.info(f"pid: {os.getpid()} idx: {idx:3} task_id: {task_id:36} recv_info: start recv")
    total_text = ""
    while True:
        try:
            # 接收服务器返回的消息
            message = await websocket.recv()
            message = json.loads(message)
            # 处理不同的响应状态
            if 'res_status' in message:
                if message['res_status'] in [2, 3]:
                    logger.info(f"pid: {os.getpid()} task_id: {task_id} recv: {message['data']['results'][0]['text']}")
                if message['res_status'] in [3, 4]:
                    total_text += message['data']['results'][0]['text']
                if message['res_status'] == 4:
                    logger.info(f"pid: {os.getpid()} task_id: {task_id} end recv: {total_text}")
                    break
        except Exception as e:
            logger.error(f"pid: {os.getpid()} task_id: {task_id} recv fail: {e}")
            return

async def uws_call(args):
    """
    主调用函数，负责连接 WebSocket 并管理发送和接收任务
    :param args: 命令行参数
    """
    idx = 0
    url = args.url
    wav_id = args.audio_path.split("/")[-1].split(".")[0]
    fs, wav = wavfile.read(args.audio_path)
    num_samples = wav.shape[0]
    audio_file = open(args.audio_path, 'rb')
    audio_file.seek(0)
    audio_file.read(44)  # 跳过 WAV 文件头
    header = {
        "Content-Type": "application/json",
        'X-APP-ID': args.X_APP_ID,
        'Authorization': args.Authorization
    }
    try:
        async with websockets.connect(url, extra_headers=header) as websocket:
            # 发送初始请求
            start_json = {
                "req_id": "test001",
                "req_status": 0,
                "sampling_rate": 16000,
                "num_channels": 1
            }
            logger.info(f"===========================start_json:{start_json}================================")
            logger.info(f"pid: {os.getpid()} idx: {idx:3} task_id: {wav_id:10} send_info: send 0")
            await websocket.send(json.dumps(start_json))
            message = await websocket.recv()
            message = json.loads(message)

            if message['code'] == 10000:
                logger.info(
                    f"pid: {os.getpid()} idx: {idx:3} task_id: {message['sid']} recv_info: {message['code']:5} {message['res_status']}")
                # 循环发送音频流
                send_time = time.perf_counter()
                loop_send_task = asyncio.ensure_future(
                    uws_loop_send([websocket, idx, wav_id, audio_file, message['sid'], CHUNK_SIZE / 2 / fs]))
                recv_task = asyncio.ensure_future(uws_recv([websocket, idx, wav_id, message['sid']]))

                await loop_send_task
                ret = loop_send_task.result()
                if ret:
                    return 1

                # 发送停止请求
                stop_json = {
                    "req_status": 2
                }
                logger.info(f"===========================stop_json:{stop_json}================================")
                await websocket.send(json.dumps(stop_json))
                await recv_task

                cost = time.perf_counter() - send_time
                audio_len_in_ms = num_samples * 1000 / fs
                rtf = float(cost * 1000) / float(audio_len_in_ms)
                logger.info(
                    f"pid: {os.getpid()} idx: {idx:3} task_id: {message['sid']} info: dur: {audio_len_in_ms} cost: {cost * 1000} rtf: {rtf}")
                ret = recv_task.result()
                if ret == 1:
                    return 1

            else:
                logger.info(f"pid: {os.getpid()} idx: {idx:3} message: {message}")

            await websocket.close()

    except Exception as e:
        logger.info(f"pid: {os.getpid()} idx: {idx:3} info: {e}")
        audio_file.close()
        return 1
    audio_file.close()
    return 0

def func(args):
    """
    运行 uws_call 函数的包装函数
    :param args: 命令行参数
    """
    if args.forever:
        while 1:
            asyncio.get_event_loop().run_until_complete(uws_call(args))
    else:
        asyncio.get_event_loop().run_until_complete(uws_call(args))

def main(args):
    """
    主函数，负责启动多个进程
    :param args: 命令行参数
    """
    process_list = []
    for i in range(args.num_processes):
        p = mp.Process(target=func, args=(args,))
        p.start()
        process_list.append(p)

    for p in process_list:
        p.join()

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--url', type=str, default='算法调用地址')
    parser.add_argument('--X_APP_ID', type=str, default='yourAppId')
    parser.add_argument('--Authorization', type=str, default='yourAuthorization')
    parser.add_argument("--num_processes", type=int, default=1)
    parser.add_argument("--audio_path", type=str, default="yourAudioPath")
    parser.add_argument("--forever", action="store_true")
    args = parser.parse_args()

    main(args)