שילוב של הגנה מוגברת על המודל עם LangChain

אפשר להשתמש ב-Model Armor עם LangChain כדי לשפר את האבטחה של אפליקציות ה-AI שלכם. המערכת סורקת באופן יזום הנחיות ותשובות של LLM. כדי למנוע קלט זדוני, לאמת את בטיחות התוכן ולאכוף מדיניות אבטחה באופן עקבי באפליקציות ה-AI, כדאי להוסיף את Model Armor לתהליכי העבודה של LangChain.

לפני שמתחילים

לפני שמטמיעים את הקוד, צריך לבצע את שלבי ההגדרה הבאים: Google Cloud

Runnables

כדי להשתמש בהגנה מוגברת על המודל עם 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

דפוסי שימוש

אפשר להשתמש בהגנה מוגברת על המודל עם LangChain באמצעות שרשור בסיסי או באמצעות תוכנת ביניים.

שרשור בסיסי

הדרך הנפוצה ביותר להשתמש בהגנה מוגברת על המודל היא על ידי שרשור של מנקה ההנחיות, המודל ומנקה התשובות באמצעות אופרטור הצינור של 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}")

מחליפים את מה שכתוב בשדות הבאים:

  • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud .
  • ‫LOCATION: המיקום של התבנית.
  • ‫TEMPLATE_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 .
  • ‫LOCATION: המיקום של התבנית.
  • ‫TEMPLATE_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()]})