jtomchak

jtomchak

Handling http response streaming

So I’m getting back a streaming response from a whisper api I set up to use fly.io GPU’s. So far I can call it, get back chunks as they come in using req’s into: parameter and put the text and timestamp on a Phoenix LiveView. What I’m struggling with is how to capture all the incoming chunks to a list so I can save the text segments to Postgres and/or to object storage so I don’t have to transcript that particular audio file again.

# calling transcribe_audio with the url
 
  def async_transcribe(%Episode{} = episode) do
    Task.Supervisor.start_child(NormanAi.TaskSupervisor, fn ->

      result =
        NormanAi.Audio.transcribe_audio(episode.enclosure_url, fn ss, text ->
          segment = %Episode.Transcript.Segment{ss: ss, text: text}
          broadcast!(episode.id, {segment, episode.id})
          IO.inspect("BROADCASTING")

          segment
        end)

      IO.inspect(result)
      # Repo.update_all(from(e in Episode, where: e.id == ^episode.id),
      #   set: [transcript: %Episode.Transcript{segments: segments}]
      # )
    end)
  end

Then the work is done here

  def transcribe_audio(url, callback) do
# internal fly.io url for making api requests to
    req =
      Req.new(
        url: "http://faster-whisper-server-patient-voice-8559.flycast/v1/audio/transcriptions",
        connect_options: [transport_opts: [inet6: true]],
        receive_timeout: 120_000
      )

# hard coded url with short sample audio
    file_path =
      file_path_from_url(
        "https://s3-us-west-2.amazonaws.com/staging-moth-social/media_attachments/files/113/041/865/730/367/819/original/9fd57c861d8f3fc7.mp3"
      )

    multipart =
      Multipart.new()
      |> Multipart.add_part(Multipart.Part.text_field("true", "stream"))
      |> Multipart.add_part(Multipart.Part.file_field(file_path, "file"))

    content_length = Multipart.content_length(multipart)
    content_type = Multipart.content_type(multipart, "multipart/form-data")

    headers = [
      {"Content-Type", content_type},
      {"Content-Length", to_string(content_length)}
    ]

    Req.post(
      req,
      headers: headers,
      body: Multipart.body_stream(multipart),
      into: fn {:data, data}, context ->
        # fn {:ok, {ss, %{chunks: [%{text: text}]}}} -> func.(ss, text) end
        results = parse(data)
        IO.inspect(results)
        IO.inspect(context, label: "CONTEXT")
        segment = Enum.at(results.segments, 0)
        callback.(trunc(segment.start), segment.text)
        {:cont, context}
      end
    )
  end

How could I gather up the incoming chunks in a list or collection so I have all the data to save to persistent?
Each returning data chunk looks like this:

%{
  text: " At Apple Park.",
  words: [],
  task: "transcribe",
  language: "en",
  duration: 0.8200000000000003,
  segments: [
    %{
      id: 14,
      start: 54.64,
      seek: 5736,
      tokens: [51695, 1711, 6373, 4964, 13, 51736],
      text: " At Apple Park.",
      end: 55.46,
      words: nil,
      temperature: 0.0,
      avg_logprob: -0.19924645728253304,
      compression_ratio: 1.7433962264150944,
      no_speech_prob: 1.9550323486328125e-5
    }
  ]
}

Marked As Solved

wojtekmach

wojtekmach

Hex Core Team

Yeah, sorry about that, it’s on my roadmap to improve documentation around this. req.private and resp.private exist solely so when writing steps and into: fun we can store intermediate state. The only restriction is private.req_* key names are reserved for Req for forward-compatibility.

Also Liked

garrison

garrison

I have never actually done this but my impression reading the docs is that you’re meant to use the response in the {req, resp} tuple (your context) as an accumulator. You could place the chunks in the :private field of resp under your own key. See :private under “Fields” at Req.Response.

Or you could use the response :body - not sure if it would get clobbered by anything else.

