Integrar com o LangChain

Use o Model Armor com LangChain para melhorar a segurança dos aplicativos de IA, verificando proativamente comandos e respostas de LLM. Para ajudar a evitar entradas maliciosas, verificar a segurança do conteúdo e aplicar políticas de segurança de maneira consistente em todos os aplicativos de IA, adicione o Model Armor aos fluxos de trabalho do LangChain.

Antes de começar

Antes de implementar o código, conclua as etapas de Google Cloud configuração a seguir:

Executáveis

Para usar o Model Armor com o LangChain, você precisa de dois executáveis principais que se encaixam na interface padrão do LangChain.

  • ModelArmorSanitizePromptRunnable: esse componente verifica a entrada do usuário (comandos) antes que ela chegue ao LLM. Ele detecta entradas maliciosas ou violações de política.

  • ModelArmorSanitizeResponseRunnable: esse componente verifica a saída gerada pelo LLM antes de ser retornada ao usuário. As modificações feitas pelo usuário na resposta após a verificação de segurança principal não são filtradas e chegam a você no estado bruto.

Instalar os pacotes Python necessários

Instale os seguintes pacotes Python:

  • langchain-google-community versão 3.0.4 ou mais recente
  • langchain-google-genai
pip install langchain-google-community langchain-google-genai

Padrões de uso

É possível usar o Model Armor com o LangChain por encadeamento básico ou usando middleware.

Encadeamento básico

A maneira mais comum de usar o Model Armor é encadeando o higienizador de comandos, o modelo e o higienizador de respostas usando o operador de pipe da linguagem de expressão do LangChain (LCEL, na sigla em inglês).

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

Substitua:

  • PROJECT_ID: o ID do Google Cloud projeto.
  • LOCATION: o local do modelo.
  • TEMPLATE_ID: o ID do modelo do Model Armor criado em Antes de começar.
  • SANITIZE_PROMPT_ERROR_HANDLING: um booleano que controla o tratamento de erros para ModelArmorSanitizePromptRunnable; True ou False.

  • SANITIZE_RESPONSE_ERROR_HANDLING: um booleano que controla o tratamento de erros para ModelArmorSanitizeResponseRunnable; True ou False.

    Ao lidar com erros, o Model Armor se comporta da seguinte maneira:

    • True: registra um aviso, mas permite que o conteúdo seja aprovado mesmo que os riscos sejam detectados.
    • False: gera um ValueError e bloqueia a execução quando conteúdo não seguro é detectado.

Middleware para agentes

Para agentes do LangChain que usam ferramentas, é possível usar ModelArmorMiddleware. Esse middleware aplica o executável de higienizador relevante em diferentes pontos de interação que o agente do LangChain passa ao usar ferramentas.

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

Substitua:

  • PROJECT_ID: o ID do Google Cloud projeto.
  • LOCATION: o local do modelo.
  • TEMPLATE_ID: o ID do modelo do Model Armor criado em Antes de começar.
  • SANITIZE_PROMPT_ERROR_HANDLING: um booleano que controla o tratamento de erros para ModelArmorSanitizePromptRunnable; True ou False.

  • SANITIZE_RESPONSE_ERROR_HANDLING: um booleano que controla o tratamento de erros para ModelArmorSanitizeResponseRunnable; True ou False.

    Ao lidar com erros, o Model Armor se comporta da seguinte maneira:

    • True: registra um aviso, mas permite que o conteúdo seja aprovado mesmo que os riscos sejam detectados.
    • False: gera um ValueError e bloqueia a execução quando conteúdo não seguro é detectado.

Essa configuração garante que, mesmo quando o agente estiver executando o raciocínio em várias etapas ou o uso de ferramentas, as entradas e saídas sejam verificadas.

Verificação de documentos

É possível usar executáveis do Model Armor para verificar documentos enviados antes de processá-los. Isso é essencial para fluxos de trabalho como a sumarização, em que o documento de origem pode conter conteúdo malicioso oculto. É possível carregar o conteúdo do arquivo e transmiti-lo pelo ModelArmorSanitizePromptRunnable antes de enviá-lo ao LLM.

Tratamento de eventos personalizados

Para observabilidade, os executáveis enviam um evento on_model_armor_finding quando conteúdo não seguro é detectado. É possível anexar um gerenciador de callback personalizado para registrar detalhes específicos (como descobertas e conteúdo) ou acionar alertas sem falhar o fluxo do aplicativo.

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