Class: OpenAI::Internal::VectorStoreFileUploader Private

Inherits:
Object
  • Object
show all
Defined in:
lib/openai/internal/vector_store_file_uploader.rb,
sig/openai/internal/vector_store_file_uploader.rbs

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Uploads files for a vector store with bounded concurrency.

The caller enumerates and snapshots stream-backed inputs before workers start, so lazy and stateful Ruby enumerables are never advanced from worker threads. Originally streamed inputs are reopened as streams for files.create, retaining its non-retryable request semantics. Results retain input order.

Constant Summary collapse

MAX_FILES_PER_BATCH =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

The API limit for files attached in one vector store batch.

Returns:

  • (Integer) 
2_000

Instance Method Summary collapse

Constructor Details

#initialize(client:, max_concurrency:, request_options:) ⇒ VectorStoreFileUploader

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of VectorStoreFileUploader.

Parameters:



29
30
31
32
33
34
35
36
37
 
# File 'lib/openai/internal/vector_store_file_uploader.rb', line 29

def initialize(client:, max_concurrency:, request_options:)
  unless max_concurrency.is_a?(Integer) && max_concurrency.positive?
    raise ArgumentError, "`max_concurrency` must be a positive integer"
  end

  @client = client
  @max_concurrency = max_concurrency
  @request_options = OpenAI::Internal::RequestOptionsScope.new(request_options)
end

Instance Method Details

#upload(files, max_files: MAX_FILES_PER_BATCH) ⇒ Array<OpenAI::Models::FileObject>

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parameters:

  • files (Enumerable<Pathname, StringIO, IO, String, OpenAI::FilePart>)
  • max_files (Integer) (defaults to: MAX_FILES_PER_BATCH) —

    Maximum number of inputs that may be uploaded.

Returns:



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/openai/internal/vector_store_file_uploader.rb', line 44

def upload(files, max_files: MAX_FILES_PER_BATCH)
  staged, temporary_files = stage(files, max_files: max_files)
  return [] if staged.empty?

  worker_count = [@max_concurrency, staged.length].min
  start_gate = Queue.new
  lock = Mutex.new
  uploaded = []
  state = {error: nil, next_index: 0, stopping: false}
  workers = []

  Thread.handle_interrupt(Exception => :never) do
    start_workers(start_gate, staged, worker_count, lock, uploaded, state, workers)
    Thread.handle_interrupt(Exception => :immediate) do
      start_gate.close
      workers.each(&:join)
    end

  ensure
    stop_workers(start_gate, workers, lock, state)
  end

  error = lock.synchronize { state[:error] }
  raise error unless error.nil?

  uploaded
ensure
  temporary_files&.each { remove_temporary_file(_1) }
end