diff --git a/README.md b/README.md index 72bb32d..c03bf0d 100644 --- a/README.md +++ b/README.md @@ -39,17 +39,18 @@ You can also do this automatically using the LiveKit CLI: lk app env -w .env ``` -Run the agent: +Run the agent in console mode: ```console -uv run python src/agent.py dev +uv run python src/agent.py console ``` + This agent requires a frontend application to communicate with. Use a [starter app](https://docs.livekit.io/agents/start/frontend/#starter-apps), our hosted [Sandbox](https://cloud.livekit.io/projects/p_/sandbox) frontends, or the [LiveKit Agents Playground](https://agents-playground.livekit.io/). -Run tests +Run evals ```console -uv run pytest +uv run pytest evals ``` \ No newline at end of file diff --git a/src/agent.py b/src/agent.py index 75caf85..10bb207 100644 --- a/src/agent.py +++ b/src/agent.py @@ -1,40 +1,115 @@ +import logging + from dotenv import load_dotenv -from livekit import agents -from livekit.agents import AgentSession, Agent, RoomInputOptions -from livekit.plugins import openai, noise_cancellation, silero, deepgram, cartesia +from livekit.agents import ( + Agent, + AgentSession, + JobContext, + JobProcess, + RoomInputOptions, + RoomOutputOptions, + RunContext, + WorkerOptions, + cli, + metrics, +) +from livekit.agents.llm import function_tool +from livekit.agents.voice import MetricsCollectedEvent +from livekit.plugins import cartesia, deepgram, openai, silero from livekit.plugins.turn_detector.multilingual import MultilingualModel +from livekit.plugins import noise_cancellation + +logger = logging.getLogger("agent") load_dotenv() class Assistant(Agent): def __init__(self) -> None: - super().__init__(instructions="You are a helpful voice AI assistant.") + super().__init__( + instructions="Your name is Kelly. You would interact with users via voice." + "with that in mind keep your responses concise and to the point." + "You are curious and friendly, and have a sense of humor.", + ) + + async def on_enter(self): + # when the agent is added to the session, it'll generate a reply + # according to its instructions + self.session.generate_reply() + + # all functions annotated with @function_tool will be passed to the LLM when this + # agent is active + @function_tool + async def lookup_weather( + self, context: RunContext, location: str, latitude: str, longitude: str + ): + """Called when the user asks for weather related information. + Ensure the user's location (city or region) is provided. + When given a location, please estimate the latitude and longitude of the location and + do not ask the user for them. + + Args: + location: The location they are asking for + latitude: The latitude of the location, do not ask user for it + longitude: The longitude of the location, do not ask user for it + """ + + logger.info(f"Looking up weather for {location}") + + return "sunny with a temperature of 70 degrees." -async def entrypoint(ctx: agents.JobContext): +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + + +async def entrypoint(ctx: JobContext): + # each log entry will include these fields + ctx.log_context_fields = { + "room": ctx.room.name, + } + session = AgentSession( - stt=deepgram.STT(), + vad=ctx.proc.userdata["vad"], + # any combination of STT, LLM, TTS, or realtime API can be used llm=openai.LLM(model="gpt-4o-mini"), + stt=deepgram.STT(model="nova-3", language="multi"), tts=cartesia.TTS(), - vad=silero.VAD.load(), + # use LiveKit's turn detection model turn_detection=MultilingualModel(), ) + # log metrics as they are emitted, and total usage after session is over + usage_collector = metrics.UsageCollector() + + @session.on("metrics_collected") + def _on_metrics_collected(ev: MetricsCollectedEvent): + metrics.log_metrics(ev.metrics) + usage_collector.collect(ev.metrics) + + async def log_usage(): + summary = usage_collector.get_summary() + logger.info(f"Usage: {summary}") + + # shutdown callbacks are triggered when the session is over + ctx.add_shutdown_callback(log_usage) + await session.start( + agent=MyAgent(), room=ctx.room, - agent=Assistant(), room_input_options=RoomInputOptions( # LiveKit Cloud enhanced noise cancellation # - If self-hosting, omit this parameter # - For telephony applications, use `BVCTelephony` for best results noise_cancellation=noise_cancellation.BVC(), ), + room_output_options=RoomOutputOptions(transcription_enabled=True), ) + # join the room when agent is ready await ctx.connect() if __name__ == "__main__": - agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint)) + cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))