Последние года 4 я использовал Apache Airflow исключительно в облаке, преимущественно в Amazon — Amazon Managed Apache Airflow. И как обычно бывает, в облаках всё так или иначе между собой связано. Логи хранятся в Cloud Watch, воркеры запускаются в изолированной среде (Amazon Fargate). С июля месяца я стал активно использовать self-hosted Airflow на своих серверах (для своих личных целей), и в целях экономии храню всё в файлах. Так уж получилось, что задачу с регулярной "чисткой" я постоянно откладывал и вот настал час X, когда всё легко из-за исчерпания inodes в файловой системе. Для этого случая я написал DAG, который каждый день в полночь чистит папки со старыми логами, делюсь с вами вдруг он пригодится:

import os
import shutil
from datetime import datetime, timedelta

import pendulum
import structlog
from airflow.sdk import DAG, task

logger = structlog.get_logger(__name__)

@task
def cleanup_airflow_logs(days_to_keep):
    log_base_path = os.environ.get("AIRFLOW_HOME", "/opt/airflow") + "/logs"
    cutoff_date = datetime.now() - timedelta(days=days_to_keep)

    for root, dirs, files in os.walk(log_base_path):
        for dir_name in dirs:
            dir_path = os.path.join(root, dir_name)
            try:
                if os.path.getmtime(dir_path) < cutoff_date.timestamp():
                    logger.info(f"Deleting old log directory: {dir_path}")
                    shutil.rmtree(dir_path)
            except Exception as e:
                logger.error(f"Error deleting directory {dir_path}: {e}")


with DAG(
    dag_id="airflow_log_cleanup_dag",
    start_date=pendulum.datetime(2025, 10, 1, tz="Asia/Almaty"),
    schedule="@daily",  # Run daily at midnight
    catchup=False,
    default_args={
        "owner": "airflow",
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
    },
    max_active_runs=1,
) as dag:
    cleanup_airflow_logs(days_to_keep=14)

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

Кстати, чтобы написать свой простейший MCP сервер очень просто:

from fastmcp import FastMCP

mcp = FastMCP("Demo 🚀")

@mcp.tool
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

FastMCP - это небольшая обертка над библиотекой Starlette, которая позволяет легко создавать MCP серверы - тут вы видите и noscript, и denoscription, и inputSchema каждого метода.

#MCP #протоколы #LLM #AI

12 Factor Agents - принципы разработки агентов

По следам от 12 Factor Apps, Декстер Хорти собрал ключевые принципы по разработке агентов, делюсь!

1) Преобразование естественного языка в вызовы инструментов. Агент интерпретирует команды в формализованные вызовы функций, а исполняет их — детерминированный код
2) Владей своими промптами (как кодом). Промпт — это артефакт. Его надо хранить, версионировать, тестировать и относиться к нему не как к важной части кода.
3) Управляй контекстом агента. Активно контролируй, что попадает в это окно. Убирай уже не нужное, суммаризируй, оставляй важное, выделяй ограниченные слоты под фичи — внимание модели ограниченно.
4) Вызов тулинга — это просто JSON. LLM генерирует название тула и параметры, остальное должен делать ваш детерминированный вход. Модель решает "что делать", а ваше приложение "как делать"
5) Держи тех.состояние и бизнес состояние вместе. Каждый шаг агента — это и логика, и данные. Всё в одном состоянии, чтобы потом не ковыряться по логам в поисках, где оно сломалось
6) API для управление циклом агента. Агент должен иметь простой API для запуска, приостановки и возобновления работы, уметь вставать на паузу и возвращаться к работе спустя время на каждом этапе жизненного цикла
7) Запрос к человеку — как инструмент. Любое обращение в процессе работы к человеку должно быть представлено агенту, как инструмент requestClarification(message:string)
8) Явно контролируйте логику исполнения. Модель лишь "говорит" что, а твой код решает стоит ли выполнять, когда и как.
9) Добавляйте ошибки исполнения инструментов в контекст. Если что-то упало, сократи информацию и добавь это в контекст агента. Пусть модель сама догадается, что дальше: повторить, спросить или изменить стратегию
10) Используйте малых специализированных агентов. Лучше много маленьких агентов, решающих узконаправленные задачи, чем один перегруженный агент всего
11) Доставляй агентов в любые интерфейсы, удобные пользователю. Slack, Telegram, веб, CLI — неважно. Главное — единый API и омниканальность. Пользователь не должен думать, где общаться с агентом. Он просто должен это делать.
12) Агент как Stateless Reducer. На входе — состояние и запрос. На выходе — новое состояние и ответ. Всё как в хорошем Redux. Никакого собственного внутреннего состояния, он должен работать как чистая функция

Бонус:
13) Предзагрузи все необходимое. Агент не должен постоянно дёргать одни и те же данные. Наполняй контекст тем, что почти наверняка пригодится.

Репа: https://github.com/humanlayer/12-factor-agents
Доклад: https://www.youtube.com/watch?v=8kMaTybvDUw

Оказывается, если вы хотите подчеркнуть, что ваш контент "NOT BY AI", то для вас уже сделали специальный trademark - https://notbyai.fyi/

Эти ребята сделали целый бизнес по продаже этих вотермарок (не знаю, насколько успешный). Для индивидуального использования вы можете просто задонатить сколько хотите - и значки отправят вам на почту. Но корпоративные юзер - расчехляйте кошельки🌚

Просто удивительно, на чем только люди не делают деньги

Как я чищу метаданные Apache Airflow 3.x

В блоге Apache Airflow есть скрипт для периодической чистки метаданных в БД Apache Airflow. Но начиная с Airflow 3.0 невозможно обращаться к БД напрямую (через модели SQLAlchemy), в связи с этим скрипт не работает. Даже вызов shell команды airflow db clean через BashOperator выдаст ошибку по типу:

Could not parse SQLAlchemy URL from string 'airflow-db-not-allowed:///': source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"

RuntimeError: Direct database access via the ORM is not allowed in Airflow 3.0

Как решить? Я нашел выход запуска через старый добрый Cron:

0 0 * * * /home/airflow/.airflow/bin/airflow db clean --clean-before-timestamp "$(date -d '7 days ago' +'%Y-%m-%d %H:%M:%S')" --skip-archive -y

В полночь команда запускается и чистит все данные , оставляя только последние 7 дней. Но это работает на моём личном self-hosted Airflow, как быть с MWAA я пока не знаю, т.к. не обновлялся и не знаю как поведёт себя DAG.

Agents 2.0: от примитивных циклов к глубокомыслящим агентам

Год назад ИИ-агент был простым while-циклом: промпт → LLM → инструмент → повтор. Решало обычные быстрые задачи типа "какая погода в Токио", но на задачах в 50 шагов, который могут выполняться несколько дней агент терялся, зацикливался, галлюцинировал.

Сейчас архитектура меняется. Deep Agents не просто реагируют, они планируют, делегируют работу специализированным саб-агентам и хранят память вне контекста.

В заметке выделяется четыре ключевые составляющие:
- Явное детальное планирование через цепочку размышлений с сохранением в markdown to-do с отслеживанием прогресса
- Иерархическая оркестрация с распределенением задач между узкими специалистами под-агентами (researcher, coder, writer) и именно они уже вызывают инструменты. Оркестратору возвращается только резальтат
- Внешняя память в файловой системе или векторных БД вместо переполненного контекста. Причём часть под-агентов могут только писать в память, а часть - только читать!
- Промпты на тысячи токенов с детальными инструкциями и указанием структур доступных инструментов и когда стоит подключить человека.

Интересное краткое отражение факта эволюции развития агентских систем.
Полезно и для понимания архитектуры и для презентаций.

#DeepAgents #Orchestrator #SubAgents
———
@tsingular

На днях в open source выпустили распределённую файловую систему, которая рассчитана на эксабайты (тысячи петабайт).

Сделали это чуваки из XTX, мощные трейдеры, которые известны двумя вещами: тем, что у них (по крайней мере недавно) был топ-3 кластер по количеству ГПУ, и тем, что их основатель, Александр Герко, так любит Лондон, что каждый год платит 500+ млн фунтов налогов на доходы как физическое лицо.

Из интересного (они выделили 9 пунктов, но только 5 мне кажутся отличительными)

Has no single point of failure in its metadata services.
Is hardware agnostic and uses TCP/IP to communicate.
Utilizes different types of storage (such as flash vs. hard disks) cost effectively.
Exposes read/write access through its own API over TCP and UDP, and a Linux kernel filesystem module.
Requires no external service and has a minimal set of build dependencies

Начали работы над системой в 2022 году, в середине 2024 мигрировали весь ML

TernFS' metadata is split into 256 logical shards. Shards never communicate with each other. This is a general principle in TernFS: Splitting the metadata into 256 shards from the get-go simplifies the design, given that horizontal scaling of metadata requires no rebalancing, just the addition of more metadata servers.

Ну и заодно свой формат сериализации разработали, чтобы разработчики передвигали не json, thrift, а что-то там свое.

Еще из интересного - обсуждение когда нужно зеркалить файлы, а когда делать Reed-Solomon coding.

Рекомендую почитать

The Annual Airflow Survey

Если вы пользуетесь Apache Airflow, то давайте поможем сообществу собрать больше информации об использовании Airflow. До 20 ноября необходимо заполнить опросник на сайте взамен вы получите возможность пройти сертификацию по Airflow от Astronomer бесплатно (я сдавал их экзамен и он хороший). Опросник небольшой, замёт не более 10 минут вашего времени.

Ребята из Qdrant запустили бесплатный семидневный курс про свою векторную базу данных: Qdrant Essentials

Меня в последнее время интересует тема векторного поиска и векторных БД, и Qdrant как раз неплохой кандидат на изучение и реализацию какого-нибудь проекта (спойлер: домашнего векторного поисковика). Курс прям как по заказу!