Add more complex agent

Author: bcherry

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
 ```

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))