Встроенные middleware: точки встраивания и порядок | Курс LangChain урок 15

"""Общая сборка модели для примеров урока 15.

Тот же модуль, что course_model.py прежних уроков, без изменений. У build_model есть
необязательный первый аргумент с именем модели, а рядом лежит gateway_kwargs()
для примеров, которые собирают модель сами.

Файл .env берётся тот же, что в уроке 0. Положите его рядом с этой папкой или
выше по дереву: load_dotenv() ищет файл начиная с папки этого модуля и поднимается
вверх.
"""

import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()


def gateway_kwargs():
    """Возвращает аргументы доступа к провайдеру: имя, адрес, ключ.

    Нужны примерам, которые вызывают init_chat_model сами. У настраиваемой модели
    имени модели при создании нет, поэтому build_model ей не подходит.
    """
    base_url = os.getenv("MODEL_BASE_URL")

    if base_url:
        return {
            "model_provider": "openai",
            "base_url": base_url,
            "api_key": os.environ["OPENAI_API_KEY"],
        }

    return {}


def build_model(model_name=None, **kwargs):
    """Собирает модель курса.

    model_name без значения означает модель из переменной MODEL_NAME. Явное имя
    нужно примерам, где моделей в приложении больше одной.

    Все именованные аргументы уходят в init_chat_model как есть: temperature,
    max_tokens, timeout, max_retries, rate_limiter, profile и прочее из раздела
    Parameters.
    """
    model_name = model_name or os.environ["MODEL_NAME"]
    access = gateway_kwargs()

    if access:
        # Путь для любого адреса, совместимого с OpenAI Chat Completions API.
        return init_chat_model(model=model_name, **access, **kwargs)

    # Путь напрямую к провайдеру.
    return init_chat_model(model_name, **kwargs)

"""Пример 1 урока 15: карта точек встраивания у встроенных middleware.

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

Приём взят из сборки агента: фабрика решает, подключать middleware к точке или нет,
сравнением метода класса с методом базового AgentMiddleware. Асинхронный
вариант хука считается тем же хуком, поэтому проверяются оба имени.
"""

from pathlib import Path

from langchain.agents.middleware import (
    AgentMiddleware,
    ContextEditingMiddleware,
    FilesystemFileSearchMiddleware,
    HumanInTheLoopMiddleware,
    LLMToolEmulator,
    LLMToolSelectorMiddleware,
    ModelCallLimitMiddleware,
    ModelFallbackMiddleware,
    ModelRetryMiddleware,
    PIIMiddleware,
    SummarizationMiddleware,
    TodoListMiddleware,
    ToolCallLimitMiddleware,
    ToolErrorMiddleware,
    ToolRetryMiddleware,
)

# Шесть точек встраивания, по колонке таблицы на каждую.
HOOKS = (
    "before_agent",
    "before_model",
    "wrap_model_call",
    "wrap_tool_call",
    "after_model",
    "after_agent",
)

SHORT = {
    "before_agent": "b_ag",
    "before_model": "b_md",
    "wrap_model_call": "w_md",
    "wrap_tool_call": "w_tl",
    "after_model": "a_md",
    "after_agent": "a_ag",
}

LAYERS = (
    ToolErrorMiddleware,
    ToolRetryMiddleware,
    ModelRetryMiddleware,
    ModelFallbackMiddleware,
    SummarizationMiddleware,
    ContextEditingMiddleware,
    ModelCallLimitMiddleware,
    ToolCallLimitMiddleware,
    TodoListMiddleware,
    LLMToolSelectorMiddleware,
    LLMToolEmulator,
    FilesystemFileSearchMiddleware,
    PIIMiddleware,
    HumanInTheLoopMiddleware,
)


def implements(layer, hook):
    """Переопределяет ли middleware этот хук, в синхронном или асинхронном виде."""
    async_hook = "a" + hook
    sync_own = getattr(layer, hook) is not getattr(AgentMiddleware, hook)
    async_own = getattr(layer, async_hook) is not getattr(AgentMiddleware, async_hook)
    return sync_own or async_own


header = "".join(f"{SHORT[hook]:>6}" for hook in HOOKS)
print(f"{'middleware':<32}{header}")

