Elixir AI Integration Tutorial: Complete Guide 2025

In this post

This tutorial builds a small chat assistant in Elixir. You call an LLM HTTP API with Req, keep one conversation per user in a GenServer, supervise those processes with OTP, show the chat in Phoenix LiveView, and test it with ExUnit and Mox.

The code targets OpenAI’s Chat Completions API, but nothing in it is specific to one provider. I have not compiled this exact project end to end. Treat it as a worked sketch: run mix test after you paste it in, and expect to adjust names to your app.

Why Elixir for AI Integration #

Most LLM features are I/O-bound: your app waits on a slow HTTP call. Python (asyncio) and Node.js both handle many waiting requests well, so concurrency alone is not a reason to switch. What Elixir adds is the shape of the runtime:

One process per conversation. Each chat gets an isolated process with its own state. A crash in one conversation does not touch the others.
Supervision. OTP supervisors restart crashed processes according to a strategy you choose.
Real-time UI without a separate front end. LiveView pushes server-rendered updates over a WebSocket.
Cheap processes. The Erlang docs say a newly spawned process uses 327 words of memory (Erlang Efficiency Guide), about 2.6 KB on a 64-bit system, so a process per conversation is affordable.

The trade-off is ecosystem size. Python has far more ML libraries. In Elixir you mostly call hosted models over HTTP, or run models locally with Nx and Bumblebee.

Comparing stacks? See our LangChain Python tutorial for the same kind of integration in Python.

When Elixir fits #

Chat-style features with many simultaneous, long-lived conversations.
Per-tenant or per-user isolation, where one bad request must not affect others.
Apps that already run on Phoenix and want AI features in the same codebase.

LangChain in Elixir #

There is a community library, langchain on Hex, described as an “Elixir implementation of a LangChain style framework”. It is not an official port. This tutorial skips it and calls the API directly, so you can see every moving part.

Setup #

Install Elixir and Erlang/OTP by following the official install guide, then install the Phoenix generator by following the Phoenix installation guide. Check the Phoenix guide for the Elixir and OTP versions the current release needs.

Create the project (LiveView is included by default in current generators):

mix phx.new elixir_ai_app
cd elixir_ai_app

Add the HTTP client and the test libraries to deps in mix.exs:

# mix.exs
defp deps do
  [
    # ...the deps phx.new generated...
    {:req, "~> 0.5"},
    {:mox, "~> 1.0", only: :test},
    {:stream_data, "~> 1.0", only: :test}
  ]
end

mix deps.get

Keep API keys out of the code #

Never commit a key. Put it in an environment variable and add any .env file to .gitignore:

# .env  (listed in .gitignore)
export OPENAI_API_KEY="sk-..."
export OPENAI_MODEL="<a model name from OpenAI's models page>"

Load it with source .env before mix phx.server. Read the values in config/runtime.exs:

# config/runtime.exs
import Config

config :elixir_ai_app, :openai,
  api_key: System.get_env("OPENAI_API_KEY"),
  model: System.get_env("OPENAI_MODEL")

The model name is left to you on purpose. Model names get deprecated: OpenAI’s own page for gpt-4.1 is marked “Deprecated” (OpenAI models). Pick a current one from the models list.

Calling the LLM API #

First a behaviour, so the rest of the app can swap the real client for a mock in tests:

# lib/elixir_ai_app/ai/ai_client_behaviour.ex
defmodule ElixirAiApp.AI.AIClientBehaviour do
  @type message :: %{role: String.t(), content: String.t()}

  @callback chat_completion([message()], keyword()) :: {:ok, String.t()} | {:error, term()}
end

Then the real client:

