Does model have state?

[lionsheep0724]

I'm using silero-vad(v3) for streaming, serve model with aiohttp server.
The server consumes every 512 bytes of audio data and feed it to the model.

When I initialize model at every request, vad process works properly.
But the issue occurs when I inject vad model to VADIterator, to avoid overhead. Its performance goes bad (I guess it splits speech to early). Since VADIterator has state, I keep this to be initialized when every reqeust comes.
Why this happen? Does vad model has state, also?

1. Good case

# This is main function of server
async def stream(request):
        audio_pipeline:AudioPipeline = AudioPipeline(session_id=session_id)
        
        await response.prepare(request)
        await stt_manager.process_request(request, response)

# VADProcessor injected to AudioPipeline
class AudioPipeline:
    def __init__(self, engine, session_id):
        self.engine = engine
        self.audio_processor = AudioProcessor()
        self.audio_buffer_processor = AudioBufferProcessor()
        self.vad_processor = VADProcessor(VAD_MODEL_PATH)

class VADProcessor:
    def __init__(self, vad_model_path: str):
        self.vad_iterator = self.get_vad_iterator(vad_model_path)

    def get_vad_iterator(self, vad_model_path: str):
        self.vad_model, VADIterator = torch.hub.load(
            source="local", repo_or_dir=vad_model_path, model="silero_vad", onnx=True
        )
        vad_iterator: VADIterator = VADIterator(self.vad_model, sampling_rate=8000,min_silence_duration_ms=MIN_SILENCE_DURATION_MS ,max_speech_duration_ms=MAX_SPEECH_DURATION_MS)
        return vad_iterator

    def detect_speech(self, audio_chunk: bytes) -> Dict[str, float]:
        audio_tensor: torch.tensor = torch.frombuffer(audio_chunk, dtype=torch.int16).float() / 32768.0
        audio_tensor: torch.tensor = audio_tensor.unsqueeze(0)
        audio_samples: int = audio_tensor.shape[1]
        if audio_samples < MINIMUM_SAMPLES_FOR_VAD:
            padding_length = MIN_SILENCE_SAMPLES - audio_samples
            padded_audio_tensor = F.pad(audio_tensor, (0, padding_length), "constant", 0)
            vad_event: Dict[str, float] = self.vad_iterator(padded_audio_tensor, return_seconds=True)
            speech_prob: float = self.vad_model(padded_audio_tensor, 8000).item()
        else:
            vad_event: Dict[str, float] = self.vad_iterator(audio_tensor, return_seconds=True)
            speech_prob = self.vad_model(audio_tensor, 8000).item()
        return speech_prob, vad_event

    def reset_states(self):
        self.vad_iterator.reset_states()

class VADIterator:
    def __init__(
        self,
        model,
        threshold: float = 0.5,
        sampling_rate: int = 8000,
        min_silence_duration_ms: int = 300,
        speech_pad_ms: int = 30,
        max_speech_duration_ms: int = 20000,
    ):
        self.model = model
        self.threshold = threshold
        self.sampling_rate = sampling_rate

        if sampling_rate not in [8000, 16000]:
            raise ValueError("VADIterator does not support sampling rates other than [8000, 16000]")

        self.min_silence_samples = sampling_rate * min_silence_duration_ms / 1000
        self.speech_pad_samples = sampling_rate * speech_pad_ms / 1000
        self.max_speech_samples = sampling_rate * max_speech_duration_ms / 1000
        self.reset_states()

    def reset_states(self):
        self.model.reset_states()
        self.triggered = False
        self.temp_end = 0
        self.current_sample = 0
        self.speech_start = 0

    def __call__(self, x, return_seconds=False):
        if not torch.is_tensor(x):
            try:
                x = torch.Tensor(x)
            except:
                raise TypeError("Audio cannot be casted to tensor. Cast it manually")

        window_size_samples = len(x[0]) if x.dim() == 2 else len(x)
        self.current_sample += window_size_samples

        speech_prob = self.model(x, self.sampling_rate).item()

        if (speech_prob >= self.threshold) and self.temp_end:
            self.temp_end = 0

        if (speech_prob >= self.threshold) and not self.triggered:
            self.triggered = True
            self.speech_start = self.current_sample - self.speech_pad_samples - window_size_samples
            return {
                "start": int(self.speech_start)
                if not return_seconds
                else round(self.speech_start / self.sampling_rate, 1)
            }

        if self.triggered:
            speech_duration = self.current_sample - self.speech_start
            if speech_duration >= self.max_speech_samples:
                speech_end = self.current_sample
                self.triggered = False
                self.temp_end = 0
                return {"end": int(speech_end) if not return_seconds else round(speech_end / self.sampling_rate, 1)}

        if (speech_prob < self.threshold - 0.15) and self.triggered:
            if not self.temp_end:
                self.temp_end = self.current_sample
            if self.current_sample - self.temp_end < self.min_silence_samples:
                return None
            else:
                speech_end = self.temp_end + self.speech_pad_samples - window_size_samples
                self.temp_end = 0
                self.triggered = False
                return {"end": int(speech_end) if not return_seconds else round(speech_end / self.sampling_rate, 1)}

        return None

2. Bad case (VADIterator stays the same)

# Triggered when server is up
async def init_app():
    app = web.Application(middlewares=[error_middleware])
    app.add_routes([web.post("/stt/pipeline/api/v1/stream", stream)])

    vad_model, VADIterator = torch.hub.load(source="local", repo_or_dir=VAD_MODEL_PATH, model="silero_vad", onnx=True)
    app['vad_model'] = vad_model
    return app