for layer in LAYERS:
    marks = "".join(f"{'X' if implements(layer, hook) else '.':>6}" for hook in HOOKS)
    print(f"{layer.__name__:<32}{marks}")

print()
print("узел графа появляется только у before_* и after_*,")
print("хуки wrap_* оборачивают вызов внутри уже существующих узлов")

# Вторая работа middleware: принести агенту свои инструменты. Это свойство экземпляра,
# а не класса, поэтому здесь middleware создаются.
print()
print("инструменты, которые middleware добавляет агенту:")

WITH_TOOLS = (
    TodoListMiddleware(),
    FilesystemFileSearchMiddleware(root_path=str(Path(__file__).parent)),
    ToolRetryMiddleware(),
)

for layer in WITH_TOOLS:
    names = [tool.name for tool in layer.tools] or ["нет"]
    print(f"  {layer.name:<32}{', '.join(names)}")

# Вывод:
# middleware                        b_ag  b_md  w_md  w_tl  a_md  a_ag
# ToolErrorMiddleware                  .     .     .     X     .     .
# ToolRetryMiddleware                  .     .     .     X     .     .
# ModelRetryMiddleware                 .     .     X     .     .     .
# ModelFallbackMiddleware              .     .     X     .     .     .
# SummarizationMiddleware              .     X     .     .     .     .
# ContextEditingMiddleware             .     .     X     .     .     .
# ModelCallLimitMiddleware             .     X     .     .     X     .
# ToolCallLimitMiddleware              .     .     .     .     X     .
# TodoListMiddleware                   .     .     X     .     X     .
# LLMToolSelectorMiddleware            .     .     X     .     .     .
# LLMToolEmulator                      .     .     .     X     .     .
# FilesystemFileSearchMiddleware       .     .     .     .     .     .
# PIIMiddleware                        .     X     .     .     X     .
# HumanInTheLoopMiddleware             .     .     .     X     X     .
#
# узел графа появляется только у before_* и after_*,
# хуки wrap_* оборачивают вызов внутри уже существующих узлов
#
# инструменты, которые middleware добавляет агенту:
#   TodoListMiddleware              write_todos
#   FilesystemFileSearchMiddleware  glob_search, grep_search
#   ToolRetryMiddleware             нет

"""Пример 2 урока 15: падение инструмента с одним middleware и без него.

Инструмент читает учётную книгу и падает на годах, которых нет в архиве.
Первый запуск идёт без middleware, второй с ToolErrorMiddleware.

Middleware не чинит инструмент. Он превращает исключение в сообщение об ошибке,
которое получает модель, и решение остаётся за ней.
"""

from langchain.agents import create_agent
from langchain.agents.middleware import ToolErrorMiddleware

from course_model import build_model


def read_ledger(year: int) -> str:
    """Читает учётную книгу склада за указанный год."""
    if year < 2019:
        raise ValueError(f"архив за {year} год не оцифрован")
    return f"Книга за {year} год: 412 приходных записей, 87 списаний."


def on_error(exc, request):
    """Что получит модель вместо исключения. None означает "пропустить дальше"."""
    if isinstance(exc, ValueError):
        name = request.tool_call["name"]
        return (
            f"Инструмент {name} завершился с ошибкой {type(exc).__name__}: "
            f"такого года в архиве нет. Самый ранний оцифрованный год, это 2019."
        )
    # Остальные исключения идут дальше и роняют агента.
    return None


SYSTEM_PROMPT = (
    "Вы архивариус склада. Отвечайте по-русски, двумя-тремя фразами. "
    "Если инструмент сообщил об ошибке, поступайте по тому, что в ней написано."
)

QUESTION = "Сколько списаний было в 2017 году?"

model = build_model(temperature=0, max_tokens=384)

plain = create_agent(model=model, tools=[read_ledger], system_prompt=SYSTEM_PROMPT)

guarded = create_agent(
    model=model,
    tools=[read_ledger],
    system_prompt=SYSTEM_PROMPT,
    middleware=[ToolErrorMiddleware(on_error)],
)

print("ЗАПУСК БЕЗ MIDDLEWARE")
try:
    result = plain.invoke({"messages": [{"role": "user", "content": QUESTION}]})
