Депешер

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

После синхронного вызова agent.invoke() приложение может отправить результат агента в Депешер. Это удобно для длительного анализа или обработки запроса: пользователь получает уведомление в почте или мессенджере.

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

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

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

Пример предполагает уже созданный объект agent и список входных messages. Создание агента и вызов invoke описаны в документации LangChain.

В 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 = agent.invoke({"messages": messages})
except Exception as error:
    try:
        send_notification(
            f"LangChain: ошибка вызова {type(error).__name__}"
        )
    except Exception:
        logging.exception("Не удалось отправить уведомление об ошибке")
    raise
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 = "LangChain завершил вызов; результат смотрите в приложении"

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

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

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

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

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

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

Для ainvoke() используйте await и асинхронный HTTP-клиент либо вынесите синхронную отправку из цикла событий. При потоковом запуске размещайте уведомление после завершения чтения потока, если нужен общий итог.

Структурированные состояния ​

Для разных получателей ошибки и результата используйте рецепт состояний AI и workflow и его Python-модуль depesher_events.py. В существующей обёртке замените текстовую отправку в ветке исключения на send_event("langchain", "failed", run_id, "Ошибка вызова агента"), а в успешной ветке — на send_event("langchain", "completed", run_id, "Вызов агента завершён"). Импортируйте send_event из модуля, задайте run_id идентификатором задачи вашего приложения и сохраните обработку ошибок отправки и повторный выброс исходного исключения.

Добавьте строку langchain в список допустимых source у трёх правил общего рецепта. Иначе эти события попадут в default. Успешный вызов означает завершение отслеживаемого этапа; ожидание ввода требует отдельной подтверждённой ветки вашего приложения. Поиск запуска и важности — история сообщений.