Middleware в LangChain 1.0: практический гайд по созданию продакшен-агентов с управлением контекстом и PII-защитой

Почему ваш LangChain-агент сливает данные и теряет контекст

Вы создали умного агента на LangChain. Он отвечает на вопросы, выполняет задачи, даже что-то запоминает. Вы запускаете его в продакшен. И тут начинается.

Пользователь спрашивает: "Сколько у меня денег на счете?" Агент радостно отвечает: "У вас, Иван Иванович (паспорт 4505 123456), на счете 123 456 рублей в Сбербанке (карта 4276 3800 1234 5678)". Все данные улетели в OpenAI. GDPR плачет в уголке.

Другой сценарий: агент обрабатывает длинную переписку. На 20-м сообщении он "забывает", что пользователь говорил в начале. Контекст переполнен, важная информация потеряна. Почему AI-агенты ломаются в продакшене? Потому что вы не контролируете что происходит между вызовами LLM.

Middleware в LangChain: не просто декоратор

Middleware в LangChain 1.0 — это не "еще одна фича". Это архитектурный паттерн, который меняет всё. Представьте цепочку вызовов:

User Input → Agent → LLM → Output

С middleware эта цепочка превращается в:

User Input → [Middleware 1] → [Middleware 2] → Agent → [Middleware 3] → LLM → [Middleware 4] → Output

Каждый middleware может:

• Модифицировать промпт перед отправкой в LLM
• Фильтровать или маскировать чувствительные данные
• Логировать все взаимодействия
• Внедрять human-in-the-loop проверки
• Управлять контекстом (что оставить, что выбросить)

Проблема в том, что документация LangChain показывает базовые примеры. А в реальном продакшене нужны сложные цепочки middleware с зависимостями между ними.

Собираем продакшен-агента с нуля

Давайте создадим финансового ассистента, который:

1. Фильтрует PII-данные (номера карт, паспортов, телефонов)
2. Управляет контекстом (не дает ему "распухнуть")
3. Просит подтверждение перед опасными операциями
4. Логирует все запросы для аудита

1 Устанавливаем окружение и импорты

Сначала подготовим всё необходимое:

# requirements.txt
langchain==0.1.0
langchain-openai==0.0.5
pydantic==2.5.0
python-dotenv==1.0.0
presidio-analyzer==2.2.0
presidio-anonymizer==2.2.0

import os
from typing import Any, Dict, List, Optional
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.middleware import RunnableLambda
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
import re
import json

2 Создаем PII-фильтр (самый важный middleware)

Вот как НЕ надо делать:

# ПЛОХО: наивная замена
class BadPIIFilter:
    def replace_pii(self, text):
        text = text.replace("карта", "[КАРТА]")
        text = text.replace("паспорт", "[ДОКУМЕНТ]")
        return text

Почему это плохо? Потому что пользователь может сказать "карточка", "кредитка", "пластик". Или написать номер паспорта без слова "паспорт".

Вот правильная реализация с Presidio (Microsoft's PII detection):

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

class PIIMiddleware:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()
        
        # Какие типы PII ищем
        self.entity_types = [
            "CREDIT_CARD",
            "PHONE_NUMBER",
            "PERSON",
            "IBAN_CODE",
            "US_PASSPORT",
            "US_SSN",
            "EMAIL_ADDRESS",
            "LOCATION"
        ]
        
        # Как маскируем (можно менять)
        self.operators = {
            "CREDIT_CARD": OperatorConfig("replace", {"new_value": "[НОМЕР_КАРТЫ]"}),
            "PHONE_NUMBER": OperatorConfig("replace", {"new_value": "[ТЕЛЕФОН]"}),
            "PERSON": OperatorConfig("replace", {"new_value": "[ИМЯ]"}),
        }
    
    def __call__(self, input_dict: Dict[str, Any]) -> Dict[str, Any]:
        """Основной метод middleware"""
        if "messages" in input_dict:
            # Обрабатываем все сообщения в цепочке
            processed_messages = []
            for msg in input_dict["messages"]:
                if hasattr(msg, "content"):
                    processed_content = self._anonymize_text(msg.content)
                    # Создаем новое сообщение с обработанным контентом
                    new_msg = type(msg)(content=processed_content)
                    processed_messages.append(new_msg)
                else:
                    processed_messages.append(msg)
            
            return {"messages": processed_messages}
        
        return input_dict
    
    def _anonymize_text(self, text: str) -> str:
        """Анонимизация текста"""
        if not text:
            return text
            
        # Анализируем текст на наличие PII
        results = self.analyzer.analyze(
            text=text,
            entities=self.entity_types,
            language="ru"
        )
        
        # Анонимизируем
        anonymized = self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators=self.operators
        )
        
        return anonymized.text

