Class: SolidLoop::Adapters::Native

Inherits:
Object
  • Object
show all
Defined in:
app/services/solid_loop/adapters/native.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(base_url:, api_token:, payload:, streaming: false, reasoning_strategies: [], dialect: nil, model_name: nil, fallback_messages_text: "", read_timeout: nil, cancellation_check: nil) ⇒ Native

Returns a new instance of Native.



6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'app/services/solid_loop/adapters/native.rb', line 6

def initialize(base_url:, api_token:, payload:, streaming: false, reasoning_strategies: [], dialect: nil, model_name: nil, fallback_messages_text: "", read_timeout: nil, cancellation_check: nil)
  @base_url = base_url
  @api_token = api_token
  @payload = payload
  @streaming = streaming
  @reasoning_strategies = Array(reasoning_strategies)
  @model_name = model_name
  @fallback_messages_text = fallback_messages_text
  # Single coherent read-timeout source: when no explicit
  # read_timeout is passed, fall back to the SAME `default_read_timeout` the
  # lease is derived from — NOT the unrelated LlmCompletionJob::READ_TIMEOUT
  # constant. Previously a configured `default_read_timeout = 60` yielded a
  # 60s-derived lease but a 600s Faraday timeout, so a healthy turn could run
  # ~600s while its lease assumed ~60s (invariant lease > client timeout
  # broken). Reading the config here keeps the derived lease and the actual
  # HTTP timeout coherent.
  @read_timeout = read_timeout || SolidLoop.config.default_read_timeout
  @cancellation_check = cancellation_check

  # Use provided dialect or default to OpenAI
  @dialect = dialect || SolidLoop::Dialects::OpenAi.new

  @log_buffer = StringIO.new
  @debug_buffer = StringIO.new

  @last_ui_update_at = Time.current
  @first_token_at = nil
  @content_started_at = nil
  @aggregator = SolidLoop::SseStreamAggregator.new
end

Instance Attribute Details

#on_chunk_proc=(value) ⇒ Object (writeonly)

Sets the attribute on_chunk_proc

Parameters:

  • value —

    the value to set the attribute on_chunk_proc to.



4
5
6
 
# File 'app/services/solid_loop/adapters/native.rb', line 4

def on_chunk_proc=(value)
  @on_chunk_proc = value
end

Instance Method Details

#build_url ⇒ Object 



108
109
110
111
112
113
114
115
116
 
# File 'app/services/solid_loop/adapters/native.rb', line 108

def build_url
  base = @dialect.completion_url(@base_url)
  if @dialect.is_a?(SolidLoop::Dialects::Gemini)
    method = @streaming ? "streamGenerateContent" : "generateContent"
    "#{base}/#{@model_name}:#{method}?key=#{@api_token}"
  else
    base
  end
end

#call ⇒ Object 



37
38
39
40
41
42
43
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'app/services/solid_loop/adapters/native.rb', line 37

def call
  @request_started_at = Time.current

  @dialect.apply_reasoning_strategies!(@payload[:messages], @reasoning_strategies)
  final_payload = @dialect.respond_to?(:render_payload) ? @dialect.render_payload(@payload) : @payload
  url = build_url

  headers = { "Content-Type" => "application/json" }
  headers.merge!(@dialect.auth_headers(@api_token))

  wire_logger = Logger.new(@log_buffer)
  wire_logger.formatter = proc { |_severity, _datetime, _progname, msg| "#{msg}\n" }

  conn = Faraday.new(url: url) do |f|
    f.options.timeout = @read_timeout
    f.options.open_timeout = 10
    f.response :logger, wire_logger, bodies: { request: true, response: true }, headers: true
    f.adapter Faraday.default_adapter
  end

  caller_thread = Thread.current
  stop_watchdog = false
  _watchdog = build_watchdog(caller_thread, stop_flag: -> { stop_watchdog })

  begin
    @response = conn.post do |req|
      req.headers = headers
      req.body = final_payload.to_json

      if @streaming
        req.options.on_data = proc do |chunk, _overall_received_bytes|
          @debug_buffer.write(chunk)
          @first_token_at ||= Time.current
          data = parse_stream_chunk(chunk)
          emit_chunk_event(data)
        end
      end
    end
  ensure
    stop_watchdog = true
  end

  duration = (Time.current - @request_started_at).to_f
  Rails.logger.info "LLM Request Finished (Duration: #{duration.round(2)}s)"

  # Final aggregation result
  raw_data =
    if @streaming && @aggregator.result.present?
      @aggregator.result
    else
      JSON.parse(@response.body) rescue {}
    end

  # DIALECT IS THE BOSS OF NORMALIZATION
  # We pass fallback strings for token estimation if provider is silent
  normalized_data = @dialect.normalize_response(
    raw_data,
    fallback_text: extract_full_text(raw_data),
    fallback_messages_text: @fallback_messages_text
  )

  {
    response: @response,
    normalized_data: normalized_data,
    duration: duration,
    aggregator: @aggregator,
    log: @log_buffer.string,
    debug_log: @debug_buffer.string
  }
end