Создание ограниченного агента LangGraph для диагностики инцидентов API

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

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

Это практическое описание схемы, а не проверенная интеграция или готовый к production-среде сервис. Приведённые ниже тела инструментов служат примерами: замените их собственными клиентами систем наблюдаемости с аутентификацией и ограниченными разрешениями.

Предварительные требования и рабочий процесс

Вам понадобятся Python, настроенная в приложении чат-модель, совместимая с LangGraph, и клиентские функции для используемых вами диагностических систем. В этом примере предполагается, что эти функции возвращают данные или вызывают исключения. Он не рассчитан на какого-либо конкретного поставщика моделей, конечную точку или интеграцию инструментов.

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

1. Соберите небольшой заранее определённый набор сигналов для затронутой службы и временного интервала.
2. Пометьте каждый результат как успешный, неудачный или недоступный.
3. Попросите модель обобщить данные и указать, что пока остаётся неизвестным.
4. Остановитесь после ограниченного числа шагов графа; прежде чем выполнять действия, меняющие состояние, запросите проверку человека.

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

Определите диагностические инструменты с явными сигналами ошибок

Контракт инструмента должен быть узким. Требуйте название службы и ограниченный временной интервал; не предоставляйте модели произвольный доступ к запросам, если достаточно фиксированного запроса. Возвращайте структурированные результаты, чтобы последующие узлы и отчёт об инциденте могли отличить отсутствие данных от ошибки инструмента.

from typing import Literal, TypedDict
from langchain.tools import tool

class Signal(TypedDict):
    status: Literal["ok", "error", "unavailable"]
    source: str
    data: str
    error: str | None


def query_logs(service: str, minutes: int) -> Signal:
    if not 1 <= minutes <= 60:
        return {
            "status": "error", "source": "logs", "data": "",
            "error": "minutes must be between 1 and 60",
        }
    try:
        # Replace with a read-only, scoped observability client call.
        rows = logs_client.search(service=service, minutes=minutes)
        return {
            "status": "ok", "source": "logs",
            "data": str(rows), "error": None,
        }
    except TimeoutError as exc:
        return {
            "status": "unavailable", "source": "logs", "data": "",
            "error": f"timeout: {exc}",
        }

@tool
def inspect_api(service: str, minutes: int = 15) -> dict:
    """Collect scoped, read-only API incident signals."""
    return query_logs(service, minutes)

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

Ограничьте выполнение и показывайте незавершённые запуски

Для агента, созданного с помощью create_agent в LangGraph, задайте лимит рекурсии графа при вызове. Этот лимит ограничивает число шагов графа, но не гарантирует полезный ответ и не заменяет тайм-ауты на уровне инструментов. Обработайте достижение лимита во внешнем приложении и возвращайте понятный статус «не завершено», а не выдавайте запуск за завершённую диагностику.

from langchain.agents import create_agent

agent = create_agent(
    model=your_configured_chat_model,
    tools=[inspect_api],
    system_prompt=(
        "Diagnose API incidents using available evidence. "
        "Report each source as ok, error, or unavailable. "
        "Do not claim a cause without supporting evidence. "
        "Do not perform remediation."
    ),
)

try:
    result = agent.invoke(
        {"messages": [{"role": "user", "content":
            "Investigate elevated errors for payments-api over the last 15 minutes."}]},
        config={"recursion_limit": 12},
    )
except Exception as exc:
    # Map known graph-limit and provider/tool failures to your app's
    # incident-run status; do not expose credentials or raw secrets.
    result = {"status": "incomplete", "reason": type(exc).__name__}

Указанное значение лимита приведено для примера и не является универсальной рекомендацией. Выбирайте его с учётом числа разрешённых инструментов и обращений к модели, а также задайте тайм-ауты и ограничения размера запросов на уровне инструментов и приложения. Записывайте ID запуска, выбранную службу и временной интервал, вызванные инструменты, их статусы и причину завершения. Эти данные позволяют оператору отличить ситуацию «ошибок не найдено» от ситуации «истёк тайм-аут серверной части журналов».

Добавьте проверку человеком перед устранением неполадок

Если позднее вы добавите инструменты, меняющие состояние production-среды, ограничьте их вызов с помощью middleware LangChain для human-in-the-loop и checkpointer. Документированное middleware может приостанавливать подходящие вызовы инструментов для подтверждения, редактирования, отклонения или прямого ответа. Состояние сохраняется, поэтому после принятия решения выполнение можно продолжить; для использования в production требуется постоянный checkpointer, а не показанный для прототипирования вариант в памяти. См. документацию LangChain по human-in-the-loop.

Практическая политика: разрешать диагностику только для чтения без прерывания, а для инструментов перезапуска, отката или изменения конфигурации требовать подтверждение. Отклоняйте предлагаемое действие именно как отклонённое: документация предупреждает, что вариант «respond» предназначен для выполнения действия от имени инструмента, а не для отказа от вызова с побочным эффектом. Проверяйте отредактированные аргументы в собственной реализации инструмента: документированное middleware не сверяет изменения с отображаемой схемой.

Какие сбои нужно учитывать

• Пустые результаты: сообщайте «подходящих событий не найдено», а не «служба работает нормально».
• Тайм-аут инструмента или ошибка авторизации: сохраняйте сведения о сбое источника и категорию ошибки; не повторяйте попытки бесконечно без уведомления.
• Повторные вызовы инструментов: ограничьте выполнение графа и задайте ограничения для входных данных инструментов. Если вы добавляете повторы, установите небольшую явную политику повторных попыток и показывайте число попыток.
• Не удаётся возобновить прерванный запуск: проверьте, настроен ли checkpointer и используется ли тот же идентификатор потока при возобновлении. В исходной документации указано, что для проверки человеком требуется сохранение контрольных точек.
• Вводящее в заблуждение итоговое резюме: требуйте, чтобы утверждения в отчёте ссылались на собранные сигналы, и перечисляйте недоступные источники и нерешённые вопросы.

Контрольный список для запуска при инциденте

Перед развертыванием убедитесь, что диагностические инструменты работают только для чтения и имеют ограниченную область доступа; временные интервалы и объём запросов ограничены; ошибки инструментов видны; выполнение графа ограничено; а каждое действие с побочными эффектами приостанавливается до решения человека. Проверьте сценарии тайм-аута, пустого результата, отклонённого действия и возобновления после прерывания. Хороший агент не просто предлагает правдоподобную первопричину — он показывает, какие данные действительно получил и где заканчиваются результаты диагностики.

Источники

• LangChain: middleware для human-in-the-loop — решения о прерывании, выборочная проверка вызовов инструментов, сохранение контрольных точек и возобновление приостановленного выполнения.