通过WebSocket连接访问实时语音识别服务

本文介绍如何通过WebSocket连接访问实时语音识别服务。

DashScope SDK目前仅支持JavaPython。若想使用其他编程语言开发Paraformer实时语音识别应用程序,可以通过WebSocket连接与服务进行通信。

WebSocket是一种支持全双工通信的网络协议。客户端和服务器通过一次握手建立持久连接,双方可以互相主动推送数据,因此在实时性和效率方面具有显著优势。

对于常用编程语言,有许多现成的WebSocket库和示例可供参考,例如:

  • Go:gorilla/websocket

  • PHP:Ratchet

  • Node.js:ws

建议您先了解WebSocket的基本原理和技术细节,再参照本文进行开发。

前提条件

开通模型服务并获取API Key,建议您API Key配置到环境变量

客户端与服务端的交互流程

按时间顺序,客户端与服务端的交互流程如下:

  1. 建立连接:客户端与服务端建立WebSocket连接。

  2. 开启任务:

    • 客户端发送run-task指令以开启任务。

    • 客户端收到服务端返回的task-started事件,标志着任务已成功开启,可以进行后续步骤。

  3. 发送音频流:

    • 客户端开始发送音频流,并同时接收服务端持续返回的result-generated事件,该事件包含语音识别结果。

  4. 通知服务端结束任务:

    • 客户端发送finish-task指令通知服务端结束任务,并继续接收服务端返回的result-generated事件。

  5. 任务结束:

    • 客户端收到服务端返回的task-finished事件,标志着任务结束。

  6. 关闭连接:客户端关闭WebSocket连接。

image

WebSocket客户端编程与消息处理

在编写WebSocket客户端代码时,为了同时发送和接收消息,通常采用异步编程。您可以按照以下步骤来编写程序:

  1. 建立WebSocket连接:首先,初始化并建立与服务器的WebSocket连接。

  2. 异步监听服务器消息:启动一个单独的线程(具体实现方式因编程语言而异)来监听服务器返回的消息,根据消息内容进行相应的操作。

  3. 发送消息:在不同于监听服务器消息的线程中(例如主线程,具体实现方式因编程语言而异),向服务器发送消息。

  4. 关闭连接:在程序结束前,确保关闭WebSocket连接以释放资源。

当然,编程思路不止这一种,您或许有更好的想法。本文主要介绍通过WebSocket连接访问服务时的鉴权细节及客户端与服务端之间的消息交互。由于篇幅有限,其他思路将不再赘述。

接下来将按照上述思路,为您详细说明。

一、建立WebSocket连接

调用WebSocket库函数(具体实现方式因编程语言或库函数而异),将请求头和URL传入以建立WebSocket连接。

请求头中需添加如下鉴权信息:

{
    "Authorization": "bearer <your_dashscope_api_key>", // 将<your_dashscope_api_key>替换成您自己的API Key
    "user-agent": "your_platform_info", //可选
    "X-DashScope-WorkSpace": workspace, // 可选
    "X-DashScope-DataInspection": "enable"
}

WebSocket URL固定如下:

wss://dashscope.aliyuncs.com/api-ws/v1/inference

二、异步监听服务器返回的消息

如上所述,您可以启动一个线程(具体实现因编程语言而异)来监听服务器返回的消息。WebSocket库通常会提供回调函数(观察者模式)来处理这些消息。您可以在回调函数中根据不同的消息类型实现相应的功能。

服务端返回给客户端的消息叫做事件,事件代表不同的处理阶段,为JSON格式,由headerpayload这两部分组成:

  • header:包含基础信息,格式较为统一。

    task-failed外,所有事件的header格式统一。

    header示例:

    {
        "header": {
            "task_id": "2bf83b9a-baeb-4fda-8d9a-xxxxxxxxxxxx",
            "event": "task-started",
            "attributes": {}
        }
    }

    header参数:

    参数

    类型

    说明

    header

    请求头

    -

    header.event

    String

    事件类型

    • task-started

    • result-generated

    • task-finished

    • task-failed

    详细说明参见下文。

    header.task_id

    String

    客户端生成的task_id

  • payload:包含基础信息外的其他信息。不同事件的payload格式可能不同。

共有如下四种事件:

1、task-started事件:语音识别任务已开启

当监听到服务端返回的task-started事件时,标志着任务已成功开启。只有在接收到该事件后,才能向服务器发送待识别音频或finish-task指令;否则,任务将执行失败。

task-started事件的payload没有内容。

示例:

{
    "header": {
        "task_id": "2bf83b9a-baeb-4fda-8d9a-xxxxxxxxxxxx",
        "event": "task-started",
        "attributes": {}
    },
    "payload": {}
}
2、result-generated事件:包含语音识别的结果

