ストリーミング STT - WebSocket

WebSocket 方式でストリーミング STT を実装する方法を説明します。

例

以下の例はローカル音声ファイルを入力とします。マイク等のデバイス入力に置き換えることで同様に利用できます。

認証トークン

認証ガイドに従ってトークンを取得してください。

パラメータ

共通 DecoderConfig/Parameters を参照してください。

レスポンス

{
  seq: integer,
  start_at: integer,
  duration: integer,
  final: boolean,
  alternatives: [
    {
      text: string,
      confidence: float,
      words?: [
        {
          text: string,
          start_at: integer,
          duration: integer,
          confidence: float
        }
      ]
    }
  ]
}

サンプルコード

Golang

package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/gorilla/websocket"
	"github.com/xfrr/goffmpeg/transcoder"
)

const ServerHost = "openapi.vito.ai"

var ClientId = os.Getenv("RTZR_CLIENT_ID")
var ClientSecret = os.Getenv("RTZR_CLIENT_SECRET")

const SAMPLE_RATE int = 8000
const BYTES_PER_SAMPLE int = 2

/*
본 예제에서는 스트리밍 입력을 음성파일을 읽어서 시뮬레이션 합니다.
실제사용시에는 마이크 입력 등의 실시간 음성 스트림이 들어와야합니다.
*/
type FileStreamer struct {
	file *os.File
}

func (fs *FileStreamer) Read(p []byte) (int, error) {
	byteSize := len(p)
	maxSize := 1024 * 1024
	if byteSize > maxSize {
		byteSize = maxSize
	}

	defer time.Sleep(time.Duration(byteSize/((SAMPLE_RATE*BYTES_PER_SAMPLE)/1000)) * time.Millisecond)
	return fs.file.Read(p[:byteSize])
}

func (fs *FileStreamer) Close() error {
	defer os.Remove(fs.file.Name())
	return fs.file.Close()
}

func OpenAudioFile(audioFile string) (io.ReadCloser, error) {
	fileName := filepath.Base(audioFile)
	i := strings.LastIndex(fileName, ".")
	audioFileName8K := filepath.Join(os.TempDir(), fileName[:i]) + fmt.Sprintf("_%d.%s", SAMPLE_RATE, "wav")
	trans := new(transcoder.Transcoder)
	if err := trans.Initialize(audioFile, audioFileName8K); err != nil {
		log.Fatal(err)
	}

	trans.MediaFile().SetAudioRate(SAMPLE_RATE)
	trans.MediaFile().SetAudioChannels(1)
	trans.MediaFile().SetSkipVideo(true)
	trans.MediaFile().SetAudioFilter("aresample=resampler=soxr")

	err := <-trans.Run(false)
	if err != nil {
		return nil, fmt.Errorf("transcode audio file failed: %w", err)
	}

	file, err := os.Open(audioFileName8K)
	if err != nil {
		return nil, fmt.Errorf("open audio file failed: %w", err)
	}

	return &FileStreamer{file: file}, nil
}

