搭配 LangChain 使用 Model Armor,主動篩選 LLM 提示詞和回覆,提升 AI 應用程式的安全性。為防範惡意輸入內容、驗證內容安全性,並在 AI 應用程式中持續強制執行安全政策,請在 LangChain 工作流程中加入 Model Armor。
事前準備
實作程式碼前,請先完成下列 Google Cloud 設定步驟:
- 選取或建立專案,並啟用帳單。
- 在專案中啟用 Model Armor API。
- 使用
gcloud或設定GOOGLE_APPLICATION_CREDENTIALS環境變數來設定驗證。 - 請確認服務帳戶具備「Model Armor 使用者」(
roles/modelarmor.user) 角色。如要管理範本,您必須有「Model Armor 管理員」(roles/modelarmor.admin) 角色 - 在 Google Cloud 控制台中建立範本。
可執行檔
如要搭配使用 Model Armor 與 LangChain,您需要兩個主要可執行檔,且必須符合標準 LangChain 介面。
ModelArmorSanitizePromptRunnable:這個元件會在使用者輸入內容 (提示) 傳送至 LLM 之前進行篩選。偵測惡意輸入內容或違反政策的行為。ModelArmorSanitizeResponseRunnable:這個元件會先檢查 LLM 生成的輸出內容,再傳回給使用者。使用者在主要安全性檢查後對回覆所做的任何修改,都不會經過篩選,而是以原始狀態傳送給您。
安裝必要的 Python 套件
安裝下列 Python 套件:
langchain-google-community3.0.4 以上版本langchain-google-genai
pip install langchain-google-community langchain-google-genai
使用模式
您可以透過基本鏈結或使用中介軟體,搭配 LangChain 使用 Model Armor。
基本鏈結
使用 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 可執行檔,在處理上傳的文件之前先進行篩選。這對於摘要等工作流程至關重要,因為來源文件可能含有隱藏的惡意內容。您可以載入檔案內容,並透過 ModelArmorSanitizePromptRunnable 傳遞內容,再傳送至 LLM。
自訂事件處理
為方便觀察,可執行檔會在偵測到不安全內容時,分派 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()]})