Packages
openai_ex
0.5.2
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.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.0
0.6.5
0.6.4
0.6.3
0.6.2
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.2
0.4.1
0.4.0
0.3.0
0.2.3
0.2.2
0.2.1
0.2.0
0.1.9
0.1.8
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
Community maintained Elixir library for OpenAI API
Current section
Files
Jump to
Current section
Files
lib/openai_ex/http_sse.ex
defmodule OpenaiEx.HttpSse do
@moduledoc false
require Logger
# based on
# https://gist.github.com/zachallaun/88aed2a0cef0aed6d68dcc7c12531649
@doc false
def post(openai = %OpenaiEx{}, url, json: json) do
request = OpenaiEx.Http.build_post(openai, url, json: json)
me = self()
ref = make_ref()
task =
Task.async(fn ->
on_chunk = fn chunk, _acc -> send(me, {:chunk, chunk, ref}) end
request |> Finch.stream(OpenaiEx.Finch, nil, on_chunk)
send(me, {:done, ref})
end)
_status = receive(do: ({:chunk, {:status, status}, ^ref} -> status))
_headers = receive(do: ({:chunk, {:headers, headers}, ^ref} -> headers))
Stream.resource(fn -> {"", ref, task} end, &next_sse/1, fn {_data, _ref, task} ->
Task.shutdown(task)
end)
end
@doc false
defp next_sse({acc, ref, task}) do
receive do
{:chunk, {:data, evt_data}, ^ref} ->
{tokens, next_acc} = tokenize_data(evt_data, acc)
{[tokens], {next_acc, ref, task}}
{:done, ^ref} ->
if acc != "" do
Logger.warning(inspect(Jason.decode!(acc)))
end
{:halt, {acc, ref, task}}
end
end
@doc false
defp tokenize_data(evt_data, acc) do
all_data = acc <> evt_data
# finds 2 (repeated) EOL characters
if Regex.match?(~r/(\r?\n|\r){2}/, evt_data) do
{remaining, token_chunks} = all_data |> String.split(~r/(\r?\n|\r){2}/) |> List.pop_at(-1)
tokens =
token_chunks
|> Enum.map(&extract_token/1)
|> Enum.filter(fn
%{data: "[DONE]"} -> false
%{data: _} -> true
# we can pass other events through but the clients will need to be rewritten
_ -> false
end)
|> Enum.map(fn %{data: data} -> %{data: Jason.decode!(data)} end)
{tokens, remaining}
else
{[], all_data}
end
end
defp extract_token(line) do
[field | rest] = String.split(line, ":", parts: 2)
value = Enum.join(rest, "") |> String.replace_prefix(" ", "")
case field do
"data" -> %{data: value}
"event" -> %{eventType: value}
"id" -> %{lastEventId: value}
"retry" -> %{retry: value}
# comment
_ -> nil
end
end
end