feat: add Pulse STT support for smallest.ai pulse (streaming + pre-recorded)

[mahimairaja]

What does this PR does?

Adds Speech-to-Text (STT) support to the livekit-plugins-smallestai plugin using Smallest AI's Pulse STT API. The existing plugin only supported TTS, this PR brings it to parity with plugins like Deepgram, ElevenLabs, and Soniox that offer both TTS and STT.

Closes #4856

Summary of Changes

New: STT class (stt.py)

• Pre-recorded transcription via HTTP POST (/api/v1/pulse/get_text)
• Real-time streaming via WebSocket (wss://waves-api.smallest.ai/api/v1/pulse/get_text)
• ~64ms TTFB streaming, word-level timestamps, speaker diarization
• 32+ languages with auto-detection (language="multi")
• Capabilities: streaming=True, interim_results=True

New: SpeechStream class (stt.py)

• WebSocket-based streaming with concurrent send/recv/keepalive tasks
• Audio chunking via AudioByteStream (~4096 byte chunks per Smallest AI docs)
• Full speech event lifecycle: START_OF_SPEECH → INTERIM_TRANSCRIPT / FINAL_TRANSCRIPT → END_OF_SPEECH
• Graceful shutdown with {"type": "end"} signaling

Usage

from livekit.plugins import smallestai

# Pre-recorded
stt = smallestai.STT(language="en")

# Streaming (used in AgentSession)
session = AgentSession(
    stt=smallestai.STT(language="en"),
    llm=...,
    tts=smallestai.TTS(),
)

Configuration via SMALLEST_API_KEY environment variable (same key used for TTS).

Testing

• Verified pre-recorded transcription with WAV audio files
• Verified real-time streaming with live microphone input via LiveKit Agents Playground
• Tested interim + final transcript emission and speech event lifecycle
• Tested with language="en" and language="multi" (auto-detection)
• Ran ruff format and check

❯ uv run ruff check .
All checks passed!

❯ uv run ruff format .
629 files left unchanged

• Ran type checking

❯ uv pip install pip && uv run mypy --install-types --non-interactive \
    -p livekit.agents \
    -p livekit.plugins.smallestai
Audited 1 package in 5ms
Success: no issues found in 169 source files

API Reference

• Pre-recorded: https://waves-docs.smallest.ai/v4.0.0/content/speech-to-text/pre-recorded/quickstart
• Streaming: https://waves-docs.smallest.ai/v4.0.0/content/speech-to-text/streaming/quickstart

[CLAassistant]

All committers have signed the CLA.

[mahimairaja]

Tested prerecorded:

import asyncio
from pathlib import Path

import aiohttp
from dotenv import load_dotenv

from livekit.agents import utils
from livekit.plugins import smallestai

load_dotenv()


async def main():
    wav = Path(__file__).resolve().parent / "sample.wav"

    async with aiohttp.ClientSession() as session:
        stt = smallestai.STT(language="en", http_session=session)
        frames = [
            f
            async for f in utils.audio.audio_frames_from_file(
                str(wav), sample_rate=16000, num_channels=1
            )
        ]
        event = await stt.recognize(frames)

    print(event.alternatives[0].text if event.alternatives else "")


if __name__ == "__main__":
    asyncio.run(main())

[mahimairaja]

Testing streaming:

from dotenv import load_dotenv

from livekit import agents
from livekit.agents import Agent, AgentServer, AgentSession, room_io
from livekit.plugins import silero
from livekit.plugins.openai.llm import LLM
from livekit.plugins.smallestai.stt import STT
from livekit.plugins.smallestai.tts import TTS
from livekit.plugins.turn_detector.english import EnglishModel

load_dotenv()


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a helpful voice AI assistant.""",
        )


server = AgentServer()


@server.rtc_session(agent_name="my-agent")
async def my_agent(ctx: agents.JobContext):
    session = AgentSession(
        stt=STT(),
        llm=LLM(model="gpt-4.1-mini"),
        tts=TTS(),
        vad=silero.VAD.load(),
        turn_detection=EnglishModel(),
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_options=room_io.RoomOptions(),
    )

    await session.generate_reply(instructions="Greet the user and offer your assistance.")


if __name__ == "__main__":
    agents.cli.run_app(server)

[mahimairaja]

After conversations with @ harshitajain165 from smallest.ai, I came to know that few more steps needed for streaming support from the smallest server. for now I am moving this PR to draft.