Integración con LangChain

Usa Model Armor con LangChain para mejorar la seguridad de tus aplicaciones de IA mediante el análisis proactivo de las instrucciones y respuestas del LLM. Para evitar entradas maliciosas, verificar la seguridad del contenido y aplicar políticas de seguridad de manera coherente en tus aplicaciones de IA, agrega Model Armor a tus flujos de trabajo de LangChain.

Antes de comenzar

Antes de implementar el código, completa los siguientes Google Cloud pasos de configuración:

Ejecutables

Para usar Model Armor con LangChain, necesitas dos ejecutables principales que se ajusten a la interfaz estándar de LangChain.

  • ModelArmorSanitizePromptRunnable: Este componente analiza la entrada del usuario (instrucciones) antes de que llegue al LLM. Detecta entradas maliciosas o incumplimientos de políticas.

  • ModelArmorSanitizeResponseRunnable: Este componente analiza el resultado generado por el LLM antes de que se devuelva al usuario. Las modificaciones que el usuario realice en la respuesta después de la verificación de seguridad principal no se filtran y te llegarán en su estado sin procesar.

Instala los paquetes obligatorios de Python

Instala los siguientes paquetes de Python:

  • langchain-google-community versión 3.0.4 o posterior
  • langchain-google-genai
pip install langchain-google-community langchain-google-genai

Patrones de uso

Puedes usar Model Armor con LangChain mediante el encadenamiento básico o el uso de middleware.

Encadenamiento básico

La forma más común de usar Model Armor es encadenar el sanitizador de instrucciones, el modelo y el sanitizador de respuestas con el operador de canalización de 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}")

Reemplaza lo siguiente:

  • PROJECT_ID: Es el ID del Google Cloud proyecto de.
  • LOCATION: la ubicación de la plantilla.
  • TEMPLATE_ID: Es el ID de la plantilla de Model Armor creada en Antes de comenzar.
  • SANITIZE_PROMPT_ERROR_HANDLING: Es un valor booleano que controla el manejo de errores para ModelArmorSanitizePromptRunnable; True o False.

  • SANITIZE_RESPONSE_ERROR_HANDLING: Es un valor booleano que controla el manejo de errores para ModelArmorSanitizeResponseRunnable; True o False.

    Cuando se manejan errores, Model Armor se comporta de la siguiente manera:

    • True: Registra una advertencia, pero permite que el contenido pase incluso si se detectan riesgos.
    • False: Genera un ValueError y bloquea la ejecución cuando se detecta contenido inseguro.

Middleware para agentes

Para los agentes de LangChain que usan herramientas, puedes usar ModelArmorMiddleware. Este middleware aplica el ejecutable de sanitizador pertinente en diferentes puntos de interacción por los que pasa el agente de LangChain cuando usa herramientas.

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

Reemplaza lo siguiente:

  • PROJECT_ID: Es el ID del Google Cloud proyecto de.
  • LOCATION: la ubicación de la plantilla.
  • TEMPLATE_ID: Es el ID de la plantilla de Model Armor creada en Antes de comenzar.
  • SANITIZE_PROMPT_ERROR_HANDLING: Es un valor booleano que controla el manejo de errores para ModelArmorSanitizePromptRunnable; True o False.

  • SANITIZE_RESPONSE_ERROR_HANDLING: Es un valor booleano que controla el manejo de errores para ModelArmorSanitizeResponseRunnable; True o False.

    Cuando se manejan errores, Model Armor se comporta de la siguiente manera:

    • True: Registra una advertencia, pero permite que el contenido pase incluso si se detectan riesgos.
    • False: Genera un ValueError y bloquea la ejecución cuando se detecta contenido inseguro.

Esta configuración garantiza que, incluso cuando el agente ejecuta un razonamiento de varios pasos o el uso de herramientas, se analicen las entradas y salidas.

Análisis de documentos

Puedes usar ejecutables de Model Armor para analizar los documentos subidos antes de procesarlos. Esto es esencial para flujos de trabajo como la generación de resúmenes, en los que el documento fuente puede contener contenido malicioso oculto. Puedes cargar el contenido del archivo y pasarlo por ModelArmorSanitizePromptRunnable antes de enviarlo al LLM.

Control de eventos personalizados

Para la observabilidad, los ejecutables envían un evento on_model_armor_finding cuando se detecta contenido inseguro. Puedes adjuntar un controlador de devolución de llamada personalizado para registrar detalles específicos (como hallazgos y contenido) o activar alertas sin bloquear el flujo de la aplicación.

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