From 5bd85013c67c4d75f4ea5f4f112258c43a8c7e99 Mon Sep 17 00:00:00 2001 From: Claudio Ortolina Date: Mon, 2 Mar 2026 13:56:46 +0000 Subject: [PATCH] Simplify OpenAI api implementation --- lib/open_ai.ex | 13 +---- lib/open_ai/api.ex | 127 ++++++++++++--------------------------------- mix.exs | 1 + mix.lock | 1 + 4 files changed, 37 insertions(+), 105 deletions(-) diff --git a/lib/open_ai.ex b/lib/open_ai.ex index 89b91977..28acdad6 100644 --- a/lib/open_ai.ex +++ b/lib/open_ai.ex @@ -2,18 +2,7 @@ defmodule OpenAI do alias OpenAI.API def gpt(completion) do - {:ok, collector} = Agent.start_link(fn -> "" end) - - API.gpt_stream(completion, api_key(), fn data -> - case get_in(data, ["choices", Access.at(0), "delta", "content"]) do - nil -> :ok - data -> Agent.update(collector, fn current -> current <> data end) - end - end) - - result = Agent.get(collector, & &1) |> JSON.decode!() - Agent.stop(collector) - {:ok, result} + API.gpt(completion, api_key()) end def chat_stream(messages, opts \\ []) when is_list(messages) do diff --git a/lib/open_ai/api.ex b/lib/open_ai/api.ex index bb474f50..461864ed 100644 --- a/lib/open_ai/api.ex +++ b/lib/open_ai/api.ex @@ -1,85 +1,30 @@ defmodule OpenAI.API do - # Lifted from https://fly.io/phoenix-files/streaming-openai-responses/ - def gpt_stream(completion, api_key, cb) do - fun = fn request, finch_request, finch_name, finch_options -> - fun = fn - {:status, status}, response -> - %{response | status: status} + def gpt(completion, api_key) do + resp = + Req.post!("https://api.openai.com/v1/chat/completions", + receive_timeout: 10_000, + connect_options: [timeout: 2_500], + json: %{ + model: completion.model, + messages: [Map.take(completion, [:content, :role])], + response_format: %{type: "json_object"}, + temperature: completion.temperature + }, + auth: {:bearer, api_key} + ) - {:headers, headers}, response -> - %{response | headers: headers} - - {:data, data}, response -> - body = - data - |> String.split("data: ") - |> Enum.map(fn str -> - str - |> String.trim() - |> decode_body(cb) - end) - |> Enum.filter(fn d -> d != :ok end) - - old_body = if response.body == "", do: [], else: response.body - - %{response | body: old_body ++ body} - end - - case Finch.stream(finch_request, finch_name, Req.Response.new(), fun, finch_options) do - {:ok, response} -> {request, response} - {:error, exception, _response} -> {request, exception} - end + if resp.status in 200..299 do + content = get_in(resp.body, ["choices", Access.at(0), "message", "content"]) + {:ok, JSON.decode!(content)} + else + {:error, resp.body} end - - Req.post!("https://api.openai.com/v1/chat/completions", - receive_timeout: 1000, - connect_options: [ - timeout: 2500 - ], - json: %{ - model: completion.model, - messages: [Map.take(completion, [:content, :role])], - response_format: %{type: "json_object"}, - stream: true, - temperature: completion.temperature - }, - auth: {:bearer, api_key}, - finch_request: fun - ) end def chat_stream(messages, instructions, model, temperature, api_key, cb) do - fun = fn request, finch_request, finch_name, finch_options -> - fun = fn - {:status, status}, response -> - %{response | status: status} - - {:headers, headers}, response -> - %{response | headers: headers} - - {:data, data}, response -> - data - |> String.split("\n") - |> Enum.each(fn line -> - line - |> String.trim() - |> decode_responses_event(cb) - end) - - response - end - - case Finch.stream(finch_request, finch_name, Req.Response.new(), fun, finch_options) do - {:ok, response} -> {request, response} - {:error, exception, _response} -> {request, exception} - end - end - case Req.post("https://api.openai.com/v1/responses", receive_timeout: 60_000, - connect_options: [ - timeout: 5_000 - ], + connect_options: [timeout: 5_000], json: %{ model: model, instructions: instructions, @@ -89,16 +34,21 @@ defmodule OpenAI.API do temperature: temperature }, auth: {:bearer, api_key}, - finch_request: fun + into: fn {:data, data}, {req, resp} -> + buffer = Req.Request.get_private(req, :sse_buffer, "") + {events, buffer} = ServerSentEvents.parse(buffer <> data) + req = Req.Request.put_private(req, :sse_buffer, buffer) + + for %{data: json} <- events do + decode_responses_event(json, cb) + end + + {:cont, {req, resp}} + end ) do - {:ok, %{status: status}} when status in 200..299 -> - :ok - - {:ok, %{body: body}} -> - {:error, "OpenAI API error: #{inspect(body)}"} - - {:error, exception} -> - {:error, "Connection error: #{Exception.message(exception)}"} + {:ok, %{status: status}} when status in 200..299 -> :ok + {:ok, %{body: body}} -> {:error, "OpenAI API error: #{inspect(body)}"} + {:error, exception} -> {:error, "Connection error: #{Exception.message(exception)}"} end end @@ -120,19 +70,10 @@ defmodule OpenAI.API do end end - defp decode_body("", _), do: :ok - defp decode_body("[DONE]", _), do: :ok - defp decode_body(json, cb), do: cb.(JSON.decode!(json)) - - defp decode_responses_event("", _cb), do: :ok - defp decode_responses_event("event:" <> _rest, _cb), do: :ok - - defp decode_responses_event("data: " <> json, cb) do + defp decode_responses_event(json, cb) do case JSON.decode!(json) do %{"type" => "response.output_text.delta", "delta" => delta} -> cb.(delta) _other -> :ok end end - - defp decode_responses_event(_line, _cb), do: :ok end diff --git a/mix.exs b/mix.exs index 0f456635..be019dd1 100644 --- a/mix.exs +++ b/mix.exs @@ -89,6 +89,7 @@ defmodule MusicLibrary.MixProject do # HTTP Clients {:finch, "~> 0.21.0"}, {:req, "~> 0.5.8"}, + {:server_sent_events, "~> 0.2"}, # Parsing {:nimble_parsec, "~> 1.4"}, diff --git a/mix.lock b/mix.lock index e9b6f51a..053ded43 100644 --- a/mix.lock +++ b/mix.lock @@ -61,6 +61,7 @@ "req": {:hex, :req, "0.5.17", "0096ddd5b0ed6f576a03dde4b158a0c727215b15d2795e59e0916c6971066ede", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0b8bc6ffdfebbc07968e59d3ff96d52f2202d0536f10fef4dc11dc02a2a43e39"}, "rewrite": {:hex, :rewrite, "1.2.0", "80220eb14010e175b67c939397e1a8cdaa2c32db6e2e0a9d5e23e45c0414ce21", [:mix], [{:glob_ex, "~> 0.1", [hex: :glob_ex, repo: "hexpm", optional: false]}, {:sourceror, "~> 1.0", [hex: :sourceror, repo: "hexpm", optional: false]}, {:text_diff, "~> 0.1", [hex: :text_diff, repo: "hexpm", optional: false]}], "hexpm", "a1cd702bbb9d51613ab21091f04a386d750fc6f4516b81900df082d78b2d8c50"}, "rustler_precompiled": {:hex, :rustler_precompiled, "0.8.4", "700a878312acfac79fb6c572bb8b57f5aae05fe1cf70d34b5974850bbf2c05bf", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:rustler, "~> 0.23", [hex: :rustler, repo: "hexpm", optional: true]}], "hexpm", "3b33d99b540b15f142ba47944f7a163a25069f6d608783c321029bc1ffb09514"}, + "server_sent_events": {:hex, :server_sent_events, "0.2.1", "f83b34f01241302a8bf451efc8dde3a36c533d5715463c31c653f3db8695f636", [:mix], [], "hexpm", "c8099ce4f9acd610eb7c8e0f89dba7d5d1c13300ea9884b0bd8662401d3cf96f"}, "sourceror": {:hex, :sourceror, "1.11.0", "df2cdaffdc323e804009ff50b50bb31e6f2d6e116d936ccf22981f592594d624", [:mix], [], "hexpm", "6e26f572bdfc21d7ad397f596b4cfbbf31d7112126fe3e902c120947073231a8"}, "spitfire": {:hex, :spitfire, "0.3.7", "d6051f94f554d33d038ab3c1d7e017293ae30429cc6b267b08cb6ad69e35e9a3", [:mix], [], "hexpm", "798ff97db02477b05fa3db8e2810cebda6ed5d90c6de6b21aa65abd577599744"}, "statistex": {:hex, :statistex, "1.1.0", "7fec1eb2f580a0d2c1a05ed27396a084ab064a40cfc84246dbfb0c72a5c761e5", [:mix], [], "hexpm", "f5950ea26ad43246ba2cce54324ac394a4e7408fdcf98b8e230f503a0cba9cf5"},