Собираем LLM-агента на Python

Команда Python for Devs подготовила перевод статьи о том, как с помощью middleware в LangChain 1.0 собирать LLM-агентов, готовых к реальному продакшену. В материале разбираются практические паттерны: управление контекстом, защита PII, human-in-the-loop, планирование задач и интеллектуальный выбор инструментов — всё то, что отличает экспериментального агента от надёжного рабочего решения.

Введение

Хотели ли вы когда-нибудь расширить своего LLM-агента дополнительными возможностями, например:

• Суммировать сообщения, чтобы укладываться в контекстное окно;

• Фильтровать PII для защиты чувствительных данных;

• Запрашивать подтверждение человека для критически важных действий,

…но не понимали, как это правильно реализовать?

Если вы пытались сделать это в LangChain v0.x, то, скорее всего, столкнулись со сложными pre/post-хуками, которые было трудно масштабировать и тестировать.

LangChain 1.0 решает эти проблемы, представляя компонуемую архитектуру middleware, которая предлагает переиспользуемые и тестируемые компоненты, построенные по тем же принципам, что и middleware в веб-серверах.

Полный исходный код и Jupyter Notebook для этого туториала доступны на GitHub. Склонируйте репозиторий и повторяйте шаги вместе с нами.

Введение в паттерн Middleware

Опираясь на основы LangChain, которые мы рассматривали ранее, LangChain 1.0 вводит middleware-компоненты, дающие детальный контроль над выполнением агентов. Каждый middleware — это изолированный компонент, который:

• Решает одну конкретную задачу (наблюдение, изменение, управление или принуждение);

• Может тестироваться независимо;

• Комбинируется с другими middleware через стандартный интерфейс.

Существует четыре категории middleware:

• Monitor: отслеживание поведения агента с помощью ��огирования, аналитики и отладки;

• Modify: преобразование промптов, выбор инструментов и форматирование вывода;

• Control: добавление повторных попыток, запасных вариантов и логики досрочного завершения;

• Enforce: применение лимитов, защитных ограничений и обнаружение PII.

В этой статье рассматриваются пять ключевых middleware-компонентов:

• Суммирование сообщений (modify): управление контекстным окном за счёт сжатия длинных диалогов;

• Фильтрация PII (enforce): защита чувствительных данных путём редактирования email-адресов и номеров телефонов;

• Human-in-the-loop (control): приостановка выполнения для критических действий, требующих одобрения;

• Планирование задач (modify): разбиение сложных запросов на управляемые подзадачи;

• Интеллектуальный выбор инструментов (modify): предварительная фильтрация инструментов для снижения затрат и повышения точности.

Давайте разберём, как каждый из этих middleware-компонентов улучшает рабочие процессы продакшен-агентов.

Установка

Установите LangChain 1.0 и интеграцию с OpenAI:

# Option 1: pip
pip install langchain langchain-openai

# Option 2: uv (faster alternative to pip)
uv add langchain langchain-openai

Примечание: если вы обновляетесь с LangChain v0.x, добавьте флаг --U:

pip install --U langchain langchain-openai

Также потребуется API-ключ OpenAI:

export OPENAI_API_KEY="your-api-key-here"

Суммирование сообщений

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

SummarizationMiddleware автоматизирует решение этой проблемы за счёт:

• Отслеживания количества токенов во всём диалоге;

• Сжатия более старых сообщений при превышении заданных порогов;

• Сохранения недавнего контекста, наиболее важного в текущий момент.

Преимущества такого подхода:

• Снижение затрат на API за счёт передачи меньшего количества токенов в каждом запросе;

• Более быстрые ответы благодаря меньшему контекстному окну;

• Сохранение полного контекста за счёт комбинации кратких сводок и полной недавней истории.

Вот пример использования SummarizationMiddleware в составе агента:

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="openai:gpt-4o",
    tools=[],
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-4o-mini",
            max_tokens_before_summary=400,
            messages_to_keep=5
        )
    ]
)

Эта конфигурация настраивает автоматическое управление диалогом:

• model="openai:gpt-4o" — основная модель, которая отвечает за ответы агента.

• max_tokens_before_summary=400 — запускает суммирование, когда диалог превышает 400 токенов.

• messages_to_keep=5 — сохраняет последние 5 сообщений целиком, без сжатия.

• model="openai:gpt-4o-mini" — использует более быструю и дешёвую модель для создания сводок.

Примечание: эти значения намеренно занижены для демонстрации, чтобы быстрее показать, как работает суммирование. В продакшене обычно используют max_tokens_before_summary=4000 и messages_to_keep=20 (рекомендуемые значения по умолчанию).

Давайте воспользуемся этим агентом, чтобы смоделировать диалог службы поддержки и отследить расход токенов.

Сначала подготовим реалистичный диалог поддержки из нескольких реплик:

# Simulate a customer support conversation
conversation_turns = [
    "I ordered a laptop last week but haven't received it yet. Order #12345.",
    "Can you check the shipping status? I need it for work next Monday.",
    "Also, I originally wanted the 16GB RAM model but ordered 8GB by mistake.",
    "Is it too late to change the order? Or should I return and reorder?",
    "What's your return policy on laptops? Do I need the original packaging?",
    "If I return it, how long does the refund take to process?",
    "Can I get expedited shipping on the replacement 16GB model?",
    "Does the 16GB version come with the same warranty as the 8GB?",
    "Are there any promotional codes I can use for the new order?",
    "What if the new laptop arrives damaged? What's the process?",
]

Далее определим вспомогательные функции, чтобы отслеживать расход токенов и проверять, что суммирование действительно срабатывает:

• estimate_token_count(): приблизительно считает количество токенов, оценивая их по числу слов во всех сообщениях;

• get_actual_tokens(): извлекает фактическое количество токенов из метаданных ответа модели;

• print_token_comparison(): выводит сравнение «оценка vs факт», чтобы было видно, в какой момент включается суммирование.

def estimate_token_count(messages):
    """Estimate total tokens in message history."""
    return sum(len(msg.content.split()) * 1.3 for msg in messages)

def get_actual_tokens(response):
    """Extract actual token count from response metadata."""
    last_ai_message = response["messages"][-1]
    if hasattr(last_ai_message, 'usage_metadata') and last_ai_message.usage_metadata:
        return last_ai_message.usage_metadata.get("input_tokens", 0)
    return None

def print_token_comparison(turn_number, estimated, actual):
    """Print token count comparison for a conversation turn."""
    if actual is not None:
        print(f"Turn {turn_number}: ~{int(estimated)} tokens (estimated) → {actual} tokens (actual)")
    else:
        print(f"Turn {turn_number}: ~{int(estimated)} tokens (estimated)")

Наконец, запустим диалог и посмотрим, как меняется расход токенов от шага к шагу:

messages = []
for i, question in enumerate(conversation_turns, 1):
    messages.append(HumanMessage(content=question))

    estimated_tokens = estimate_token_count(messages)
    response = agent.invoke({"messages": messages})
    messages.extend(response["messages"][len(messages):])

    actual_tokens = get_actual_tokens(response)
    print_token_comparison(i, estimated_tokens, actual_tokens)

Вывод:

Turn 1: ~16 tokens (estimated) → 24 tokens (actual)
Turn 2: ~221 tokens (estimated) → 221 tokens (actual)
Turn 3: ~408 tokens (estimated) → 415 tokens (actual)
Turn 4: ~646 tokens (estimated) → 509 tokens (actual)
Turn 5: ~661 tokens (estimated) → 524 tokens (actual)
Turn 6: ~677 tokens (estimated) → 379 tokens (actual)
Turn 7: ~690 tokens (estimated) → 347 tokens (actual)
Turn 8: ~705 tokens (estimated) → 184 tokens (actual)
Turn 9: ~721 tokens (estimated) → 204 tokens (actual)
Turn 10: ~734 tokens (estimated) → 195 tokens (actual)

Обратите внимание на закономерность в количестве токенов:

• Шаги 1–3: токены растут равномерно (24 → 221 → 415) по мере того, как диалог «обрастает» контекстом.

• Шаг 4: подключается суммирование — фактическое число токенов падает до 509, хотя по оценке должно было быть 646.

• Шаг 8: самое заметное снижение — отправлено всего 184 фактических токена вместо 705 по оценке (сокращение на 74%).

После прохождения порога в 400 токенов middleware автоматически сжимает более старые сообщения, сохраняя при этом последние 5 реплик целиком. Благодаря этому расход токенов остаётся низким, даже когда диалог продолжается.

Обнаружение и фильтрация PII

В диалогах службы поддержки часто встречаются чувствительные данные: email-адреса, номера телефонов, идентификаторы аккаунтов. Если логировать или хранить такую информацию без редактирования, это создаёт риски для соответствия требованиям и безопасности.

PIIMiddleware автоматически защищает персональные данные (PII) за счёт:

• Встроенных детекторов для распространённых типов PII (email, банковские карты, IP-адреса);

• Пользовательских regex-шаблонов для чувствительных данных, специфичных для домена;

• Нескольких стратегий защиты: редактирование, маскирование, хэширование или блокировка;

• Автоматического применения ко всем сообщениям до того, как их обработает модель.

Сначала настроим агента с несколькими детекторами PII:

Каждый детектор в этом примере демонстрирует свою стратегию защиты:

• Детектор email: использует встроенный шаблон и стратегию redact (полная замена).

• Детектор телефона: использует пользовательский regex bd{3}-d{3}-d{4}b и стратегию mask (частичная видимость).

• Детектор ID аккаунта: использует пользовательский шаблон b[A-Z]{2}d{8}b и стратегию redact (полное удаление).

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain_core.messages import HumanMessage

agent = create_agent(
    model="openai:gpt-4o",
    tools=[],
    middleware=[
        # Built-in email detector - replaces emails with [REDACTED_EMAIL]
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        # Custom phone number pattern - shows only last 4 digits
        PIIMiddleware(
            "phone",
            detector=r"bd{3}-d{3}-d{4}b",
            strategy="mask",
            apply_to_input=True,
        ),
        # Custom regex pattern for account IDs (e.g., AB12345678)
        PIIMiddleware(
            "account_id",
            detector=r"b[A-Z]{2}d{8}b",
            strategy="redact",
            apply_to_input=True,
        ),
    ],
)

Далее создадим сообщение с чувствительными данными и вызовем агента:

# Create a message with PII
original_message = HumanMessage(content="My email is john@example.com, phone is 555-123-4567, and account is AB12345678")
print(f"Original message: {original_message.content}")

# Invoke the agent
response = agent.invoke({"messages": [original_message]})

Вывод:

Original message: My email is john@example.com, phone is 555-123-4567, and account is AB12345678

Теперь посмотрим, какое сообщение фактически было отправлено в модель, чтобы убедиться, что редактирование сработало:

# Check what was actually sent to the model (after PII redaction)
input_message = response["messages"][0]
print(f"Message sent to model: {input_message.content}")

Вывод:

Message sent to model: My email is [REDACTED_EMAIL], phone is ****4567, and account is [REDACTED_ACCOUNT_ID]

Middleware корректно обработал все три типа чувствительной информации:

• Email: полностью отредактирован и заменён на [REDACTED_EMAIL].

• Телефон: замаскирован с сохранением только последних 4 цифр (****4567).

• ID аккаунта: полностью удалён и заменён на [REDACTED_ACCOUNT_ID].

Human-in-the-loop

Автономные агенты могут выполнять чувствительные действия — например, обрабатывать возвраты или менять настройки аккаунта. Если делать это без контроля человека, возрастает риск ошибок или злоупотреблений.

HumanInTheLoopMiddleware автоматизирует процесс согласования: он ставит выполнение на паузу и ждёт одобрения, прежде чем продолжить:

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver


@tool
def process_refund(amount: float, reason: str) -> str:
    """Process a customer refund. Use this when a customer requests a refund."""
    return f"Refund of ${amount} processed for reason: {reason}"


# Create memory checkpointer for state persistence
memory = MemorySaver()

agent = create_agent(
    model="openai:gpt-4o",
    tools=[process_refund],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"process_refund": True})],
    checkpointer=memory,  # Required for state persistence
    system_prompt="You are a customer support agent. Use the available tools to help customers. When a customer asks for a refund, use the process_refund tool.",
)

Эта конфигурация настраивает агента, который:

• Использует HumanInTheLoopMiddleware, чтобы приостанавливать выполнение перед вызовом process_refund;

• Использует чекпойнтер (MemorySaver), чтобы сохранять состояние агента во время прерываний и позволять продолжить выполнение после одобрения.

Теперь вызовем агента с запросом на возврат:

# Agent pauses before executing sensitive tools
response = agent.invoke(
    {"messages": [("user", "I need a refund of $100 for my damaged laptop")]},
    config={"configurable": {"thread_id": "user-123"}},
)

Агент поставит выполнение на паузу в момент, когда попытается обработать возврат. Чтобы убедиться, что пауза действительно произошла, определим вспомогательные функции для обнаружения прерываний.

def has_interrupt(response):
    """Check if response contains an interrupt."""
    return "__interrupt__" in response

def display_action(action):
    """Display pending action details."""
    print(f"Pending action: {action['name']}")
    print(f"Arguments: {action['args']}")
    print()

def get_user_approval():
    """Prompt user for approval and return decision."""
    approval = input("Approve this action? (yes/no): ")
    if approval.lower() == "yes":
        print("✓ Action approved")
        return True
    else:
        print("✗ Action rejected")
        return False

Теперь воспользуемся этими функциями, чтобы проверить наличие прерываний и обработать одобрение:

if has_interrupt(response):
    print("Execution interrupted - waiting for approvaln")

    interrupts = response["__interrupt__"]
    for interrupt in interrupts:
        for action in interrupt.value["action_requests"]:
            display_action(action)
            approved = get_user_approval()

Вывод:

Execution interrupted - waiting for approval

Pending action: process_refund
Arguments: {'amount': 100, 'reason': 'Damaged Laptop'}

Approve this action? (yes/no): yes
✓ Action approved

Middleware успешно перехватил вызов инструмента process_refund до выполнения, показав все необходимые детали (название действия и аргументы) для проверки человеком. И только после явного одобрения агент продолжает выполнение чувствительной операции.

Планирование задач

Сложные задачи вроде «отрефакторить мой код» или «проанализировать этот датасет» требуют разбиения на более мелкие и управляемые шаги. Без явного планирования агенты часто начинают хаотично перескакивать между подзадачами или вовсе пропускают критически важные этапы.

TodoListMiddleware обеспечивает структурированное управление задачами за счёт:

• Автоматического предоставления инструмента write_todos для планирования;

• Отслеживания статуса выполнения в многошаговых сценариях;

• Возврата структурированного списка задач в результатах работы агента.

Преимущества такого подхода:

• Более качественная декомпозиция задач благодаря пошаговому планированию;

• Прозрачное отслеживание прогресса в сложных рабочих процессах;

• Снижение числа ошибок из-за пропущенных или забытых подзадач.

Вот как включить планирование для агента:

from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain_core.tools import tool

@tool
def analyze_code(file_path: str) -> str:
    """Analyze code quality and find issues."""
    return f"Analyzed {file_path}: Found 3 code smells, 2 security issues"

@tool
def refactor_code(file_path: str, changes: str) -> str:
    """Refactor code with specified changes."""
    return f"Refactored {file_path}: {changes}"

agent = create_agent(
    model="openai:gpt-4o",
    tools=[analyze_code, refactor_code],
    middleware=[TodoListMiddleware()]
)

Эта конфигурация автоматически добавляет агенту возможности планирования.

Теперь попросим агента выполнить многошаговую задачу по рефакторингу:

from langchain_core.messages import HumanMessage

response = agent.invoke({
    "messages": [HumanMessage("I need to refactor my authentication module. First analyze it, then suggest improvements, and finally implement the changes.")]
})