# This is main function of server
async def stream(request):
        vad_model = request.app["vad_model"]
        audio_pipeline:AudioPipeline = AudioPipeline(vad_model=vad_model, session_id=session_id)
        
        await response.prepare(request)
        await stt_manager.process_request(request, response)

# VADProcessor injected to AudioPipeline
class AudioPipeline:
    def __init__(self, engine, vad_model, session_id):
        self.engine = engine
        self.audio_processor = AudioProcessor()
        self.audio_buffer_processor = AudioBufferProcessor()
        self.vad_processor = VADProcessor(vad_model)

# vad_model directly injected to VADProcessor
class VADProcessor:
    def __init__(self,vad_model):
        self.vad_model=vad_model
        self.vad_iterator:VADIterator = VADIterator(vad_model, sampling_rate=8000,min_silence_duration_ms=MIN_SILENCE_DURATION_MS ,max_speech_duration_ms=MAX_SPEECH_DURATION_MS)
        pass

    def detect_speech(self, audio_chunk: bytes) -> Dict[str, float]:
        audio_tensor: torch.tensor = torch.frombuffer(audio_chunk, dtype=torch.int16).float() / 32768.0
        audio_tensor: torch.tensor = audio_tensor.unsqueeze(0)
        audio_samples: int = audio_tensor.shape[1]
        if audio_samples < MINIMUM_SAMPLES_FOR_VAD:
            padding_length = MIN_SILENCE_SAMPLES - audio_samples
            padded_audio_tensor = F.pad(audio_tensor, (0, padding_length), "constant", 0)
            vad_event: Dict[str, float] = self.vad_iterator(padded_audio_tensor, return_seconds=True)
            speech_prob: float = self.vad_model(padded_audio_tensor, 8000).item()
        else:
            vad_event: Dict[str, float] = self.vad_iterator(audio_tensor, return_seconds=True)
            speech_prob = self.vad_model(audio_tensor, 8000).item()
        return speech_prob, vad_event

    def reset_states(self):
        self.vad_iterator.reset_states()

[snakers4]

Why this happen? Does vad model has state, also?

Yes, it does, the VAD considers previous chunks.

State reset in ONNX wrapper, for example:

silero-vad/src/silero_vad/utils_vad.py

Lines 46 to 50 in 102e6d0

	def reset_states(self, batch_size=1):
	self._state = torch.zeros((2, batch_size, 128)).float()
	self._context = torch.zeros(0)
	self._last_sr = 0
	self._last_batch_size = 0

When I initialize model at every request, vad process works properly.

The state should be reset each time the audio stream changes or ends.

Also for your use-case, you may find this discussion thread useful - #570 (comment)

[lionsheep0724]

Thank you for quick reply. Could you explain me more? I don't understand your first comment.

the only really correct way is to have a separate process / worker and to pass state to this worker back-and-forth, like shown in ONNX wrapper.

[snakers4]

The ONNX wrapper manually resets the state and manually updates it explicitly in python code, unlike the JIT model, where it is handled internally.

So if you want to build some RPC service using VAD, you can "make" your workers kind of stateless by using the ONNX model and passing the model state back to your publisher from your worker.

silero-vad/src/silero_vad/utils_vad.py

Lines 52 to 87 in 102e6d0

	def __call__(self, x, sr: int):
	x, sr = self._validate_input(x, sr)
	num_samples = 512 if sr == 16000 else 256
	if x.shape[-1] != num_samples:
	raise ValueError(f"Provided number of samples is {x.shape[-1]} (Supported values: 256 for 8000 sample rate, 512 for 16000)")
	batch_size = x.shape[0]
	context_size = 64 if sr == 16000 else 32
	if not self._last_batch_size:
	self.reset_states(batch_size)
	if (self._last_sr) and (self._last_sr != sr):
	self.reset_states(batch_size)
	if (self._last_batch_size) and (self._last_batch_size != batch_size):
	self.reset_states(batch_size)
	if not len(self._context):
	self._context = torch.zeros(batch_size, context_size)
	x = torch.cat([self._context, x], dim=1)
	if sr in [8000, 16000]:
	ort_inputs = {'input': x.numpy(), 'state': self._state.numpy(), 'sr': np.array(sr, dtype='int64')}
	ort_outs = self.session.run(None, ort_inputs)
	out, state = ort_outs
	self._state = torch.from_numpy(state)
	else:
	raise ValueError()
	self._context = x[..., -context_size:]
	self._last_sr = sr
	self._last_batch_size = batch_size
	out = torch.from_numpy(out)
	return out

[snakers4]

Publisher is some service that "demands" some job to be done from a worker / consumer

[lionsheep0724]

When using your methodology, how do I mange concurrency? Do I need to adjust message queue system, like rabbit mq or kafka, to record states of multiple sessions?

[snakers4]

I believe kafka is about large regular data pipelines. But the answer to your question lies in the following aspects:

• How much load there will be;
• Is production grade quality required or it is just a one-off data processing job;

I would order the possible known solutions as follows:

• Just launching several local processes for one time job or writing a batch script (batching is possible, but not recommended by us)
• Python's built in ProcessPoolExecutor (with custom initialization procedure) or similar as quick and dirty local solution
• celery or rabbit-mq with RPC as a more production ready solution, dockerized of course, with scaling controls and compute isolation (ideally each VAD should have a separate CPU thread)
• Maybe FastAPI can be made to load 1 instance of each model propely (it has several processes + async and / or threads, so tread carefully)
• gRPC (not for the faint-hearted)