客户端发送待识别音频和finish-task指令的同时,服务端持续返回result-generated事件,该事件包含语音识别的结果。

可以通过result-generated事件中的payload.sentence.endTime是否为空来判断该结果是中间结果还是最终结果。

示例:

{
  "header": {
    "task_id": "2bf83b9a-baeb-4fda-8d9a-xxxxxxxxxxxx",
    "event": "result-generated",
    "attributes": {}
  },
  "payload": {
    "output": {
      "sentence": {
        "begin_time": 170,
        "end_time": null,
        "text": "好,我们的一个",
        "words": [
          {
            "begin_time": 170,
            "end_time": 295,
            "text": "好",
            "punctuation": ","
          },
          {
            "begin_time": 295,
            "end_time": 503,
            "text": "我们",
            "punctuation": ""
          },
          {
            "begin_time": 503,
            "end_time": 711,
            "text": "的一",
            "punctuation": ""
          },
          {
            "begin_time": 711,
            "end_time": 920,
            "text": "个",
            "punctuation": ""
          }
        ]
      }
    },
    "usage": null
  }
}

payload参数说明:

参数

类型

说明

output

Object

output.sentence为识别结果,详细内容见下文。

usage

Object

固定为null。

payload.output.sentence格式如下:

参数

类型

说明

begin_time

Long

句子开始时间。

end_time

Long

句子结束时间,如果为中间识别结果则为null。

text

String

识别文本。

words

List<Word>

字时间戳信息。

payload.output.sentence.words为字时间戳列表,其中每一个word格式如下:

参数

类型

说明

begin_time

Long

字开始时间。

end_time

Long

字结束时间。

text

String

字。

punctuation

String

标点。

3、task-finished事件:语音识别任务已结束

当监听到服务端返回的task-finished事件时,说明任务已结束。此时可以关闭WebSocket连接并结束程序。

示例:

{
    "header": {
        "task_id": "2bf83b9a-baeb-4fda-8d9a-xxxxxxxxxxxx",
        "event": "task-finished",
        "attributes": {}
    },
    "payload": {
        "output": {},
        "usage": null
    }
}
4、task-failed事件:任务失败

如果接收到task-failed事件,表示任务失败。此时需要关闭WebSocket连接并处理错误。通过分析报错信息,如果是由于编程问题导致的任务失败,您可以调整代码进行修正。

示例:

{
    "header": {
        "task_id": "2bf83b9a-baeb-4fda-8d9a-xxxxxxxxxxxx",
        "event": "task-failed",
        "error_code": "CLIENT_ERROR",
        "error_message": "request timeout after 23 seconds.",
        "attributes": {}
    },
    "payload": {}
}

header参数说明:

参数

类型

说明

header.error_code

String

报错类型描述。

header.error_message

String

具体报错原因。

三、给服务器发送消息

在与监听服务器消息不同的线程中(比如主线程,具体实现因编程语言而异),向服务器发送消息。

客户端发送给服务端的消息有两种:

  1. 音频流。

  2. 指令:以Text Frame方式发送的JSON格式的数据,用于控制任务的起止和标识任务边界。

    指令由headerpayload这两部分组成:

    • header:包含基础信息,格式统一。

      header示例:

      {
          "header": {
              "action": "run-task",
              "task_id": "2bf83b9a-baeb-4fda-8d9a-xxxxxxxxxxxx", // 随机uuid
              "streaming": "duplex"
          }
      }

      header参数:

      参数

      类型

      是否必选

      说明

      header

      请求头

      -

      -

      header.action

      String

      指令类型,可以选填

      • "run-task"

      • "finish-task"

      用法参见下文。

      header.task_id

      String

      当次任务ID,随机生成的32位唯一ID。

      32位通用唯一识别码(UUID),由32个随机生成的字母和数字组成。可以带横线(如 "2bf83b9a-baeb-4fda-8d9a-xxxxxxxxxxxx")或不带横线(如 "2bf83b9abaeb4fda8d9axxxxxxxxxxxx")。大多数编程语言都内置了生成UUIDAPI,例如Python:

      import uuid
      
      def generateTaskId(self):
          # 生成随机UUID
          return uuid.uuid4().hex

      header.streaming

      String

      固定字符串:"duplex"

    • payload:包含基础信息外的其他信息。不同指令的payload格式可能不同。

向服务器发送消息需要遵循如下时序,否则会导致任务失败:首先发送run-task指令,待监听到服务器返回的task-started事件后,再发送待识别的音频流。在音频流发送结束后,发送finish-task指令。

1、发送run-task指令:开启语音识别任务(支持定制热词)

该指令用于开启语音识别任务。task_id在后续发送finish-task指令时也需要使用,必须保持一致。

示例:

{
    "header": {
        "action": "run-task",
        "task_id": "2bf83b9a-baeb-4fda-8d9a-xxxxxxxxxxxx", // 随机uuid
        "streaming": "duplex"
    },
    "payload": {
        "task_group": "audio",
        "task": "asr",
        "function": "recognition",
        "model": "paraformer-realtime-v2",
        "parameters": {
            "format": "pcm", // 音频格式
            "sample_rate": 16000, // 采样率
            "vocabulary_id": "vocab-xxx-24ee19fa8cfb4d52902170a0xxxxxxxx", // paraformer-realtime-v2支持的热词ID
            "disfluency_removal_enabled": false, // 过滤语气词
            "language_hints": [
                "en"
            ] // 指定语言,仅支持paraformer-realtime-v2模型
        },
        "resources": [
            {
                "resource_id": "xxxxxxxxxxxx", // paraformer-realtime-v1支持的热词ID
                "resource_type": "asr_phrase"
            }
        ],
        "input": {}
    }
}

payload参数说明:

参数

类型

是否必选

说明

payload.task_group

String

固定字符串:"audio"。

payload.task

String

固定字符串:"asr"。

payload.function

String

固定字符串:"recognition"。

payload.model

String

模型名称:"paraformer-realtime-v2",支持的模型请参考实时语音识别API详情

payload.input

Object

固定格式:{}。

payload.parameters

format

String

音频编码格式,支持'pcm'、'wav'、'opus'、'speex'、'aac'、'amr'等多种音频格式。

sample_rate

Integer

识别音频采样率支持8000Hz以及16000Hz。

  • paraformer-realtime-v2 支持任意采样率。

  • paraformer-realtime-v1 仅支持16000Hz采样。

  • paraformer-realtime-8k-v1 仅支持8000Hz采样率。

vocabulary_id

String

最新热词ID,支持最新v2系列模型并配置语种信息,此次语音识别中生效此热词ID对应的热词信息。默认不启用。使用方法请参考定制热词

disfluency_removal_enabled

Boolean

过滤语气词,默认关闭。

language_hints

List[String]

指定识别语音中语言的代码列表。支持的语言代码:

  • zh: 中文。

  • en: 英文。

  • ja: 日语。

  • yue: 粤语。

  • ko: 韩语。

仅对支持多语言的模型生效。如果不填写则模型会自动识别语种。

payload.resources(内容为列表)

resource_id

String

热词ID,此次语音识别中生效此热词ID对应的热词信息。默认不启用。

注:phrase_id为旧版本热词方案,不支持v2及后续系列模型。支持旧版本热词的模型列表请参考Paraformer语音识别热词定制与管理

resource_type

String

固定字符串"asr_phrase",和"resource_id"需同时使用。

2、发送待识别音频流

客户端需在收到task-started事件后,再发送待识别的音频流。

音频通过WebSocket的二进制通道上传。建议每次发送100ms的音频,并间隔100ms。

3、发送finish-task指令:结束语音识别任务

该指令用于结束语音识别任务。音频发送完毕后,客户端可以发送此指令以结束任务。

示例:

{
    "header": {
        "action": "finish-task",
        "task_id": "2bf83b9a-baeb-4fda-8d9a-xxxxxxxxxxxx",
        "streaming": "duplex"
    },
    "payload": {
        "input": {}
    }
}

payload参数说明:

参数

类型

是否必选

说明

payload.input

Object

固定格式:{}。

四、关闭WebSocket连接

在程序正常结束、运行中出现异常或接收到task-finishedtask-failed事件时,关闭WebSocket连接。通常通过调用工具库中的close函数来实现。

示例代码

如下示例中,使用的音频文件为asr_example.wav

Go

package main

import (
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"time"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)

const (
	wsURL     = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/" // WebSocket服务器地址
	audioFile = "asr_example.wav"                                   // 替换为您的音频文件路径
)

var dialer = websocket.DefaultDialer

func main() {
	// 若没有将API Key配置到环境变量,可将下行替换为:apiKey := "your_api_key"。不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
	apiKey := os.Getenv("DASHSCOPE_API_KEY")

	// 连接WebSocket服务
	conn, err := connectWebSocket(apiKey)
	if err != nil {
		log.Fatal("连接WebSocket失败:", err)
	}
	defer closeConnection(conn)

	// 启动一个goroutine来接收结果
	taskStarted := make(chan bool)
	taskDone := make(chan bool)
	startResultReceiver(conn, taskStarted, taskDone)

	// 发送run-task指令
	taskID, err := sendRunTaskCmd(conn)
	if err != nil {
		log.Fatal("发送run-task指令失败:", err)
	}

	// 等待task-started事件
	waitForTaskStarted(taskStarted)

	// 发送待识别音频文件流
	if err := sendAudioData(conn); err != nil {
		log.Fatal("发送音频失败:", err)
	}

	// 发送finish-task指令
	if err := sendFinishTaskCmd(conn, taskID); err != nil {
		log.Fatal("发送finish-task指令失败:", err)
	}

	// 等待任务完成或失败
	<-taskDone
}