func main() {
	flag.Parse()
	log.SetFlags(0)

	interrupt := make(chan os.Signal, 1)
	signal.Notify(interrupt, os.Interrupt)

	u := url.URL{Scheme: "wss", Host: ServerHost, Path: "/v1/transcribe:streaming"}

	query := u.Query()
	query.Set("sample_rate", strconv.Itoa(SAMPLE_RATE))
	query.Set("encoding", "LINEAR16")
	query.Set("use_itn", "true")
	query.Set("use_disfluency_filter", "true")
	query.Set("use_profanity_filter", "false")
	u.RawQuery = query.Encode()

	log.Printf("connecting to %s", u.String())

	audioFile := flag.Arg(0)
	streamingFile, err := OpenAudioFile(audioFile)
	if err != nil {
		log.Fatal(err)
	}
	defer streamingFile.Close()

	data := map[string][]string{
		"client_id":     []string{ClientId},
		"client_secret": []string{ClientSecret},
	}
	resp, _ := http.PostForm("https://openapi.vito.ai/v1/authenticate", data)
	if resp.StatusCode != 200 {
		panic("Failed to authenticate")
	}

	bytes, _ := io.ReadAll(resp.Body)
	var result struct {
		Token string `json:"access_token"`
	}
	json.Unmarshal(bytes, &result)
	requestHeader := http.Header{
		"Authorization": []string{fmt.Sprintf("Bearer %s", result.Token)},
	}
	c, res, err := websocket.DefaultDialer.Dial(u.String(), requestHeader)
	if err != nil {
		log.Printf("status: %s, body: %v", res.Status, res.Body)
		log.Fatal("dial:", err)
	}
	defer c.Close()

	start := time.Now()
	go func() {
		buf := make([]byte, 1024)
		for {
			n, err := streamingFile.Read(buf)
			if err == io.EOF {
				// Nothing else to pipe, close the stream.
				log.Println("send EOS")
				if err := c.WriteMessage(websocket.TextMessage, []byte("EOS")); err != nil {
					log.Fatalf("Could not close stream: %v", err)
				}
				return
			}
			if err != nil {
				log.Printf("Could not read from %s: %v", audioFile, err)
				continue
			}
			err = c.WriteMessage(websocket.BinaryMessage, buf[:n])
			if err != nil {
				log.Println("write:", err)
			}

			select {
			case <-interrupt:
				log.Println("interrupt")
				if err := c.WriteMessage(websocket.TextMessage, []byte("EOS")); err != nil {
					log.Fatalf("Could not close: %v", err)
				}
				return
			default:

			}
		}
	}()

	start2 := time.Now()
	for {
		_, message, err := c.ReadMessage()
		if err != nil {
			log.Printf("elapsed[%v]\n", time.Since(start))
			log.Println("read:", err)
			return
		}
		log.Printf("[%v]recv: %s", time.Since(start2), message)
		start2 = time.Now()
	}

}

Python

import argparse
import asyncio
import json
import logging
import os
import tempfile
import time
from io import DEFAULT_BUFFER_SIZE

import websockets
from pydub import AudioSegment
from requests import Session

API_BASE = "https://openapi.vito.ai"
CLIENT_ID = os.environ.get("RTZR_CLIENT_ID")
CLIENT_SECRET = os.environ.get("RTZR_CLIENT_SECRET")

if not CLIENT_ID or not CLIENT_SECRET:
    raise ValueError("RTZR_CLIENT_ID and RTZR_CLIENT_SECRET must be set")


SAMPLE_RATE = 8000
BYTES_PER_SAMPLE = 2


# In this example, we simulate streaming input by reading an audio file.
# When using the API, you need to provide a real-time audio stream, such as microphone input.
class FileStreamer:
    def __init__(self, file):
        file_name = os.path.basename(file)
        i = file_name.rindex(".")
        audio_file_8k_path = os.path.join(tempfile.gettempdir(), file_name[:i]) + "_" + str(SAMPLE_RATE) + ".wav"
        self.filepath = audio_file_8k_path
        audio = AudioSegment.from_file(file=file, format=file[i + 1 :])
        audio = audio.set_frame_rate(SAMPLE_RATE)
        audio = audio.set_channels(1)
        audio.export(audio_file_8k_path, format="wav")
        self.file = open(audio_file_8k_path, "rb")

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.file.close()
        os.remove(self.filepath)

    async def read(self, size):
        if size > 1024 * 1024:
            size = 1024 * 1024
        await asyncio.sleep(size / (SAMPLE_RATE * BYTES_PER_SAMPLE))
        content = self.file.read(size)
        return content


