AI i Machine Learning w Elixirze - Nx, Axon i Bumblebee bez Pythona

„Chcemy dodać AI do naszego systemu" - słyszymy to coraz częściej. Zwykle oznacza to: postawić osobny serwis w Pythonie (Flask/FastAPI), podłączyć przez HTTP, zarządzać kolejnym deploymentem, monitorować kolejny serwer, debugować komunikację między dwoma runtime'ami.

A co jeśli model ML mógłby działać w tym samym procesie co Twoja aplikacja? Bez Pythona, bez osobnego serwisu, bez HTTP między komponentami?

Od 2022 roku Elixir ma pełny ekosystem ML: Nx (obliczenia numeryczne), Axon (sieci neuronowe), Bumblebee (pretrained modele z Hugging Face). Wszystko natywne, wszystko w BEAM.

Ekosystem ML w Elixirze

graph TD
    APP["Twoja aplikacja
(Phoenix + LiveView)"]
    BB["Bumblebee
Pretrained modele
(GPT, BERT, Whisper, Stable Diffusion)"]
    AX["Axon
Budowanie i trenowanie
własnych sieci neuronowych"]
    NX["Nx
Tensory, operacje numeryczne
(odpowiednik NumPy)"]
    EXLA["EXLA / Torchx
Backendy obliczeniowe (Google XLA / LibTorch)
GPU (CUDA) lub CPU"]

    APP --> BB --> AX --> NX --> EXLA

Bumblebee - pretrained modele bez trenowania

Bumblebee to biblioteka, która pozwala uruchomić modele z Hugging Face bezpośrednio w Elixirze. Bez Pythona, bez konwersji, bez HTTP.

Klasyfikacja tekstu (analiza sentymentu)

# Załaduj model sentymentu - raz, przy starcie aplikacji
{:ok, model_info} = Bumblebee.load_model({:hf, "finiteautomata/bertweet-base-sentiment-analysis"})
{:ok, tokenizer} = Bumblebee.load_tokenizer({:hf, "vinai/bertweet-base"})

serving = Bumblebee.Text.text_classification(model_info, tokenizer,
  compile: [batch_size: 8],
  defn_options: [compiler: EXLA]
)

# Użycie - wywołanie funkcji, nie HTTP request
Nx.Serving.run(serving, "Produkt świetny, szybka dostawa, polecam!")
#=> %{predictions: [%{label: "POS", score: 0.98}]}

Nx.Serving.run(serving, "Towar uszkodzony, zwracam i żądam zwrotu pieniędzy")
#=> %{predictions: [%{label: "NEG", score: 0.95}]}

Integracja z Phoenix - Nx.Serving jako proces BEAM

# application.ex - model uruchamiany jako proces supervisowany
defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      MyApp.Repo,
      {Phoenix.PubSub, name: MyApp.PubSub},
      # Model ML jako proces BEAM - supervise, restart, monitoring
      {Nx.Serving, serving: sentiment_serving(), name: MyApp.SentimentAnalysis},
      MyAppWeb.Endpoint
    ]

    Supervisor.start_link(children, strategy: :one_for_one)
  end

  defp sentiment_serving do
    {:ok, model_info} = Bumblebee.load_model({:hf, "finiteautomata/bertweet-base-sentiment-analysis"})
    {:ok, tokenizer} = Bumblebee.load_tokenizer({:hf, "vinai/bertweet-base"})

    Bumblebee.Text.text_classification(model_info, tokenizer,
      compile: [batch_size: 8],
      defn_options: [compiler: EXLA]
    )
  end
end

# Użycie w dowolnym miejscu aplikacji:
Nx.Serving.batched_run(MyApp.SentimentAnalysis, review_text)

Model ML działa w tym samym drzewie supervisorów co reszta aplikacji. Jeśli padnie - supervisor go zrestartuje. Jeśli potrzebuje aktualizacji - hot code reload. Monitoring, logi, telemetria - te same narzędzia co dla reszty systemu.

Automatyczne batching

# Nx.Serving automatycznie grupuje requesty w batche
# Jeśli 8 użytkowników wyśle tekst do analizy w ciągu 50 ms
# → model przetworzy je jako 1 batch (szybciej niż 8 osobnych wywołań)

# Konfiguracja:
Bumblebee.Text.text_classification(model_info, tokenizer,
  compile: [batch_size: 16],          # Max rozmiar batcha
  defn_options: [compiler: EXLA],
  batch_timeout: 50                    # Czekaj max 50 ms na batch
)

# Efekt:
# - Niskie obciążenie: requesty przetwarzane natychmiast
# - Wysokie obciążenie: batching zwiększa throughput 4-8×
# - Automatyczne, zero konfiguracji po stronie kodu

Praktyczne zastosowania w systemach biznesowych

1. Automatyczna kategoryzacja zgłoszeń helpdesk

defmodule MyApp.Tickets.AutoCategorizer do
  @categories ["billing", "technical", "shipping", "returns", "general"]

  def categorize(ticket_text) do
    result = Nx.Serving.batched_run(MyApp.TextClassifier, ticket_text)

    %{label: category, score: confidence} =
      Enum.max_by(result.predictions, & &1.score)

    if confidence > 0.75 do
      {:ok, category}
    else
      {:uncertain, category, confidence}
    end
  end
end

# Integracja z systemem ticketowym:
defmodule MyApp.Tickets do
  def create_ticket(attrs) do
    with {:ok, ticket} <- insert_ticket(attrs),
         {:ok, category} <- AutoCategorizer.categorize(ticket.description) do
      assign_to_team(ticket, category)
    else
      {:uncertain, suggested, confidence} ->
        # Zaproponuj kategorię, ale niech człowiek potwierdzi
        update_ticket(ticket, %{
          suggested_category: suggested,
          auto_confidence: confidence,
          needs_review: true
        })
    end
  end
end

2. Wykrywanie anomalii w danych biznesowych

defmodule MyApp.AnomalyDetector do
  @doc "Wykryj nietypowe zamówienia (potencjalne oszustwa lub błędy)"
  def check_order(order) do
    features = extract_features(order)
    prediction = Nx.Serving.batched_run(MyApp.AnomalyModel, features)

    case prediction do
      %{anomaly_score: score} when score > 0.8 ->
        {:flagged, "Zamówienie znacząco odbiega od wzorca", score}

      %{anomaly_score: score} when score > 0.5 ->
        {:review, "Zamówienie wymaga weryfikacji", score}

      _ ->
        :ok
    end
  end

  defp extract_features(order) do
    Nx.tensor([
      Decimal.to_float(order.total_price),
      order.item_count,
      hour_of_day(order.inserted_at),
      day_of_week(order.inserted_at),
      customer_order_frequency(order.customer_id),
      customer_avg_order_value(order.customer_id),
      distance_from_avg_for_product(order)
    ])
  end
end

# Automatyczne flagowanie podejrzanych zamówień:
# - Zamówienie o 3:00 w nocy na 50 000 PLN od nowego klienta → flagged
# - Zamówienie 10× większe niż średnia tego klienta → review
# - Normalne zamówienie → ok

3. Predykcja popytu (demand forecasting)

defmodule MyApp.DemandForecast do
  @doc "Przewiduj zapotrzebowanie na produkt na następne 7 dni"
  def predict(product_id) do
    # Pobierz historię sprzedaży
    history = Sales.daily_quantities(product_id, last_days: 90)

    # Przygotuj dane wejściowe (ostatnie 30 dni + cechy)
    features =
      history
      |> Enum.take(30)
      |> Enum.map(fn day ->
        [
          day.quantity,
          day.day_of_week,
          day.is_holiday,
          day.promotion_active,
          day.temperature_avg
        ]
      end)
      |> Nx.tensor()

    # Predykcja na 7 dni
    prediction = Nx.Serving.batched_run(MyApp.DemandModel, features)

    prediction
    |> Nx.to_flat_list()
    |> Enum.zip(next_7_days())
    |> Enum.map(fn {qty, date} -> %{date: date, predicted_quantity: round(qty)} end)
  end
end

# Użycie w LiveView - dashboard z prognozą:
defmodule MyAppWeb.InventoryDashboardLive do
  def mount(_params, _session, socket) do
    products = Inventory.list_low_stock_products()

    forecasts =
      Enum.map(products, fn product ->
        forecast = DemandForecast.predict(product.id)
        %{product: product, forecast: forecast, reorder_needed: needs_reorder?(product, forecast)}
      end)

    {:ok, assign(socket, forecasts: forecasts)}
  end
end

4. Analiza dokumentów i OCR

defmodule MyApp.DocumentProcessor do
  @doc "Wyciągnij dane z faktury (zdjęcie/PDF)"
  def process_invoice(image_path) do
    # OCR - rozpoznaj tekst z obrazu
    text = Nx.Serving.batched_run(MyApp.OCRModel, File.read!(image_path))

    # Ekstrakcja strukturalnych danych z tekstu
    extracted = Nx.Serving.batched_run(MyApp.EntityExtractor, text)

    %{
      vendor_name: extracted.entities["vendor"],
      invoice_number: extracted.entities["invoice_number"],
      total_amount: extracted.entities["total"],
      vat_amount: extracted.entities["vat"],
      issue_date: extracted.entities["date"],
      confidence: extracted.overall_confidence
    }
  end
end

# Automatyzacja: faktura przychodzi mailem → system parsuje →
# tworzy wpis w książce kosztowej → do akceptacji jeśli confidence < 90%

5. Wyszukiwanie semantyczne (embeddings)

defmodule MyApp.SemanticSearch do
  @doc "Wyszukaj produkty po znaczeniu, nie po słowach kluczowych"
  def search(query) do
    # Zamień zapytanie na wektor (embedding)
    query_embedding = Nx.Serving.batched_run(MyApp.EmbeddingModel, query)

    # Wyszukaj najbliższe wektory w PostgreSQL (pgvector)
    Repo.all(
      from p in Product,
      order_by: fragment("embedding <-> ?", ^Pgvector.new(query_embedding)),
      limit: 10
    )
  end
end

# "coś do ochrony stóp na budowie" → buty robocze, trzewiki ochronne, ochraniacze
# "narzędzie do kręcenia śrub" → wkrętarka, klucz nasadowy, klucz dynamometryczny
# Tradycyjne wyszukiwanie full-text nie złapie tych związków

Nx.Serving - skalowanie ML na BEAM

Kluczowa innowacja: Nx.Serving traktuje model ML jak GenServer - proces BEAM z interfejsem, stanem i supervisorem:

# Model jako proces - pełna integracja z ekosystemem BEAM

# 1. Supervision - restart po awarii
children = [
  {Nx.Serving, serving: sentiment_model(), name: MyApp.Sentiment},
  {Nx.Serving, serving: embedding_model(), name: MyApp.Embeddings},
  {Nx.Serving, serving: anomaly_model(), name: MyApp.AnomalyDetector}
]
# Jeśli model Sentiment padnie → supervisor restartuje
# Embeddings i AnomalyDetector działają dalej

# 2. Telemetry - monitoring wydajności
:telemetry.attach("ml-metrics", [:nx, :serving, :execute, :stop],
  fn _event, measurements, metadata, _config ->
    Logger.info("Model #{metadata.name}: #{measurements.duration}ms, batch: #{metadata.batch_size}")
  end)

# 3. Distributed - klaster BEAM
# Model na serwerze 1, aplikacja na serwerze 2:
Nx.Serving.batched_run({MyApp.Sentiment, :"ml@server1"}, text)
# Automatyczna komunikacja między nodami - zero HTTP, zero kolejek

Python vs Elixir - porównanie architektury ML

Tradycyjne podejście: Python microservice

graph LR
    PHOENIX["Phoenix App"]
    PYTHON["Python
FastAPI + PyTorch"]
    GPU["GPU Server"]

    PHOENIX -- "HTTP/gRPC" --> PYTHON
    PYTHON -- "JSON" --> PHOENIX
    PYTHON --> GPU

Problemy:
├── 2 deploymenty (Elixir + Python)
├── Serializacja/deserializacja JSON przy każdym wywołaniu
├── Latencja sieciowa (+5-20 ms per request)
├── 2 systemy monitoringu
├── 2 zestawy logów
├── Python GIL = osobny proces per request
├── Osobna infrastruktura dla Pythona (pip, venv, CUDA drivers)
└── Debugging rozproszony - "gdzie jest bug?"

Podejście Elixir: Nx.Serving w tej samej aplikacji

graph TD
    subgraph APP["Phoenix App"]
        LV["LiveView Dashboard"] --> NX["Nx.Serving
(model ML)"]
        NX --> EXLA["EXLA
(CPU/GPU)"]
    end

Korzyści:
├── 1 deployment
├── Zero serializacji (dane w pamięci)
├── Zero latencji sieciowej
├── 1 system monitoringu (Telemetry)
├── 1 zestaw logów (Logger)
├── BEAM scheduler = automatyczne batching
├── Supervisor = restart modelu po awarii
└── Debugging w jednym procesie

Kiedy Elixir ML, a kiedy Python

Reguła: jeśli potrzebujesz inference (uruchamianie gotowego modelu) w aplikacji webowej - Elixir. Jeśli potrzebujesz trenować nowe modele od zera lub prowadzić badania - Python.

Najczęstszy scenariusz dla firm: trenuj w Pythonie, servuj w Elixirze. Wytrenuj model w Jupyter Notebook, wyeksportuj przez ONNX, załaduj w Bumblebee - najlepsze z obu światów.

Co to oznacza dla Twojego systemu

Dodanie AI do istniejącej aplikacji Phoenix:

Jeśli Twoja aplikacja działa na Elixirze - AI jest jednym mix deps.get od wdrożenia. Zero nowych serwerów, zero nowych technologii, zero nowych deploymentów.

Chcesz dodać inteligentne funkcje do swojego systemu? Porozmawiajmy - pokażemy, jak Nx i Bumblebee mogą wzbogacić Twoją aplikację bez komplikowania infrastruktury.