Skip to content

feat: add run_in_parallel parameter to input guardrails #1986

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 9 commits into
base: main
Choose a base branch
from
+1,220 −7
Open

feat: add run_in_parallel parameter to input guardrails #1986

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
d28f527
feat: add run_in_parallel parameter to input guardrails
RVV-karma Oct 23, 2025
d4bc971
docs: clarify blocking vs parallel guardrail behavior
RVV-karma Oct 23, 2025
0050939
fix: blocking guardrails now prevent agent execution in streaming mode
RVV-karma Oct 23, 2025
721e9ce
fix: add queue sentinel before raising blocking guardrail exception i…
RVV-karma Nov 15, 2025
e35aa7c
test: add comprehensive tests for blocking vs parallel guardrail exec…
RVV-karma Nov 15, 2025
f2c34eb
docs: revert guardrails.md changes for separate PR
RVV-karma Nov 15, 2025
339895f
docs: revert input_guardrails.py docstring changes
RVV-karma Nov 16, 2025
db58955
fix: cancel remaining guardrails when blocking guardrail triggers
RVV-karma Nov 16, 2025
900241a
test: add cancellation tests for blocking guardrails
RVV-karma Nov 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 16 additions & 2 deletions src/agents/guardrail.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ class OutputGuardrailResult:

@dataclass
class InputGuardrail(Generic[TContext]):
"""Input guardrails are checks that run in parallel to the agent's execution.
"""Input guardrails are checks that run either in parallel with the agent or before it starts.
They can be used to do things like:
- Check if input messages are off-topic
- Take over control of the agent's execution if an unexpected input is detected
Expand All @@ -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).
"""
Comment on lines +100 to +103
Copy link

Copilot AI Oct 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The class-level docstring for InputGuardrail states 'Input guardrails are checks that run in parallel to the agent's execution' but this is no longer accurate since guardrails can now run sequentially. The docstring should be updated to reflect that guardrails can run either in parallel or sequentially based on the run_in_parallel parameter.

Copilot uses AI. Check for mistakes.
Copy link
Author

@RVV-karma RVV-karma Oct 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated in 1ad513a


def get_name(self) -> str:
if self.name:
return self.name
Expand Down Expand Up @@ -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],
Expand All @@ -221,6 +227,7 @@ def input_guardrail(
| None = None,
*,
name: str | None = None,
run_in_parallel: bool = True,
) -> (
InputGuardrail[TContext_co]
| Callable[
Expand All @@ -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(
Expand All @@ -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:
Expand Down
70 changes: 65 additions & 5 deletions src/agents/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -601,11 +601,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,
),
Expand All @@ -622,6 +642,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,
Expand Down Expand Up @@ -941,6 +964,11 @@ async def _run_input_guardrails_with_queue(
for done in asyncio.as_completed(guardrail_tasks):
result = await done
if result.output.tripwire_triggered:
# Cancel all remaining guardrail tasks if a tripwire is triggered.
for t in guardrail_tasks:
t.cancel()
# Wait for cancellations to propagate by awaiting the cancelled tasks.
await asyncio.gather(*guardrail_tasks, return_exceptions=True)
_error_tracing.attach_error_to_span(
parent_span,
SpanError(
Expand All @@ -951,14 +979,19 @@ async def _run_input_guardrails_with_queue(
},
),
)
queue.put_nowait(result)
guardrail_results.append(result)
break
queue.put_nowait(result)
guardrail_results.append(result)
except Exception:
for t in guardrail_tasks:
t.cancel()
raise

streamed_result.input_guardrail_results = guardrail_results
streamed_result.input_guardrail_results = (
streamed_result.input_guardrail_results + guardrail_results
)

@classmethod
async def _start_streaming(
Comment on lines 989 to 997
Copy link

@chatgpt-codex-connector chatgpt-codex-connector bot Nov 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Cancel blocking guardrails when one trips in streaming mode

Blocking input guardrails are cancelled in the non‑streaming path (_run_input_guardrails cancels remaining tasks once a result trips the tripwire) so no additional guardrail calls run after failure. The new streaming path executes blocking guardrails through _run_input_guardrails_with_queue, then just loops over the accumulated results and raises if any tripwire triggered. Because _run_input_guardrails_with_queue never cancels the remaining guardrail tasks, all blocking guardrails continue running even after the first one fails. In scenarios where guardrails call out to external services this wastes time and can cause unintended side effects, and it contradicts the stated motivation of blocking guardrails providing better cost efficiency. Consider mirroring the non‑streaming behaviour by cancelling outstanding guardrail tasks once a tripwire fires before starting the agent.

Useful? React with 👍 / 👎.

Copy link
Author

@RVV-karma RVV-karma Nov 16, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fix: db58955
Test for the same: 900241a

Expand Down Expand Up @@ -1050,11 +1083,36 @@ 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.
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,
)
# Check if any blocking guardrail triggered and raise before starting agent.
for result in streamed_result.input_guardrail_results:
if result.output.tripwire_triggered:
streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
raise InputGuardrailTripwireTriggered(result)

# 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,
Expand Down Expand Up @@ -1619,6 +1677,8 @@ async def _run_input_guardrails(
# Cancel all guardrail tasks if a tripwire is triggered.
for t in guardrail_tasks:
t.cancel()
# Wait for cancellations to propagate by awaiting the cancelled tasks.
await asyncio.gather(*guardrail_tasks, return_exceptions=True)
_error_tracing.attach_error_to_current_span(
SpanError(
message="Guardrail tripwire triggered",
Expand Down
Loading