Депешер

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

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

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

Сначала создайте и подтвердите исходящий канал: Email, Telegram, группу Telegram или Пачку. Затем создайте входящий канал для этого инструмента, свяжите его с выбранным получателем и скопируйте публичный 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-клиент либо вынесите синхронную отправку из цикла событий. При потоковом запуске размещайте уведомление после завершения чтения потока, если нужен общий итог.