メインコンテンツへスキップ
見出し画像

OpenAI Agents SDK 入門 (4) - Guardrail ・ Orchestration ・ Model

    「OpenAI Agents SDK」の「Guardrail」「Orchestration」「Model」についてまとめました。

    ・Guardrail
    ・
    Orchestrating multiple agents
    ・
    Models
    ・
    Configuring the SDK

    前回

    1. Guardrail

    1-1. Guardrail

    「Guardrail」はAgentと並行して実行され、ユーザー入力のチェックやバリデーションを行うためのしくみです。

    例えば、非常に高度(=遅くて高コスト)なモデルを使用して顧客対応を行うAgentがあるとします。しかし、悪意のあるユーザーがこのモデルを使って数学の宿題を解かせようとした場合、それを防ぎたいと考えます。

    そこで、高速かつ低コストなモデルを使った「Guardrail」を実行します。「Guardrail」が不正な利用を検出すると、即座にエラーを発生させ、コストの高いモデルの実行を防ぐことができます。これにより、時間とコストを節約できます。

    ガードレールの種類は、次のとおりです。

    ・Input Guardrail
    ユーザーの最初の入力 に対して実行され、不適切なリクエストを検出・ブロックします。

    ・Output Guardrail
    Agentの最終的な出力 に対して実行され、不適切な回答が生成されないように制御します。

    1-2. Input Guardrail

    「Input Guardrail」は、以下の3つのステップで実行されます。

    (1) Agentに渡されたものと同じ入力を、Guardrailも受け取る。
    (2) Guardrail関数が実行され、GuardrailFunctionOutput を生成し、それを InputGuardrailResult にラップ。
    (3) .tripwire_triggered が true かどうかをチェック。
    true の場合、InputGuardrailTripwireTriggered 例外が発生します。これにより、適切なユーザー対応や例外処理が可能になります。

    1-3. Output Guardrail

    「Output Guardrail」は、以下の3つのステップ で実行されます。

    (1) Agentに渡されたものと同じ入力を、Guardrailも受け取る。
    (2) Guardrail関数が実行され、GuardrailFunctionOutput を生成し、それを OutputGuardrailResult にラップ。
    (3) .tripwire_triggered が true かどうかをチェック。

    true の場合、OutputGuardrailTripwireTriggered 例外が発生します。これにより、適切なユーザー対応や例外処理が可能になります。

    1-4. Tripwires

    入力または出力がGuardrailのチェックに 失敗 した場合、Guardrailは「Tripwire」 を作動させて警告を出すことができます。

    GuardrailがTripwireを検出すると、すぐに {Input,Output}GuardrailTripwireTriggered 例外を発生させ、Agentの実行を停止 します。

    1-5. Guardrail の実装方法

    「Guardrail」を実装するには、入力を受け取り、GuardrailFunctionOutput を返す関数を提供 する必要があります。

    以下の例では、Agentをバックグラウンドで実行することで、このGuardrailを実装します。

    from pydantic import BaseModel
    from agents import (
        Agent,
        GuardrailFunctionOutput,
        InputGuardrailTripwireTriggered,
        RunContextWrapper,
        Runner,
        TResponseInputItem,
        input_guardrail,
    )
    
    class MathHomeworkOutput(BaseModel):
        is_math_homework: bool
        reasoning: str
    
    guardrail_agent = Agent( 
        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( 
        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, 
            tripwire_triggered=result.final_output.is_math_homework,
        )
    
    
    agent = Agent(  
        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")

    「Output Guardrail」も同様です。

    from pydantic import BaseModel
    from agents import (
        Agent,
        GuardrailFunctionOutput,
        OutputGuardrailTripwireTriggered,
        RunContextWrapper,
        Runner,
        output_guardrail,
    )
    class MessageOutput(BaseModel): 
        response: str
    
    class MathOutput(BaseModel): 
        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(  
        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( 
        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")

    2. Orchestration

    2-1. Orchestration

    「Orchestration」とは、アプリ内でのAgentの流れのことで、どのAgentが実行されるのか、それらがどの順番で動作するのか、そして次に何をすべきかをどのように決定するのかを管理することを指します。

    AgentをOrchestrationする主な方法は2つあります。

    (1) LLMによるOrchestration
    LLMの知能を活用して、計画を立て、Reasoningし、どのステップを取るべきかを決定する方法です。

    (2) コードによるOrchestration
    コードによってAgentの流れを決定する方法です。

    これらの方法を組み合わせて使用することも可能です。それぞれにメリットとデメリットがあります。

    2-2. LLMによるOrchestration

    「Agent」とは、指示・Tool・タスクの受け渡し機能を備えた LLMのことです。これにより、Agentは自由度の高いタスクを与えられた際に、自律的に計画を立て、Toolを使用してアクションを実行したりデータを取得したり、またサブAgentにタスクを委任することができます。

    例えば、リサーチAgentには以下のようなToolを搭載できます。

    ・Web search : オンライン上で情報を検索
    ・File search : 独自のデータや接続情報を調査
    ・Computer use : コンピュータ上でアクションを実行
    ・Code execution:データ分析のためのコード実行
    ・Handoff : 計画立案やレポート作成に特化したAgentへ業務を委任

    このパターンは、タスクがオープンエンドで、LLMの知性に頼りたい場合に最適です。ここで最も重要な戦術は次のとおりです。

    (1) 良いプロンプトに投資する。利用可能なTool、使用方法、操作するパラメータを明確にします。
    (2) アプリを監視し、繰り返します。問題がどこにあるかを確認し、プロンプトを繰り返します。
    (3) Agentが内省して改善することを許可します。たとえば、ループで実行して、それ自体を批判させるか、エラーメッセージを提供して改善させるか。
    (4) 何でも得意と期待される汎用Agentではなく、1つのタスクに優れた専門Agentを持つ。
    (5) 評価に投資する。これにより、Agentがタスクを改善し、より良くなるように学習することができます。

    2-3. コードによるOrchestration

    LLMによるOrchestrationは強力ですが、コードによるOrchestrationは、速度・コスト・性能の点で、タスクをより決定的で予測可能にします。ここでの一般的なパターンは次のとおりです。

    ・構造化出力を使用して、コードで検査できる適切に形成されたデータを生成します。たとえば、Agentにタスクをいくつかのカテゴリに分類してもらい、そのカテゴリに基づいて次のAgentを選択するように依頼できます。
    ・1つの出力を次の入力に変換して、複数のAgentを連鎖させる。ブログ投稿を書くなどのタスクは、調査、概要の作成、ブログ投稿の作成、批評、改善など、一連のステップに分解できます。
    ・評価者が出力が特定の基準に合格すると言うまで、評価とフィードバックを提供するAgentと、whileループでタスクを実行するAgentを実行します。
    ・複数のAgentを並行して実行します。たとえば、asyncio.gatherのようなPythonプリミティブを介して実行します。これは、互いに依存しない複数のタスクがある場合の速度に役立ちます。

    examples/agent_patternsにはいくつかの例があります。

    3. Model

    3-1. Model

    「Agents SDK」は、2つのフレーバーのOpenAIモデルをサポートします。

    ・OpenAIResponsesModel
    Responses APIを使用してOpenAI APIを呼び出す。(推奨)

    ・OpenAIChatCompletionsModel
    ChatCompletions APIを使用してOpenAI APIを呼び出す。

    3-2. モデルの混合とマッチング

    1 つのワークフローでは、Agentごとに異なるモデルを使用することができます。たとえば、トリアージにはより小さく、より高速なモデルを使用し、複雑なタスクにはより大きく、より有能なモデルを使用できます。「Agent」を設定する際、次のいずれかの方法で特定のモデルを選択できます。

    (1) OpenAIモデルの名前を渡す。
    (2) 任意のモデル名と、それをモデルインスタンスにマッピングできる ModelProvider を渡す。
    (3) 直接 Model の実装を渡す。

    from agents import Agent, Runner, AsyncOpenAI, OpenAIChatCompletionsModel
    import asyncio
    
    spanish_agent = Agent(
        name="Spanish agent",
        instructions="You only speak Spanish.",
        model="o3-mini", 
    )
    
    english_agent = Agent(
        name="English agent",
        instructions="You only speak English",
        model=OpenAIChatCompletionsModel( 
            model="gpt-4o",
            openai_client=AsyncOpenAI()
        ),
    )
    
    triage_agent = Agent(
        name="Triage agent",
        instructions="Handoff to the appropriate agent based on the language of the request.",
        handoffs=[spanish_agent, english_agent],
        model="gpt-3.5-turbo",
    )
    
    async def main():
        result = await Runner.run(triage_agent, input="Hola, ¿cómo estás?")
        print(result.final_output)

    3-3. LLMプロバイダーの使用

    LLMプロバイダーを使用する方法は3つあります。

    (1) set_default_openai_client を使用
    AsyncOpenAI のインスタンスをグローバルにLLMクライアントとして使用したい場合に便利です。これは、LLM ProvideerがOpenAI互換のAPIエンドポイントを持っており、base_url と api_key を設定できる場合に適用されます。
    例: examples/model_providers/custom_example_global.py

    (2) ModelProvider を Runner.run レベルで指定
    「この実行におけるすべてのAgentでカスタムモデルプロバイダーを使用する」と指定できます。
    例: examples/model_providers/custom_example_provider.py

    (3) Agent.model を使用して特定のAgentにモデルを指定
    異なるAgentごとに異なるプロバイダーを組み合わせて使用することが可能になります。
    例: examples/model_providers/custom_example_agent.py

    また、platform.openai.com から取得したAPIキーを持っていない場合は、set_tracing_disabled() を使用してトレーシングを無効にするか、別のトレーシングプロセッサを設定することを推奨します。

    3-4. 他のLLMプロバイダーを使用する際の一般的な問題

    ・Tracing client error 401
    トレーシングに関連するエラーが発生する場合、これはトレースが OpenAI のサーバーにアップロードされるため、OpenAIのAPIキーがないことが原因です。
    この問題を解決する方法は以下の3つです。

    (1) トレーシングを完全に無効化
    set_tracing_disabled(True)
    (2) トレーシング用の OpenAI API キーを設定
    これはトレースのアップロード専用のAPI キーであり、platform.openai.com から取得する必要があります。
    set_tracing_export_api_key("your_openai_api_key")
    (3) OpenAI 以外のトレースプロセッサを使用
    詳しくは トレーシングのドキュメント を参照してください。

    ・Responses API のサポート
    SDKはデフォルトで Responses API を使用しますが、多くの他の LLM プロバイダーはまだこれをサポートしていません。そのため、404 エラー などの問題が発生する可能性があります。
    解決策として、以下の2つの方法があります。

    (1) set_default_openai_api("chat_completions") を呼び出す
    環境変数 OPENAI_API_KEY と OPENAI_BASE_URL を設定している場合に機能します。
    set_default_openai_api("chat_completions")

    (2) OpenAIChatCompletionsModel を使用
    詳しい例は こちら にあります。

    ・構造化出力のサポート
    一部のモデルプロバイダーは構造化出力をサポートしていません。
    これにより、以下のようなエラーが発生することがあります。

    BadRequestError: Error code: 400 - {'error': {'message': "'response_format.type' : value is not one of the allowed values ['text','json_object']", 'type': 'invalid_request_error'}}

    この問題は、一部のモデルプロバイダーが JSON出力をサポートしているものの、出力に使用する json_schema を指定できないことに起因しています。現在、この問題の修正に取り組んでいますが、JSONスキーマ出力をサポートするプロバイダーを使用することを推奨します。そうしないと、不正なJSONによってアプリケーションが頻繁にエラーを起こす可能性があります。

    次回


     
     

    npaka

     
     
    プログラマー。iPhone / Android / Unity / ROS / AI / AR / VR / RasPi / ロボット / ガジェット。年2冊ペースで技術書を執筆。アニソン / カラオケ / ギター / 猫 twitter : @npaka123

    あなたへのおすすめ