// 定义结构体来表示JSON数据
type Header struct {
	Action       string                 `json:"action"`
	TaskID       string                 `json:"task_id"`
	Streaming    string                 `json:"streaming"`
	Event        string                 `json:"event"`
	ErrorCode    string                 `json:"error_code,omitempty"`
	ErrorMessage string                 `json:"error_message,omitempty"`
	Attributes   map[string]interface{} `json:"attributes"`
}

type Output struct {
	Sentence struct {
		BeginTime int64  `json:"begin_time"`
		EndTime   *int64 `json:"end_time"`
		Text      string `json:"text"`
		Words     []struct {
			BeginTime   int64  `json:"begin_time"`
			EndTime     *int64 `json:"end_time"`
			Text        string `json:"text"`
			Punctuation string `json:"punctuation"`
		} `json:"words"`
	} `json:"sentence"`
	Usage interface{} `json:"usage"`
}

type Payload struct {
	TaskGroup  string     `json:"task_group"`
	Task       string     `json:"task"`
	Function   string     `json:"function"`
	Model      string     `json:"model"`
	Parameters Params     `json:"parameters"`
	Resources  []Resource `json:"resources"`
	Input      Input      `json:"input"`
	Output     Output     `json:"output,omitempty"`
}

type Params struct {
	Format                   string   `json:"format"`
	SampleRate               int      `json:"sample_rate"`
	VocabularyID             string   `json:"vocabulary_id"`
	DisfluencyRemovalEnabled bool     `json:"disfluency_removal_enabled"`
	LanguageHints            []string `json:"language_hints"`
}

type Resource struct {
	ResourceID   string `json:"resource_id"`
	ResourceType string `json:"resource_type"`
}

type Input struct {
}

type Event struct {
	Header  Header  `json:"header"`
	Payload Payload `json:"payload"`
}

// 连接WebSocket服务
func connectWebSocket(apiKey string) (*websocket.Conn, error) {
	header := make(http.Header)
	header.Add("X-DashScope-DataInspection", "enable")
	header.Add("Authorization", fmt.Sprintf("bearer %s", apiKey))
	conn, _, err := dialer.Dial(wsURL, header)
	return conn, err
}

// 启动一个goroutine异步接收WebSocket消息
func startResultReceiver(conn *websocket.Conn, taskStarted chan<- bool, taskDone chan<- bool) {
	go func() {
		for {
			_, message, err := conn.ReadMessage()
			if err != nil {
				log.Println("解析服务器消息失败:", err)
				return
			}
			var event Event
			err = json.Unmarshal(message, &event)
			if err != nil {
				log.Println("解析事件失败:", err)
				continue
			}
			if handleEvent(conn, event, taskStarted, taskDone) {
				return
			}
		}
	}()
}

// 发送run-task指令
func sendRunTaskCmd(conn *websocket.Conn) (string, error) {
	runTaskCmd, taskID, err := generateRunTaskCmd()
	if err != nil {
		return "", err
	}
	err = conn.WriteMessage(websocket.TextMessage, []byte(runTaskCmd))
	return taskID, err
}

// 生成run-task指令
func generateRunTaskCmd() (string, string, error) {
	taskID := uuid.New().String()
	runTaskCmd := Event{
		Header: Header{
			Action:    "run-task",
			TaskID:    taskID,
			Streaming: "duplex",
		},
		Payload: Payload{
			TaskGroup: "audio",
			Task:      "asr",
			Function:  "recognition",
			Model:     "paraformer-realtime-v2",
			Parameters: Params{
				Format:     "pcm",
				SampleRate: 16000,
			},
			Input: Input{},
		},
	}
	runTaskCmdJSON, err := json.Marshal(runTaskCmd)
	return string(runTaskCmdJSON), taskID, err
}

// 等待task-started事件
func waitForTaskStarted(taskStarted chan bool) {
	select {
	case <-taskStarted:
		fmt.Println("任务开启成功")
	case <-time.After(10 * time.Second):
		log.Fatal("等待task-started超时,任务开启失败")
	}
}

