Ruby chatgpt chat api when stream mode, deal with error handle and chunk break.

plain_chatgpt_client.rb

class Chatgpt::PlainChatgptClient
  def self.chat(parameters)
    @client ||= generate_client

    res = @client.post("/v1/chat/completions") do |req|
      if parameters[:stream].is_a?(Proc)
        req.options.on_data = to_json_stream(user_proc: parameters[:stream])
        parameters[:stream] = true
      end

      req.body = parameters
    end

    res.body
  end

  def self.to_json_stream(user_proc:)
    last_data_part = ""

    proc do |chunk, overall_received_bytes, env|
      if chunk.include?('"error": ')
        user_proc.call(JSON.parse(chunk))
        next
      end

      chunk = last_data_part.strip + chunk
      last_data_part = ""

      chunk.each_line do |line|
        if line.start_with?("data: {") && line.end_with?("}\n")
          begin
            data = JSON.parse(line.delete_prefix("data: "))
            user_proc.call(data)
          rescue JSON::ParserError
            Rails.logger.error("Parse error line: #{line} in chunk: #{chunk}")
          end
          next
        end

        # When not complete json
        if line.start_with?("data: {")
          last_data_part = line
          next
        end

        # Ignore
        if line.blank? || line.start_with?("data: [DONE]")
          next
        end

        Rails.logger.warn("Unkown line: #{line}")
      end
    end
  end

  def self.generate_client
    Faraday.new("https://api.openai.com/") do |conn|
      conn.request :json
      conn.request :authorization, 'Bearer', ENV["OPENAI_ACCESS_TOKEN"]
      conn.response :logger, nil, headers: false, body: true, bodies: { request: true, response: false }
      conn.response :json
      conn.adapter Faraday.default_adapter
    end
  end
end