except Exception as exc:  # noqa: BLE001 - пример печатает тип падения
    print(f"  запуск упал: {type(exc).__name__}: {exc}")
else:
    print("  запуск дошёл до конца:", result["messages"][-1].text)

print()
print("ЗАПУСК С MIDDLEWARE")
result = guarded.invoke({"messages": [{"role": "user", "content": QUESTION}]})

for message in result["messages"]:
    if message.type == "tool":
        print(f"  {message.name}: status={message.status}")

print("  ответ:", result["messages"][-1].text.replace("\n", " "))

# Вывод:
# ЗАПУСК БЕЗ MIDDLEWARE
#   запуск упал: ValueError: архив за 2017 год не оцифрован
#
# ЗАПУСК С MIDDLEWARE
#   read_ledger: status=error
#   ответ: В архиве нет данных за 2017 год — самый ранний оцифрованный год, это 2019.

"""Пример 3 урока 15: тот же набор middleware в двух порядках.

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

Инструмент падает всегда и считает свои запуски. По отношению числа запусков
к числу вызовов, которые сделала модель, видно, отработали повторы или нет.
"""

from langchain.agents import create_agent
from langchain.agents.middleware import ToolErrorMiddleware, ToolRetryMiddleware

from course_model import build_model

attempts = {"count": 0}


def fetch_index(path: str) -> str:
    """Загружает индекс модулей репозитория по пути."""
    attempts["count"] += 1
    raise ConnectionError(f"хранилище не ответило на запрос {path}")


def on_error(exc, request):
    """Превращает отказ хранилища в сообщение для модели."""
    if isinstance(exc, ConnectionError):
        return (
            f"Инструмент {request.tool_call['name']} недоступен "
            f"({type(exc).__name__}). Повторять вызов не нужно, сообщите об "
            f"этом пользователю."
        )
    return None


def retry_layer():
    """Повтор с короткой паузой, чтобы пример отработал за секунды."""
    return ToolRetryMiddleware(
        max_retries=2,
        initial_delay=0.2,
        backoff_factor=1.0,
        jitter=False,
        on_failure="error",
    )


SYSTEM_PROMPT = (
    "Вы помощник по разбору легаси. Отвечайте по-русски одной фразой. "
    "Если инструмент сообщил об ошибке, не повторяйте вызов."
)

QUESTION = "Загрузите индекс модулей по пути /src и скажите, что получилось."

model = build_model(temperature=0, max_tokens=256)


def run(label, middleware):
    """Собирает агента с заданным порядком middleware, отправляет вопрос и печатает строку отчёта."""
    attempts["count"] = 0
    agent = create_agent(
        model=model,
        tools=[fetch_index],
        system_prompt=SYSTEM_PROMPT,
        middleware=middleware,
    )
    result = agent.invoke({"messages": [{"role": "user", "content": QUESTION}]})
    calls = sum(1 for message in result["messages"] if message.type == "tool")
    per_call = attempts["count"] / calls if calls else 0
    print(f"{label:<24}{calls:>8}{attempts['count']:>10}{per_call:>12.1f}")
    return result


print(f"{'порядок middleware':<24}{'вызовов':>8}{'запусков':>10}{'на вызов':>12}")

first = run("повтор, обработка", [retry_layer(), ToolErrorMiddleware(on_error)])
second = run("обработка, повтор", [ToolErrorMiddleware(on_error), retry_layer()])

print()
print("ответ при порядке 'повтор, обработка':", first["messages"][-1].text)
print("ответ при порядке 'обработка, повтор':", second["messages"][-1].text)

# Вывод:
# порядок middleware       вызовов  запусков    на вызов
# повтор, обработка              1         1         1.0
# обработка, повтор              1         3         3.0
#
# ответ при порядке 'повтор, обработка': Инструмент сообщил об ошибке подключения — не удалось загрузить индекс модулей по пути `/src`.
# ответ при порядке 'обработка, повтор': Инструмент не смог загрузить индекс модулей по пути /src из-за ошибки подключения (ConnectionError).

"""Пример 4 урока 15: запасная модель вместо упавшей основной.

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

Запасная модель передаётся готовым объектом, а не строкой: строку middleware отдаёт
в init_chat_model сам, без адреса и ключа шлюза курса.
"""

from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware

from course_model import build_model

SYSTEM_PROMPT = "Вы помощник разработчика. Отвечайте по-русски одной фразой."
QUESTION = "Назовите две причины, по которым вызов модели может упасть."

broken = build_model("no-such-model-in-this-gateway", temperature=0, max_tokens=256)
spare = build_model(temperature=0, max_tokens=256)

plain = create_agent(model=broken, tools=[], system_prompt=SYSTEM_PROMPT)

with_fallback = create_agent(
    model=broken,
    tools=[],
    system_prompt=SYSTEM_PROMPT,
    middleware=[ModelFallbackMiddleware(spare)],
)


def answered_by(message):
    """Имя модели из ответа. Провайдер может его не прислать."""
    return message.response_metadata.get("model_name", "провайдер не сообщил")


print("ЗАПУСК БЕЗ MIDDLEWARE")
try:
    result = plain.invoke({"messages": [{"role": "user", "content": QUESTION}]})
except Exception as exc:  # noqa: BLE001 - пример печатает тип падения
    print(f"  запуск упал: {type(exc).__name__}")
else:
    print("  запуск дошёл до конца:", result["messages"][-1].text)

print()
print("ЗАПУСК С MIDDLEWARE")
result = with_fallback.invoke({"messages": [{"role": "user", "content": QUESTION}]})
answer = result["messages"][-1]
print("  отвечала модель:", answered_by(answer))
print("  ответ:", answer.text.replace("\n", " "))

# Вывод:
# ЗАПУСК БЕЗ MIDDLEWARE
#   запуск упал: OpenAIInvalidRequestError
#
# ЗАПУСК С MIDDLEWARE
#   отвечала модель: deepseek/deepseek-v4-flash
#   ответ: Вызов модели может упасть из-за сбоя в работе серверной инфраструктуры или превышения лимита запросов.

"""Пример 5 урока 15: список задач, который агент ведёт сам.

TodoListMiddleware добавляет агенту инструмент write_todos и кусок системного
промпта про то, как им пользоваться. Поле todos хранится в состоянии агента.

Задача требует пройти по всем модулям репозитория и для каждого узнать три
вещи по отдельности, а потом свести это в один вывод. Шаги внутри модуля
зависят друг от друга: владельца и тесты спрашивают про уже найденный модуль.
Модулей несколько, поэтому такую задачу удобнее вести по списку.
"""

from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware

from course_model import build_model

MODULES = {
    "billing": {"lines": 4120, "owner": "Северов", "tests": 12},
    "auth": {"lines": 890, "owner": "Липатова", "tests": 41},
    "reports": {"lines": 7305, "owner": "владелец не назначен", "tests": 0},
}


def list_modules() -> str:
    """Возвращает список модулей репозитория."""
    return ", ".join(sorted(MODULES))


def module_size(name: str) -> str:
    """Возвращает число строк кода в модуле."""
    if name not in MODULES:
        return f"Модуля {name} в репозитории нет."
    return f"{name}: {MODULES[name]['lines']} строк."


def module_owner(name: str) -> str:
    """Возвращает владельца модуля и число тестов на него."""
    if name not in MODULES:
        return f"Модуля {name} в репозитории нет."
    info = MODULES[name]
    return f"{name}: владелец {info['owner']}, тестов {info['tests']}."


SYSTEM_PROMPT = (
    "Вы разбираете чужой репозиторий. Отвечайте по-русски. "
    "Работу над многошаговой задачей начинайте с плана."
)

QUESTION = (
    "Проверьте все модули репозитория по очереди: для каждого узнайте размер, "
    "владельца и число тестов. Отдельно выпишите модули без владельца и "
    "модули, где тестов меньше десяти. В конце скажите, какой модуль чинить "
    "первым и почему."
)

agent = create_agent(
    model=build_model(temperature=0, max_tokens=768),
    tools=[list_modules, module_size, module_owner],
    system_prompt=SYSTEM_PROMPT,
    middleware=[TodoListMiddleware()],
)

result = agent.invoke({"messages": [{"role": "user", "content": QUESTION}]})

tool_calls = [
    call["name"]
    for message in result["messages"]
    if message.type == "ai"
    for call in message.tool_calls
]
print("вызовы инструментов:", ", ".join(tool_calls) or "их не было")

print()
print("список задач в конце запуска:")
for item in result.get("todos") or []:
    print(f"  [{item['status']:<11}] {item['content']}")

print()
print("ответ:", result["messages"][-1].text.replace("\n", " "))

# Вывод:
# вызовы инструментов: list_modules, module_size, module_size, module_size,
#   module_owner, module_owner, module_owner
#
# список задач в конце запуска:
#
# ответ: Все данные собраны. Подведу итоги.  ---  ## Сводка по модулям  | Модуль
#   | Размер (строк) | Владелец | Тестов | ... ## Какой модуль чинить первым и почему
#   **Первым нужно чинить модуль `reports`.**  Причины: 1. **Нет владельца** ...
#   2. **Нет тестов (0)** ... Таким образом, `reports` — самый уязвимый модуль ...
# [вывод подрезан: ответ модели идёт таблицей и разбором на три десятка строк]

"""Пример 6 урока 15: поиск по файлам без единого своего инструмента.

FilesystemFileSearchMiddleware вовсе не встраивается в цикл. Он приносит
агенту два инструмента, glob_search и grep_search, и ограничивает их корнем,
за который они не выходят.

Корень здесь, это папка с примерами урока, то есть агент ищет по тем файлам,
которые вы только что запускали. Поиск идёт средствами Python: так результат
не зависит от того, установлен у вас ripgrep или нет.
"""

from pathlib import Path

from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemFileSearchMiddleware

from course_model import build_model

ROOT = Path(__file__).parent

SYSTEM_PROMPT = (
    "Вы разбираете незнакомый репозиторий. Пользуйтесь поиском по файлам, "
    "а не догадками. Отвечайте по-русски, двумя-тремя фразами."
)

QUESTION = (
    "Сколько в репозитории файлов с расширением .py и в каких из них "
    "встречается ToolRetryMiddleware?"
)

agent = create_agent(
    model=build_model(temperature=0, max_tokens=512),
    tools=[],
    system_prompt=SYSTEM_PROMPT,
    middleware=[
        FilesystemFileSearchMiddleware(root_path=str(ROOT), use_ripgrep=False),
    ],
)

result = agent.invoke({"messages": [{"role": "user", "content": QUESTION}]})

print("что агент спрашивал у поиска:")
for message in result["messages"]:
    if message.type == "ai":
        for call in message.tool_calls:
            print(f"  {call['name']}({call['args']})")

print()
print("ответ:", result["messages"][-1].text.replace("\n", " "))

# Вывод:
# что агент спрашивал у поиска:
#   glob_search({'pattern': '**/*.py'})
#   grep_search({'pattern': 'ToolRetryMiddleware', 'include': '*.py'})
#
# ответ: В репозитории 7 файлов с расширением `.py`: `course_model.py`, `05_todo_list.py`, `06_file_search.py`, `04_model_fallback.py`, `03_order.py`, `02_tool_error.py`, `01_hook_map.py`. Из них `ToolRetryMiddleware` встречается в двух файлах: `01_hook_map.py` и `06_file_search.py`.

# Неправильно: гасящий middleware оказался внутри повторяющего
agent = create_agent(
    model=model,
    tools=[search_db],
    middleware=[
        ToolRetryMiddleware(max_retries=3, on_failure="error"),
        ToolErrorMiddleware(on_error),
    ],
)

# Правильно: гасящий middleware снаружи, повторяющий внутри
agent = create_agent(
    model=model,
    tools=[search_db],
    middleware=[
        ToolErrorMiddleware(on_error),
        ToolRetryMiddleware(max_retries=3, on_failure="error"),
    ],
)

# Неправильно: обработчик ждёт ошибку разбора аргументов
def on_error(exc, request):
    if isinstance(exc, ValidationError):
        return "Аргументы не подошли, проверьте типы."
    return None

# Правильно: обработчик разбирает исключения своего кода
def on_error(exc, request):
    if isinstance(exc, (ConnectionError, TimeoutError)):
        return f"Источник недоступен ({type(exc).__name__}), попробуйте позже."
    return None