From 5d570b425078851f75150748ac96ee85c4e51ca9 Mon Sep 17 00:00:00 2001 From: Showmick Das Date: Mon, 31 Aug 2026 02:16:39 -0400 Subject: [PATCH 1/2] fix: add run_in_parallel support to OutputGuardrail for sequential execution --- src/agents/guardrail.py | 8 +++++ src/agents/run_internal/guardrails.py | 28 ++++++++++++--- tests/test_output_guardrail_parallel.py | 48 +++++++++++++++++++++++++ 3 files changed, 80 insertions(+), 4 deletions(-) create mode 100644 tests/test_output_guardrail_parallel.py diff --git a/src/agents/guardrail.py b/src/agents/guardrail.py index 07475c8183..5164d86993 100644 --- a/src/agents/guardrail.py +++ b/src/agents/guardrail.py @@ -156,6 +156,11 @@ class OutputGuardrail(Generic[TContext]): function's name. """ + run_in_parallel: bool = True + """Whether the guardrail runs concurrently with other guardrails (True, default) or before + subsequent guardrails (False). + """ + def get_name(self) -> str: if self.name: return self.name @@ -296,6 +301,7 @@ def output_guardrail( def output_guardrail( *, name: str | None = None, + run_in_parallel: bool = True, ) -> Callable[ [_OutputGuardrailFuncSync[TContext_co] | _OutputGuardrailFuncAsync[TContext_co]], OutputGuardrail[TContext_co], @@ -308,6 +314,7 @@ def output_guardrail( | None = None, *, name: str | None = None, + run_in_parallel: bool = True, ) -> ( OutputGuardrail[TContext_co] | Callable[ @@ -333,6 +340,7 @@ def decorator( guardrail_function=f, # Guardrail name defaults to function's name when not specified (None). name=name if name else f.__name__, + run_in_parallel=run_in_parallel, ) if func is not None: diff --git a/src/agents/run_internal/guardrails.py b/src/agents/run_internal/guardrails.py index dc09f0cd7a..6116ecb530 100644 --- a/src/agents/run_internal/guardrails.py +++ b/src/agents/run_internal/guardrails.py @@ -184,10 +184,8 @@ async def run_output_guardrails( if not guardrails: return [] - guardrail_tasks = [ - asyncio.create_task(run_single_output_guardrail(guardrail, agent, agent_output, context)) - for guardrail in guardrails - ] + sequential_guardrails = [g for g in guardrails if not g.run_in_parallel] + parallel_guardrails = [g for g in guardrails if g.run_in_parallel] guardrail_results: list[OutputGuardrailResult] = [] @@ -196,6 +194,28 @@ def record(result: OutputGuardrailResult) -> None: if results_sink is not None: results_sink.append(result) + # Run sequential guardrails first, one by one + for guardrail in sequential_guardrails: + result = await run_single_output_guardrail(guardrail, agent, agent_output, context) + if result.output.tripwire_triggered: + record(result) + _error_tracing.attach_error_to_current_span( + SpanError( + message="Guardrail tripwire triggered", + data={"guardrail": result.guardrail.get_name()}, + ) + ) + raise OutputGuardrailTripwireTriggered(result) + record(result) + + if not parallel_guardrails: + return guardrail_results + + guardrail_tasks = [ + asyncio.create_task(run_single_output_guardrail(guardrail, agent, agent_output, context)) + for guardrail in parallel_guardrails + ] + try: for done in asyncio.as_completed(guardrail_tasks): result = await done diff --git a/tests/test_output_guardrail_parallel.py b/tests/test_output_guardrail_parallel.py new file mode 100644 index 0000000000..23a1f37fd8 --- /dev/null +++ b/tests/test_output_guardrail_parallel.py @@ -0,0 +1,48 @@ +import asyncio + +import pytest + +from agents.agent import Agent +from agents.guardrail import GuardrailFunctionOutput, output_guardrail +from agents.run_context import RunContextWrapper +from agents.run_internal.guardrails import run_output_guardrails + + +@pytest.mark.asyncio +async def test_output_guardrail_sequential_execution(): + execution_order = [] + + @output_guardrail(run_in_parallel=False, name="seq1") + async def seq_guardrail_1(ctx, agent, output): + await asyncio.sleep(0.05) + execution_order.append("seq1") + return GuardrailFunctionOutput(tripwire_triggered=False, output_info=None) + + @output_guardrail(run_in_parallel=False, name="seq2") + async def seq_guardrail_2(ctx, agent, output): + execution_order.append("seq2") + return GuardrailFunctionOutput(tripwire_triggered=False, output_info=None) + + @output_guardrail(run_in_parallel=True, name="par1") + async def par_guardrail_1(ctx, agent, output): + await asyncio.sleep(0.02) + execution_order.append("par1") + return GuardrailFunctionOutput(tripwire_triggered=False, output_info=None) + + @output_guardrail(run_in_parallel=True, name="par2") + async def par_guardrail_2(ctx, agent, output): + execution_order.append("par2") + return GuardrailFunctionOutput(tripwire_triggered=False, output_info=None) + + agent = Agent(name="test") + ctx = RunContextWrapper(context=None) + guardrails = [par_guardrail_1, seq_guardrail_1, par_guardrail_2, seq_guardrail_2] + + results = await run_output_guardrails( + guardrails=guardrails, agent=agent, agent_output="test_output", context=ctx, results_sink=[] + ) + + assert len(results) == 4 + # Sequential ones should run first and in order, then parallel ones run concurrently + # par2 will finish before par1 because par1 sleeps + assert execution_order == ["seq1", "seq2", "par2", "par1"] From 50e1a5c498761896f2e5029fd351eba7f2a941d8 Mon Sep 17 00:00:00 2001 From: Showmick Das Date: Mon, 31 Aug 2026 02:46:30 -0400 Subject: [PATCH 2/2] fix: honor output guardrail parallel flag in RealtimeSession --- src/agents/realtime/session.py | 40 +++++++++++++++++++++++++++++----- tests/realtime/test_session.py | 12 +++++----- 2 files changed, 40 insertions(+), 12 deletions(-) diff --git a/src/agents/realtime/session.py b/src/agents/realtime/session.py index ef5f18cdfd..6993456ef2 100644 --- a/src/agents/realtime/session.py +++ b/src/agents/realtime/session.py @@ -1660,7 +1660,10 @@ async def _run_output_guardrails( triggered_results = [] - for guardrail in output_guardrails: + sequential_guardrails = [g for g in output_guardrails if not getattr(g, "run_in_parallel", True)] + parallel_guardrails = [g for g in output_guardrails if getattr(g, "run_in_parallel", True)] + + async def _run_single(guardrail: Any) -> Any: try: result = await guardrail.run( # TODO (rm) Remove this cast, it's wrong @@ -1668,10 +1671,7 @@ async def _run_output_guardrails( cast(Agent[Any], source_agent), text, ) - if self._closing or self._closed: - return False - if result.output.tripwire_triggered: - triggered_results.append(result) + return result except Exception as exc: log_model_and_tool_action_warning( logger, @@ -1679,7 +1679,35 @@ async def _run_output_guardrails( exc, diagnostic_extra=partial(_guardrail_diagnostic_extra, guardrail), ) - continue + return None + + # Run sequential guardrails first + for guardrail in sequential_guardrails: + result = await _run_single(guardrail) + if self._closing or self._closed: + return False + if result and result.output.tripwire_triggered: + triggered_results.append(result) + break + + # Run parallel guardrails only if sequential didn't trip + if not triggered_results and parallel_guardrails: + tasks = [asyncio.create_task(_run_single(g)) for g in parallel_guardrails] + try: + for done in asyncio.as_completed(tasks): + result = await done + if self._closing or self._closed: + return False + if result and result.output.tripwire_triggered: + triggered_results.append(result) + for t in tasks: + t.cancel() + break + finally: + for t in tasks: + if not t.done(): + t.cancel() + await asyncio.gather(*tasks, return_exceptions=True) if triggered_results: # Double-check: bail if already interrupted for this response diff --git a/tests/realtime/test_session.py b/tests/realtime/test_session.py index 178d37fa8e..9bcf4b0e52 100644 --- a/tests/realtime/test_session.py +++ b/tests/realtime/test_session.py @@ -5722,20 +5722,20 @@ def guardrail_func(context, agent, output): # Wait for async guardrail tasks to complete await self._wait_for_guardrail_tasks(session) - # Should have interrupted and sent message with both guardrail names + # Should have interrupted and sent message with the first tripped guardrail assert mock_model.interrupts_called == 1 assert len(mock_model.sent_messages) == 1 message = mock_model.sent_messages[0] - assert "guardrail_1" in message and "guardrail_2" in message + # Because we fail fast, only one guardrail will finish tripping before others are cancelled. + assert "guardrail_1" in message or "guardrail_2" in message - # Should have emitted event with both guardrail results + # Should have emitted event with the guardrail result 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 len(guardrail_events[0].guardrail_results) == 2 + event = next(e for e in events if isinstance(e, RealtimeGuardrailTripped)) + assert len(event.guardrail_results) == 1 @pytest.mark.asyncio async def test_agent_output_guardrails_triggered(self, mock_model, triggered_guardrail):