语音 AI Agent 实践

import asyncio

import numpy as np

import srtc

RATE = 16000


async def respond(uid: str, utterance: np.ndarray) -> bytes:
    """ASR → LLM → TTS，返回要播放的 PCM（16k 单声道）。替换成你的实现。"""
    text = await asr.transcribe(utterance, RATE)
    reply = await llm.chat(uid, text)
    return await tts.synthesize(reply, sample_rate=RATE)


async def main(token: str):
    async with await srtc.Channel.join(token, auto_subscribe_audio=True,
                                       audio_format=srtc.AudioFormat(RATE, 1)) as ch:
        speaker = await ch.publish_audio(desc="agent", audio_format=srtc.AudioFormat(RATE, 1))
        vad = MyVad()                      # 你的 VAD，按 uid 维护状态
        reply_task: asyncio.Task | None = None

        async def reply(uid, utterance):
            await speaker.write(await respond(uid, utterance))

        async for frame in ch.audio_frames():
            speaking, utterance = vad.feed(frame.uid, frame.to_numpy()[:, 0])

            # 用户开口时 agent 还在播：立即打断
            if speaking and speaker.buffered_seconds > 0:
                speaker.clear()
                if reply_task:
                    reply_task.cancel()

            # 一句话说完：异步生成回复，不阻塞收音
            if utterance is not None:
                reply_task = asyncio.create_task(reply(frame.uid, utterance))

class Router(srtc.ChannelHandler):
    def __init__(self):
        self.ch: srtc.Channel | None = None

    async def on_track_added(self, track: srtc.TrackInfo):
        if track.kind == srtc.TrackKind.AUDIO and track.uid.startswith("teacher"):
            await self.ch.subscribe_audio(track.uid, track.id)


router = Router()
ch = await srtc.Channel.join(token, handler=router)     # 不开 auto_subscribe_audio
router.ch = ch
for user in ch.users:                                   # 入会前就已发布的轨道也要订
    for track in user.stream_tracks:
        await router.on_track_added(track)