Посмотрим на список задач агента, чтобы понять, как он спланировал работу:

# Access the structured todo list from the response
if "todos" in response:
    print("Agent's Task Plan:")
    for i, todo in enumerate(response["todos"], 1):
        status = todo.get("status", "pending")
        print(f"{i}. [{status}] {todo['content']}")

Вывод:

Agent's Task Plan:
1. [in_progress] Analyze the authentication module code to identify quality issues and areas for improvement.
2. [pending] Suggest improvements based on the analysis of the authentication module.
3. [pending] Implement the suggested improvements in the authentication module code.

Отлично! Агент автоматически разложил многошаговый запрос на рефакторинг на три отдельные задачи: одна уже в работе, две ожидают выполнения. Такой структурированный подход обеспечивает последовательное выполнение без пропуска критически важных шагов.

Интеллектуальный выбор инструментов

Агенты с большим количеством инструментов (10+) сталкиваются с проблемой масштабирования: если в каждом запросе отправлять описания всех инструментов, это впустую расходует токены и ухудшает производительность. Модели приходится «переваривать» нерелевантные варианты, что увеличивает задержки и стоимость.

LLMToolSelectorMiddleware решает это, используя более компактную модель для предварительного отбора релевантных инструментов:

• Использует вторую LLM (отдельную от основной модели агента), чтобы заранее отфильтровать и ограничить набор инструментов, отправляемых в основную модель;

• Позволяет всегда включать в выбор критически важные инструменты;

• Анализирует запросы и оставляет только релевантные инструменты.

Преимущества:

• Более низкие затраты за счёт передачи меньшего числа описаний инструментов в каждом запросе;

• Более быстрые ответы благодаря меньшему «инструментальному» контексту;

• Более высокая точность, потому что модель не отвлекается на нерелевантные варианты.

Давайте создадим агента с большим набором инструментов для сценария службы поддержки:

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware
from langchain_core.tools import tool

# Define multiple tools for different support scenarios
@tool
def lookup_order(order_id: str) -> str:
    """Look up order details and shipping status."""
    return f"Order {order_id}: Shipped on 2025-01-15"

@tool
def process_refund(order_id: str, amount: float) -> str:
    """Process a customer refund."""
    return f"Refund of ${amount} processed for order {order_id}"

@tool
def check_inventory(product_id: str) -> str:
    """Check product inventory levels."""
    return f"Product {product_id}: 42 units in stock"

@tool
def update_address(order_id: str, new_address: str) -> str:
    """Update shipping address for an order."""
    return f"Address updated for order {order_id}"

@tool
def cancel_order(order_id: str) -> str:
    """Cancel an existing order."""
    return f"Order {order_id} cancelled"

@tool
def track_shipment(tracking_number: str) -> str:
    """Track package location."""
    return f"Package {tracking_number}: Out for delivery"

@tool
def apply_discount(order_id: str, code: str) -> str:
    """Apply discount code to order."""
    return f"Discount {code} applied to order {order_id}"

@tool
def schedule_delivery(order_id: str, date: str) -> str:
    """Schedule delivery for specific date."""
    return f"Delivery scheduled for {date}"

Настроим агента с интеллектуальным выбором инструментов:

agent = create_agent(
    model="openai:gpt-4o",
    tools=[
        lookup_order, process_refund, check_inventory,
        update_address, cancel_order, track_shipment,
        apply_discount, schedule_delivery
    ],
    middleware=[
        LLMToolSelectorMiddleware(
            model="openai:gpt-4o-mini",  # Use cheaper model for selection
            max_tools=3,  # Limit to 3 most relevant tools
            always_include=["lookup_order"],  # Always include order lookup
        )
    ]
)

Эта конфигурация создаёт эффективную систему фильтрации:

• model="openai:gpt-4o-mini" — использует более компактную и быструю модель для выбора инструментов.

• max_tools=3 — ограничивает набор тремя наиболее релевантными инструментами для каждого запроса.

• always_include=["lookup_order"] — гарантирует, что инструмент поиска заказа всегда будет доступен.

