fix: add run_in_parallel support to OutputGuardrail for sequential ex…

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:

src/agents/realtime/session.py

@@ -1660,26 +1660,54 @@ 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
                     self._context_wrapper,
                     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,
                     "Output guardrail raised an exception; skipping it",
                     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

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

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

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"]