コンテンツにスキップ

ガードレール

ガードレールを使用すると、ユーザー入力とエージェント出力のチェックおよび検証を行えます。たとえば、顧客からのリクエストに対応するため、非常に高性能である一方、低速かつ高コストなモデルを使用するエージェントがあるとします。悪意のあるユーザーに、数学の宿題を手伝うようモデルへ依頼されることは避けたいでしょう。そのため、高速で低コストなモデルを使用してガードレールを実行できます。ガードレールが悪意のある利用を検出した場合、直ちにエラーを発生させ、時間とコストを節約できます。ブロッキング実行では、高コストなモデルが起動しないことが保証されます。一方、並列実行では、ガードレールが完了する前に高コストなモデルがすでに起動している可能性があります。詳細については、以下の「実行モード」を参照してください。

ガードレールには、次の 2 種類があります。

  1. 入力ガードレールは、最初のユーザー入力に対して実行されます
  2. 出力ガードレールは、エージェントの最終出力に対して実行されます

ワークフローの境界

ガードレールはエージェントとツールに関連付けられますが、ワークフロー内ですべてが同じタイミングに実行されるわけではありません。

  • 入力ガードレール は、チェーン内の最初のエージェントに対してのみ実行されます。
  • 出力ガードレール は、最終出力を生成するエージェントに対してのみ実行されます。
  • ツールガードレール は、カスタム関数ツールが呼び出されるたびに実行されます。入力ガードレールは実行前、出力ガードレールは実行後に実行されます。

マネージャー、ハンドオフ、または委任先の専門エージェントを含むワークフローで、各カスタム関数ツールの呼び出し前後にチェックが必要な場合は、エージェントレベルの入力／出力ガードレールだけに依存せず、ツールガードレールを使用してください。

入力ガードレール

入力ガードレールは、次の 3 ステップで実行されます。

  1. 最初に、ガードレールはエージェントに渡されたものと同じ入力を受け取ります。
  2. 次に、ガードレール関数が実行され、GuardrailFunctionOutput が生成されます。その後、これは InputGuardrailResult にラップされます
  3. 最後に、.tripwire_triggered が true かどうかを確認します。true の場合、InputGuardrailTripwireTriggered 例外が発生するため、ユーザーに適切に応答するか、例外を処理できます。

Note

入力ガードレールはユーザー入力に対して実行することを意図しているため、エージェントのガードレールは、そのエージェントが 最初 のエージェントである場合にのみ実行されます。なぜ guardrails プロパティが Runner.run に渡されるのではなく、エージェント上にあるのか疑問に思うかもしれません。これは、ガードレールが実際のエージェントに関連する傾向があるためです。エージェントごとに異なるガードレールを実行するため、コードを同じ場所に配置すると可読性が向上します。

実行モード

入力ガードレールは、次の 2 つの実行モードをサポートしています。

  • 並列実行 （デフォルト、 run_in_parallel=True ）：ガードレールはエージェントの実行と同時に実行されます。両方が同時に開始されるため、レイテンシーを最小限に抑えられます。ただし、ガードレールのトリップワイヤーが作動した場合、キャンセルされる前にエージェントがすでにトークンを消費し、ツールを実行している可能性があります。

  • ブロッキング実行 （ run_in_parallel=False ）：ガードレールは、エージェントが起動する 前 に実行され、完了します。ガードレールのトリップワイヤーが作動した場合、エージェントは一切実行されないため、トークンの消費とツールの実行を防止できます。これは、コストを最適化したい場合や、ツール呼び出しによる潜在的な副作用を回避したい場合に最適です。

出力ガードレール

出力ガードレールは、次の 3 ステップで実行されます。

  1. 最初に、ガードレールはエージェントが生成した出力を受け取ります。
  2. 次に、ガードレール関数が実行され、GuardrailFunctionOutput が生成されます。その後、これは OutputGuardrailResult にラップされます
  3. 最後に、.tripwire_triggered が true かどうかを確認します。true の場合、OutputGuardrailTripwireTriggered 例外が発生するため、ユーザーに適切に応答するか、例外を処理できます。

Note

出力ガードレールはエージェントの最終出力に対して実行することを意図しているため、エージェントのガードレールは、そのエージェントが 最後 のエージェントである場合にのみ実行されます。入力ガードレールと同様に、これはガードレールが実際のエージェントに関連する傾向があるためです。エージェントごとに異なるガードレールを実行するため、コードを同じ場所に配置すると可読性が向上します。

