feat(guardrail): add conversation monitoring for AgentSession

examples/voice_agents/guardrail_agent.py

@@ -0,0 +1,114 @@
+"""
+Guardrail example: Restaurant reservation with safety monitoring.
+
+WHY GUARDRAILS?
+The main agent prompt can be huge (10k+ tokens) with complex business logic.
+Guardrails are small, focused monitors that run in parallel - they don't add
+latency to the main conversation but catch critical issues.
+
+Here: main agent handles full reservation flow, guardrail watches for ONE
+safety-critical thing: dietary restrictions/allergies.
+
+Run: python examples/voice_agents/guardrail_agent.py console
+"""
+
+import logging
+
+from dotenv import load_dotenv
+
+from livekit.agents import Agent, AgentServer, AgentSession, JobContext, cli, room_io
+from livekit.agents.voice import Guardrail
+from livekit.plugins import silero
+
+logger = logging.getLogger("guardrail-agent")
+
+load_dotenv()
+
+
+class RestaurantAgent(Agent):
+    def __init__(self) -> None:
+        super().__init__(
+            instructions="""You are the host at "The Golden Fork" fine dining restaurant.
+
+RESERVATION FLOW:
+1. Greet warmly, ask how you can help
+2. Get party size and preferred date/time
+3. Check availability (always say "let me check... yes we have that available")
+4. Get name and contact number for confirmation
+5. Mention dress code (smart casual, no sportswear)
+6. Explain our 15-minute late policy
+7. Offer to note any special requests
+8. Confirm all details and thank them
+
+UPSELLING (be subtle):
+- Parties of 6+ → suggest private dining room ($50 extra)
+- Weekend bookings → mention live jazz on Saturday nights
+- If they mention celebration → offer complimentary dessert
+
+ALSO HANDLE:
+- Cancellations (need 24hr notice)
+- Modifications to existing bookings
+- Questions about menu, parking, accessibility
+- Directions to the restaurant
+
+Keep responses SHORT (1-2 sentences). Sound professional but warm.
+Don't use emojis or markdown.""",
+        )
+
+    async def on_enter(self):
+        self.session.generate_reply(
+            instructions="Greet the caller warmly and ask how you can help today."
+        )
+
+
+server = AgentServer()
+
+
+def prewarm(proc):
+    proc.userdata["vad"] = silero.VAD.load()
+
+
+server.setup_fnc = prewarm
+
+
+@server.rtc_session()
+async def entrypoint(ctx: JobContext):
+    session = AgentSession(
+        stt="deepgram/nova-3",
+        llm="openai/gpt-4o-mini",
+        tts="cartesia/sonic-2",
+        vad=ctx.proc.userdata["vad"],
+        guardrails=[
+            Guardrail(
+                name="safety",
+                instructions="""You monitor restaurant reservations for FOOD SAFETY.
+
+CRITICAL: Before confirming ANY new reservation, agent MUST ask about
+dietary restrictions and allergies. This is a safety requirement.
+
+Intervene if:
+- Agent is about to confirm without asking about allergies/dietary needs
+- Agent collected name, date, time, party size but skipped dietary question
+
+Do NOT intervene if:
+- Customer already mentioned dietary needs ("I'm vegetarian", "nut allergy")
+- Agent already asked about restrictions
+- Customer is just asking questions, not making a reservation
+- Customer is canceling or modifying existing booking""",
+                llm="openai/gpt-4o-mini",
+                eval_interval=3,
+                max_interventions=2,
+                cooldown=15.0,
+            ),
+        ],
+    )
+
+    await session.start(
+        agent=RestaurantAgent(),
+        room=ctx.room,
+        room_options=room_io.RoomOptions(),
+    )
+
+
+if __name__ == "__main__":
+    cli.run_app(server)

livekit-agents/livekit/agents/voice/__init__.py

@@ -16,6 +16,7 @@
     UserInputTranscribedEvent,
     UserStateChangedEvent,
 )
+from .guardrail import Guardrail
 from .room_io import (
     _ParticipantAudioOutput,
     _ParticipantStreamTranscriptionOutput,
@@ -30,6 +31,7 @@
     "Agent",
     "ModelSettings",
     "AgentTask",
+    "Guardrail",
     "SpeechHandle",
     "RunContext",
     "UserInputTranscribedEvent",

livekit-agents/livekit/agents/voice/agent_session.py

@@ -53,6 +53,7 @@
     UserState,
     UserStateChangedEvent,
 )
+from .guardrail import Guardrail, _GuardrailRunner
 from .ivr import IVRActivity
 from .recorder_io import RecorderIO
 from .run_result import RunResult
@@ -160,6 +161,7 @@ def __init__(
         tts_text_transforms: NotGivenOr[Sequence[TextTransforms] | None] = NOT_GIVEN,
         preemptive_generation: bool = False,
         ivr_detection: bool = False,
+        guardrails: list[Guardrail] | None = None,
         conn_options: NotGivenOr[SessionConnectOptions] = NOT_GIVEN,
         loop: asyncio.AbstractEventLoop | None = None,
         # deprecated
@@ -246,6 +248,10 @@ def __init__(
                 Defaults to ``False``.
             ivr_detection (bool): Whether to detect if the agent is interacting with an IVR system.
                 Default ``False``.
+            guardrails (list[Guardrail], optional): List of conversation guardrails.
+                Each guardrail runs independently, monitoring conversations and injecting
+                advice to the agent when specific situations are detected. Multiple
+                guardrails can watch for different things (compliance, sentiment, etc.).
             conn_options (SessionConnectOptions, optional): Connection options for
                 stt, llm, and tts.
             loop (asyncio.AbstractEventLoop, optional): Event loop to bind the
@@ -360,6 +366,10 @@ def __init__(
         # ivr activity
         self._ivr_activity: IVRActivity | None = None
 
+        # guardrails
+        self._guardrails = guardrails or []
+        self._guardrail_runners: list[_GuardrailRunner] = []
+
     def emit(self, event: EventTypes, arg: AgentEvent) -> None:  # type: ignore
         self._recorded_events.append(arg)
         super().emit(event, arg)
@@ -607,6 +617,12 @@ async def start(
                         asyncio.create_task(self._ivr_activity.start(), name="_ivr_activity_start")
                     )
 
+            for guardrail in self._guardrails:
+                runner = _GuardrailRunner(guardrail, self)
+                runner.start()
+                self._guardrail_runners.append(runner)
+
+            if job_ctx:
                 current_span.set_attribute(trace_types.ATTR_ROOM_NAME, job_ctx.room.name)
                 current_span.set_attribute(trace_types.ATTR_JOB_ID, job_ctx.job.id)
                 current_span.set_attribute(trace_types.ATTR_AGENT_NAME, job_ctx.job.agent_name)
@@ -761,6 +777,10 @@ async def _aclose_impl(
             self._closing = True
             self._cancel_user_away_timer()
 
+            for runner in self._guardrail_runners:
+                runner.stop()
+            self._guardrail_runners.clear()
+
             if self._activity is not None:
                 if not drain:
                     try: