Уведомления из LangGraph

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

Подготовьте Депешер

Сначала создайте и подтвердите исходящий канал: Email, Telegram, группу Telegram, Пачку или Depesher Desktop. Затем создайте входящий канал для этого инструмента, свяжите его с выбранным получателем и скопируйте публичный URL. Выполните проверку из статьи «Первое уведомление».

Добавьте отправку в приложение

Пример предполагает уже скомпилированный объект graph и список входных messages. Граф должен использовать поле messages в состоянии. Сохраните аргументы конфигурации своего вызова: если используется checkpointer, передайте прежний config с thread_id; ниже показан базовый вызов без сохранения состояния. Основы описаны в документации LangGraph.

В Python-примере замените DEPESHER_URL адресом своего канала. Для защищённого канала заполните DEPESHER_TOKEN токеном без префикса Bearer; иначе оставьте пустую строку. Нужен Python 3, дополнительных Python-пакетов для HTTP-запроса не требуется.

Разместите функцию отправки в коде приложения:

from urllib.request import Request, urlopen

DEPESHER_URL = "https://depesher.ru/w/ВАШ_PUBLIC_ID"
DEPESHER_TOKEN = ""


def send_notification(message):
    headers = {"Content-Type": "text/plain"}
    if DEPESHER_TOKEN:
        headers["Authorization"] = f"Bearer {DEPESHER_TOKEN}"

    request = Request(
        DEPESHER_URL,
        data=message.encode("utf-8"),
        headers=headers,
        method="POST",
    )
    with urlopen(request, timeout=15):
        pass

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

import logging

try:
    result = graph.invoke({"messages": messages})
except Exception as error:
    try:
        send_notification(
            f"LangGraph: ошибка вызова {type(error).__name__}"
        )
    except Exception:
        logging.exception("Не удалось отправить уведомление об ошибке")
    raise
else:
    if result.get("__interrupt__"):
        notification = "LangGraph ожидает продолжения после interrupt"
    else:
        result_messages = result.get("messages", [])
        content = result_messages[-1].content if result_messages else ""
        if isinstance(content, str) and content.strip():
            notification = content
        else:
            notification = "LangGraph завершил вызов; результат смотрите в приложении"

    try:
        send_notification(notification)
    except Exception:
        logging.exception("Не удалось отправить результат в Депешер")

Проверка типа content нужна, потому что содержимое сообщения может быть строкой или набором структурированных блоков. В примере для блоков отправляется короткий статус; извлечение текста настройте под формат своего приложения.

Проверьте результат и ошибку

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

Если отправка не проходит, проверьте URL, токен и подключение получателей. Ветка ошибки передаёт тип исключения; подробности остаются в приложении.

Другие режимы выполнения

Возврат из invoke() не всегда означает завершение всего процесса: граф может остановиться на interrupt и ожидать ввода человека. Для стандартного результата с __interrupt__ пример отправляет отдельный текст. Если вы используете другую версию формата результата или собственное состояние паузы, адаптируйте проверку. Возобновление графа описано в документации interrupts. Для ainvoke() используйте асинхронную отправку или вынесите синхронный HTTP-запрос из цикла событий.

Передайте состояние как JSON

Если ошибка, ожидание и завершение нужны разным людям, настройте рецепт состояний AI и workflow. Сохраните из него модуль depesher_events.py рядом с приложением и задайте URL/токен через переменные окружения. Следующий код заменяет текстовую обёртку выше; одновременно обе использовать не нужно.

Пример рассчитан на стандартный graph.invoke() с результатом-словарём. Он сохраняет config с thread_id для вашего уже скомпилированного графа с checkpointer. Один run_id связывает первоначальный вызов и последующие возобновления в вашем приложении.

import logging

from depesher_events import send_event


def invoke_with_notification(graph, graph_input, config, run_id):
    def notify(status, message):
        try:
            send_event("langgraph", status, run_id, message)
        except Exception:
            logging.exception("Не удалось отправить состояние в Депешер")

    try:
        result = graph.invoke(graph_input, config=config)
    except Exception as error:
        notify("failed", f"Ошибка вызова: {type(error).__name__}")
        raise

    if result.get("__interrupt__"):
        notify("waiting_for_input", "Нужно решение человека в приложении")
    else:
        notify("completed", "Вызов графа завершён; результат смотрите в приложении")
    return result

Первый запуск и возобновление:

from langgraph.types import Command

run_id = "analysis-42"
config = {"configurable": {"thread_id": run_id}}

result = invoke_with_notification(
    graph,
    {"messages": messages},
    config,
    run_id,
)

После решения человека выполните отдельно с тем же config и run_id. True — пример для графа, который ожидает логическое подтверждение:

resumed = invoke_with_notification(
    graph,
    Command(resume=True),
    config,
    run_id,
)

Содержимое Command(resume=...) выбирает ваше приложение по контракту своего interrupt. В стандартном invoke() пауза отражается в __interrupt__; потоковые и другие форматы требуют своей проверки, как указано в документации LangGraph.

Проверьте три отдельных пути: возврат без interrupt → completed; пауза → waiting_for_input; контролируемое исключение → failed. Убедитесь, что обёртка возвращает исходный результат и сохраняет исходное исключение даже при сбое уведомления. Затем отдельно возобновите паузу и проверьте новое событие. Уведомление об ожидании не возобновляет граф и не закрывается автоматически последующим сообщением.