LangChain と統合する

LangChain悪意のある入力の防止、コンテンツの安全性の検証、AI アプリケーション全体でのセキュリティ ポリシーの一貫した適用を行うには、LangChain ワークフローに Model Armor を追加します。

始める前に

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

カスタム イベント処理

可観測性のため、危険なコンテンツが検出されると、実行可能ファイルは 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()]})