class RTZROpenAPIClient:
    def __init__(self, client_id, client_secret):
        super().__init__()
        self._logger = logging.getLogger(__name__)
        self.client_id = client_id
        self.client_secret = client_secret
        self._sess = Session()
        self._token = None

    @property
    def token(self):
        if self._token is None or self._token["expire_at"] < time.time():
            resp = self._sess.post(
                API_BASE + "/v1/authenticate",
                data={"client_id": self.client_id, "client_secret": self.client_secret},
            )
            resp.raise_for_status()
            self._token = resp.json()
        return self._token["access_token"]

    async def streaming_transcribe(self, filename, config=None):
        if config is None:
            config = dict(
                sample_rate=str(SAMPLE_RATE),
                encoding="LINEAR16",
                use_itn="true",
                use_disfluency_filter="false",
                use_profanity_filter="false",
            )

        STREAMING_ENDPOINT = "wss://{}/v1/transcribe:streaming?{}".format(
            API_BASE.split("://")[1], "&".join(map("=".join, config.items()))
        )
        conn_kwargs = dict(extra_headers={"Authorization": "bearer " + self.token})

        async def streamer(websocket):
            with FileStreamer(filename) as f:
                while True:
                    buff = await f.read(DEFAULT_BUFFER_SIZE)
                    if buff is None or len(buff) == 0:
                        break
                    await websocket.send(buff)
                await websocket.send("EOS")

        async def transcriber(websocket):
            async for msg in websocket:
                msg = json.loads(msg)
                print(msg)
                if msg["final"]:
                    print("final ended with " + msg["alternatives"][0]["text"])

        async with websockets.connect(STREAMING_ENDPOINT, **conn_kwargs) as websocket:
            await asyncio.gather(
                streamer(websocket),
                transcriber(websocket),
            )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("stream", help="File to stream to the API")
    args = parser.parse_args()

    client = RTZROpenAPIClient(CLIENT_ID, CLIENT_SECRET)
    asyncio.run(client.streaming_transcribe(args.stream))

Java

// full version
// https://github.com/vito-ai/java-sample

/* This example supports the following file formats:
WAV, AU, AIFF
https://docs.oracle.com/javase/8/docs/technotes/guides/sound/index.html
*/

package ai.vito.openapi.stream;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.HashMap;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.concurrent.CountDownLatch;

import javax.sound.sampled.AudioFormat;
import javax.sound.sampled.AudioInputStream;
import javax.sound.sampled.AudioSystem;
import javax.sound.sampled.UnsupportedAudioFileException;

import com.fasterxml.jackson.databind.ObjectMapper;

import okhttp3.FormBody;
import okhttp3.HttpUrl;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;
import okio.ByteString;

/*
In this example, we simulate streaming input by reading an audio file.
When using the API, you need to provide a real-time audio stream, such as microphone input.
*/
final class FileStreamer {
    private AudioInputStream audio8KStream;
    private int SAMPLE_RATE = 8000;
    private int BITS_PER_SAMPLE = 16;

    public FileStreamer(String filePath) throws IOException, UnsupportedAudioFileException {
        File file = new File(filePath);
        try {
            AudioInputStream originalAudioStream = AudioSystem.getAudioInputStream(file);
            AudioFormat originalFormat = originalAudioStream.getFormat();
            AudioFormat newFormat = new AudioFormat(
                    AudioFormat.Encoding.PCM_SIGNED,
                    SAMPLE_RATE,
                    BITS_PER_SAMPLE,
                    1,
                    1 * (BITS_PER_SAMPLE / 8),
                    SAMPLE_RATE,
                    originalFormat.isBigEndian());

            this.audio8KStream = AudioSystem.getAudioInputStream(newFormat, originalAudioStream);
        } catch (IOException | UnsupportedAudioFileException e) {
            throw e;
        }
    }

    public int read(byte[] b) throws IOException, InterruptedException {
        int maxSize = 1024 * 1024;
        int byteSize = Math.min(b.length, maxSize);
        try {
            Thread.sleep(byteSize / ((SAMPLE_RATE * (BITS_PER_SAMPLE / 8)) / 1000));
        } catch (InterruptedException e) {
            throw e;
        }
        return this.audio8KStream.read(b, 0, byteSize);
    }

    public void close() throws IOException {
        this.audio8KStream.close();
    }
}

public class RTZRSttWebSocketClient {
    public static String getAccessToken() throws IOException {
        OkHttpClient client = new OkHttpClient();
        RequestBody formBody = new FormBody.Builder()
                .add("client_id", "{RTZR_CLIENT_ID}")
                .add("client_secret", "{RTZR_CLIENT_SECRET}")
                .build();

        Request request = new Request.Builder()
                .url("https://openapi.vito.ai/v1/authenticate")
                .post(formBody)
                .build();

        Response response = client.newCall(request).execute();
        ObjectMapper objectMapper = new ObjectMapper();
        HashMap<String, String> map = objectMapper.readValue(response.body().string(), HashMap.class);

        return map.get("access_token");
    }

    public static void main(String[] args) throws Exception {
        Logger logger = Logger.getLogger(RTZRSttWebSocketClient.class.getName());
        OkHttpClient client = new OkHttpClient();

        String token = getAccessToken();

        HttpUrl.Builder httpBuilder = HttpUrl.get("https://openapi.vito.ai/v1/transcribe:streaming").newBuilder();
        httpBuilder.addQueryParameter("sample_rate", "8000");
        httpBuilder.addQueryParameter("encoding", "LINEAR16");
        httpBuilder.addQueryParameter("use_itn", "true");
        httpBuilder.addQueryParameter("use_disfluency_filter", "true");
        httpBuilder.addQueryParameter("use_profanity_filter", "false");

        String url = httpBuilder.toString().replace("https://", "wss://");

        Request request = new Request.Builder()
                .url(url)
                .addHeader("Authorization", "Bearer " + token)
                .build();

        RTZRWebSocketListener webSocketListener = new RTZRWebSocketListener();
        WebSocket rtzrWebSocket = client.newWebSocket(request, webSocketListener);

        FileStreamer fileStreamer = new FileStreamer("sample.wav");

        byte[] buffer = new byte[1024];
        int readBytes;
        while ((readBytes = fileStreamer.read(buffer)) != -1) {
            boolean sent = rtzrWebSocket.send(ByteString.of(buffer, 0, readBytes));
            if (!sent) {
                logger.log(Level.WARNING, "Send buffer is full. Cannot complete request. Increase sleep interval.");
                System.exit(1);
            }
        }
        fileStreamer.close();
        rtzrWebSocket.send("EOS");

        webSocketListener.waitClose();
        client.dispatcher().executorService().shutdown();
    }
}

class RTZRWebSocketListener extends WebSocketListener {
    private static final Logger logger = Logger.getLogger(RTZRWebSocketListener.class.getName());
    private static final int NORMAL_CLOSURE_STATUS = 1000;
    private CountDownLatch latch = null;

    private static void log(Level level, String msg, Object... args) {
        logger.log(level, msg, args);
    }

    @Override
    public void onOpen(WebSocket webSocket, Response response) {
        log(Level.INFO, "Open " + response.message());
        latch = new CountDownLatch(1);
    }

    @Override
    public void onMessage(WebSocket webSocket, String text) {
        System.out.println(text);
    }

    @Override
    public void onMessage(WebSocket webSocket, ByteString bytes) {
        System.out.println(bytes.hex());
    }

    @Override
    public void onClosing(WebSocket webSocket, int code, String reason) {
        webSocket.close(NORMAL_CLOSURE_STATUS, null);
        log(Level.INFO, "Closing {0} {1}", code, reason);
    }

    @Override
    public void onClosed(WebSocket webSocket, int code, String reason) {
        webSocket.close(NORMAL_CLOSURE_STATUS, null);
        log(Level.INFO, "Closed {0} {1}", code, reason);
        latch.countDown();
    }

    @Override
    public void onFailure(WebSocket webSocket, Throwable t, Response response) {
        t.printStackTrace();
        latch.countDown();
    }

    public void waitClose() throws InterruptedException {
        log(Level.INFO, "Wait for finish");
        latch.await();
    }
}

エラーコード

WebSocket ダイアル時のステータスコードでエラーを示します。

HttpStatus	Code	Notes
400	H0001	パラメータ不正
401	H0002	認証失敗
429	A0001	使用量超過
500	E500	サーバーエラー

備考

音声ファイルの変換は、バッチ STT の方が簡便な場合があります。