出力ガードレールは常にエージェントの完了後に実行されるため、 run_in_parallel パラメーターをサポートしていません。

出力トリップワイヤーと、ガードレール関数によって発生した例外では、セッションの動作が異なります。トリップワイヤーは、最終出力の候補を拒否します。トリップワイヤーが作動すると、ランナーは設定済みのセッションに対して、すでに完了したツール呼び出しとツール出力の項目を、それらの呼び出しの再実行に必要な推論コンテキストとともに永続化するよう要求します。このとき、拒否された最終出力の候補は除外されます。ランナーは、このトリップワイヤーのルールをストリーミング実行と非ストリーミング実行の両方に適用します。ガードレール関数がトリップワイヤーの結果を返す代わりに例外を発生させた場合、ランナーは判定を不明として扱い、ガードレール例外を通知する前に、完了した最終ターンの項目を永続化するよう設定済みのセッションに要求します。そのセッションへの書き込みも失敗した場合は、セッション書き込みエラーが優先されます。ストリーミング実行では、非ストリーミング実行と同じ永続化順序が使用され、 stream_events() から終端例外が発生します。出力ガードレールの実行中に RunResultStreaming.cancel() を直ちに呼び出すと、実行中のガードレールがキャンセルされ、最終ターンのセッション書き込みは開始されません。

ツールガードレール

ツールガードレールは FunctionTool のインスタンス をラップし、それらのツールの呼び出しを実行前後に検証またはブロックできるようにします。ツール自体に設定され、そのツールが呼び出されるたびに実行されます。

  • 入力ツールガードレールはツールの実行前に実行され、呼び出しのスキップ、メッセージによる出力の置き換え、またはトリップワイヤーの作動が可能です。
  • 出力ツールガードレールはツールの実行後に実行され、出力の置き換えまたはトリップワイヤーの作動が可能です。
  • 関数ツールに承認が必要な場合、入力ツールガードレールは通常、承認後かつ実行直前に実行されます。保留中の承認による中断が発生する前にこれらの入力チェックを実行するには、RunConfig.tool_execution を ToolExecutionConfig(pre_approval_tool_input_guardrails=True) に設定してください。この事前承認チェックに合格した呼び出しも、ツールの実行前に承認後のチェックを再度受けます。
  • ツールガードレールは、function_tool で作成された関数ツールにのみ適用されます。ハンドオフは通常の関数ツールパイプラインではなく SDK のハンドオフパイプラインを通じて実行されるため、ツールガードレールはハンドオフ呼び出し自体には適用されません。OpenAI がホストするツール（ WebSearchTool 、 FileSearchTool 、 HostedMCPTool 、 CodeInterpreterTool 、 ImageGenerationTool ）および組み込み実行ツール（ ComputerTool 、 ShellTool 、 ApplyPatchTool 、 LocalShellTool ）も、このガードレールパイプラインを使用しません。また、Agent.as_tool() は現在、ツールガードレールのオプションを直接公開していません。

詳細については、以下のコードスニペットを参照してください。

トリップワイヤー

エージェントの入力または出力がガードレールのチェックに失敗した場合、ガードレールはトリップワイヤーによってそれを通知できます。ランナーは直ちに InputGuardrailTripwireTriggered または OutputGuardrailTripwireTriggered 例外を発生させ、エージェントの実行を停止します。ツールガードレールでは、対応する ToolInputGuardrailTripwireTriggered および ToolOutputGuardrailTripwireTriggered 例外が使用されます。

エージェントレベルのトリップワイヤーでは、例外の guardrail_result によって、トリップワイヤーを作動させたガードレールを特定できます。ランナーによって入力トリップワイヤーが発生した場合、 exception.run_data.input_guardrail_results には、実行が停止する前に完了したすべての入力ガードレールの結果が含まれます。これには、トリップワイヤーを作動させた結果も含まれます。出力トリップワイヤーでは、同等の累積結果が exception.run_data.output_guardrail_results を通じて提供されます。

一方、ツールトリップワイヤー例外では、作動の原因となった guardrail と output が直接公開されます。その run_data.tool_input_guardrail_results および run_data.tool_output_guardrail_results のリストには、失敗前に完了したターンから累積された結果が保持されます。作動の原因となった結果は、例外の output を通じて取得できます。 MaxTurnsExceeded など、ランナーが管理するその他の失敗でも、完了したツールガードレールの結果がこれらのリストに保持されます。 stream_events() が例外を発生させた後、ストリーミング実行結果には、同じく累積されたエージェントおよびツールガードレールの結果リストが公開されます。ランナーが管理する実行パスの外部で例外が発生した場合、 run_data は None になることがあります。

