feat: add run_in_parallel parameter to input guardrails

Author: RVV-karma

docs/guardrails.md

@@ -1,6 +1,6 @@
 # Guardrails
 
-Guardrails run _in parallel_ to your agents, enabling you to do checks and validations of user input. For example, imagine you have an agent that uses a very smart (and hence slow/expensive) model to help with customer requests. You wouldn't want malicious users to ask the model to help them with their math homework. So, you can run a guardrail with a fast/cheap model. If the guardrail detects malicious usage, it can immediately raise an error, which stops the expensive model from running and saves you time/money.
+Guardrails enable you to do checks and validations of user input and agent output. For example, imagine you have an agent that uses a very smart (and hence slow/expensive) model to help with customer requests. You wouldn't want malicious users to ask the model to help them with their math homework. So, you can run a guardrail with a fast/cheap model. If the guardrail detects malicious usage, it can immediately raise an error and prevent the expensive model from running, saving you time and money.
 
 There are two kinds of guardrails:
 
@@ -19,6 +19,14 @@ Input guardrails run in 3 steps:
 
     Input guardrails are intended to run on user input, so an agent's guardrails only run if the agent is the *first* agent. You might wonder, why is the `guardrails` property on the agent instead of passed to `Runner.run`? It's because guardrails tend to be related to the actual Agent - you'd run different guardrails for different agents, so colocating the code is useful for readability.
 
+### Execution modes
+
+Input guardrails support two execution modes:
+
+- **Parallel execution** (default, `run_in_parallel=True`): The guardrail runs concurrently with the agent's execution. This provides the best latency since both start at the same time. However, if the guardrail fails, the agent may have already consumed tokens and executed tools before being cancelled.
+
+- **Blocking execution** (`run_in_parallel=False`): The guardrail runs and completes *before* the agent starts. If the guardrail tripwire is triggered, the agent never executes, preventing token consumption and tool execution. This is ideal for cost optimization and when you want to avoid potential side effects from tool calls.
+
 ## Output guardrails
 
 Output guardrails run in 3 steps:
@@ -31,6 +39,8 @@ Output guardrails run in 3 steps:
 
     Output guardrails are intended to run on the final agent output, so an agent's guardrails only run if the agent is the *last* agent. Similar to the input guardrails, we do this because guardrails tend to be related to the actual Agent - you'd run different guardrails for different agents, so colocating the code is useful for readability.
 
+    Output guardrails always run after the agent completes, so they don't support the `run_in_parallel` parameter.
+
 ## Tripwires
 
 If the input or output fails the guardrail, the Guardrail can signal this with a tripwire. As soon as we see a guardrail that has triggered the tripwires, we immediately raise a `{Input,Output}GuardrailTripwireTriggered` exception and halt the Agent execution.

examples/agent_patterns/input_guardrails.py

@@ -17,14 +17,18 @@
 """
 This example shows how to use guardrails.
 
-Guardrails are checks that run in parallel to the agent's execution.
+Guardrails are checks that run either in parallel with the agent or before the agent starts.
 They can be used to do things like:
 - Check if input messages are off-topic
 - Check that input messages don't violate any policies
 - Take over control of the agent's execution if an unexpected input is detected
 
 In this example, we'll setup an input guardrail that trips if the user is asking to do math homework.
 If the guardrail trips, we'll respond with a refusal message.
+
+By default, guardrails run in parallel with the agent for better latency.
+You can set run_in_parallel=False to run the guardrail before the agent starts,
+which saves token costs if the guardrail fails (the agent never starts).
 """
 
 

src/agents/guardrail.py