# lib/elixir_ai_app/ai/openai_client.ex
defmodule ElixirAiApp.AI.OpenAIClient do
  @moduledoc "Minimal OpenAI Chat Completions client built on Req."
  @behaviour ElixirAiApp.AI.AIClientBehaviour

  alias ElixirAiApp.AI.RateLimiter

  @url "https://api.openai.com/v1/chat/completions"

  @impl true
  def chat_completion(messages, opts \\ []) when is_list(messages) do
    with :ok <- RateLimiter.check_rate_limit() do
      request(messages, opts)
    end
  end

  defp request(messages, opts) do
    config = Application.get_env(:elixir_ai_app, :openai, [])
    api_key = config[:api_key] || raise "OPENAI_API_KEY is not set"
    model = opts[:model] || config[:model] || raise "OPENAI_MODEL is not set"

    body = %{
      model: model,
      messages: [%{role: "system", content: "You are a helpful Elixir assistant."} | messages]
    }

    case Req.post(@url,
           auth: {:bearer, api_key},
           json: body,
           receive_timeout: 50_000,
           retry: :transient
         ) do
      {:ok, %Req.Response{status: 200, body: body}} -> parse_response(body)
      {:ok, %Req.Response{status: status, body: body}} -> {:error, {:http, status, body}}
      {:error, exception} -> {:error, {:request, exception}}
    end
  end

  defp parse_response(%{"choices" => [%{"message" => %{"content" => content}} | _]})
       when is_binary(content),
       do: {:ok, content}

  defp parse_response(_body), do: {:error, :unexpected_response}
end

Notes on the Req options, checked against the Req docs:

json: encodes the body and sets the content type. auth: {:bearer, key} sets the Authorization header.
Req’s default retry only covers safe methods (GET and HEAD). retry: :transient extends retries to POST for HTTP 408/429/500/502/503/504 and common transport errors, with exponential backoff and jitter, up to 3 retries by default. Retrying a POST can send the same prompt twice, so be aware of the cost.
The default receive timeout is short for LLM calls, so the example raises it.

Try it in iex -S mix after source .env:

ElixirAiApp.AI.OpenAIClient.chat_completion([%{role: "user", content: "Say hello in five words."}])
# {:ok, "..."}

This needs the rate limiter from the production section to be running, because the client calls it. If you want to try the client first, remove the with wrapper temporarily.

Validating responses with pattern matching #

Pattern matching keeps response handling flat:

# lib/elixir_ai_app/ai/response_validator.ex
defmodule ElixirAiApp.AI.ResponseValidator do
  @moduledoc "Turn client results into values the rest of the app can rely on."

  def validate({:ok, ""}), do: {:error, :empty_response}

  def validate({:ok, text}) when is_binary(text) and byte_size(text) > 10_000,
    do: {:error, :response_too_long}

  def validate({:ok, text}) when is_binary(text),
    do: {:ok, %{content: String.trim(text), length: String.length(text)}}

  def validate({:error, {:http, 429, _body}}), do: {:error, :rate_limit}
  def validate({:error, reason}), do: {:error, {:unknown_error, reason}}
end

The 10,000-byte cap is an arbitrary example. Choose your own limit.

One GenServer per conversation #

A GenServer holds the history for one conversation and sends it with each request, so the model sees earlier turns:

# lib/elixir_ai_app/ai/assistant.ex
defmodule ElixirAiApp.AI.Assistant do
  @moduledoc "Holds one conversation and calls the LLM client."
  use GenServer
  require Logger

  @client Application.compile_env(:elixir_ai_app, :ai_client, ElixirAiApp.AI.OpenAIClient)
  @call_timeout 60_000

  # Client API

  def start_link(opts \\ []) do
    gen_opts = if name = opts[:name], do: [name: name], else: []
    GenServer.start_link(__MODULE__, opts, gen_opts)
  end

  def ask(assistant, message) when is_binary(message),
    do: GenServer.call(assistant, {:ask, message}, @call_timeout)

  def get_history(assistant), do: GenServer.call(assistant, :get_history)

  def clear_history(assistant), do: GenServer.cast(assistant, :clear_history)

  # Server callbacks

  @impl true
  def init(opts) do
    {:ok, %{history: [], max_history: Keyword.get(opts, :max_history, 10), model: opts[:model]}}
  end

  @impl true
  def handle_call({:ask, message}, _from, state) do
    messages = state.history ++ [%{role: "user", content: message}]

    case @client.chat_completion(messages, model: state.model) do
      {:ok, reply} ->
        history = Enum.take(messages ++ [%{role: "assistant", content: reply}], -state.max_history)
        {:reply, {:ok, reply}, %{state | history: history}}

      {:error, reason} ->
        Logger.error("AI request failed: #{inspect(reason)}")
        {:reply, {:error, reason}, state}
    end
  end

  def handle_call(:get_history, _from, state), do: {:reply, state.history, state}

  @impl true
  def handle_cast(:clear_history, state), do: {:noreply, %{state | history: []}}
