dnnai/asr_agent/agent.py

"""
LiveKit ASR Agent — publishes user speech transcriptions to the room.
"""

from __future__ import annotations

import logging

from livekit import agents, rtc
from livekit.agents import Agent, AgentSession, JobContext, cli, inference
from livekit.agents.inference import TurnDetector
from livekit.plugins import noise_cancellation, silero

from qwen_stt import QwenSTT

logger = logging.getLogger("asr-agent")

LIVEKIT_URL = "ws://localhost:7880"
LIVEKIT_API_KEY = "devkey"
LIVEKIT_API_SECRET = "secretsecretsecretsecretsecret12"


class ASRAgent(Agent):
    """
    Speech-to-text agent. Transcribes user speech and sends results
    back to the room via data messages / agent utterances.

    No LLM or TTS — pure transcription for now.
    """

    def __init__(self) -> None:
        super().__init__(
            instructions="将用户语音实时转写为文字。"
        )

    async def on_enter(self):
        # Send a welcome message to the room chat
        await self.session.send_utterance(
            instructions="ASR 转写已就绪，请开始说话。"
        )


server = agents.AgentServer()


@server.rtc_session()
async def entrypoint(ctx: JobContext):
    logger.info(f"new session: room={ctx.room.name}")

    session = AgentSession(
        stt=QwenSTT(
            model_id="Qwen/Qwen3-ASR-0.6B",
            silence_repeats=3,
            silence_seconds=3.0,
        ),
        vad=inference.VAD(),
        turn_detection=TurnDetector(),
    )

    await session.start(
        agent=ASRAgent(),
        room=ctx.room,
        room_input_options=agents.RoomInputOptions(
            noise_cancellation=noise_cancellation.BVC(),
        ),
    )


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