ガードレールの実装

入力を受け取り、GuardrailFunctionOutput を返す関数を用意する必要があります。この例では、内部でエージェントを実行することによって、これを実現します。

from pydantic import BaseModel
from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    TResponseInputItem,
)
from agents.decorators import input_guardrail

class MathHomeworkOutput(BaseModel):
    is_math_homework: bool
    reasoning: str

guardrail_agent = Agent( # (1)!
    name="Guardrail check",
    instructions="Check if the user is asking you to do their math homework.",
    output_type=MathHomeworkOutput,
)


@input_guardrail
async def math_guardrail( # (2)!
    ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    result = await Runner.run(guardrail_agent, input, context=ctx.context)

    return GuardrailFunctionOutput(
        output_info=result.final_output, # (3)!
        tripwire_triggered=result.final_output.is_math_homework,
    )


agent = Agent(  # (4)!
    name="Customer support agent",
    instructions="You are a customer support agent. You help customers with their questions.",
    input_guardrails=[math_guardrail],
)

async def main():
    # This should trip the guardrail
    try:
        await Runner.run(agent, "Hello, can you help me solve for x: 2x + 3 = 11?")
        print("Guardrail didn't trip - this is unexpected")

    except InputGuardrailTripwireTriggered:
        print("Math homework guardrail tripped")
  1. このエージェントをガードレール関数で使用します。
  2. これは、エージェントの入力／コンテキストを受け取り、結果を返すガードレール関数です。
  3. ガードレールの結果には、追加情報を含めることができます。
  4. これは、ワークフローを定義する実際のエージェントです。

出力ガードレールも同様です。

from pydantic import BaseModel
from agents import (
    Agent,
    GuardrailFunctionOutput,
    OutputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
)
from agents.decorators import output_guardrail
class MessageOutput(BaseModel): # (1)!
    response: str

class MathOutput(BaseModel): # (2)!
    reasoning: str
    is_math: bool

guardrail_agent = Agent(
    name="Guardrail check",
    instructions="Check if the output includes any math.",
    output_type=MathOutput,
)

@output_guardrail
async def math_guardrail(  # (3)!
    ctx: RunContextWrapper, agent: Agent, output: MessageOutput
) -> GuardrailFunctionOutput:
    result = await Runner.run(guardrail_agent, output.response, context=ctx.context)

    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_math,
    )

agent = Agent( # (4)!
    name="Customer support agent",
    instructions="You are a customer support agent. You help customers with their questions.",
    output_guardrails=[math_guardrail],
    output_type=MessageOutput,
)

async def main():
    # This should trip the guardrail
    try:
        await Runner.run(agent, "Hello, can you help me solve for x: 2x + 3 = 11?")
        print("Guardrail didn't trip - this is unexpected")

    except OutputGuardrailTripwireTriggered:
        print("Math output guardrail tripped")
  1. これは、実際のエージェントの出力型です。
  2. これは、ガードレールの出力型です。
  3. これは、エージェントの出力を受け取り、結果を返すガードレール関数です。
  4. これは、ワークフローを定義する実際のエージェントです。

最後に、ツールガードレールの例を示します。

import json
from agents import (
    Agent,
    Runner,
    ToolGuardrailFunctionOutput,
)
from agents.decorators import tool, tool_input_guardrail, tool_output_guardrail

@tool_input_guardrail
def block_secrets(data):
    args = json.loads(data.context.tool_arguments or "{}")
    if "sk-" in json.dumps(args):
        return ToolGuardrailFunctionOutput.reject_content(
            "Remove secrets before calling this tool."
        )
    return ToolGuardrailFunctionOutput.allow()


@tool_output_guardrail
def redact_output(data):
    text = str(data.output or "")
    if "sk-" in text:
        return ToolGuardrailFunctionOutput.reject_content("Output contained sensitive data.")
    return ToolGuardrailFunctionOutput.allow()


@tool(
    tool_input_guardrails=[block_secrets],
    tool_output_guardrails=[redact_output],
)
def classify_text(text: str) -> str:
    """Classify text for internal routing."""
    return f"length:{len(text)}"


agent = Agent(name="Classifier", tools=[classify_text])
result = Runner.run_sync(agent, "hello world")
print(result.final_output)