// 发送音频数据
func sendAudioData(conn *websocket.Conn) error {
	file, err := os.Open(audioFile)
	if err != nil {
		return err
	}
	defer file.Close()

	buf := make([]byte, 1024) // 假设100ms的音频数据大约为1024字节
	for {
		n, err := file.Read(buf)
		if n == 0 {
			break
		}
		if err != nil && err != io.EOF {
			return err
		}
		err = conn.WriteMessage(websocket.BinaryMessage, buf[:n])
		if err != nil {
			return err
		}
		time.Sleep(100 * time.Millisecond)
	}
	return nil
}

// 发送finish-task指令
func sendFinishTaskCmd(conn *websocket.Conn, taskID string) error {
	finishTaskCmd, err := generateFinishTaskCmd(taskID)
	if err != nil {
		return err
	}
	err = conn.WriteMessage(websocket.TextMessage, []byte(finishTaskCmd))
	return err
}

// 生成finish-task指令
func generateFinishTaskCmd(taskID string) (string, error) {
	finishTaskCmd := Event{
		Header: Header{
			Action:    "finish-task",
			TaskID:    taskID,
			Streaming: "duplex",
		},
		Payload: Payload{
			Input: Input{},
		},
	}
	finishTaskCmdJSON, err := json.Marshal(finishTaskCmd)
	return string(finishTaskCmdJSON), err
}

// 处理事件
func handleEvent(conn *websocket.Conn, event Event, taskStarted chan<- bool, taskDone chan<- bool) bool {
	switch event.Header.Event {
	case "task-started":
		fmt.Println("收到task-started事件")
		taskStarted <- true
	case "result-generated":
		if event.Payload.Output.Sentence.Text != "" {
			fmt.Println("识别结果:", event.Payload.Output.Sentence.Text)
		}
	case "task-finished":
		fmt.Println("任务完成")
		taskDone <- true
		return true
	case "task-failed":
		handleTaskFailed(event, conn)
		taskDone <- true
		return true
	default:
		log.Printf("预料之外的事件:%v", event)
	}
	return false
}

// 处理任务失败事件
func handleTaskFailed(event Event, conn *websocket.Conn) {
	if event.Header.ErrorMessage != "" {
		log.Fatalf("任务失败:%s", event.Header.ErrorMessage)
	} else {
		log.Fatal("未知原因导致任务失败")
	}
}

// 关闭连接
func closeConnection(conn *websocket.Conn) {
	if conn != nil {
		conn.Close()
	}
}

C#

示例代码如下:

using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;

class Program {
    private static ClientWebSocket _webSocket = new ClientWebSocket();
    private static CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
    private static bool _taskStartedReceived = false;
    private static bool _taskFinishedReceived = false;
    // 若没有将API Key配置到环境变量,可将下行替换为:private const string ApiKey="your_api_key"。不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
    private static readonly string ApiKey = Environment.GetEnvironmentVariable("DASHSCOPE_API_KEY") ?? throw new InvalidOperationException("DASHSCOPE_API_KEY environment variable is not set.");

    // WebSocket服务器地址
    private const string WebSocketUrl = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/";
    // 替换为您的音频文件路径
    private const string AudioFilePath = "asr_example.wav";

    static async Task Main(string[] args) {
        // 建立WebSocket连接,配置headers进行鉴权
        _webSocket.Options.SetRequestHeader("Authorization", ApiKey);
        _webSocket.Options.SetRequestHeader("X-DashScope-DataInspection", "enable");

        await _webSocket.ConnectAsync(new Uri(WebSocketUrl), _cancellationTokenSource.Token);

        // 启动线程异步接收WebSocket消息
        var receiveTask = ReceiveMessagesAsync();

        // 发送run-task指令
        string _taskId = Guid.NewGuid().ToString("N"); // 生成32位随机ID
        var runTaskJson = GenerateRunTaskJson(_taskId);
        await SendAsync(runTaskJson);

        // 等待task-started事件
        while (!_taskStartedReceived) {
            await Task.Delay(100, _cancellationTokenSource.Token);
        }

        // 读取本地文件,向服务器发送待识别音频流
        await SendAudioStreamAsync(AudioFilePath);

        // 发送finish-task指令结束任务
        var finishTaskJson = GenerateFinishTaskJson(_taskId);
        await SendAsync(finishTaskJson);

        // 等待task-finished事件
        while (!_taskFinishedReceived && !_cancellationTokenSource.IsCancellationRequested) {
            try {
                await Task.Delay(100, _cancellationTokenSource.Token);
            } catch (OperationCanceledException) {
                // 任务已被取消,退出循环
                break;
            }
        }

        // 关闭连接
        if (!_cancellationTokenSource.IsCancellationRequested) {
            await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", _cancellationTokenSource.Token);
        }

        _cancellationTokenSource.Cancel();
        try {
            await receiveTask;
        } catch (OperationCanceledException) {
            // 忽略操作取消异常
        }
    }

