Guardrails в LangChain | Курс LangChain Agents урок 9

Цель урока: ограничить агента тематикой задачи и защитить персональные данные пользователей от утечки в логи и контекст.

Теория

Что такое guardrails

Агент по умолчанию ответит на любой запрос. Без ограничений он может обсуждать темы вне своей области, повторять чужие персональные данные в логах, или следовать инструкциям из пользовательского ввода (prompt injection).

Guardrails - это барьеры, которые перехватывают запросы или ответы и блокируют нежелательное поведение ещё до того, как оно произошло.

Два подхода:

Детерминированные - regex и проверка по словарю. Надёжные, предсказуемые, не тратят токены. Хорошо подходят для структурированных данных: номера карт, email, IP адреса, ключи API.

Основанные на модели - LLM как судья. Гибкие, понимают смысл и контекст. Нужны когда нельзя описать ограничение формальными правилами: тема разговора, тон, намерение запроса.

Примеры кода

PIIMiddleware: защита персональных данных

LangChain включает готовый middleware для обнаружения и обработки персональных данных (PII). Он перехватывает сообщения до и после вызова модели.

import os
from typing import Any
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import PIIMiddleware, before_model
from langchain_openai import ChatOpenAI
from langgraph.runtime import Runtime

load_dotenv()


@before_model
def debug_input(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """Показывает что именно получит модель после обработки PIIMiddleware."""
    print("Модель получит:", state["messages"][-1].content)
    return None


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        # Кастомный детектор карт: 16 цифр с любыми разделителями
        PIIMiddleware(
            "card",
            detector=r"\d{4}[\s\-]?\d{4}[\s\-]?\d{4}[\s\-]?\d{4}",
            strategy="mask",
            apply_to_input=True,
        ),
        debug_input,
    ],
)

response = agent.invoke({
    "messages": [{"role": "user", "content":
        "Повтори мои данные: email user@example.com, карта 4111-1111-1111-1234."}]
})
print(response["messages"][-1].content)

Встроенные типы: email, credit_card, ip, mac_address, url.

Стратегии обработки:

"redact" - заменяет на [REDACTED_EMAIL]
"mask" - частично скрывает: ****-****-****-1234
"hash" - детерминированный хэш
"block" - выбрасывает исключение

apply_to_input=True проверяет входящие сообщения пользователя. Можно добавить apply_to_output=True для проверки ответов модели и apply_to_tool_results=True для результатов инструментов.

Кастомный детектор PII

Встроенные типы покрывают не все случаи. Для специфических паттернов можно передать свой regex или функцию.

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain_openai import ChatOpenAI

load_dotenv()

model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{20,}",
            strategy="redact",
            apply_to_input=True,
        ),
        PIIMiddleware(
            "phone_ru",
            detector=r"\+?7[\s\-]?\(?\d{3}\)?[\s\-]?\d{3}[\s\-]?\d{2}[\s\-]?\d{2}",
            strategy="mask",
            apply_to_input=True,
        ),
    ],
)

response = agent.invoke({
    "messages": [{"role": "user", "content":
        "Мой ключ sk-abcdef1234567890abcd, телефон +7 (999) 123-45-67"}]
})
print(response["messages"][-1].content)

Кастомный guardrail: блокировка тем

Для ограничения тематики используйте @before_model с параметром can_jump_to=["end"]. Если входящее сообщение нарушает политику, middleware завершает агента с заранее подготовленным ответом.

import os
from typing import Any
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langchain.messages import AIMessage
from langchain_openai import ChatOpenAI
from langgraph.runtime import Runtime

load_dotenv()

ALLOWED_TOPICS = ["langchain", "python", "агент", "инструмент", "модель", "llm", "ai"]
OFF_TOPIC_KEYWORDS = ["политика", "религия", "криптовалюта", "ставки", "казино"]


@before_model(can_jump_to=["end"])
def topic_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    from langchain_core.messages import HumanMessage
    last_message = state["messages"][-1]
    if not isinstance(last_message, HumanMessage):
        return None
    content = str(last_message.content).lower()

    for keyword in OFF_TOPIC_KEYWORDS:
        if keyword in content:
            return {
                "messages": [AIMessage(
                    f"Я специализируюсь на LangChain и разработке AI агентов. "
                    f"Этот вопрос выходит за рамки моей области."
                )],
                "jump_to": "end",
            }
    return None


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[],
    middleware=[topic_guardrail],
)

# Разрешённый запрос
r1 = agent.invoke({"messages": [{"role": "user", "content": "Что такое LangChain?"}]})
print("OK:", r1["messages"][-1].content[:60])

# Заблокированный запрос
r2 = agent.invoke({"messages": [{"role": "user", "content": "Посоветуй ставки на спорт."}]})
print("Заблокирован:", r2["messages"][-1].content)

can_jump_to=["end"] обязателен, без него возврат jump_to игнорируется. Ключ "jump_to": "end" завершает выполнение немедленно, не вызывая модель.

Guardrail на основе модели

Когда тему нельзя описать ключевыми словами, используйте модель как судью. Отдельный быстрый вызов LLM проверяет запрос до основного агента.

import os
from typing import Any
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langchain.messages import AIMessage
from langchain_openai import ChatOpenAI
from langgraph.runtime import Runtime

load_dotenv()

judge_model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"), temperature=0)