It would be nice if the docs for :into actually spelled this out as I was looking into this exact thing a couple weeks ago and had to infer that this was the intended approach. Seems like it would be a very common use case.

jtomchak

jtomchak

ok. so I should be able to use resp.private to append the chunks as they come back into a list. sweet. I’ll try it out and post the results. I tried with an Agent and managed to get the results back correctly, probably not idiomatic Elixir.

  def transcribe_audio(url, callback) do
    req =
      Req.new(
        url: "http://faster-whisper-server-patient-voice-8559.flycast/v1/audio/transcriptions",
        connect_options: [transport_opts: [inet6: true]],
        receive_timeout: 120_000
      )

    file_path =
      file_path_from_url(
        "https://s3-us-west-2.amazonaws.com/staging-moth-social/media_attachments/files/113/041/865/730/367/819/original/9fd57c861d8f3fc7.mp3"
      )

    multipart =
      Multipart.new()
      |> Multipart.add_part(Multipart.Part.text_field("true", "stream"))
      |> Multipart.add_part(Multipart.Part.file_field(file_path, "file"))

    content_length = Multipart.content_length(multipart)
    content_type = Multipart.content_type(multipart, "multipart/form-data")

    headers = [
      {"Content-Type", content_type},
      {"Content-Length", to_string(content_length)}
    ]

    # Initialize buffer state
    {:ok, agent} = Agent.start_link(fn -> [] end)

    Req.post(
      req,
      headers: headers,
      body: Multipart.body_stream(multipart),
      into: fn {:data, data}, context ->
        # fn {:ok, {ss, %{chunks: [%{text: text}]}}} -> func.(ss, text) end
        results = parse(data)
        segment = Enum.at(results.segments, 0)
        callback.(trunc(segment.start), segment.text)
        :ok = Agent.update(agent, fn state -> [results | state] end)
        {:cont, context}
      end
    )

    # Make sure we shut the agent down
    data_results = Agent.get(agent, & &1) |> Enum.reverse()
    :ok = Agent.stop(agent)
    data_results
  end
wojtekmach

wojtekmach

Hex Core Team

put_private, update_private, etc return updated response, you need to reassign the resp variable.

Btw you can nowadays drop Multipart dependency since we have this capability built in, see encode_body/1 step.

jtomchak

jtomchak

Yeah I saw the :private field but was scared off with the detail

a map reserved for libraries and frameworks to use. Prefix the keys with the name of your project to avoid any future conflicts. Only accepts atom/0 keys.

But it’s worth a shot

jtomchak

jtomchak

I don’t seem to be able to append/update :private from within the into: fun

 def transcribe_audio(url, callback) do
    req =
      Req.new(
        url: "http://faster-whisper-server-patient-voice-8559.flycast/v1/audio/transcriptions",
        connect_options: [transport_opts: [inet6: true]],
        receive_timeout: 120_000
      )

    file_path =
      file_path_from_url(
        "https://s3-us-west-2.amazonaws.com/moth-social/media_attachments/files/113/087/363/649/463/883/original/6f9b1ad7ff1c0018.mp3"
      )

    multipart =
      Multipart.new()
      |> Multipart.add_part(Multipart.Part.text_field("true", "stream"))
      |> Multipart.add_part(Multipart.Part.file_field(file_path, "file"))

    content_length = Multipart.content_length(multipart)
    content_type = Multipart.content_type(multipart, "multipart/form-data")

    headers = [
      {"Content-Type", content_type},
      {"Content-Length", to_string(content_length)}
    ]

    {:ok, response} =
      Req.post(
        req,
        headers: headers,
        body: Multipart.body_stream(multipart),
        into: fn {:data, data}, {req, resp} ->
          # try to add any value to the response
          Req.Response.put_private(resp, :transcription, [{%{:b => "moe"}}])
          results = parse(data)
          segment = Enum.at(results.segments, 0)
          callback.(trunc(segment.start), segment.text)
          # update private response
          Req.Response.update_private(resp, :transcript, [], fn state -> [results | state] end)
          IO.inspect(resp.private)
          {:cont, {req, resp}}
        end
      )

    Req.Response.put_private(response, :something_else, [{%{:a => "ted"}}])

    # Make sure we shut the agent down
    # data_results = Agent.get(agent, & &1) |> Enum.reverse()
    # :ok = Agent.stop(agent)
    # data_results
  end