    private static async Task ReceiveMessagesAsync() {
        try {
            while (_webSocket.State == WebSocketState.Open && !_cancellationTokenSource.IsCancellationRequested) {
                var message = await ReceiveMessageAsync(_cancellationTokenSource.Token);
                if (message != null) {
                    var eventValue = message["header"]?["event"]?.GetValue<string>();
                    switch (eventValue) {
                        case "task-started":
                            Console.WriteLine("任务开启成功");
                            _taskStartedReceived = true;
                            break;
                        case "result-generated":
                            Console.WriteLine($"识别结果:{message["payload"]?["output"]?["sentence"]?["text"]?.GetValue<string>()}");
                            break;
                        case "task-finished":
                            Console.WriteLine("任务完成");
                            _taskFinishedReceived = true;
                            _cancellationTokenSource.Cancel();
                            break;
                        case "task-failed":
                            Console.WriteLine($"任务失败:{message["header"]?["error_message"]?.GetValue<string>()}");
                            _cancellationTokenSource.Cancel();
                            break;
                    }
                }
            }
        } catch (OperationCanceledException) {
            // 忽略操作取消异常
        }
    }

    private static async Task<JsonNode?> ReceiveMessageAsync(CancellationToken cancellationToken) {
        var buffer = new byte[1024 * 4];
        var segment = new ArraySegment<byte>(buffer);
        var result = await _webSocket.ReceiveAsync(segment, cancellationToken);

        if (result.MessageType == WebSocketMessageType.Close) {
            await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", cancellationToken);
            return null;
        }

        var message = Encoding.UTF8.GetString(buffer, 0, result.Count);
        return JsonNode.Parse(message);
    }

    private static async Task SendAsync(string message) {
        var buffer = Encoding.UTF8.GetBytes(message);
        var segment = new ArraySegment<byte>(buffer);
        await _webSocket.SendAsync(segment, WebSocketMessageType.Text, true, _cancellationTokenSource.Token);
    }

    private static async Task SendAudioStreamAsync(string filePath) {
        using (var audioStream = File.OpenRead(filePath)) {
            var buffer = new byte[1024]; // 每次发送100ms的音频数据
            int bytesRead;

            while ((bytesRead = await audioStream.ReadAsync(buffer, 0, buffer.Length)) > 0) {
                var segment = new ArraySegment<byte>(buffer, 0, bytesRead);
                await _webSocket.SendAsync(segment, WebSocketMessageType.Binary, true, _cancellationTokenSource.Token);
                await Task.Delay(100); // 间隔100ms
            }
        }
    }

    private static string GenerateRunTaskJson(string taskId) {
        var runTask = new JsonObject {
            ["header"] = new JsonObject {
                ["action"] = "run-task",
                ["task_id"] = taskId,
                ["streaming"] = "duplex"
            },
            ["payload"] = new JsonObject {
                ["task_group"] = "audio",
                ["task"] = "asr",
                ["function"] = "recognition",
                ["model"] = "paraformer-realtime-v2",
                ["parameters"] = new JsonObject {
                    ["format"] = "pcm",
                    ["sample_rate"] = 16000,
                    ["vocabulary_id"] = "vocab-xxx-24ee19fa8cfb4d52902170a0xxxxxxxx",
                    ["disfluency_removal_enabled"] = false
                },
                ["resources"] = new JsonArray {
                    new JsonObject {
                        ["resource_id"] = "xxxxxxxxxxxx",
                        ["resource_type"] = "asr_phrase"
                    }
                },
                ["input"] = new JsonObject()
            }
        };
        return JsonSerializer.Serialize(runTask);
    }

    private static string GenerateFinishTaskJson(string taskId) {
        var finishTask = new JsonObject {
            ["header"] = new JsonObject {
                ["action"] = "finish-task",
                ["task_id"] = taskId,
                ["streaming"] = "duplex"
            },
            ["payload"] = new JsonObject {
                ["input"] = new JsonObject()
            }
        };
        return JsonSerializer.Serialize(finishTask);
    }
}

PHP

示例代码目录结构为:

my-php-project/

├── composer.json

├── vendor/

└── index.php

composer.json内容如下,相关依赖的版本号请根据实际情况自行决定:

{
    "require": {
        "react/event-loop": "^1.3",
        "react/socket": "^1.11",
        "react/stream": "^1.2",
        "react/http": "^1.1",
        "ratchet/pawl": "^0.4"
    },
    "autoload": {
        "psr-4": {
            "App\\": "src/"
        }
    }
}

index.php内容如下:

<?php

require __DIR__ . '/vendor/autoload.php';

use Ratchet\Client\Connector;
use React\EventLoop\Loop;
use React\Socket\Connector as SocketConnector;
use Ratchet\rfc6455\Messaging\Frame;

# 若没有将API Key配置到环境变量,可将下行替换为:$api_key="your_api_key"。不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
$api_key = getenv("DASHSCOPE_API_KEY");
$websocket_url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/'; // WebSocket服务器地址
$audio_file_path = 'asr_example.wav'; // 替换为您的音频文件路径

$loop = Loop::get();

// 创建自定义的连接器
$socketConnector = new SocketConnector($loop, [
    'tcp' => [
        'bindto' => '0.0.0.0:0',
    ],
    'tls' => [
        'verify_peer' => false,
        'verify_peer_name' => false,
    ],
]);

$connector = new Connector($loop, $socketConnector);

$headers = [
    'Authorization' => 'bearer ' . $api_key,
    'X-DashScope-DataInspection' => 'enable'
];

$connector($websocket_url, [], $headers)->then(function ($conn) use ($loop, $audio_file_path) {
    echo "连接到WebSocket服务器\n";

    // 启动异步接收WebSocket消息的线程
    $conn->on('message', function($msg) use ($conn, $loop, $audio_file_path) {
        $response = json_decode($msg, true);

        if (isset($response['header']['event'])) {
            handleEvent($conn, $response, $loop, $audio_file_path);
        } else {
            echo "未知的消息格式\n";
        }
    });

    // 监听连接关闭
    $conn->on('close', function($code = null, $reason = null) {
        echo "连接已关闭\n";
        if ($code !== null) {
            echo "关闭代码: " . $code . "\n";
        }
        if ($reason !== null) {
            echo "关闭原因:" . $reason . "\n";
        }
    });

    // 生成任务ID
    $taskId = generateTaskId();

    // 发送 run-task 指令
    sendRunTaskMessage($conn, $taskId);

}, function ($e) {
    echo "无法连接:{$e->getMessage()}\n";
});

$loop->run();

/**
 * 生成任务ID
 * @return string
 */
function generateTaskId(): string {
    return bin2hex(random_bytes(16));
}

/**
 * 发送 run-task 指令
 * @param $conn
 * @param $taskId
 */
function sendRunTaskMessage($conn, $taskId) {
    $runTaskMessage = json_encode([
        "header" => [
            "action" => "run-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
        ],
        "payload" => [
            "task_group" => "audio",
            "task" => "asr",
            "function" => "recognition",
            "model" => "paraformer-realtime-v2",
            "parameters" => [
                "format" => "pcm",
                "sample_rate" => 16000
            ],
            "resources" => [
                [
                    "resource_id" => "xxxxxxxxxxxx",
                    "resource_type" => "asr_phrase"
                ]
            ],
            "input" => []
        ]
    ]);
    echo "准备发送run-task指令:" . $runTaskMessage . "\n";
    $conn->send($runTaskMessage);
    echo "run-task指令已发送\n";
}

/**
 * 读取音频文件
 * @param string $filePath
 * @return bool|string
 */
function readAudioFile(string $filePath) {
    $voiceData = file_get_contents($filePath);
    if ($voiceData === false) {
        echo "无法读取音频文件\n";
    }
    return $voiceData;
}

/**
 * 分割音频数据
 * @param string $data
 * @param int $chunkSize
 * @return array
 */
function splitAudioData(string $data, int $chunkSize): array {
    return str_split($data, $chunkSize);
}

/**
 * 发送 finish-task 指令
 * @param $conn
 * @param $taskId
 */
function sendFinishTaskMessage($conn, $taskId) {
    $finishTaskMessage = json_encode([
        "header" => [
            "action" => "finish-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
        ],
        "payload" => [
            "input" => []
        ]
    ]);
    echo "准备发送finish-task指令: " . $finishTaskMessage . "\n";
    $conn->send($finishTaskMessage);
    echo "finish-task指令已发送\n";
}

/**
 * 处理事件
 * @param $conn
 * @param $response
 * @param $loop
 * @param $audio_file_path
 */
