Guardrails

GuardrailFunctionOutput dataclass

The output of a guardrail function.

源代码位于 src/agents/guardrail.py

@dataclass
class GuardrailFunctionOutput:
    """The output of a guardrail function."""

    output_info: Any
    """
    Optional information about the guardrail's output. For example, the guardrail could include
    information about the checks it performed and granular results.
    """

    tripwire_triggered: bool
    """
    Whether the tripwire was triggered. If triggered, the agent's execution will be halted.
    """

output_info instance-attribute

output_info: Any

关于 guardrail 输出的可选信息。例如，guardrail 可以包含有关其执行的检查和粒度结果的信息。

tripwire_triggered instance-attribute

tripwire_triggered: bool

是否触发了 tripwire。如果触发，代理的执行将被停止。

InputGuardrailResult dataclass

guardrail 运行的结果。

源代码位于 src/agents/guardrail.py

@dataclass
class InputGuardrailResult:
    """The result of a guardrail run."""

    guardrail: InputGuardrail[Any]
    """
    The guardrail that was run.
    """

    output: GuardrailFunctionOutput
    """The output of the guardrail function."""

guardrail instance-attribute

guardrail: InputGuardrail[Any]

运行的 guardrail。

输出 实例属性

output: GuardrailFunctionOutput

guardrail 函数的输出。

OutputGuardrailResult dataclass

guardrail 运行的结果。

源代码位于 src/agents/guardrail.py

@dataclass
class OutputGuardrailResult:
    """The result of a guardrail run."""

    guardrail: OutputGuardrail[Any]
    """
    The guardrail that was run.
    """

    agent_output: Any
    """
    The output of the agent that was checked by the guardrail.
    """

    agent: Agent[Any]
    """
    The agent that was checked by the guardrail.
    """

    output: GuardrailFunctionOutput
    """The output of the guardrail function."""

guardrail instance-attribute

guardrail: OutputGuardrail[Any]

运行的 guardrail。

agent_output instance-attribute

agent_output: Any

被 guardrail 检查的代理的输出。

agent 实例属性

agent: Agent[Any]

被 guardrail 检查的代理。

输出 实例属性

output: GuardrailFunctionOutput

guardrail 函数的输出。

InputGuardrail dataclass

基类: Generic[TContext]

输入 guardrail 是在与代理并行运行或在代理启动之前运行的检查。它们可用于执行以下操作：- 检查输入消息是否偏离主题 - 如果检测到意外输入，则接管代理执行的控制权

您可以使用 @input_guardrail() 装饰器将函数转换为 InputGuardrail，或手动创建 InputGuardrail。

Guardrails 返回一个 GuardrailResult。如果 result.tripwire_triggered 为 True，代理的执行将立即停止，并引发 InputGuardrailTripwireTriggered 异常

源代码位于 src/agents/guardrail.py

@dataclass
class InputGuardrail(Generic[TContext]):
    """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

    You can use the `@input_guardrail()` decorator to turn a function into an `InputGuardrail`, or
    create an `InputGuardrail` manually.

    Guardrails return a `GuardrailResult`. If `result.tripwire_triggered` is `True`,
    the agent's execution will immediately stop, and
    an `InputGuardrailTripwireTriggered` exception will be raised
    """

    guardrail_function: Callable[
        [RunContextWrapper[TContext], Agent[Any], str | list[TResponseInputItem]],
        MaybeAwaitable[GuardrailFunctionOutput],
    ]
    """A function that receives the agent input and the context, and returns a
     `GuardrailResult`. The result marks whether the tripwire was triggered, and can optionally
     include information about the guardrail's output.
    """

    name: str | None = None
    """The name of the guardrail, used for tracing. If not provided, we'll use the guardrail
    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

        return self.guardrail_function.__name__

    async def run(
        self,
        agent: Agent[Any],
        input: str | list[TResponseInputItem],
        context: RunContextWrapper[TContext],
    ) -> InputGuardrailResult:
        if not callable(self.guardrail_function):
            raise UserError(f"Guardrail function must be callable, got {self.guardrail_function}")

        output = self.guardrail_function(context, agent, input)
        if inspect.isawaitable(output):
            return InputGuardrailResult(
                guardrail=self,
                output=await output,
            )

        return InputGuardrailResult(
            guardrail=self,
            output=output,
        )

guardrail_function instance-attribute

guardrail_function: Callable[
    [
        RunContextWrapper[TContext],
        Agent[Any],
        str | list[TResponseInputItem],
    ],
    MaybeAwaitable[GuardrailFunctionOutput],
]

接收代理输入和上下文，并返回 GuardrailResult 的函数。结果标记 tripwire 是否被触发，并可以包含有关 guardrail 输出的信息。

name class-attribute instance-attribute

name: str | None = None

guardrail 的名称，用于跟踪。如果未提供，我们将使用 guardrail 函数的名称。

run_in_parallel class-attribute instance-attribute

run_in_parallel: bool = True

guardrail 是否与代理并行运行（True，默认）或在代理启动之前运行（False）。

OutputGuardrail dataclass

基类: Generic[TContext]

输出 guardrail 是在代理的最终输出上运行的检查。它们可用于检查输出是否通过某些验证标准

您可以使用 @output_guardrail() 装饰器将函数转换为 OutputGuardrail，或手动创建 OutputGuardrail。

Guardrails 返回一个 GuardrailResult。如果 result.tripwire_triggered 为 True，将引发 OutputGuardrailTripwireTriggered 异常。

源代码位于 src/agents/guardrail.py

@dataclass
class OutputGuardrail(Generic[TContext]):
    """Output guardrails are checks that run on the final output of an agent.
    They can be used to do check if the output passes certain validation criteria

    You can use the `@output_guardrail()` decorator to turn a function into an `OutputGuardrail`,
    or create an `OutputGuardrail` manually.

    Guardrails return a `GuardrailResult`. If `result.tripwire_triggered` is `True`, an
    `OutputGuardrailTripwireTriggered` exception will be raised.
    """

    guardrail_function: Callable[
        [RunContextWrapper[TContext], Agent[Any], Any],
        MaybeAwaitable[GuardrailFunctionOutput],
    ]
    """A function that receives the final agent, its output, and the context, and returns a
     `GuardrailResult`. The result marks whether the tripwire was triggered, and can optionally
     include information about the guardrail's output.
    """

    name: str | None = None
    """The name of the guardrail, used for tracing. If not provided, we'll use the guardrail
    function's name.
    """

    def get_name(self) -> str:
        if self.name:
            return self.name

        return self.guardrail_function.__name__

    async def run(
        self, context: RunContextWrapper[TContext], agent: Agent[Any], agent_output: Any
    ) -> OutputGuardrailResult:
        if not callable(self.guardrail_function):
            raise UserError(f"Guardrail function must be callable, got {self.guardrail_function}")

        output = self.guardrail_function(context, agent, agent_output)
        if inspect.isawaitable(output):
            return OutputGuardrailResult(
                guardrail=self,
                agent=agent,
                agent_output=agent_output,
                output=await output,
            )

        return OutputGuardrailResult(
            guardrail=self,
            agent=agent,
            agent_output=agent_output,
            output=output,
        )

guardrail_function instance-attribute

guardrail_function: Callable[
    [RunContextWrapper[TContext], Agent[Any], Any],
    MaybeAwaitable[GuardrailFunctionOutput],
]

接收最终代理、其输出和上下文，并返回 GuardrailResult 的函数。结果标记 tripwire 是否被触发，并可以包含有关 guardrail 输出的信息。

name class-attribute instance-attribute

name: str | None = None

guardrail 的名称，用于跟踪。如果未提供，我们将使用 guardrail 函数的名称。

input_guardrail

input_guardrail(
    func: _InputGuardrailFuncSync[TContext_co],
) -> InputGuardrail[TContext_co]

input_guardrail(
    func: _InputGuardrailFuncAsync[TContext_co],
) -> InputGuardrail[TContext_co]

input_guardrail(
    *, name: str | None = None, run_in_parallel: bool = True
) -> Callable[
    [
        _InputGuardrailFuncSync[TContext_co]
        | _InputGuardrailFuncAsync[TContext_co]
    ],
    InputGuardrail[TContext_co],
]

input_guardrail(
    func: _InputGuardrailFuncSync[TContext_co]
    | _InputGuardrailFuncAsync[TContext_co]
    | None = None,
    *,
    name: str | None = None,
    run_in_parallel: bool = True,
) -> (
    InputGuardrail[TContext_co]
    | Callable[
        [
            _InputGuardrailFuncSync[TContext_co]
            | _InputGuardrailFuncAsync[TContext_co]
        ],
        InputGuardrail[TContext_co],
    ]
)

将同步或异步函数转换为 InputGuardrail 的装饰器。可以直接使用（无括号），也可以使用关键字参数，例如：

@input_guardrail
def my_sync_guardrail(...): ...

@input_guardrail(name="guardrail_name", run_in_parallel=False)
async def my_async_guardrail(...): ...

参数

名称	类型	描述	默认
func	_InputGuardrailFuncSync[TContext_co] | _InputGuardrailFuncAsync[TContext_co] | None	要包装的 guardrail 函数。	None
name	str | None	guardrail 的可选名称。如果未提供，则使用函数的名称。	None
run_in_parallel	bool	是否与代理并行运行 guardrail（True，默认）或在代理启动之前运行（False）。	True

源代码位于 src/agents/guardrail.py

def input_guardrail(
    func: _InputGuardrailFuncSync[TContext_co]
    | _InputGuardrailFuncAsync[TContext_co]
    | None = None,
    *,
    name: str | None = None,
    run_in_parallel: bool = True,
) -> (
    InputGuardrail[TContext_co]
    | Callable[
        [_InputGuardrailFuncSync[TContext_co] | _InputGuardrailFuncAsync[TContext_co]],
        InputGuardrail[TContext_co],
    ]
):
    """
    Decorator that transforms a sync or async function into an `InputGuardrail`.
    It can be used directly (no parentheses) or with keyword args, e.g.:

        @input_guardrail
        def my_sync_guardrail(...): ...

        @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(
        f: _InputGuardrailFuncSync[TContext_co] | _InputGuardrailFuncAsync[TContext_co],
    ) -> InputGuardrail[TContext_co]:
        return InputGuardrail(
            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:
        # Decorator was used without parentheses
        return decorator(func)

    # Decorator used with keyword arguments
    return decorator

output_guardrail

output_guardrail(
    func: _OutputGuardrailFuncSync[TContext_co],
) -> OutputGuardrail[TContext_co]

output_guardrail(
    func: _OutputGuardrailFuncAsync[TContext_co],
) -> OutputGuardrail[TContext_co]

output_guardrail(
    *, name: str | None = None
) -> Callable[
    [
        _OutputGuardrailFuncSync[TContext_co]
        | _OutputGuardrailFuncAsync[TContext_co]
    ],
    OutputGuardrail[TContext_co],
]

output_guardrail(
    func: _OutputGuardrailFuncSync[TContext_co]
    | _OutputGuardrailFuncAsync[TContext_co]
    | None = None,
    *,
    name: str | None = None,
) -> (
    OutputGuardrail[TContext_co]
    | Callable[
        [
            _OutputGuardrailFuncSync[TContext_co]
            | _OutputGuardrailFuncAsync[TContext_co]
        ],
        OutputGuardrail[TContext_co],
    ]
)

将同步或异步函数转换为 OutputGuardrail 的装饰器。可以直接使用（无括号），也可以使用关键字参数，例如：

@output_guardrail
def my_sync_guardrail(...): ...

@output_guardrail(name="guardrail_name")
async def my_async_guardrail(...): ...

源代码位于 src/agents/guardrail.py

def output_guardrail(
    func: _OutputGuardrailFuncSync[TContext_co]
    | _OutputGuardrailFuncAsync[TContext_co]
    | None = None,
    *,
    name: str | None = None,
) -> (
    OutputGuardrail[TContext_co]
    | Callable[
        [_OutputGuardrailFuncSync[TContext_co] | _OutputGuardrailFuncAsync[TContext_co]],
        OutputGuardrail[TContext_co],
    ]
):
    """
    Decorator that transforms a sync or async function into an `OutputGuardrail`.
    It can be used directly (no parentheses) or with keyword args, e.g.:

        @output_guardrail
        def my_sync_guardrail(...): ...

        @output_guardrail(name="guardrail_name")
        async def my_async_guardrail(...): ...
    """

    def decorator(
        f: _OutputGuardrailFuncSync[TContext_co] | _OutputGuardrailFuncAsync[TContext_co],
    ) -> OutputGuardrail[TContext_co]:
        return OutputGuardrail(
            guardrail_function=f,
            # Guardrail name defaults to function's name when not specified (None).
            name=name if name else f.__name__,
        )

    if func is not None:
        # Decorator was used without parentheses
        return decorator(func)

    # Decorator used with keyword arguments
    return decorator