@@ -97,6 +97,11 @@ class InputGuardrail(Generic[TContext]):
     function's name.
     """
 
+    run_in_parallel: bool = True
+    """Whether the guardrail runs concurrently with the agent (True, default) or before
+    the agent starts (False).
+    """
+
     def get_name(self) -> str:
         if self.name:
             return self.name
@@ -209,6 +214,7 @@ def input_guardrail(
 def input_guardrail(
     *,
     name: str | None = None,
+    run_in_parallel: bool = True,
 ) -> Callable[
     [_InputGuardrailFuncSync[TContext_co] | _InputGuardrailFuncAsync[TContext_co]],
     InputGuardrail[TContext_co],
@@ -221,6 +227,7 @@ def input_guardrail(
     | None = None,
     *,
     name: str | None = None,
+    run_in_parallel: bool = True,
 ) -> (
     InputGuardrail[TContext_co]
     | Callable[
@@ -235,8 +242,14 @@ def input_guardrail(
         @input_guardrail
         def my_sync_guardrail(...): ...
 
-        @input_guardrail(name="guardrail_name")
+        @input_guardrail(name="guardrail_name", run_in_parallel=False)
         async def my_async_guardrail(...): ...
+
+    Args:
+        func: The guardrail function to wrap.
+        name: Optional name for the guardrail. If not provided, uses the function's name.
+        run_in_parallel: Whether to run the guardrail concurrently with the agent (True, default)
+            or before the agent starts (False).
     """
 
     def decorator(
@@ -246,6 +259,7 @@ def decorator(
             guardrail_function=f,
             # If not set, guardrail name uses the function’s name by default.
             name=name if name else f.__name__,
+            run_in_parallel=run_in_parallel,
         )
 
     if func is not None:

src/agents/run.py

@@ -599,11 +599,31 @@ async def run(
                     )
 
                     if current_turn == 1:
+                        # Separate guardrails based on execution mode.
+                        all_input_guardrails = starting_agent.input_guardrails + (
+                            run_config.input_guardrails or []
+                        )
+                        sequential_guardrails = [
+                            g for g in all_input_guardrails if not g.run_in_parallel
+                        ]
+                        parallel_guardrails = [g for g in all_input_guardrails if g.run_in_parallel]
+
+                        # Run blocking guardrails first, before agent starts.
+                        # (will raise exception if tripwire triggered).
+                        sequential_results = []
+                        if sequential_guardrails:
+                            sequential_results = await self._run_input_guardrails(
+                                starting_agent,
+                                sequential_guardrails,
+                                _copy_str_or_list(prepared_input),
+                                context_wrapper,
+                            )
+
+                        # Run parallel guardrails + agent together.
                         input_guardrail_results, turn_result = await asyncio.gather(
                             self._run_input_guardrails(
                                 starting_agent,
-                                starting_agent.input_guardrails
-                                + (run_config.input_guardrails or []),
+                                parallel_guardrails,
                                 _copy_str_or_list(prepared_input),
                                 context_wrapper,
                             ),
@@ -620,6 +640,9 @@ async def run(
                                 server_conversation_tracker=server_conversation_tracker,
                             ),
                         )
+
+                        # Combine sequential and parallel results.
+                        input_guardrail_results = sequential_results + input_guardrail_results
                     else:
                         turn_result = await self._run_single_turn(
                             agent=current_agent,
@@ -997,11 +1020,31 @@ async def _start_streaming(
                     break
 
                 if current_turn == 1:
-                    # Run the input guardrails in the background and put the results on the queue
+                    # Separate guardrails based on execution mode.
+                    all_input_guardrails = starting_agent.input_guardrails + (
+                        run_config.input_guardrails or []
+                    )
+                    sequential_guardrails = [
+                        g for g in all_input_guardrails if not g.run_in_parallel
+                    ]
+                    parallel_guardrails = [g for g in all_input_guardrails if g.run_in_parallel]
+
+                    # Run sequential guardrails first (will raise exception if tripwire triggered).
+                    if sequential_guardrails:
+                        await cls._run_input_guardrails_with_queue(
+                            starting_agent,
+                            sequential_guardrails,
+                            ItemHelpers.input_to_new_input_list(prepared_input),
+                            context_wrapper,
+                            streamed_result,
+                            current_span,
+                        )
+
+                    # Run parallel guardrails in background.
                     streamed_result._input_guardrails_task = asyncio.create_task(
                         cls._run_input_guardrails_with_queue(
                             starting_agent,
-                            starting_agent.input_guardrails + (run_config.input_guardrails or []),
+                            parallel_guardrails,
                             ItemHelpers.input_to_new_input_list(prepared_input),
                             context_wrapper,
                             streamed_result,

tests/test_guardrails.py

@@ -260,3 +260,58 @@ async def test_output_guardrail_decorators():
     assert not result.output.tripwire_triggered
     assert result.output.output_info == "test_4"
     assert guardrail.get_name() == "Custom name"
+
+
+@pytest.mark.asyncio
+async def test_input_guardrail_run_in_parallel_default():
+    guardrail = InputGuardrail(
+        guardrail_function=lambda ctx, agent, input: GuardrailFunctionOutput(
+            output_info=None, tripwire_triggered=False
+        )
+    )
+    assert guardrail.run_in_parallel is True
+
+
+@pytest.mark.asyncio
+async def test_input_guardrail_run_in_parallel_false():
+    guardrail = InputGuardrail(
+        guardrail_function=lambda ctx, agent, input: GuardrailFunctionOutput(
+            output_info=None, tripwire_triggered=False
+        ),
+        run_in_parallel=False,
+    )
+    assert guardrail.run_in_parallel is False
+
+
+@pytest.mark.asyncio
+async def test_input_guardrail_decorator_with_run_in_parallel():
+    @input_guardrail(run_in_parallel=False)
+    def blocking_guardrail(
+        context: RunContextWrapper[Any], agent: Agent[Any], input: str | list[TResponseInputItem]
+    ) -> GuardrailFunctionOutput:
+        return GuardrailFunctionOutput(
+            output_info="blocking",
+            tripwire_triggered=False,
+        )
+
+    assert blocking_guardrail.run_in_parallel is False
+    result = await blocking_guardrail.run(
+        agent=Agent(name="test"), input="test", context=RunContextWrapper(context=None)
+    )
+    assert not result.output.tripwire_triggered
+    assert result.output.output_info == "blocking"
+
+
+@pytest.mark.asyncio
+async def test_input_guardrail_decorator_with_name_and_run_in_parallel():
+    @input_guardrail(name="custom_name", run_in_parallel=False)
+    def named_blocking_guardrail(
+        context: RunContextWrapper[Any], agent: Agent[Any], input: str | list[TResponseInputItem]
+    ) -> GuardrailFunctionOutput:
+        return GuardrailFunctionOutput(
+            output_info="named_blocking",
+            tripwire_triggered=False,
+        )
+
+    assert named_blocking_guardrail.get_name() == "custom_name"
+    assert named_blocking_guardrail.run_in_parallel is False