end

What this does:

History is stored oldest first and trimmed to the last max_history messages, which caps the prompt size.
A failed request leaves the history unchanged.
The GenServer blocks during a request, so one user’s questions are answered one at a time. Other users are unaffected because they have their own process.

Give each user an assistant with a DynamicSupervisor #

# lib/elixir_ai_app/ai/assistant_supervisor.ex
defmodule ElixirAiApp.AI.AssistantSupervisor do
  @moduledoc "Starts one Assistant per user, registered by user id."
  use DynamicSupervisor

  alias ElixirAiApp.AI.Assistant

  def start_link(init_arg), do: DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__)

  @doc "Return the user's assistant, starting it if needed."
  def ensure_assistant(user_id, opts \\ []) do
    spec = %{
      id: Assistant,
      start: {Assistant, :start_link, [Keyword.put(opts, :name, via_tuple(user_id))]},
      restart: :transient
    }

    case DynamicSupervisor.start_child(__MODULE__, spec) do
      {:ok, pid} -> {:ok, pid}
      {:error, {:already_started, pid}} -> {:ok, pid}
      other -> other
    end
  end

  @impl true
  def init(_init_arg), do: DynamicSupervisor.init(strategy: :one_for_one)

  defp via_tuple(user_id), do: {:via, Registry, {AssistantRegistry, user_id}}
end

Add the registry, supervisor and rate limiter to the children in lib/elixir_ai_app/application.ex, next to the children Phoenix generated:

children = [
  # ...generated children (Telemetry, PubSub, Endpoint...)...
  {Registry, keys: :unique, name: AssistantRegistry},
  {ElixirAiApp.AI.RateLimiter, max_requests: 100, window_ms: 60_000},
  ElixirAiApp.AI.AssistantSupervisor
]

If an assistant crashes, the supervisor starts a fresh one. Because the history lives in process memory, a restarted assistant starts with an empty history. If you need history to survive crashes or deploys, store it in your database and load it in init/1.

Parallel requests with Task.async_stream #

For batches of independent prompts, Task.async_stream/3 runs them with a cap on concurrency:

# lib/elixir_ai_app/ai/batch_processor.ex
defmodule ElixirAiApp.AI.BatchProcessor do
  alias ElixirAiApp.AI.OpenAIClient

  def process_batch(prompts, opts \\ []) do
    prompts
    |> Task.async_stream(
      fn prompt -> OpenAIClient.chat_completion([%{role: "user", content: prompt}]) end,
      max_concurrency: Keyword.get(opts, :max_concurrency, 5),
      timeout: Keyword.get(opts, :timeout, 60_000),
      on_timeout: :kill_task
    )
    |> Enum.map(fn
      {:ok, result} -> result
      {:exit, reason} -> {:error, {:task_exit, reason}}
    end)
  end
end

Results come back in input order. max_concurrency is your main lever: your provider’s rate limits, not the BEAM, will usually be the bottleneck. I did not benchmark this, so I make no speed claim.

A LiveView chat page #