Теперь протестируем агента на разных запросах клиентов.

Сначала определим вспомогательную функцию, которая покажет, какие инструменты использовались:

def show_tools_used(response):
    """Display which tools were called during agent execution."""
    tools_used = []
    for msg in response["messages"]:
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            for tool_call in msg.tool_calls:
                tools_used.append(tool_call["name"])

    if tools_used:
        print(f"Tools used: {', '.join(tools_used)}")
    print(f"Response: {response['messages'][-1].content}n")

Проверим запрос на отслеживание посылки:

# Example 1: Package tracking query
response = agent.invoke({
    "messages": [HumanMessage("Where is my package? Tracking number is 1Z999AA10123456784")]
})
show_tools_used(response)

Вывод:

Tools used: track_shipment
Response: Your package with tracking number 1Z999AA10123456784 is currently out for delivery.

Проверим запрос на возврат:

# Example 2: Refund request
response = agent.invoke({
    "messages": [HumanMessage("I need a refund of $50 for order ORD-12345")]
})
show_tools_used(response)

Вывод:

Tools used: lookup_order, process_refund
Response: The refund of $50 for order ORD-12345 has been successfully processed.

Проверим запрос на наличие товара:

# Example 3: Inventory check
response = agent.invoke({
    "messages": [HumanMessage("Do you have product SKU-789 in stock?")]
})
show_tools_used(response)

Вывод:

Tools used: check_inventory
Response: Yes, we currently have 42 units of product SKU-789 in stock.

Middleware показал точный выбор инструментов для разных типов запросов:

• track_shipment — для трекинг-номеров;

• lookup_order + process_refund — для запросов на возврат;

• check_inventory — для запросов о наличии на складе.

В каждом запросе middleware отфильтровал 5+ нерелевантных инструментов, передав основной модели только то, что действительно нужно.

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

Давайте объединим три middleware-компонента и соберём готового к продакшену агента поддержки, который обрабатывает реалистичный сценарий: у клиента длинная история переписки, он просит вернуть деньги и одновременно сообщает свой email.

from langchain.agents import create_agent
from langchain.agents.middleware import (
    SummarizationMiddleware,
    PIIMiddleware,
    HumanInTheLoopMiddleware
)
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver

@tool
def process_refund(amount: float, reason: str) -> str:
    """Process a customer refund."""
    return f"Refund of ${amount} processed for reason: {reason}"

# Create agent with three middleware components
agent = create_agent(
    model="openai:gpt-4o",
    tools=[process_refund],
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-4o-mini",
            max_tokens_before_summary=400,
            messages_to_keep=5
        ),
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        HumanInTheLoopMiddleware(interrupt_on={"process_refund": True})
    ],
    checkpointer=MemorySaver()
)

Теперь протестируем агента на реалистичном взаимодействии с клиентом, обрабатывая каждое сообщение и показывая, как middleware с ними работает.

Сначала определим вспомогательную функцию, чтобы отслеживать поведение middleware, используя ранее определённые хелперы:

def process_message_with_tracking(agent, messages, thread_id, turn_num):
    """Process messages and show middleware behavior."""
    print(f"n--- Turn {turn_num} ---")
    print(f"User: {messages[-1][1]}")

    response = agent.invoke(
        {"messages": messages},
        config={"configurable": {"thread_id": thread_id}}
    )

    # Check for interrupts (human-in-the-loop)
    if has_interrupt(response):
        print(" Execution paused for approval")
    else:
        # Show agent response
        agent_message = response["messages"][-1].content
        print(f"Agent: {agent_message}")

    # Check for PII redaction
    full_response = str(response["messages"])
    if "[REDACTED_EMAIL]" in full_response:
        print(" PII detected and redacted")

    return response

Теперь смоделируем диалог с клиентом так, чтобы продемонстрировать все три middleware-компонента:

• Шаги 1–3: обычный диалог о повреждённом ноутбуке.

• Шаг 4: клиент сообщает email и просит подтверждение (проверяем редактирование в PIIMiddleware).

