ChatManager

answer_from_text

from __future__ import annotations

import logging
import asyncio
from livekit import rtc
from livekit.agents import AutoSubscribe, JobContext, WorkerOptions, cli, llm
from livekit.agents.multimodal import MultimodalAgent
from livekit.plugins import openai
from dotenv import load_dotenv
from datetime import datetime

load_dotenv()

logger = logging.getLogger("myagent")
logger.setLevel(<http://logging.INFO|logging.INFO>)

async def entrypoint(ctx: JobContext):
    <http://logger.info|logger.info>("starting entrypoint")

    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    participant = await ctx.wait_for_participant()

    model = openai.realtime.RealtimeModel(
        instructions="You are a helpful assistant and you love kittens",
        voice="shimmer",
        temperature=0.8,
        modalities=["audio", "text"],
    )
    assistant = MultimodalAgent(model=model)
    assistant.start(ctx.room)

    <http://logger.info|logger.info>("starting agent")

    session = model.sessions[0]
    session.conversation.item.create(
      llm.ChatMessage(
        role="assistant",
        content="Please begin the interaction with the user in a manner consistent with your instructions.",
      )
    )
    session.response.create()

    # Initialize ChatManager for text input
    chat = rtc.ChatManager(ctx.room)
    <http://logger.info|logger.info>("ChatManager initialized")

    # Listen for incoming chat messages
    @chat.on("message_received")
    def on_chat_received(msg: rtc.ChatMessage):
        <http://logger.info|logger.info>("message_received event fired")
        <http://logger.info|logger.info>(f"Received message object: {msg}")
        if msg.message:
            <http://logger.info|logger.info>(f"Received chat message: {msg.message}")
            asyncio.create_task(answer_from_text(msg.message))

    async def answer_from_text(txt: str):
        chat_ctx = assistant.chat_ctx_copy()
        chat_ctx.append(role="user", text=txt)
        await assistant.set_chat_ctx(chat_ctx)
        session.response.create()

    # Handle user speech committed event
    @assistant.on("user_speech_committed")
    def on_user_speech_committed(msg: llm.ChatMessage):
        if isinstance(msg.content, list):
            msg.content = "\n".join(
                "[image]" if isinstance(x, llm.ChatImage) else x for x in msg.content
            )
        <http://logger.info|logger.info>(f"[{datetime.now()}] USER:\n{msg.content}\n\n")

    # Handle agent speech committed event
    @assistant.on("agent_speech_committed")
    def on_agent_speech_committed(msg: llm.ChatMessage):
        <http://logger.info|logger.info>(f"Agent responded with: {msg.content}")

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

MultimodalAgent

from __future__ import annotations

import logging
import asyncio
from livekit import rtc
from livekit.agents import AutoSubscribe, JobContext, WorkerOptions, cli, llm
from livekit.agents.multimodal import MultimodalAgent
from livekit.plugins import openai
from dotenv import load_dotenv
from datetime import datetime

load_dotenv()

logger = logging.getLogger("myagent")
logger.setLevel(<http://logging.INFO|logging.INFO>)

async def entrypoint(ctx: JobContext):
    <http://logger.info|logger.info>("starting entrypoint")

    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    participant = await ctx.wait_for_participant()

    model = openai.realtime.RealtimeModel(
        instructions="You are a helpful assistant and you want to help people",
        voice="shimmer",
        temperature=0.8,
        modalities=["audio", "text"],
    )
    assistant = MultimodalAgent(model=model)
    assistant.start(ctx.room)

    <http://logger.info|logger.info>("starting agent")

    session = model.sessions[0]
    session.conversation.item.create(
      llm.ChatMessage(
        role="assistant",
        content="Please begin the interaction with the user in a manner consistent with your instructions.",
      )
    )
    session.response.create()

    # Initialize ChatManager for text input
    chat = rtc.ChatManager(ctx.room)
    <http://logger.info|logger.info>("ChatManager initialized")

    # Listen for incoming chat messages
    @chat.on("message_received")
    def on_chat_received(msg: rtc.ChatMessage):
        <http://logger.info|logger.info>("message_received event fired")
        <http://logger.info|logger.info>(f"Received message object: {msg}")
        if msg.message:
            <http://logger.info|logger.info>(f"Received chat message: {msg.message}")
            asyncio.create_task(answer_from_text(msg.message))

    async def answer_from_text(txt: str):
        chat_ctx = assistant.chat_ctx_copy()
        chat_ctx.append(role="user", text=txt)
        await assistant.set_chat_ctx(chat_ctx)
        session.response.create()

    # Handle user speech committed event
    @assistant.on("user_speech_committed")
    def on_user_speech_committed(msg: llm.ChatMessage):
        if isinstance(msg.content, list):
            msg.content = "\n".join(
                "[image]" if isinstance(x, llm.ChatImage) else x for x in msg.content
            )
        <http://logger.info|logger.info>(f"[{datetime.now()}] USER:\n{msg.content}\n\n")

    # Handle agent speech committed event
    @assistant.on("agent_speech_committed")
    def on_agent_speech_committed(msg: llm.ChatMessage):
        <http://logger.info|logger.info>(f"Agent responded with: {msg.content}")

    await asyncio.sleep(1)
    await assistant.say("Hello, I am here to assist you in completing a Geriatric Depression Scale screening. The GDS-15 is a series of 15 questions that will help assess your emotional well-being. For each question, please answer with either 'yes' or 'no'. If you're unsure, please feel free to ask me for clarification. Let's get started!", allow_interruptions=True)

if __name__ == "__main__":

cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))