the put_private or update_private from the into: fun are not on the return response, but the final put_private is added

#Req.Response<
  status: 200,
  headers: %{
    "content-type" => ["text/event-stream; charset=utf-8"],
    "date" => ["Thu, 05 Sep 2024 23:39:26 GMT"],
    "fly-request-id" => ["01J728X25018F47F3W0KD31EJ1-ord"],
    "server" => ["Fly/813cb6e67 (2024-09-03)"],
    "transfer-encoding" => ["chunked"],
    "via" => ["1.1 fly.io"]
  },
  body: "",
  trailers: %{},
  private: %{something_else: [{%{a: "ted"}}]},
  ...
>

Where Next?

Popular in Questions Top

JDanielMartinez
Hi! May someone helps me, please! I have two apps into an umbrella project: the first one is Database, which manages queries, and the se...
New
albydarned
Hello all! I am typing this post from my new MacBook Pro with the M1 chip. I’m loving it so far, and will probably use it as my daily dr...
New
quazar
How to set Jason to encode all fields in ecto schema, I don’t care about security and implementing only is taking long list of attributes...
New
fireproofsocks
I’m working on defining a simple Ecto schema for a table (in PostGres), but I don’t see where I can define a column as NOT NULL. Conside...
New
lastday4you
I wanted to check elixir version in phoenix because i found that my elixir is 1.5 but when i use Enum.chunk_by it said the function is un...
New
script
If I have a string “1000 cfu/ml” . I want to remove the characters and / and space . So the string is like this "1000" What is the ...
New
vac
Hi, I'm quite new in Elixir and I'm trying to format a string to a PEM format. I have the certificate value like MIIDBTCCAe2...... and ...
New
Patoshizzle
After calling mix ecto.create I get this error: 17:00:32.162 [error] GenServer #PID&lt;0.412.0&gt; terminating ** (Postgrex.Error) FATAL...
New
belgoros
I’m not a pro in using Regex and can’t figure out why the following behaviour happens, especially if we take into account the difference ...
New
lanycrost
Hi everyone! I need implement if…else if…else condition from my elixir code, and anymore of this control flow structures not work proper...
New

Other popular topics Top

senggen
Erlang/OTP 25 [erts-13.2.2] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] 15:22:35.803 [error] gen_event {lager_file_backend...
New
dotdotdotPaul
Okay, I'm having a heck of a time trying to figure out how to best handle the validation of belongs_to associations in Ecto. I'm sure I'...
New
vonH
In asking this question I am more interested about the expressiveness of the language itself and less concerned about the availability of...
New
vac
Hi, I'm quite new in Elixir and I'm trying to format a string to a PEM format. I have the certificate value like MIIDBTCCAe2...... and ...
New
aadeshere1
I have a another noob question about loop. Since elixir is immutable, while loop is not directly possible. total = 10 while total != 0 ...
New
minhajuddin
I have seen a lot of code which picks the first element from a list using Enum.at(0) instead of List.first. Is there a reason why people ...
New
shahryarjb
Hello, I have map which I want to convert it to string like this: the map: %{last_name: "tavakkoli", name: "shahryar"} the string I ne...
New
aesmail
Hello guys, I have finally made it. I created an admin interface for a framework. It’s been on my todo list for years and with the curre...
New
siddhant3030
Hi, I have to write a raw query for one of my project. But till now I have used ecto queries and don’t have much experience writing raw ...
New
joeerl
Hello again - after a longish gap I’ve decided I really must dig into Elixir and see what’s been happening here - so I have a few questio...
New

We're in Beta

About us Mission Statement