@before_model(can_jump_to=["end"])
def llm_topic_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    from langchain_core.messages import HumanMessage
    last_message = state["messages"][-1]
    if not isinstance(last_message, HumanMessage):
        return None
    content = str(last_message.content)

    verdict = judge_model.invoke([
        {"role": "system", "content":
            "Ты - фильтр запросов. Отвечай ТОЛЬКО словом ALLOWED или BLOCKED.\n"
            "ALLOWED: вопросы о Python, LangChain, AI, программировании.\n"
            "BLOCKED: всё остальное."},
        {"role": "user", "content": content},
    ])

    if "BLOCKED" in verdict.content:
        return {
            "messages": [AIMessage(
                "Я специализируюсь на LangChain и разработке AI агентов. "
                "Обращайтесь с вопросами по этой теме."
            )],
            "jump_to": "end",
        }
    return None


model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

agent = create_agent(
    model=model,
    tools=[],
    middleware=[llm_topic_guardrail],
)

r1 = agent.invoke({"messages": [{"role": "user", "content": "Как создать агента в LangChain?"}]})
print("OK:", r1["messages"][-1].content[:60])

r2 = agent.invoke({"messages": [{"role": "user", "content": "Расскажи про историю Рима."}]})
print("Заблокирован:", r2["messages"][-1].content)

Недостаток: каждый запрос тратит дополнительные токены на вызов модели-судьи. Используйте LLM guardrail только для тех случаев, где regex недостаточен.

Сквозной проект: агент-аналитик с guardrails

import asyncio
import os
import sys
from pathlib import Path
from typing import Any
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import (
    PIIMiddleware,
    SummarizationMiddleware,
    ModelCallLimitMiddleware,
    before_model,
)
from langchain.messages import AIMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime

load_dotenv()

SERVER_PATH = str(Path(__file__).parent.parent / "agents_course" / "tools_server.py")

OFF_TOPIC_KEYWORDS = ["политика", "религия", "криптовалюта", "ставки", "казино"]


@before_model(can_jump_to=["end"])
def topic_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    from langchain_core.messages import HumanMessage
    last_message = state["messages"][-1]
    if not isinstance(last_message, HumanMessage):
        return None
    content = str(last_message.content).lower()
    for keyword in OFF_TOPIC_KEYWORDS:
        if keyword in content:
            return {
                "messages": [AIMessage(
                    "Я аналитик данных. Обращайтесь с аналитическими задачами."
                )],
                "jump_to": "end",
            }
    return None


async def main():
    client = MultiServerMCPClient({
        "analyst_tools": {
            "transport": "stdio",
            "command": sys.executable,
            "args": [SERVER_PATH],
        }
    })

    tools = await client.get_tools()
    model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))
    summary_model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

    agent = create_agent(
        model=model,
        tools=tools,
        checkpointer=InMemorySaver(),
        system_prompt="Ты агент-аналитик данных.",
        middleware=[
            topic_guardrail,
            PIIMiddleware("email", strategy="redact", apply_to_input=True),
            PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
            SummarizationMiddleware(
                model=summary_model,
                trigger=("messages", 12),
                keep=("messages", 4),
            ),
            ModelCallLimitMiddleware(run_limit=10, exit_behavior="end"),
        ],
    )

    config = {"configurable": {"thread_id": "analyst-final"}}

    # Разрешённый запрос
    r1 = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Найди информацию об Anthropic."}]},
        config,
    )
    print("Аналитик:", r1["messages"][-1].content[:80])

    # Заблокированный запрос
    r2 = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Посоветуй ставки на спорт."}]},
        {"configurable": {"thread_id": "analyst-blocked"}},
    )
    print("Guardrail:", r2["messages"][-1].content)

if __name__ == "__main__":
    asyncio.run(main())

Частые ошибки

can_jump_to не указан

# Неправильно: jump_to в возвращаемом словаре молча игнорируется
@before_model
def my_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    return {"messages": [...], "jump_to": "end"}  # не сработает

# Правильно: явно указать разрешённые переходы
@before_model(can_jump_to=["end"])
def my_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    return {"messages": [...], "jump_to": "end"}

LLM guardrail на каждый запрос

# Дорого: LLM-судья на каждом запросе
middleware=[llm_topic_guardrail, ...]

# Лучше: сначала быстрая проверка по ключевым словам,
# LLM только для пограничных случаев
@before_model(can_jump_to=["end"])
def combined_guardrail(state, runtime):
    content = state["messages"][-1].content.lower()
    # Сначала быстрая детерминированная проверка
    for kw in HARD_BLOCKED:
        if kw in content:
            return {"messages": [AIMessage("Недопустимый запрос.")], "jump_to": "end"}
    # Только для неоднозначных случаев LLM
    if any(kw in content for kw in AMBIGUOUS):
        # вызов LLM...
        pass
    return None

Задание

Добавьте к агенту из урока 8 два guardrail:

PIIMiddleware для типа "phone_ru" с кастомным regex для российских номеров телефонов, стратегия "mask"
Детерминированный @before_model guardrail, который блокирует запросы содержащие слова из списка по вашему выбору

Убедитесь что guardrail срабатывает до вызова модели: в LangSmith должен быть виден короткий тред без вызова LLM при заблокированном запросе.

<< урок 8

урок 10 >>

Подписывайтесь на мой Telegram канал

Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия