Packages
fnord
0.9.17
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/endpoint.ex
defmodule AI.Endpoint do
@moduledoc """
API endpoint abstraction.
This behavior encapsulates the common mechanics for calling API endpoints via
`Http.post_json/3` and applying API-level retry semantics.
In particular, OpenAI rate limiting is surfaced as HTTP `429` with a JSON
body containing an error code (commonly `"rate_limit_exceeded"` or
`"rate_limit"`).
Callers implement `endpoint_path/0` and then call `AI.Endpoint.post_json/3`.
"""
@type headers :: list({String.t(), String.t()})
@type payload :: map()
@type http_status :: integer()
@type http_error :: {:http_error, {http_status, String.t()}}
@type transport_error :: {:transport_error, any()}
@type success :: {:ok, %{body: map(), headers: headers(), status: http_status()}}
@type response :: success | http_error | transport_error
@type endpoint :: module()
@retry_limit 3
@backoff_base_ms 200
@backoff_cap_ms 2_000
# ----------------------------------------------------------------------------
# Behaviour Callbacks
# ----------------------------------------------------------------------------
@callback endpoint_path() :: String.t()
# ----------------------------------------------------------------------------
# API
# ----------------------------------------------------------------------------
@doc "The fixed retry limit for API calls."
@spec retry_limit() :: pos_integer()
def retry_limit, do: @retry_limit
@doc """
Perform a JSON POST request against the endpoint module's `endpoint_path/0`.
Retries up to `#{@retry_limit}` times when the server indicates throttling.
"""
@spec post_json(endpoint, headers, payload) :: response
def post_json(endpoint_module, headers, payload) when is_atom(endpoint_module) do
url = endpoint_module.endpoint_path()
do_post_json(url, headers, payload, 1)
end
defp do_post_json(url, headers, payload, attempt) do
result = Http.post_json(url, headers, payload)
model = model_from_payload(payload)
Store.APIUsage.record_for_model(model, result)
case result do
{:http_error, {429, body}} ->
# Retry on throttling errors up to the retry limit
if attempt < retry_limit() and throttling_error?(body) do
Services.BgIndexingControl.note_throttle(model)
usage_wait = usage_wait_ms(model)
body_wait = throttling_delay_ms(body)
backoff_wait = backoff_delay_ms(attempt)
delay =
[usage_wait, body_wait, backoff_wait]
|> Enum.reject(&is_nil/1)
|> Enum.max()
UI.warn(
"[AI.Endpoint] Throttled (429), model=#{inspect(model)}, attempt #{attempt}/#{retry_limit()}, retrying in #{delay}ms"
)
maybe_sleep(delay)
do_post_json(url, headers, payload, attempt + 1)
else
# Non-throttling or max retries reached: return HTTP error
{:http_error, {429, body}}
end
{:ok, %{body: _body} = payload} ->
Services.BgIndexingControl.note_success(model)
{:ok, payload}
{:http_error, error} ->
{:http_error, error}
{:transport_error, reason} ->
{:transport_error, reason}
end
end
defp throttling_error?(body) when is_binary(body) do
case SafeJson.decode(body) do
{:ok, %{"error" => %{"code" => code}}} when code in ["rate_limit_exceeded", "rate_limit"] ->
true
_ ->
false
end
end
defp model_from_payload(payload) when is_map(payload) do
cond do
is_binary(Map.get(payload, :model)) -> Map.get(payload, :model)
is_binary(Map.get(payload, "model")) -> Map.get(payload, "model")
true -> nil
end
end
defp usage_wait_ms(nil), do: nil
defp usage_wait_ms(model) when is_binary(model) do
case Store.APIUsage.check(model) do
:ok -> 0
{:wait, ms} when is_integer(ms) and ms >= 0 -> ms
{:error, _} -> nil
end
end
defp throttling_delay_ms(body) when is_binary(body) do
# OpenAI commonly returns a message like:
# "Please try again in 959ms."
case Regex.run(~r/try again in (\d+)ms/i, body) do
[_, ms] ->
ms
|> String.to_integer()
|> max(1)
_ ->
nil
end
end
# Exponential backoff with jitter (+/- 20%), capped at 2 seconds.
defp backoff_delay_ms(attempt) when attempt >= 1 do
jitter = 0.8 + :rand.uniform() * 0.4
@backoff_cap_ms
|> min(@backoff_base_ms * :math.pow(2, attempt - 1))
|> trunc()
|> then(&(&1 * jitter))
|> min(@backoff_cap_ms)
|> max(1)
|> round()
end
defp maybe_sleep(ms) do
if Services.Globals.get_env(:fnord, :http_retry_skip_sleep, false) do
:ok
else
Process.sleep(ms)
end
end
end