Packages
fnord
0.8.89
0.9.40
0.9.39
0.9.38
0.9.37
0.9.36
0.9.35
0.9.34
0.9.33
0.9.32
0.9.31
0.9.30
0.9.29
0.9.28
0.9.27
0.9.26
0.9.25
0.9.24
0.9.23
0.9.22
0.9.21
0.9.20
0.9.19
0.9.18
0.9.17
0.9.16
0.9.15
0.9.14
0.9.13
0.9.12
0.9.11
0.9.10
0.9.9
0.9.8
0.9.7
0.9.6
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.99
0.8.98
0.8.97
0.8.96
0.8.95
0.8.94
0.8.93
0.8.92
0.8.91
0.8.90
0.8.89
0.8.88
0.8.87
0.8.86
0.8.85
0.8.84
0.8.83
0.8.82
0.8.81
0.8.80
0.8.79
0.8.78
0.8.77
0.8.76
0.8.75
0.8.74
0.8.73
0.8.72
0.8.71
0.8.70
0.8.69
0.8.68
0.8.67
0.8.66
0.8.65
0.8.64
0.8.63
0.8.62
0.8.61
0.8.60
0.8.59
0.8.58
0.8.57
0.8.56
0.8.55
0.8.54
0.8.53
0.8.52
0.8.51
0.8.50
0.8.49
0.8.48
0.8.47
0.8.46
0.8.45
0.8.44
0.8.43
0.8.42
0.8.41
0.8.40
0.8.39
0.8.38
0.8.37
0.8.36
0.8.35
0.8.34
0.8.33
0.8.32
0.8.31
0.8.30
0.8.29
0.8.27
0.8.26
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.1
0.8.0
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.3
0.7.2
0.7.1
0.7.0
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.1
0.6.0
0.5.9
0.5.8
0.5.7
0.5.6
0.5.5
0.5.4
0.5.3
0.5.2
0.5.1
0.5.0
0.4.44
0.4.43
0.4.42
0.4.41
0.4.40
0.4.39
0.4.38
0.4.37
0.4.36
0.4.35
0.4.34
0.4.33
0.4.32
0.4.30
0.4.29
0.4.28
0.4.27
0.4.26
0.4.25
0.4.24
0.4.23
0.4.22
0.4.21
0.4.20
0.4.19
0.4.18
0.4.17
0.4.16
0.4.15
0.4.14
0.4.13
0.4.12
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.0
0.2.0
0.1.0
AI code archaeology
Current section
Files
Jump to
Current section
Files
lib/ai/embeddings.ex
defmodule AI.Embeddings do
@endpoint "https://api.openai.com/v1/embeddings"
@retry_interval 250
@rate_limit_backoff_base 200
@rate_limit_backoff_cap 2_000
@rate_limit_max_attempts 5
@max_attempts 3
@model "text-embedding-3-large"
@batch_size 300_000
@batch_reduction_factor 0.75
@chunk_size 8192
@type embedding :: list(float())
@type embeddings :: list(embedding())
@type error ::
{:error, :max_attempts_reached}
| {:error, :http_error}
| {:error, :transport_error}
| {:error, :rate_limited}
| {:error, String.t()}
@type attempt :: non_neg_integer()
@type inputs :: list(String.t())
@spec get(String.t()) :: {:ok, embeddings} | error
def get(input) do
input
|> split_into_batches()
|> get(1, [])
end
@spec get(inputs, attempt, embeddings) :: {:ok, embeddings} | error
defp get(batches, attempt, acc)
defp get([], _attempt, acc), do: {:ok, acc}
defp get(_batches, attempt, _acc) when attempt > @max_attempts do
{:error, :max_attempts_reached}
end
defp get([batch | rest], attempt, acc) do
if attempt > 1, do: Process.sleep(@retry_interval)
batch
|> split_into_chunks(attempt)
|> endpoint()
|> case do
{:ok, embeddings} ->
get(rest, 1, merge_embeddings(embeddings, acc))
{:error, :token_limit_exceeded} ->
get([batch | rest], attempt + 1, acc)
other ->
{:error, other}
end
end
@spec split_into_batches(String.t()) :: inputs
defp split_into_batches(input) do
tokens = AI.PretendTokenizer.guesstimate_tokens(input)
if tokens >= @batch_size do
AI.PretendTokenizer.chunk(input, @batch_size, @batch_reduction_factor)
else
[input]
end
end
@spec split_into_chunks(String.t(), attempt) :: inputs
defp split_into_chunks(batch, attempt) do
reduction_factor = token_reduction_factor(attempt)
AI.PretendTokenizer.chunk(batch, @chunk_size, reduction_factor)
end
# -----------------------------------------------------------------------------
# For each dimension, find the maximum value across all embeddings. This
# isn't necessarily the _most_ accurate, but it selects the highest rating
# for each dimension found in the file, which should be reasonable for
# semantic searching.
# -----------------------------------------------------------------------------
@spec merge_embeddings(embeddings, embedding) :: embeddings | error
defp merge_embeddings([], acc), do: acc
defp merge_embeddings([first | rest], []), do: merge_embeddings(rest, first)
defp merge_embeddings([first | rest], acc) do
merged = Enum.zip_with(acc, first, &max/2)
merge_embeddings(rest, merged)
end
@spec get_api_key!() :: String.t()
defp get_api_key!() do
["FNORD_OPENAI_API_KEY", "OPENAI_API_KEY"]
|> Enum.find_value(&System.get_env(&1, nil))
|> case do
nil ->
raise "Either FNORD_OPENAI_API_KEY or OPENAI_API_KEY environment variable must be set"
api_key ->
api_key
end
end
# -----------------------------------------------------------------------------
# Calculate the token reduction factor based on the number of attempts. This
# is used to dial back the number of (estimated) tokens sent to the endpoint
# when retrying requests.
# -----------------------------------------------------------------------------
@spec token_reduction_factor(attempt) :: float()
defp token_reduction_factor(attempt) do
case attempt do
1 -> 0.75
2 -> 0.50
_ -> 0.25
end
end
@spec token_limit_error?(String.t()) :: boolean
defp token_limit_error?(body) do
String.contains?(body, "maximum context length")
end
@spec endpoint(inputs) ::
{:ok, embeddings}
| {:error, :token_limit_exceeded}
| {:error, :http_error}
| {:error, :transport_error}
| {:error, :rate_limited}
defp endpoint(input) do
endpoint(input, 1)
end
@spec endpoint(inputs, attempt) ::
{:ok, embeddings}
| {:error, :token_limit_exceeded}
| {:error, :http_error}
| {:error, :transport_error}
| {:error, :rate_limited}
defp endpoint(input, attempt) do
api_key = get_api_key!()
headers = [
{"Authorization", "Bearer #{api_key}"},
{"Content-Type", "application/json"}
]
payload = %{
encoding_format: "float",
model: @model,
input: input
}
Http.post_json(@endpoint, headers, payload)
|> case do
{:ok, %{"data" => embeddings}} ->
{:ok, Enum.map(embeddings, &Map.get(&1, "embedding"))}
{:http_error, {status_code, body}} ->
error_map =
case Jason.decode(body) do
{:ok, json} -> json
_ -> nil
end
if status_code == 429 and rate_limit_error?(error_map) do
if attempt < @rate_limit_max_attempts do
backoff =
min(@rate_limit_backoff_cap, @rate_limit_backoff_base * :math.pow(2, attempt - 1))
factor = 0.8 + :rand.uniform() * 0.4
delay = round(backoff * factor)
delay = if delay < 1, do: 1, else: delay
UI.warn("[AI.Embeddings] Rate limited, attempt #{attempt}, retrying in #{delay}ms")
Process.sleep(delay)
endpoint(input, attempt + 1)
else
UI.warn("[AI.Embeddings] Rate limit exceeded after #{attempt} attempts")
{:error, :rate_limited}
end
else
if token_limit_error?(body) do
{:error, :token_limit_exceeded}
else
UI.warn("[AI.Embeddings] Error getting embeddings: #{status_code} - #{body}")
{:error, :http_error}
end
end
{:transport_error, error} ->
UI.warn("[AI.Embeddings] Error getting embeddings: #{inspect(error)}")
{:error, :transport_error}
end
end
defp rate_limit_error?(%{"error" => %{"code" => "rate_limit_exceeded"}}), do: true
defp rate_limit_error?(_), do: false
end