Data2Good | Про данные, финансы, технологии и жизнь

Кстати это очень важный шаг, так как работа с RAG и грамотная комбинация классического ML с LLM моделями - это залог конкурентного преимущества

S3 Vector Bucket для демократизации RAG В Amazon S3 появился новый тип «vector bucket» (часть превью-функции S3 Vectors). Он хранит и индексирует эмбеддинги так же дёшево и надёжно, как обычные объекты S3, при этом предоставляет быстрый поиск и отдельный API. По расчётам AWS, перенос «холодных» векторов из постоянно работающей базы OpenSearch в S3 снижает совокупные расходы на хранение и поиск до 90 % — именно то, чего не хватало массовым RAG-системам, где объём эмбеддингов растёт быстрее, чем сами модели. Retrieval-Augmented Generation (RAG) добавляет внешние знания к LLM, но требует постоянного пополнения и переиндексации эмбеддингов. Vector Buckets делают векторизацию «коммодити»-услугой: вместо собственных кластеров Milvus/Faiss теперь достаточно S3. Это удешевляет вход в RAG-проекты, а значит, ускорит внедрение корпоративных ассистентов, поисковых движков и агентных ИИ-систем. Следующий логичный шаг — появление «умных» объектных сторажей, где векторный поиск, версионирование и AI-функции войдут в ядро, а не будут внешними аддонами. И именно над такой гибридной архитектурой мы в BAUM уже работаем 🧠 Подробнее - в блоге BAUM! ❤️ #BaumTechPulse

Кстати это очень важный шаг, так как работа с RAG и грамотная комбинация классического ML с LLM моделями - это залог конкурентного преимущества.

а это пример задачек по цифровой трансформации

Сегодня тестировал GPT-5 для написания кода, не могу сказать, что есть кардинальные отличия... И кстати в Gemini контекст расширяется до 2 млн токенов... Так что сейчас становится ощущение, что прокачка GPT, это как айфон. Разница между 15 и 16 настолько мала, что почти не видна. Или нет?🤔

У меня есть цель: Создать ИИ платформу, которой будут пользоваться десятки тысяч пользователей И в последней итерации наша платформа обросла реальным маркетплейсом, где каждый желающий может создавать свои плагины, и творить ИИ своей мечты... И вот что я нашел! ЭТО 500 ИИ АГЕНТОВ на всех (ну почти на все) случаи жизни) дублирую ссылку: https://github.com/Lasttrader/ai-agents-marketplace И вот порой кажется, что всё сходиться - продукт, рынок, пользователи... Но вклиниваются события, которые призваны тебя отбросить еще на какое-то расстояние назад... и опять отдалить тебя от цели... 🤔 P.S. Пользуйтесь репозиторием, и не забывайте ставить звездочки 🫡🌟 И возможно, в скором времени увидимся с вами на просторах платформы ИИ, где уже вы сможете создать ассистентов и агентов своей мечты.

Коллеги, всем привет! Создал свой небольшой блог в телеге. Где собираю контент из свей личной и профессиональной жизни Язык блога: Русский и Английский С удовольствием приглашаю вас присоединиться! https://t.me/sgv1985

Цифра дня: 8️⃣0️⃣🔤 Топ менеджеров компаний России не понимают ценности цифровизации бизнеса и пользы от ИИ... P.s. кстати, мы даже были вынуждены создать воркшопы по повышению квалификации ключевых сотрудников компаний, чтобы говорить с ними на одном языке...🤷‍♀️

Входной билет будет символическим (в районе 3000 руб. скинуться на токены и пиццу команде)

Современный маркетинг — это что-то невероятное. покупал кухню — оказалась бракованная столешница. Пришлось ехать менять. Чтобы время не терять, заказал срочную доставку через «Яндекс». Продиктовал номер… и буквально через 10-15 минут мне уже звонит робот! Предлагает «экспресс-ремонт» и строительные услуги. Вот она — скорость реакции. Вот оно — будущее, где технологии настигают тебя мгновенно. Человеку в этом мире будет всё труднее сопротивляться. Мы уже внутри системы, где алгоритмы знают о нас больше, чем мы сами. 💭 Как думаете, это тотальный контрольи или тотальное удобство?🤔

Ну и на последок - боль перфекциониста )))

От всей ИИ души )

# Задача: получить файл из S3 fetch_s3_task = PythonOperator( task_id='fetch_s3', python_callable=fetch_s3 ) # Задача: получить файл с NFS fetch_nfs_file_task = PythonOperator( task_id='fetch_nfs_file', python_callable=fetch_nfs_file ) # Задача: получить локальный файл fetch_local_file_task = PythonOperator( task_id='fetch_local_file', python_callable=fetch_local_file ) # Задача: агрегировать и сформировать финальный ответ aggregate_task = PythonOperator( task_id='aggregate_and_respond', python_callable=aggregate_and_respond, provide_context=True # Передаёт context (для xcom_pull) ) # Устанавливаем зависимости: все fetch задачи → агрегация [ fetch_postgresql_task, fetch_clickhouse_task, fetch_s3_task, fetch_nfs_file_task, fetch_local_file_task ] >> aggregate_task

Ну а бонус, я тут наклепал (не без помощи ИИ😐) пример готового кода ассистента с графами! # Импортируем необходимые компоненты из airflow from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime # Импортируем библиотеки для работы с разными источниками данных import psycopg2 # Для PostgreSQL import boto3 # Для подключения к Amazon S3 import fsspec # Универсальный доступ к файловым системам, включая NFS from clickhouse_driver import Client # Для подключения к ClickHouse import os # Работа с файловой системой # Функция для получения данных из PostgreSQL def fetch_postgresql(**kwargs): # Устанавливаем соединение с Postgres conn = psycopg2.connect( dbname='your_db', user='user', password='pass', host='localhost', port=5432 ) # Создаём курсор для выполнения SQL-запроса cur = conn.cursor() # Выполняем SQL-запрос cur.execute("SELECT * FROM your_table LIMIT 10;") # Забираем результаты result = cur.fetchall() # Закрываем соединение cur.close() conn.close() # Возвращаем результат return result # Функция для получения данных из ClickHouse def fetch_clickhouse(**kwargs): # Подключаемся к ClickHouse-серверу client = Client('localhost') # Выполняем SQL-запрос data = client.execute('SELECT * FROM your_table LIMIT 10') # Возвращаем результат return data # Функция для получения данных из AWS S3 def fetch_s3(**kwargs): # Создаём клиент автомата S3 s3 = boto3.client('s3') # Загружаем объект (например, JSON-файл) obj = s3.get_object(Bucket='your-bucket', Key='example.json') # Считываем содержимое файла data = obj['Body'].read() # Возвращаем данные (в байтах) return data # Функция для чтения файла с NFS-ресурса def fetch_nfs_file(**kwargs): # Путь к файлу на подключённом сетевом хранилище file_path = '/mnt/nfs/data_file.txt' # Читаем содержимое файла with open(file_path, 'r') as f: return f.read() # Функция для чтения локального файла def fetch_local_file(**kwargs): # Путь к файлу в локальной ФС file_path = 'local_data.txt' # Читаем содержимое with open(file_path, 'r') as f: return f.read() # Функция для агрегации данных из всех источников def aggregate_and_respond(**context): # Вытаскиваем данные из предыдущих задач через XCom pg_data = context['ti'].xcom_pull(task_ids='fetch_postgresql') ch_data = context['ti'].xcom_pull(task_ids='fetch_clickhouse') s3_data = context['ti'].xcom_pull(task_ids='fetch_s3') nfs_data = context['ti'].xcom_pull(task_ids='fetch_nfs_file') file_data = context['ti'].xcom_pull(task_ids='fetch_local_file') # Возвращаем агрегированные данные в виде словаря return { "PostgreSQL": pg_data, "ClickHouse": ch_data, "S3": s3_data.decode() if isinstance(s3_data, bytes) else s3_data, # Преобразуем байты в строку "NFS": nfs_data, "File": file_data, } # Настройки DAG: владелец, дата запуска (в прошлом, чтобы не было catchup) default_args = { 'start_date': datetime(2023, 1, 1), 'owner': 'ai_assistant', } # Определяем DAG — главный объект пайплайна with DAG( 'multi_source_dag_pipeline', # Уникальное имя пайплайна default_args=default_args, # Параметры по умолчанию schedule_interval=None, # Запускается вручную catchup=False # Не выполняет предыдущие пропущенные запуски ) as dag: # Задача: получить данные из PostgreSQL fetch_postgresql_task = PythonOperator( task_id='fetch_postgresql', # ID задачи python_callable=fetch_postgresql # Указание, какую функцию вызывать ) # Задача: получить данные из ClickHouse fetch_clickhouse_task = PythonOperator( task_id='fetch_clickhouse', python_callable=fetch_clickhouse )

Коллеги, привет! Фуф, последние 1,5 года усиленно пилим ИИ ассистентов, для разных бизнес сфер. Это вылилось у создание программно-аппаратного комплекса, где во всей красе сияют наша СХД и платформа ИИ. Хочу подсветить, как мы пилим ИИ ассистентов на базе DAG. 1. Что такое DAG pipeline? DAG (Directed Acyclic Graph) pipeline — это «умная» схема обработки данных, где каждый этап (узел) чётко связан с последующими и работает только после завершения своих «родителей». Такой подход часто используют в современных дата платформах (например, Airflow) для оркестрации сложных процессов. 2. Архитектура пайплайна для ИИ-ассистента Схема типового пайплайна для ассистента: 1. Получение запроса пользователя 2. Нормализация и обработка текста 3. Параллельный поиск в разных источниках: - ClickHouse - PostgreSQL - S3 - NFS - Локальные файлы 4. Агрегация и ранжирование результатов 5. Генерация финального ответа 6. Логирование и аналитика (по желанию) 3. Пример кода на Python

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import psycopg2, boto3, fsspec, os
from clickhouse_driver import Client

def fetch_postgresql(**kwargs):
    # Соединение и запрос к PostgreSQL
    ...

def fetch_clickhouse(**kwargs):
    # Соединение и запрос к ClickHouse
    ...

def fetch_s3(**kwargs):
    # Скачать файл с S3
    ...

def fetch_nfs_file(**kwargs):
    # Прочитать файл из сетевого хранилища (NFS)
    ...

def fetch_local_file(**kwargs):
    # Прочитать файл из локальной ФС
    ...

def aggregate_and_respond(**context):
    # Агрегация результатов от всех fetch_* функций
    ...

default_args = {
    'start_date': datetime(2023, 1, 1),
    'owner': 'ai_assistant',
}

with DAG('multi_source_dag_pipeline', default_args=default_args, schedule_interval=None, catchup=False) as dag:
    fetch_postgresql_task = PythonOperator(task_id='fetch_postgresql', python_callable=fetch_postgresql)
    fetch_clickhouse_task = PythonOperator(task_id='fetch_clickhouse', python_callable=fetch_clickhouse)
    fetch_s3_task = PythonOperator(task_id='fetch_s3', python_callable=fetch_s3)
    fetch_nfs_file_task = PythonOperator(task_id='fetch_nfs_file', python_callable=fetch_nfs_file)
    fetch_local_file_task = PythonOperator(task_id='fetch_local_file', python_callable=fetch_local_file)
    aggregate_task = PythonOperator(task_id='aggregate_and_respond', python_callable=aggregate_and_respond, provide_context=True)

    [fetch_postgresql_task, fetch_clickhouse_task, fetch_s3_task, fetch_nfs_file_task, fetch_local_file_task] >> aggregate_task

4. В чём смысл и основные плюсы? - Масштабируемость: Хотите добавить ещё одну базу знаний? Просто добавь воды! Просто добавьте новый узел! - Гибкость: Можно легко изменить этапы агрегации или поиска. - Наглядность: Логика пайплайна прозрачна — всегда видно, на каком этапе и в каком источнике возникли проблемы. - Параллельность: Данные собираются из нескольких систем одновременно — экономия времени! 5. Расширения - Добавить обработку текста с помощью LLM (например, OpenAI или YandexGPT) на этапе агрегации. - Встроить ассистента в Telegram-бот. - Реализовать мониторинг выполнения шагов пайплайна и автоматическую обработку ошибок.

Новый ИИ бот, пингует пользователей телеги? 🤔