3 Middleware для управления контекстом

Самая частая ошибка — отправлять весь исторический контекст в LLM. Токены стоят денег, а модель может "потерять" важное в море старого текста.

Решение — умное сжатие контекста:

class ContextManagerMiddleware:
    def __init__(self, max_tokens: int = 4000, summary_ratio: float = 0.3):
        self.max_tokens = max_tokens
        self.summary_ratio = summary_ratio  # Какую часть старого контекста суммировать
        
    def __call__(self, input_dict: Dict[str, Any]) -> Dict[str, Any]:
        if "messages" not in input_dict:
            return input_dict
            
        messages = input_dict["messages"]
        
        # Оцениваем количество токенов (грубая оценка)
        total_chars = sum(len(str(msg.content)) for msg in messages if hasattr(msg, "content"))
        estimated_tokens = total_chars // 4  # Примерная оценка
        
        if estimated_tokens <= self.max_tokens:
            return input_dict  # Всё в порядке
            
        # Нужно сжать контекст
        return self._compress_context(messages, input_dict)
    
    def _compress_context(self, messages, original_input):
        """Сжимаем старые сообщения, оставляя новые нетронутыми"""
        if len(messages) <= 2:
            return original_input
            
        # Новые сообщения оставляем как есть
        new_messages = messages[-2:]  # Последние 2 сообщения
        
        # Старые сообщения суммируем
        old_messages = messages[:-2]
        old_text = "\n".join([str(msg.content) for msg in old_messages if hasattr(msg, "content")])
        
        # Создаем промпт для суммирования
        summary_prompt = f"""Суммируй следующий диалог, сохраняя:
        1. Ключевые факты
        2. Принятые решения
        3. Важные детали о пользователе
        4. Контекст текущей задачи
        
        Диалог:
        {old_text}
        
        Краткое содержание:"""
        
        # Здесь нужно вызвать LLM для суммирования
        # Для простоты пока возвращаем урезанный контекст
        summary_msg = SystemMessage(
            content=f"[СЖАТЫЙ КОНТЕКСТ] Предыдущий диалог содержал {len(old_messages)} сообщений. Ключевые моменты сохранены."
        )
        
        # Возвращаем сжатый контекст + новые сообщения
        return {
            "messages": [summary_msg] + new_messages,
            **{k: v for k, v in original_input.items() if k != "messages"}
        }

Но это наивная реализация. В реальном продакшене нужно:

1. Использовать точный подсчет токенов (tiktoken для OpenAI)
2. Суммировать через ту же LLM, но с меньшим контекстом
3. Сохранять структурированные данные (даты, числа, имена) отдельно

4 Human-in-the-loop middleware

Некоторые операции слишком опасны для полной автоматизации. Перевод денег, изменение настроек, удаление данных — требуют человеческого подтверждения.

class HumanApprovalMiddleware:
    def __init__(self, approval_callback=None):
        """
        approval_callback: функция, которая возвращает bool (подтверждено/нет)
        В продакшене это может быть HTTP-вызов к UI
        """
        self.approval_callback = approval_callback or self._default_callback
        
        # Какие действия требуют подтверждения
        self.dangerous_patterns = [
            r"переведи\s+деньги",
            r"удали\s+данные",
            r"измени\s+настройки",
            r"купи\s+акции",
            r"продай\s+акции",
            r"подпишись\s+на",
        ]
    
    def __call__(self, input_dict: Dict[str, Any]) -> Dict[str, Any]:
        """Проверяем, не опасный ли запрос"""
        if "messages" not in input_dict:
            return input_dict
            
        last_message = input_dict["messages"][-1]
        if not hasattr(last_message, "content"):
            return input_dict
            
        user_input = str(last_message.content).lower()
        
        # Проверяем паттерны
        requires_approval = False
        for pattern in self.dangerous_patterns:
            if re.search(pattern, user_input):
                requires_approval = True
                break
        
        if not requires_approval:
            return input_dict
        
        # Требуем подтверждения
        approved = self.approval_callback(user_input)
        
        if not approved:
            # Прерываем цепочку, возвращаем отказ
            raise ValueError("Действие требует подтверждения оператора")
        
        return input_dict
    
    def _default_callback(self, action: str) -> bool:
        """Заглушка для демо"""
        print(f"\n⚠️  ТРЕБУЕТСЯ ПОДТВЕРЖДЕНИЕ: {action}")
        print("Разрешить выполнение? (yes/no): ")
        response = input().strip().lower()
        return response == "yes"

Собираем всё вместе: продакшен-агент

Теперь создаем агента со всей цепочкой middleware:

