与 LangChain 集成

将 Model Armor 与 LangChain 搭配使用，通过主动过滤 LLM 提示和回答来增强 AI 应用的安全性。如需帮助防止恶意输入、验证内容安全性并在 AI 应用中以一致的方式实施安全政策，请将 Model Armor 添加到 LangChain 工作流中。

准备工作

在实现代码之前，请完成以下 Google Cloud 设置步骤：

可运行对象

如需将 Model Armor 与 LangChain 搭配使用，您需要两个主要的可运行对象，它们符合标准 LangChain 接口。

  • 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 表达式语言 (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()]})