In LangChain einbinden

Verwenden Sie Model Armor mit LangChain, um die Sicherheit Ihrer KI-Anwendungen zu verbessern, indem Sie LLM-Prompts und ‑Antworten proaktiv prüfen. Um schädliche Eingaben zu verhindern, die Inhaltssicherheit zu überprüfen und Sicherheitsrichtlinien in Ihren KI-Anwendungen einheitlich durchzusetzen, fügen Sie Model Armor in Ihre LangChain-Workflows ein.

Hinweis

Führen Sie vor der Implementierung des Codes die folgenden Google Cloud Einrichtungsschritte aus:

Runnables

Wenn Sie Model Armor mit LangChain verwenden möchten, benötigen Sie zwei primäre Runnables, die in die standardmäßige LangChain-Schnittstelle passen.

  • ModelArmorSanitizePromptRunnable: Diese Komponente prüft Nutzereingaben (Prompts), bevor sie das LLM erreichen. Es erkennt schädliche Eingaben oder Richtlinienverstöße.

  • ModelArmorSanitizeResponseRunnable: Diese Komponente prüft die vom LLM generierte Ausgabe, bevor sie an den Nutzer zurückgegeben wird. Alle Änderungen, die der Nutzer nach der primären Sicherheitsprüfung an der Antwort vornimmt, werden nicht gefiltert und erreichen Sie im Rohzustand.

Erforderliche Python-Pakete installieren

Installieren Sie die folgenden Python-Pakete:

  • langchain-google-community-Version 3.0.4 oder höher
  • langchain-google-genai
pip install langchain-google-community langchain-google-genai

Nutzungsmuster

Sie können Model Armor mit LangChain entweder durch grundlegende Verkettung oder durch Verwendung von Middleware verwenden.

Einfaches Chaining

Model Armor wird am häufigsten verwendet, indem der Prompt-Sanitizer, das Modell und der Antwort-Sanitizer mit dem Pipe-Operator der LangChain Expression Language (LCEL) verkettet werden.

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}")

Ersetzen Sie Folgendes:

  • PROJECT_ID: Projekt-ID in Google Cloud .
  • LOCATION: Der Speicherort der Vorlage.
  • TEMPLATE_ID: die ID der Model Armor-Vorlage, die Sie unter Vorbereitung erstellt haben.
  • SANITIZE_PROMPT_ERROR_HANDLING: Ein boolescher Wert, der die Fehlerbehandlung für ModelArmorSanitizePromptRunnable steuert; True oder False.

  • SANITIZE_RESPONSE_ERROR_HANDLING: Ein boolescher Wert, der die Fehlerbehandlung für ModelArmorSanitizeResponseRunnable steuert; True oder False.

    So geht Model Armor mit Fehlern um:

    • True: Es wird eine Warnung protokolliert, aber Inhalte werden auch dann zugelassen, wenn Risiken erkannt werden.
    • False: Löst eine ValueError aus und blockiert die Ausführung, wenn unsichere Inhalte erkannt werden.

Middleware für KI-Agenten

Für LangChain-Agents, die Tools verwenden, können Sie ModelArmorMiddleware verwenden. Diese Middleware wendet den entsprechenden Sanitizer-Runnable an verschiedenen Interaktionspunkten an, die der LangChain-Agent bei der Verwendung von Tools durchläuft.

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],
)

Ersetzen Sie Folgendes:

  • PROJECT_ID: Projekt-ID in Google Cloud .
  • LOCATION: Der Speicherort der Vorlage.
  • TEMPLATE_ID: die ID der Model Armor-Vorlage, die Sie unter Vorbereitung erstellt haben.
  • SANITIZE_PROMPT_ERROR_HANDLING: Ein boolescher Wert, der die Fehlerbehandlung für ModelArmorSanitizePromptRunnable steuert; True oder False.

  • SANITIZE_RESPONSE_ERROR_HANDLING: Ein boolescher Wert, der die Fehlerbehandlung für ModelArmorSanitizeResponseRunnable steuert; True oder False.

    So geht Model Armor mit Fehlern um:

    • True: Es wird eine Warnung protokolliert, aber Inhalte werden auch dann zugelassen, wenn Risiken erkannt werden.
    • False: Löst eine ValueError aus und blockiert die Ausführung, wenn unsichere Inhalte erkannt werden.

So wird sichergestellt, dass Ein- und Ausgaben auch dann überprüft werden, wenn der Agent mehrstufige Schlussfolgerungen zieht oder Tools verwendet.

Dokumentprüfung

Sie können Model Armor-Runnables verwenden, um hochgeladene Dokumente zu prüfen, bevor Sie sie verarbeiten. Dies ist für Workflows wie die Zusammenfassung unerlässlich, da das Quelldokument möglicherweise verborgene schädliche Inhalte enthält. Sie können den Dateiinhalt laden und ihn durch ModelArmorSanitizePromptRunnable leiten, bevor Sie ihn an das LLM senden.

Benutzerdefinierte Ereignisverarbeitung

Zur Beobachtbarkeit senden die Runnables ein on_model_armor_finding-Ereignis, wenn unsichere Inhalte erkannt werden. Sie können einen benutzerdefinierten Callback-Handler anhängen, um bestimmte Details (z. B. Ergebnisse und Inhalte) zu protokollieren oder Benachrichtigungen auszulösen, ohne den Anwendungsablauf zu unterbrechen.

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()]})