def create_production_agent():
    """Создает агента с полным стеком middleware"""
    
    # 1. Инициализируем LLM
    llm = ChatOpenAI(
        model="gpt-4-turbo-preview",
        temperature=0.1,  # Низкая температура для консистентности
        max_tokens=1000
    )
    
    # 2. Создаем цепочку middleware
    middleware_chain = (
        RunnableLambda(PIIMiddleware())  # Сначала фильтруем PII
        | RunnableLambda(ContextManagerMiddleware(max_tokens=3000))  # Управляем контекстом
        | RunnableLambda(HumanApprovalMiddleware())  # Проверяем опасные действия
    )
    
    # 3. Системный промпт (инструкции для агента)
    system_prompt = """Ты финансовый ассистент. Твои правила:
    1. Никогда не называй реальные номера карт, счетов, документов
    2. Если пользователь спрашивает о чувствительных данных, говори что информация скрыта
    3. Перед переводом денег всегда спрашивай подтверждение
    4. Будь вежливым и профессиональным
    """
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        ("placeholder", "{messages}"),
    ])
    
    # 4. Создаем цепочку: middleware → промпт → LLM
    chain = (
        {"messages": lambda x: x["messages"]}
        | middleware_chain
        | prompt
        | llm
    )
    
    return chain

# Использование
agent = create_production_agent()

# Тестируем с опасным запросом
messages = [
    HumanMessage(content="Переведи 5000 рублей с карты 4276 1234 5678 9012 на счет Ивана")
]

try:
    response = agent.invoke({"messages": messages})
    print(response.content)
except ValueError as e:
    print(f"Действие прервано: {e}")

Продвинутые техники: о чем молчит документация

Middleware с состоянием

Иногда middleware нужно запоминать что-то между вызовами. Например, счетчик запросов или кэш PII-масок.

class StatefulMiddleware:
    def __init__(self):
        self.request_count = 0
        self.pii_cache = {}  # Кэш масок (оригинал → маска)
        
    def __call__(self, input_dict: Dict[str, Any]) -> Dict[str, Any]:
        self.request_count += 1
        
        # Пример: кэшируем маскированные значения
        if "messages" in input_dict:
            for msg in input_dict["messages"]:
                if hasattr(msg, "content"):
                    content = msg.content
                    if content in self.pii_cache:
                        # Используем кэшированную маску
                        msg.content = self.pii_cache[content]
                    else:
                        # Создаем новую маску и кэшируем
                        masked = self._mask_pii(content)
                        self.pii_cache[content] = masked
                        msg.content = masked
        
        return input_dict

Middleware для аудита и мониторинга

Каждый запрос нужно логировать. Но не просто в stdout, а в структурированном виде:

class AuditMiddleware:
    def __init__(self, log_destination="elasticsearch"):
        self.log_destination = log_destination
        
    def __call__(self, input_dict: Dict[str, Any]) -> Dict[str, Any]:
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "input": self._sanitize_for_log(input_dict),
            "user_id": self._extract_user_id(input_dict),  # Из заголовков или контекста
            "has_pii": self._check_for_pii(input_dict),
        }
        
        # Отправляем в Elasticsearch/S3/базу данных
        self._send_to_destination(log_entry)
        
        return input_dict  # Пропускаем дальше

Распространенные ошибки и как их избежать

Когда middleware недостаточно: архитектурные решения

Middleware решает много проблем, но не все. Если ваш агент становится слишком сложным, рассмотрите:

1. Суб-агенты для специализированных задач. Не пытайтесь запихнуть всю логику в одну цепочку. Разделяйте ответственность: один агент для анализа, другой для выполнения, третий для валидации. Вот реальные сценарии использования суб-агентов.
2. Внешние сервисы валидации. Вынесите PII-детекцию в отдельный микросервис с SLA и мониторингом.
3. Паттерн Circuit Breaker. Если LLM или middleware начинает тормозить, временно отключайте несущественные проверки.

Помните историю про уязвимости локальных AI-агентов? Middleware — это ваша первая линия защиты. Но не последняя.

Что дальше? Middleware 2.0

Текущая реализация middleware в LangChain хороша, но ограничена. Что будет в следующих версиях (или в альтернативных фреймворках):

• Graph-based middleware. Вместо линейной цепочки — граф, где middleware могут выполняться параллельно или условно.
• Declarative middleware. Описываете политики ("фильтровать PII", "логировать запросы", "сжимать контекст"), а система сама строит цепочку.
• Middleware marketplace. Загружаете готовые middleware из репозитория (как Docker images).

Пока этого нет, используйте текущие возможности на полную. И помните: лучший middleware — тот, который вы написали под свои конкретные требования, а не скопировали из блога.

Кстати, если LangChain кажется вам слишком тяжелым, посмотрите на альтернативы вроде Cogitator. Иногда проще написать своё, чем бороться с чужим API.