本文介绍如何通过WebSocket连接访问实时语音识别服务。
DashScope SDK目前仅支持Java和Python。若想使用其他编程语言开发Paraformer实时语音识别应用程序,可以通过WebSocket连接与服务进行通信。
WebSocket是一种支持全双工通信的网络协议。客户端和服务器通过一次握手建立持久连接,双方可以互相主动推送数据,因此在实时性和效率方面具有显著优势。
对于常用编程语言,有许多现成的WebSocket库和示例可供参考,例如:
Go:
gorilla/websocket
PHP:
Ratchet
Node.js:
ws
建议您先了解WebSocket的基本原理和技术细节,再参照本文进行开发。
前提条件
客户端与服务端的交互流程
按时间顺序,客户端与服务端的交互流程如下:
建立连接:客户端与服务端建立WebSocket连接。
开启任务:
客户端发送
run-task
指令以开启任务。客户端收到服务端返回的
task-started
事件,标志着任务已成功开启,可以进行后续步骤。
发送音频流:
客户端开始发送音频流,并同时接收服务端持续返回的
result-generated
事件,该事件包含语音识别结果。
通知服务端结束任务:
客户端发送
finish-task
指令通知服务端结束任务,并继续接收服务端返回的result-generated
事件。
任务结束:
客户端收到服务端返回的
task-finished
事件,标志着任务结束。
关闭连接:客户端关闭WebSocket连接。
WebSocket客户端编程与消息处理
在编写WebSocket客户端代码时,为了同时发送和接收消息,通常采用异步编程。您可以按照以下步骤来编写程序:
建立WebSocket连接:首先,初始化并建立与服务器的WebSocket连接。
异步监听服务器消息:启动一个单独的线程(具体实现方式因编程语言而异)来监听服务器返回的消息,根据消息内容进行相应的操作。
发送消息:在不同于监听服务器消息的线程中(例如主线程,具体实现方式因编程语言而异),向服务器发送消息。
关闭连接:在程序结束前,确保关闭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格式,由header
和payload
这两部分组成:
payload
:包含基础信息外的其他信息。不同事件的payload
格式可能不同。
共有如下四种事件:
1、task-started事件:语音识别任务已开启
2、result-generated事件:包含语音识别的结果
3、task-finished事件:语音识别任务已结束
4、task-failed事件:任务失败
三、给服务器发送消息
在与监听服务器消息不同的线程中(比如主线程,具体实现因编程语言而异),向服务器发送消息。
客户端发送给服务端的消息有两种:
音频流。
向服务器发送消息需要遵循如下时序,否则会导致任务失败:首先发送run-task
指令,待监听到服务器返回的task-started
事件后,再发送待识别的音频流。在音频流发送结束后,发送finish-task
指令。
1、发送run-task指令:开启语音识别任务(支持定制热词)
2、发送待识别音频流
3、发送finish-task指令:结束语音识别任务
四、关闭WebSocket连接
在程序正常结束、运行中出现异常或接收到task-finished
、task-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);
});