function handleEvent($conn, $response, $loop, $audio_file_path) {
    static $taskId;
    static $chunks;
    static $allChunksSent = false;

    if (is_null($taskId)) {
        $taskId = generateTaskId();
    }

    switch ($response['header']['event']) {
        case 'task-started':
            echo "任务开始,发送音频数据...\n";
            // 读取音频文件
            $voiceData = readAudioFile($audio_file_path);
            if ($voiceData === false) {
                echo "无法读取音频文件\n";
                $conn->close();
                return;
            }

            // 分割音频数据
            $chunks = splitAudioData($voiceData, 1024);

            // 定义发送函数
            $sendChunk = function() use ($conn, &$chunks, $loop, &$sendChunk, &$allChunksSent, $taskId) {
                if (!empty($chunks)) {
                    $chunk = array_shift($chunks);
                    $binaryMsg = new Frame($chunk, true, Frame::OP_BINARY);
                    $conn->send($binaryMsg);
                    // 100ms后发送下一个片段
                    $loop->addTimer(0.1, $sendChunk);
                } else {
                    echo "所有数据块已发送\n";
                    $allChunksSent = true;

                    // 发送 finish-task 指令
                    sendFinishTaskMessage($conn, $taskId);
                }
            };

            // 开始发送音频数据
            $sendChunk();
            break;
        case 'result-generated':
            $result = $response['payload']['output']['sentence'];
            echo "识别结果:" . $result['text'] . "\n";
            break;
        case 'task-finished':
            echo "任务完成\n";
            $conn->close();
            break;
        case 'task-failed':
            echo "任务失败\n";
            echo "错误代码:" . $response['header']['error_code'] . "\n";
            echo "错误信息:" . $response['header']['error_message'] . "\n";
            $conn->close();
            break;
        case 'error':
            echo "错误:" . $response['payload']['message'] . "\n";
            break;
        default:
            echo "未知事件:" . $response['header']['event'] . "\n";
            break;
    }

    // 如果所有数据已发送且任务已完成,关闭连接
    if ($allChunksSent && $response['header']['event'] == 'task-finished') {
        // 等待1秒以确保所有数据都已传输完毕
        $loop->addTimer(1, function() use ($conn) {
            $conn->close();
            echo "客户端关闭连接\n";
        });
    }
}

Node.js

需安装相关依赖:

npm install ws
npm install uuid

示例代码如下:

const fs = require('fs');
const WebSocket = require('ws');
const { v4: uuidv4 } = require('uuid'); // 用于生成UUID

// 若没有将API Key配置到环境变量,可将下行替换为:apiKey = 'your_api_key'。不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
const apiKey = process.env.DASHSCOPE_API_KEY;
const url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/'; // WebSocket服务器地址
const audioFile = 'asr_example.wav'; // 替换为您的音频文件路径

// 生成32位随机ID
const TASK_ID = uuidv4().replace(/-/g, '').slice(0, 32);

// 创建WebSocket客户端
const ws = new WebSocket(url, {
  headers: {
    Authorization: `bearer ${apiKey}`,
    'X-DashScope-DataInspection': 'enable'
  }
});

let taskStarted = false; // 标记任务是否已启动

// 连接打开时发送run-task指令
ws.on('open', () => {
  console.log('连接到服务器');
  sendRunTask();
});

// 接收消息处理
ws.on('message', (data) => {
  const message = JSON.parse(data);
  switch (message.header.event) {
    case 'task-started':
      console.log('任务开始');
      taskStarted = true;
      sendAudioStream();
      break;
    case 'result-generated':
      console.log('识别结果:', message.payload.output.sentence.text);
      break;
    case 'task-finished':
      console.log('任务完成');
      ws.close();
      break;
    case 'task-failed':
      console.error('任务失败:', message.header.error_message);
      ws.close();
      break;
    default:
      console.log('未知事件:', message.header.event);
  }
});

// 如果没有收到task-started事件,关闭连接
ws.on('close', () => {
  if (!taskStarted) {
    console.error('任务未启动,关闭连接');
  }
});

// 发送run-task指令
function sendRunTask() {
  const runTaskMessage = {
    header: {
      action: 'run-task',
      task_id: TASK_ID,
      streaming: 'duplex'
    },
    payload: {
      task_group: 'audio',
      task: 'asr',
      function: 'recognition',
      model: 'paraformer-realtime-v2',
      parameters: {},
      input: {}
    }
  };
  ws.send(JSON.stringify(runTaskMessage));
}

// 发送音频流
function sendAudioStream() {
  const audioStream = fs.createReadStream(audioFile);
  let chunkCount = 0;

  function sendNextChunk() {
    const chunk = audioStream.read();
    if (chunk) {
      ws.send(chunk);
      chunkCount++;
      setTimeout(sendNextChunk, 100); // 每100ms发送一次
    }
  }

  audioStream.on('readable', () => {
    sendNextChunk();
  });

  audioStream.on('end', () => {
    console.log('音频流结束');
    sendFinishTask();
  });

  audioStream.on('error', (err) => {
    console.error('读取音频文件错误:', err);
    ws.close();
  });
}

// 发送finish-task指令
function sendFinishTask() {
  const finishTaskMessage = {
    header: {
      action: 'finish-task',
      task_id: TASK_ID,
      streaming: 'duplex'
    },
    payload: {
      input: {}
    }
  };
  ws.send(JSON.stringify(finishTaskMessage));
}

// 错误处理
ws.on('error', (error) => {
  console.error('WebSocket错误:', error);
});