From f8bc0897826f97df951bbe28a8c8993c7535d045 Mon Sep 17 00:00:00 2001 From: James Hills <70035505+jhills20@users.noreply.github.com> Date: Tue, 5 Aug 2025 21:14:11 -0400 Subject: [PATCH] fix: avoid running duplicate realtime guardrails --- docs/ja/realtime/guide.md | 18 ++++++++++ docs/realtime/guide.md | 18 ++++++++++ src/agents/realtime/agent.py | 6 ++++ src/agents/realtime/session.py | 12 ++++++- tests/realtime/test_session.py | 62 ++++++++++++++++++++++++++++++++++ 5 files changed, 115 insertions(+), 1 deletion(-) diff --git a/docs/ja/realtime/guide.md b/docs/ja/realtime/guide.md index 70185e05c..93abfcf5c 100644 --- a/docs/ja/realtime/guide.md +++ b/docs/ja/realtime/guide.md @@ -134,6 +134,24 @@ main_agent = RealtimeAgent( Realtime エージェントでは出力ガードレールのみがサポートされます。パフォーマンス低下を防ぐためにデバウンスされ、リアルタイム生成中に毎単語ではなく定期的に実行されます。デフォルトのデバウンス長は 100 文字で、設定可能です。 +ガードレールは `RealtimeAgent` に直接設定することも、セッションの `run_config` で指定することもできます。両方のリストが同時に実行されます。 + +```python +from agents.guardrail import GuardrailFunctionOutput, OutputGuardrail + +def sensitive_data_check(context, agent, output): + return GuardrailFunctionOutput( + tripwire_triggered="password" in output, + output_info=None, + ) + +agent = RealtimeAgent( + name="Assistant", + instructions="...", + output_guardrails=[OutputGuardrail(guardrail_function=sensitive_data_check)], +) +``` + ガードレールがトリガーされると `guardrail_tripped` イベントが生成され、エージェントの現在の応答を中断できます。デバウンス動作により安全性とリアルタイム性能のバランスを取ります。テキストエージェントとは異なり、realtime エージェントはガードレールがトリップしても Exception を発生させません。 ## オーディオ処理 diff --git a/docs/realtime/guide.md b/docs/realtime/guide.md index cefa5d688..6d0648f60 100644 --- a/docs/realtime/guide.md +++ b/docs/realtime/guide.md @@ -130,6 +130,24 @@ For complete event details, see [`RealtimeSessionEvent`][agents.realtime.events. Only output guardrails are supported for realtime agents. These guardrails are debounced and run periodically (not on every word) to avoid performance issues during real-time generation. The default debounce length is 100 characters, but this is configurable. +Guardrails can be attached directly to a `RealtimeAgent` or provided via the session's `run_config`. Guardrails from both sources run together. + +```python +from agents.guardrail import GuardrailFunctionOutput, OutputGuardrail + +def sensitive_data_check(context, agent, output): + return GuardrailFunctionOutput( + tripwire_triggered="password" in output, + output_info=None, + ) + +agent = RealtimeAgent( + name="Assistant", + instructions="...", + output_guardrails=[OutputGuardrail(guardrail_function=sensitive_data_check)], +) +``` + When a guardrail is triggered, it generates a `guardrail_tripped` event and can interrupt the agent's current response. The debounce behavior helps balance safety with real-time performance requirements. Unlike text agents, realtime agents do **not** raise an Exception when guardrails are tripped. ## Audio processing diff --git a/src/agents/realtime/agent.py b/src/agents/realtime/agent.py index 30e80a95b..29483ac27 100644 --- a/src/agents/realtime/agent.py +++ b/src/agents/realtime/agent.py @@ -7,6 +7,7 @@ from typing import Any, Callable, Generic, cast from ..agent import AgentBase +from ..guardrail import OutputGuardrail from ..handoffs import Handoff from ..lifecycle import AgentHooksBase, RunHooksBase from ..logger import logger @@ -62,6 +63,11 @@ class RealtimeAgent(AgentBase, Generic[TContext]): modularity. """ + output_guardrails: list[OutputGuardrail[TContext]] = field(default_factory=list) + """A list of checks that run on the final output of the agent, after generating a response. + Runs only if the agent produces a final output. + """ + hooks: RealtimeAgentHooks | None = None """A class that receives callbacks on various lifecycle events for this agent. """ diff --git a/src/agents/realtime/session.py b/src/agents/realtime/session.py index e3ad21c5b..3e6747910 100644 --- a/src/agents/realtime/session.py +++ b/src/agents/realtime/session.py @@ -443,7 +443,17 @@ def _get_new_history( async def _run_output_guardrails(self, text: str) -> bool: """Run output guardrails on the given text. Returns True if any guardrail was triggered.""" - output_guardrails = self._run_config.get("output_guardrails", []) + combined_guardrails = self._current_agent.output_guardrails + self._run_config.get( + "output_guardrails", [] + ) + seen_ids: set[int] = set() + output_guardrails = [] + for guardrail in combined_guardrails: + guardrail_id = id(guardrail) + if guardrail_id not in seen_ids: + output_guardrails.append(guardrail) + seen_ids.add(guardrail_id) + if not output_guardrails or self._interrupted_by_guardrail: return False diff --git a/tests/realtime/test_session.py b/tests/realtime/test_session.py index 921af744b..e5d2d5d45 100644 --- a/tests/realtime/test_session.py +++ b/tests/realtime/test_session.py @@ -104,6 +104,7 @@ def mock_agent(): agent.get_all_tools = AsyncMock(return_value=[]) type(agent).handoffs = PropertyMock(return_value=[]) + type(agent).output_guardrails = PropertyMock(return_value=[]) return agent @@ -1063,6 +1064,37 @@ async def test_transcript_delta_triggers_guardrail_at_threshold( assert len(guardrail_events) == 1 assert guardrail_events[0].message == "this is more than ten characters" + @pytest.mark.asyncio + async def test_agent_and_run_config_guardrails_not_run_twice(self, mock_model): + """Guardrails shared by agent and run config should execute once.""" + + call_count = 0 + + def guardrail_func(context, agent, output): + nonlocal call_count + call_count += 1 + return GuardrailFunctionOutput(output_info={}, tripwire_triggered=False) + + shared_guardrail = OutputGuardrail( + guardrail_function=guardrail_func, name="shared_guardrail" + ) + + agent = RealtimeAgent(name="agent", output_guardrails=[shared_guardrail]) + run_config: RealtimeRunConfig = { + "output_guardrails": [shared_guardrail], + "guardrails_settings": {"debounce_text_length": 5}, + } + + session = RealtimeSession(mock_model, agent, None, run_config=run_config) + + await session.on_event( + RealtimeModelTranscriptDeltaEvent(item_id="item_1", delta="hello", response_id="resp_1") + ) + + await self._wait_for_guardrail_tasks(session) + + assert call_count == 1 + @pytest.mark.asyncio async def test_transcript_delta_multiple_thresholds_same_item( self, mock_model, mock_agent, triggered_guardrail @@ -1210,6 +1242,36 @@ def guardrail_func(context, agent, output): assert len(guardrail_events) == 1 assert len(guardrail_events[0].guardrail_results) == 2 + @pytest.mark.asyncio + async def test_agent_output_guardrails_triggered(self, mock_model, triggered_guardrail): + """Test that guardrails defined on the agent are executed.""" + agent = RealtimeAgent(name="agent", output_guardrails=[triggered_guardrail]) + run_config: RealtimeRunConfig = { + "guardrails_settings": {"debounce_text_length": 10}, + } + + session = RealtimeSession(mock_model, agent, None, run_config=run_config) + + transcript_event = RealtimeModelTranscriptDeltaEvent( + item_id="item_1", delta="this is more than ten characters", response_id="resp_1" + ) + + await session.on_event(transcript_event) + await self._wait_for_guardrail_tasks(session) + + assert session._interrupted_by_guardrail is True + assert mock_model.interrupts_called == 1 + assert len(mock_model.sent_messages) == 1 + assert "triggered_guardrail" in mock_model.sent_messages[0] + + events = [] + while not session._event_queue.empty(): + events.append(await session._event_queue.get()) + + guardrail_events = [e for e in events if isinstance(e, RealtimeGuardrailTripped)] + assert len(guardrail_events) == 1 + assert guardrail_events[0].message == "this is more than ten characters" + class TestModelSettingsIntegration: """Test suite for model settings integration in RealtimeSession."""