This version waits for the full reply and shows a “thinking” state. It does not stream tokens. Streaming needs server-sent-event parsing on top of this; the Req docs describe the into: option for consuming a response body as it arrives.

start_async/3 runs the slow call outside the LiveView process, so the page stays responsive (LiveView docs):

# lib/elixir_ai_app_web/live/chat_live.ex
defmodule ElixirAiAppWeb.ChatLive do
  use ElixirAiAppWeb, :live_view

  alias ElixirAiApp.AI.{Assistant, AssistantSupervisor}

  # Assumes your authentication puts "user_id" in the session.
  @impl true
  def mount(_params, %{"user_id" => user_id}, socket) do
    {:ok, pid} = AssistantSupervisor.ensure_assistant(user_id)
    {:ok, assign(socket, assistant: pid, messages: [], loading: false)}
  end

  @impl true
  def handle_event("send_message", %{"message" => message}, socket) do
    message = String.trim(message)

    if message == "" or socket.assigns.loading do
      {:noreply, socket}
    else
      pid = socket.assigns.assistant

      {:noreply,
       socket
       |> update(:messages, &(&1 ++ [{:user, message}]))
       |> assign(:loading, true)
       |> start_async(:ai_reply, fn -> Assistant.ask(pid, message) end)}
    end
  end

  def handle_event("clear_chat", _params, socket) do
    Assistant.clear_history(socket.assigns.assistant)
    {:noreply, assign(socket, :messages, [])}
  end

  @impl true
  def handle_async(:ai_reply, {:ok, {:ok, reply}}, socket) do
    {:noreply, socket |> update(:messages, &(&1 ++ [{:assistant, reply}])) |> assign(:loading, false)}
  end

  def handle_async(:ai_reply, {:ok, {:error, reason}}, socket) do
    {:noreply, socket |> update(:messages, &(&1 ++ [{:error, "Request failed: #{inspect(reason)}"}])) |> assign(:loading, false)}
  end

  def handle_async(:ai_reply, {:exit, reason}, socket) do
    {:noreply, socket |> update(:messages, &(&1 ++ [{:error, "Request crashed: #{inspect(reason)}"}])) |> assign(:loading, false)}
  end

  @impl true
  def render(assigns) do
    ~H"""
    <div class="mx-auto max-w-3xl p-4">
      <h1 class="text-2xl font-bold">Elixir AI Assistant</h1>

      <div id="chat-messages" class="h-96 overflow-y-auto border p-4">
        <div :for={{role, content} <- @messages} class="mb-3">
          <div class="text-xs font-semibold"><%= role_name(role) %></div>
          <div class="whitespace-pre-wrap"><%= content %></div>
        </div>
        <div :if={@loading} class="animate-pulse text-gray-500">Thinking...</div>
      </div>

      <form phx-submit="send_message" class="mt-4 flex gap-2">
        <input type="text" name="message" autocomplete="off" class="flex-1 border px-3 py-2" />
        <button type="submit" disabled={@loading} class="border px-4 py-2">Send</button>
        <button type="button" phx-click="clear_chat" class="border px-4 py-2">Clear</button>
      </form>
    </div>
    """
  end

  defp role_name(:user), do: "You"
  defp role_name(:assistant), do: "Assistant"
  defp role_name(:error), do: "Error"
end

Add the route inside a :browser scope in router.ex:

live "/chat", ChatLive, :index

Testing with ExUnit and Mox #

Never call a paid API from the default test run. Point the :ai_client setting at a Mox mock in config/test.exs:

# config/test.exs
config :elixir_ai_app, :ai_client, ElixirAiApp.AI.MockAIClient

# test/test_helper.exs
Mox.defmock(ElixirAiApp.AI.MockAIClient, for: ElixirAiApp.AI.AIClientBehaviour)
ExUnit.start(exclude: [:integration])

Mox expectations are private to the test process by default. The assistant runs in its own process, so these tests use global mode, which is why they are not async (Mox docs):