• Шаг 5: клиент просит вернуть $1200 (срабатывает согласование в HumanInTheLoopMiddleware).

messages = []

# Turn 1: Initial complaint
messages.append(("user", "I ordered a laptop but it arrived damaged."))
process_message_with_tracking(agent, messages, "customer-456", 1)

# Turn 2: Additional details
messages.append(("user", "I already tried troubleshooting but it won't turn on."))
process_message_with_tracking(agent, messages, "customer-456", 2)

# Turn 3: More context
messages.append(("user", "The screen is cracked and there's a dent on the corner."))
process_message_with_tracking(agent, messages, "customer-456", 3)

# Turn 4: PII exposure - test if middleware hides email from model
messages.append(("user", "My email is customer@example.com. Can you confirm my email address?"))
response = process_message_with_tracking(agent, messages, "customer-456", 4)

# Turn 5: Sensitive action request - triggers human approval
messages.append(("user", "Can I get a full refund of $1200?"))
response = process_message_with_tracking(agent, messages, "customer-456", 5)

# If interrupted, show approval flow
if has_interrupt(response):
    interrupts = response["__interrupt__"]
    for interrupt in interrupts:
        for action in interrupt.value["action_requests"]:
            display_action(action)
            approved = get_user_approval()

Вывод:

--- Turn 1 ---
User: I ordered a laptop but it arrived damaged.
Agent: I'm sorry to hear that your laptop arrived damaged. To resolve this issue, I can assist you in processing a refund. Could you please provide the amount of the purchase and any specific reason you would like to include for the refund?

--- Turn 2 ---
User: I already tried troubleshooting but it won't turn on.
Agent: Thank you for sharing that information. I'll proceed with processing a refund for the damaged laptop. Could you please provide the purchase amount, so I can include it in the refund request?

--- Turn 3 ---
User: The screen is cracked and there's a dent on the corner.
Agent: I'm sorry to hear about the condition of your laptop. I will help you process a refund. Could you please let me know the purchase amount, so I can proceed with the refund request?

--- Turn 4 ---
User: My email is customer@example.com. Can you confirm my email address?
Agent: I'm sorry, but I'm unable to confirm or access email addresses for privacy and security reasons. However, I can assist you with processing a refund. Could you please provide the amount you paid for the laptop so that I can proceed with the refund request?
 PII detected and redacted

--- Turn 5 ---
User: Can I get a full refund of $1200?
 Execution paused for approval

Pending action: process_refund
Arguments: {'amount': 1200, 'reason': 'Laptop arrived damaged with a cracked screen and dent on the corner, and it will not turn on after troubleshooting.'}

Approve this action? (yes/no): yes
✓ Action approved

Результат демонстрирует корректные меры безопасности:

• Шаг 4: агент сообщает, что «не может подтвердить или получить доступ к email-адресам», что подтверждает: PIIMiddleware успешно заменил customer@example.com на [REDACTED_EMAIL].

• Защита email: модель так и не увидела реальный адрес — это снижает риск утечек и попадания PII в логи.

• Подтверждение возврата: операция на $1200 не выполнялась до тех пор, пока человек явно не дал одобрение.

Если вам нужно координировать несколько агентов с общим состоянием и рабочими процессами, изучите наш туториал по LangGraph.

Итоговые мысли

Создание LLM-агентов, готовых к продакшену, с использованием middleware LangChain 1.0 требует минимум инфраструктурного кода. Каждый компонент отвечает за одну конкретную задачу: управление контекстным окном, защита чувствительных данных, контроль потока выполнения или структурирование сложных задач.

Лучше всего двигаться постепенно. Добавляйте по одному middleware за раз, проверяйте, как он себя ведёт, а затем сочетайте его с другими. Такой модульный подход позволяет начать с простого решения и расширять его по мере того, как растут требования к агенту.

Русскоязычное сообщество про Python

Друзья! Эту статью подготовила команда Python for Devs — канала, где каждый день выходят самые свежие и полезные материалы о Python и его экосистеме. Подписывайтесь, чтобы ничего не пропустить!