AI в энтерпрайзе

Мнение.

Основные проблемы в корпоративном IT это, как и прежде: инфраструктура, безопасность и работа с данными.

AI и, так называемые, Агенты AI, в этой сфере, в ближайшие 2-3 года, мало что поменяют.

В корпоративном секторе столько неэффективности и реакционности, что буст продуктивности, который принесет AI станет каплей в море.

Миграция с Oracle на Postgresql или переезд с Lotus Domino, для большинства крупных не-IT компаний принесет больше пользы, чем внедрение AI, здесь и сейчас.

Без современной инфраструктуры и стека данных, внедрение AI не отобьет OPEX и тем более CAPEX.

Единственное революционное измерение, которые стоит ожидать уже в этом году – западные корпорации будут отказывается от аутотренинг, особенно в Индии.

Вот кстати пример, того как можно автоматизировать работу с данными.

Представим себе, что нам на почту присылают письма с вложенными csv, json и xml файлами. Скрипт, который запускается по расписанию и ищет письма с словом «documents» в теме и вложенным файлом в формате csv. Сохраняет эти файлы в папку.

import imaplib import email from email.header import decode_header import os import csv  IMAP_SERVER = "imap.yandex.com" EMAIL = "email@yandex.com" PASSWORD = "password" DOWNLOAD_FOLDER = "attachments"  if not os.path.exists(DOWNLOAD_FOLDER):     os.makedirs(DOWNLOAD_FOLDER)  def connect_to_imap():     try:         print(f"Connecting to IMAP server: {IMAP_SERVER}")         mail = imaplib.IMAP4_SSL(IMAP_SERVER, port=993)         mail.login(EMAIL, PASSWORD)         mail.select("INBOX")         print("Connected successfully.")         return mail     except imaplib.IMAP4.error as e:         print(f"IMAP error: {e}")         print("Check your email and password, or enable app passwords if 2FA is enabled.")         return None     except Exception as e:         print(f"Unexpected error: {e}")         return None  def decode_header_value(encoded_value):     decoded_parts = decode_header(encoded_value)     result = ""     for part, encoding in decoded_parts:         if isinstance(part, bytes):             try:                 result += part.decode(encoding or "utf-8")             except LookupError:                 result += part.decode("latin-1")             except UnicodeDecodeError:                 result += part.decode("utf-8", errors="ignore")         else:             result += part     return result  def extract_csv_attachments(msg):     for part in msg.walk():         if part.get_content_maintype() == "multipart":             continue         if part.get("Content-Disposition") is None:             continue          filename = part.get_filename()         if filename and filename.lower().endswith(".csv"):             filename = decode_header_value(filename)             print(f"Found CSV attachment: {filename}")              filepath = os.path.join(DOWNLOAD_FOLDER, filename)             with open(filepath, "wb") as f:                 f.write(part.get_payload(decode=True))             print(f"Saved CSV attachment: {filename}")  def fetch_emails():     mail = connect_to_imap()     if not mail:         return      try:         status, messages = mail.search(None, 'SUBJECT "documents"')         if status != "OK":             print("No emails found with 'documents' in the subject.")             return          email_ids = messages[0].split()         print(f"Found {len(email_ids)} emails with 'documents' in the subject.")          for mail_id in reversed(email_ids):             status, data = mail.fetch(mail_id, "(RFC822)")             if status != "OK":                 continue              msg = email.message_from_bytes(data[0][1])             subject = decode_header_value(msg["Subject"])             print(f"Processing email: {subject}")              extract_csv_attachments(msg)     finally:         mail.logout()  if __name__ == "__main__":     fetch_emails() 

Другой скрипт забирает csv и перекладывает их в duckDB – ультрабыстрая колоночная RDBMS, которая пользуется популярностью среди дата-аналитиков:

import os import duckdb  DUCKDB_PATH = "my_database.duckdb"  ATTACHMENTS_DIR = "attachments"  def send_csv_to_duckdb():     conn = duckdb.connect(DUCKDB_PATH)     print(f"Connected to DuckDB database: {DUCKDB_PATH}")      for filename in os.listdir(ATTACHMENTS_DIR):         if filename.endswith(".csv"):             filepath = os.path.join(ATTACHMENTS_DIR, filename)             table_name = os.path.splitext(filename)[0]             print(f"Processing file: {filename}")              quoted_table_name = f'"{table_name}"'              conn.execute(                 f"CREATE OR REPLACE TABLE {quoted_table_name} AS SELECT * FROM read_csv_auto('{filepath}')"             )             print(f"Data from {filename} sent to DuckDB table {quoted_table_name}.")      print("All CSV files processed.")  if __name__ == "__main__":     send_csv_to_duckdb() 

Metabase – один из лучших инструментов с открытым исходным кодом для аналитики и визуализации данных.

docker run -d -p 3000:3000 -v "$(pwd)":/srv --name metabase metabase/metabase 

Для работы с duckdb, необходимо установить плагин:

https://github.com/MotherDuck-Open-Source/metabase_duckdb_driver 

Дальше в дело вступает Sling это инструмент для репликации между базами и различными типами хранилищ. Вместо нагромождения UI и микросервисов вроде Airbyte, эти ребята решили пойти по простому пути и написали rsync или rclone для работы с большими данными.

curl -LO 'https://github.com/slingdata-io/sling-cli/releases/latest/download/sling_linux_amd64.tar.gz' \   && tar xf sling_linux_amd64.tar.gz \   && rm -f sling_linux_amd64.tar.gz \   && chmod +x sling  mv sling /usr/local/bin/ 

$ sling conns list +--------------------------+-----------------+-------------------+ | CONN NAME                | CONN TYPE       | SOURCE            | +--------------------------+-----------------+-------------------+ | AWS_S3                   | FileSys - S3    | sling env yaml    | | DO_SPACES                | FileSys - S3    | sling env yaml    | | LOCALHOST_DEV            | DB - PostgreSQL | dbt profiles yaml | | MSSQL                    | DB - SQLServer  | sling env yaml    | | MYSQL                    | DB - MySQL      | sling env yaml    | | ORACLE_DB                | DB - Oracle     | env variable      | | MY_PG                    | DB - PostgreSQL | sling env yaml    | +--------------------------+-----------------+-------------------+ 

Дата пайплайн можно описать в простом YAML файле. В данном случае мы будем брать csv файл и перекладывать его в PostgreSQL и minio s3 для долгосрочного хранения:

version: 1 sources:   duckdb_source:     type: duckdb     path: /path/to/your_database.duckdb  targets:   minio_target:     type: s3     bucket: <bucket>     access_key_id: <access_key_id>     secret_access_key: '<secret_access_key>'     endpoint: '<endpoint>'     url_style: path    postgres_target:     type: postgres     host: postgres_host     port: 5432     database: mydb     username: user     password: password  pipelines:   - name: duckdb_to_minio     source: duckdb_source     target: minio_target     query: "SELECT * FROM sales"     mode: overwrite    - name: duckdb_to_postgres     source: duckdb_source     target: postgres_target     query: "SELECT * FROM sales"     mode: overwrite 

Ну и наконец AI. Куда же без него?
Ollama это очень популярный проект для работы с AI моделями локально, предоставляет не только CLI но и отличную библиотеку на Python.

Сначала нужно установить зависимости и скачать модельку:

pip install ollama graphvizollama run qwen2.5-coder:7b 

Для своей весовой категории qwen2.5 это одна из лучших открытых LLM.

Скрипт с помощью qwen сделает краткое изложение csv на 10 гигабайт, а если надо нарисует графики и сконвертирует этот файл в нужный формат.

Все зависит от того, какой вы отправите промпт:

Скрипт:

#!/usr/bin/env python3 import csv import ollama import argparse import logging from typing import Optional  logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')  PROMPT_TEMPLATE = """[INST] <<SYS>> You are a data analysis expert. Analyze this CSV data and:  1. Summarize the data 2. Identify trends or patterns 3. Provide insights and recommendations  Format response with: - Markdown analysis - Key insights - Recommendations for further analysis <</SYS>>  CSV Data: {data} [/INST]"""  def read_csv(file_path: str) -> str:     try:         with open(file_path, "r") as file:             reader = csv.reader(file)             rows = [",".join(row) for row in reader]         return "\n".join(rows)     except Exception as e:         logging.error(f"Error reading CSV file: {e}")         raise  def analyze_csv(csv_file: str, context_size: Optional[int] = None) -> str:     try:         logging.info(f"Reading CSV file: {csv_file}")         csv_text = read_csv(csv_file)         logging.info(f"Extracted CSV text: {csv_text[:100]}...")          options = {}         if context_size:             options['max_tokens'] = context_size          logging.info("Sending data to Ollama model...")         response = ollama.chat(             model='qwen2.5-coder:7b',             messages=[{'role': 'user', 'content': PROMPT_TEMPLATE.format(data=csv_text)}],             options=options         )['message']['content']         logging.info("Received response from Ollama model")          return response      except Exception as e:         logging.error(f"An error occurred: {e}")         raise  if __name__ == "__main__":     parser = argparse.ArgumentParser(description='CSV AI Analyzer')     parser.add_argument('csv_file', help='Input CSV file')     parser.add_argument('--context-size', type=int, help='Optional context size (max tokens) for the Ollama model')     args = parser.parse_args()      try:         report = analyze_csv(args.csv_file, args.context_size)         print("\nAnalysis Report:")         print(report)     except Exception as e:         logging.error(f"Script failed: {e}")