# test/elixir_ai_app/ai/assistant_test.exs
defmodule ElixirAiApp.AI.AssistantTest do
  use ExUnit.Case, async: false

  import Mox

  alias ElixirAiApp.AI.{Assistant, MockAIClient}

  setup :set_mox_global
  setup :verify_on_exit!

  setup do
    {:ok, pid} = Assistant.start_link(max_history: 4)
    %{pid: pid}
  end

  test "returns the client's reply", %{pid: pid} do
    expect(MockAIClient, :chat_completion, fn [%{role: "user", content: "Hello"}], _opts ->
      {:ok, "Hi!"}
    end)

    assert {:ok, "Hi!"} = Assistant.ask(pid, "Hello")
  end

  test "sends earlier turns with the next question", %{pid: pid} do
    expect(MockAIClient, :chat_completion, fn _messages, _opts -> {:ok, "First answer"} end)

    expect(MockAIClient, :chat_completion, fn messages, _opts ->
      assert Enum.map(messages, & &1.content) == ["First", "First answer", "Second"]
      {:ok, "Second answer"}
    end)

    assert {:ok, _} = Assistant.ask(pid, "First")
    assert {:ok, _} = Assistant.ask(pid, "Second")
  end

  test "keeps only the last max_history messages", %{pid: pid} do
    expect(MockAIClient, :chat_completion, 3, fn _messages, _opts -> {:ok, "Reply"} end)

    for n <- 1..3, do: Assistant.ask(pid, "Message #{n}")

    assert length(Assistant.get_history(pid)) == 4
  end

  test "leaves history alone when the client fails", %{pid: pid} do
    expect(MockAIClient, :chat_completion, fn _messages, _opts -> {:error, :boom} end)

    assert {:error, :boom} = Assistant.ask(pid, "Hi")
    assert Assistant.get_history(pid) == []
  end

  test "clear_history empties the conversation", %{pid: pid} do
    expect(MockAIClient, :chat_completion, fn _messages, _opts -> {:ok, "Reply"} end)

    Assistant.ask(pid, "Hi")
    Assistant.clear_history(pid)

    assert Assistant.get_history(pid) == []
  end
end

A property test for the validator, using StreamData:

# test/elixir_ai_app/ai/response_validator_test.exs
defmodule ElixirAiApp.AI.ResponseValidatorTest do
  use ExUnit.Case, async: true
  use ExUnitProperties

  alias ElixirAiApp.AI.ResponseValidator

  property "accepts any non-empty short string" do
    check all text <- string(:printable, min_length: 1, max_length: 1000) do
      assert {:ok, _} = ResponseValidator.validate({:ok, text})
    end
  end

  test "rejects an empty response" do
    assert {:error, :empty_response} = ResponseValidator.validate({:ok, ""})
  end

  test "maps HTTP 429 to :rate_limit" do
    assert {:error, :rate_limit} = ResponseValidator.validate({:error, {:http, 429, %{}}})
  end
end

One opt-in test against the real API, excluded from the default run by exclude: [:integration]:

# test/elixir_ai_app/ai/openai_client_integration_test.exs
defmodule ElixirAiApp.AI.OpenAIClientIntegrationTest do
  use ExUnit.Case, async: false

  @moduletag :integration

  test "gets a reply from the real API" do
    start_supervised!({ElixirAiApp.AI.RateLimiter, max_requests: 10, window_ms: 60_000})

    assert {:ok, reply} =
             ElixirAiApp.AI.OpenAIClient.chat_completion([%{role: "user", content: "Say hi."}])

    assert is_binary(reply)
  end
end

mix test                      # unit tests, no network
mix test --only integration   # needs OPENAI_API_KEY and OPENAI_MODEL
mix test --cover              # built-in coverage report

Production notes #

A rate limiter process #

A simple sliding-window limiter keeps you under your provider’s request limits. This one counts requests per node, so with several nodes the limit applies to each node separately:

# lib/elixir_ai_app/ai/rate_limiter.ex
defmodule ElixirAiApp.AI.RateLimiter do
  @moduledoc "Sliding-window rate limiter for outgoing API calls."
  use GenServer

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)

  @doc "Returns :ok if a request is allowed, {:error, :rate_limited} otherwise."
  def check_rate_limit, do: GenServer.call(__MODULE__, :check)

  @impl true
  def init(opts) do
    {:ok,
     %{
       max_requests: Keyword.fetch!(opts, :max_requests),
       window_ms: Keyword.fetch!(opts, :window_ms),
       requests: []
     }}
  end

  @impl true
  def handle_call(:check, _from, state) do
    now = System.monotonic_time(:millisecond)
    recent = Enum.filter(state.requests, &(&1 > now - state.window_ms))

    if length(recent) < state.max_requests do
      {:reply, :ok, %{state | requests: [now | recent]}}
    else
      {:reply, {:error, :rate_limited}, %{state | requests: recent}}
    end
  end
end

The limiter is a single process, so all checks go through one mailbox. That is fine at modest request rates; measure before relying on it at high rates.

Supervision strategy #

Your top-level supervisor in application.ex uses :one_for_one by default: a crashing child restarts alone. Use :rest_for_one if later children depend on earlier ones, and :one_for_all only when children cannot work without each other. The Supervisor docs describe all three.

Releases #

Phoenix can generate release files with mix phx.gen.release. Then:

MIX_ENV=prod mix release
_build/prod/rel/elixir_ai_app/bin/elixir_ai_app start

Set OPENAI_API_KEY, OPENAI_MODEL, SECRET_KEY_BASE and PHX_HOST in the production environment. config/runtime.exs reads them when the release boots.

Troubleshooting #

OPENAI_API_KEY is not set: the variable was not in the environment when the app started. Check System.get_env("OPENAI_API_KEY") in iex -S mix and Application.get_env(:elixir_ai_app, :openai). runtime.exs runs at boot, so restart after changing the variable.

{:error, {:http, 401, _}}: the key is wrong or missing.

{:error, {:http, 404, _}} mentioning the model: OPENAI_MODEL names a model that does not exist or was retired. Check the models list.

{:error, :rate_limited}: your local limiter said no. Raise max_requests or queue the work. Separately, a {:http, 429, _} means the provider rejected the request; Req already retries 429 with backoff.

An assistant keeps crashing: look at the crash report in the logs. :observer.start() and Process.info(pid, :current_stacktrace) help for a live process.

Frequently Asked Questions #

Is Elixir good for AI development?

For calling hosted models and serving real-time chat, yes. For training models or using the wider ML ecosystem, Python has more libraries. Elixir has Nx and Bumblebee for running pre-trained models locally.

Can you use LangChain with Elixir?

There is a community library, langchain on Hex, that follows LangChain’s design. It is not an official LangChain project. You can also call the provider API directly, as in this tutorial.

How do I integrate the OpenAI API with Phoenix?

Call it with an HTTP client such as Req, wrap per-user state in a GenServer, and run the slow call from LiveView with start_async/3.

Why a GenServer per conversation?

It gives each conversation isolated state and its own failure boundary. Remember that state lives in memory: persist it if it must survive a restart.

Next steps #

Stream tokens to the page by consuming the response with Req’s into: option.
Persist conversation history in your database.
Run models locally with Nx and Bumblebee.
Compare with the Python approach.

If you want help shipping AI features on an existing app, see how we work.

Sources #

All fetched 2026-10-30.

Erlang Efficiency Guide: Processes: process memory (327 words).
Req.Steps (retry): :transient retry, backoff, defaults.
Phoenix.LiveView: start_async/3, handle_async/3.
Mox: private and global modes.
Supervisor: restart strategies.
OpenAI GPT-4.1 model page: model marked Deprecated.
langchain on Hex and Bumblebee on Hex: package descriptions.