session.say()

participant_identity

RoomInputOptions.participant_kinds

from livekit import rtc
from livekit.agents import cli, JobContext, WorkerOptions
from livekit.agents.voice import VoiceAssistant
from livekit.agents.voice.room_io import RoomIO, RoomInputOptions
import asyncio

class SelectorAgent:
    """
    Listens to all participants and responds only to the participant
    matching custom criteria (e.g., identity or keyword).
    """

    def __init__(self, session: VoiceAssistant = None):
        self.session = session

    async def decide_whether_to_respond(self, text: str, participant_identity: str) -> bool:
        # Example decision logic:
        # - respond if identity equals "target_user"
        # - or if text contains "hey bot"
        if participant_identity == "target_user":
            return True
        if "hey bot" in (text or "").lower():
            return True
        return False

    async def on_user_transcribed(self, evt):
        """
        Handler for user transcription events. The event object typically
        contains text and participant metadata — adjust field names to match your SDK version.
        """
        user_text = getattr(evt, "text", None) or getattr(evt, "transcript", None) or ""
        participant_identity = getattr(evt, "participant_identity", None) or "unknown"

        should_respond = await self.decide_whether_to_respond(user_text, participant_identity)
        if not should_respond:
            # ignore
            return

        # Generate or send a reply targeted to the chosen participant.
        reply_text = f"Hello {participant_identity}, I heard you: \"{user_text}\""
        # Publish TTS reply into the room so everyone can hear; session.say() publishes the agent's audio
        await self.session.say(reply_text)

async def entrypoint(ctx: JobContext):
    # Connect and auto-subscribe to audio so the agent hears participants
    await ctx.connect(auto_subscribe=rtc.AutoSubscribe.AUDIO_ONLY)

    # Listen to all participant kinds (optional; include AGENT if agents should hear other agents)
    room_input_options = RoomInputOptions(
        participant_kinds=[
            rtc.ParticipantKind.PARTICIPANT_KIND_STANDARD,
            rtc.ParticipantKind.PARTICIPANT_KIND_SIP,
            rtc.ParticipantKind.PARTICIPANT_KIND_AGENT,
        ]
    )

    # Create RoomIO without specifying participant_identity (listens to all)
    room_io = RoomIO(ctx.room, input_options=room_input_options)

    # Create assistant (use your STT/LLM/TTS implementations)
    assistant = VoiceAssistant(
        vad=await rtc.VAD.load(),
        stt=await rtc.STT.create(),
        llm=await rtc.LLM.create(),
        tts=await rtc.TTS.create(),
    )

    # Create session bound to room_io
    session = assistant.create_session(room=ctx.room, room_io=room_io)
    await session.start()

    # Attach selector logic
    selector = SelectorAgent(session=session)

    # Hook into transcription events. Adjust event registration to match your SDK:
    # Many SDKs expose an event like `on_user_speech` or `on_transcription`.
    # Here we assume session.context.on("user_speech", handler) style; adapt if needed.
    try:
        session.context.on("user_speech", selector.on_user_transcribed)  # type: ignore
    except Exception:
        # Fallback: if the SDK exposes a different event name, adapt accordingly
        session.context.on("user_input_transcribed", selector.on_user_transcribed)  # type: ignore

    # Keep running
    await ctx.wait_for_shutdown()

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))