LangChain と統合する

LangChain で Model Armor を使用して、LLM のプロンプトとレスポンスを事前にスクリーニングすることで、AI アプリケーションのセキュリティを強化します。悪意のある入力を防ぎ、コンテンツの安全性を検証し、AI アプリケーション全体でセキュリティ ポリシーを一貫して適用するには、LangChain ワークフローに Model Armor を追加します。

始める前に

コードを実装する前に、次の Google Cloud 設定手順を完了します。

  • プロジェクトを選択または作成して、課金を有効にします。
  • プロジェクトで Model Armor API を有効にします。
  • gcloud を使用するか、GOOGLE_APPLICATION_CREDENTIALS 環境変数を設定して認証を設定します。
  • サービス アカウントに Model Armor ユーザー（roles/modelarmor.user）ロールがあることを確認します。テンプレートを管理するには、Model Armor 管理者（roles/modelarmor.admin）ロールが必要です。
  • Google Cloud コンソールでテンプレートを作成します。

実行可能ファイル

LangChain で Model Armor を使用するには、標準の LangChain インターフェースに適合する 2 つのプライマリ実行可能ファイルが必要です。

  • ModelArmorSanitizePromptRunnable: このコンポーネントは、ユーザー入力（プロンプト）が LLM に到達する前にスクリーニングします。悪意のある入力やポリシー違反を検出します。

  • ModelArmorSanitizeResponseRunnable: このコンポーネントは、LLM が生成した出力をユーザーに返す前にスクリーニングします。ユーザーがプライマリ セキュリティ チェック後に回答に加えた変更はフィルタされず、未加工の状態で届きます。

必要な Python パッケージをインストールする

次の Python パッケージをインストールします。

  • langchain-google-community バージョン 3.0.4 以降
  • langchain-google-genai
pip install langchain-google-community langchain-google-genai

使用パターン

Model Armor を LangChain で使用するには、基本的なチェーン処理またはミドルウェアを使用します。

基本的なチェーン

Model Armor の最も一般的な使用方法は、LangChain Expression Language（LCEL）パイプ演算子を使用して、プロンプト サニタイザー、モデル、レスポンス サニタイザーを連結することです。

from langchain_google_community.model_armor import (
    ModelArmorSanitizePromptRunnable,
    ModelArmorSanitizeResponseRunnable,
)
from langchain_google_genai import ChatGoogleGenerativeAI

# Initialize Model Armor runnables
sanitize_prompt = ModelArmorSanitizePromptRunnable(
    project="PROJECT_ID",
    location="LOCATION",
    template_id="TEMPLATE_ID",
    fail_open="SANITIZE_PROMPT_ERROR_HANDLING"
)

sanitize_response = ModelArmorSanitizeResponseRunnable(
    project="PROJECT_ID",
    location="LOCATION",
    template_id="TEMPLATE_ID",
    fail_open="SANITIZE_RESPONSE_ERROR_HANDLING"
)

# Initialize the model
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", <var>...</var>)

# Build the chain
chain = sanitize_prompt | llm | sanitize_response

# Invoke
try:
    result = chain.invoke("<var>YOUR_USER_PROMPT_HERE</var>")
    print(result)
except ValueError as e:
    print(f"Content blocked: {e}")

次のように置き換えます。

  • PROJECT_ID: 実際の Google Cloud プロジェクト ID。
  • LOCATION: テンプレートのロケーション。
  • TEMPLATE_ID: 始める前にで作成した Model Armor テンプレートの ID。
  • SANITIZE_PROMPT_ERROR_HANDLING: ModelArmorSanitizePromptRunnable、True、False のエラー処理を制御するブール値。

  • SANITIZE_RESPONSE_ERROR_HANDLING: ModelArmorSanitizeResponseRunnable、True、False のエラー処理を制御するブール値。

    エラーを処理する場合、Model Armor は次のように動作します。

    • True: 警告をログに記録しますが、リスクが検出されてもコンテンツを通過させます。
    • False: 安全でないコンテンツが検出されると、ValueError を発生させ、実行をブロックします。

エージェントのミドルウェア

ツールを使用する LangChain エージェントの場合は、ModelArmorMiddleware を使用できます。このミドルウェアは、LangChain エージェントがツールを使用するときに通過するさまざまなインタラクション ポイントで、関連するサニタイザー実行可能ファイルを適用します。

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_google_genai import ChatGoogleGenerativeAI

from langchain_google_community.model_armor import (
    ModelArmorMiddleware,
    ModelArmorSanitizePromptRunnable,
    ModelArmorSanitizeResponseRunnable,
)

# Create Model Armor sanitizers (same configuration as earlier sections)
prompt_sanitizer = ModelArmorSanitizePromptRunnable(
    project="PROJECT_ID",
    location="LOCATION",
    template_id="TEMPLATE_ID",
    fail_open="SANITIZE_PROMPT_ERROR_HANDLING",
)

response_sanitizer = ModelArmorSanitizeResponseRunnable(
    project="PROJECT_ID",
    location="LOCATION",
    template_id="TEMPLATE_ID",
    fail_open="SANITIZE_RESPONSE_ERROR_HANDLING",
)

# Wrap them in the official Model Armor middleware
model_armor_middleware = ModelArmorMiddleware(
    prompt_sanitizer=prompt_sanitizer,
    response_sanitizer=response_sanitizer,
)

# --- Create agent with Model Armor middleware ---
agent = create_agent(
    model="SOME_GEMINI_VERSION",
    tools=[],
    middleware=[model_armor_middleware],
)

次のように置き換えます。

  • PROJECT_ID: 実際の Google Cloud プロジェクト ID。
  • LOCATION: テンプレートのロケーション。
  • TEMPLATE_ID: 始める前にで作成した Model Armor テンプレートの ID。
  • SANITIZE_PROMPT_ERROR_HANDLING: ModelArmorSanitizePromptRunnable、True、False のエラー処理を制御するブール値。

  • SANITIZE_RESPONSE_ERROR_HANDLING: ModelArmorSanitizeResponseRunnable、True、False のエラー処理を制御するブール値。

    エラーを処理する場合、Model Armor は次のように動作します。

    • True: 警告をログに記録しますが、リスクが検出されてもコンテンツを通過させます。
    • False: 安全でないコンテンツが検出されると、ValueError を発生させ、実行をブロックします。

この設定により、エージェントがマルチステップの推論やツールの使用を実行している場合でも、入力と出力がスクリーニングされます。

ドキュメントのスクリーニング

Model Armor 実行可能ファイルを使用すると、アップロードされたドキュメントを処理前にスクリーニングできます。これは、ソースドキュメントに隠れた悪意のあるコンテンツが含まれている可能性がある要約などのワークフローに不可欠です。ファイル コンテンツを読み込んで ModelArmorSanitizePromptRunnable を通してから、LLM に送信できます。

カスタム イベント処理

オブザーバビリティのために、安全でないコンテンツが検出されると、実行可能ファイルは on_model_armor_finding イベントをディスパッチします。カスタム コールバック ハンドラをログにアタッチして、特定の詳細（検出結果やコンテンツなど）を記録したり、アプリケーション フローをクラッシュさせることなくアラートをトリガーしたりできます。

from langchain_core.callbacks import BaseCallbackHandler

class SecurityHandler(BaseCallbackHandler):
    def on_custom_event(self, name, data, **kwargs):
        if name == "on_model_armor_finding":
            print(f"Alert: {data['findings']} detected in {data['template_id']}")

# Attach to config
chain.invoke(user_input, config={"callbacks": [SecurityHandler()]})