Депешер

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

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

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

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

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

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

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

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

python
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

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

python
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-запрос из цикла событий.