Class: Langfuse::ScoreClient Private

Inherits:
Object
  • Object
show all
Defined in:
lib/langfuse/score_client.rb

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.

Client for creating and batching Langfuse scores

Handles thread-safe queuing, batching, and sending of score events to the Langfuse ingestion API. Scores are batched and sent automatically based on batch_size and flush_interval configuration.

rubocop:disable Metrics/ClassLength

Examples:

Basic usage

score_client = ScoreClient.new(api_client: api_client, config: config)
score_client.create(name: "quality", value: 0.85, trace_id: "abc123...")

With OTel integration

Langfuse.observe("operation") do |obs|
  score_client.score_active_observation(name: "accuracy", value: 0.92)
end

Constant Summary collapse

HEX_TRACE_ID_PATTERN =

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. 

/\A[0-9a-f]{32}\z/

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(api_client:, config:) ⇒ ScoreClient

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.

Initialize a new ScoreClient

Parameters:

  • api_client (ApiClient) —

    The API client for sending batches

  • config (Config) —

    Configuration object with batch_size and flush_interval



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/langfuse/score_client.rb', line 46

def initialize(api_client:, config:)
  @api_client = api_client
  @config = config
  @logger = config.logger
  @queue = PendingScoreQueue.new(capacity: config.score_queue_capacity)
  @shutdown_mutex = Mutex.new
  @flush_mutex = Mutex.new
  @flush_thread = nil
  @shutdown = false
  # Match the immutable tracing setup contract: once this client exists, later config
  # mutations must not change score sampling without rebuilding the client.
  @score_sampler = Sampling.build_sampler(config.sample_rate)

  start_flush_timer
  ForkSafety.register(self)
end

Instance Attribute Details

#api_client ⇒ ApiClient (readonly)

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 The API client for sending batches.

Returns:

  • (ApiClient) —

    The API client for sending batches



29
30
31
 
# File 'lib/langfuse/score_client.rb', line 29

def api_client
  @api_client
end

#config ⇒ Config (readonly)

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 Configuration object.

Returns:

  • (Config) —

    Configuration object



32
33
34
 
# File 'lib/langfuse/score_client.rb', line 32

def config
  @config
end

#logger ⇒ Logger (readonly)

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 Logger instance.

Returns:

  • (Logger) —

    Logger instance



35
36
37
 
# File 'lib/langfuse/score_client.rb', line 35

def logger
  @logger
end

Instance Method Details

#create(name:, value:, id: nil, trace_id: nil, session_id: nil, observation_id: nil, comment: nil, metadata: nil, environment: nil, data_type: :numeric, dataset_run_id: nil, config_id: nil) ⇒ void

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.

This method returns an undefined value.

Create a score event and queue it for batching

rubocop:disable Metrics/ParameterLists

Examples:

Numeric score

create(name: "quality", value: 0.85, trace_id: "abc123", data_type: :numeric)

Boolean score

create(name: "passed", value: true, trace_id: "abc123", data_type: :boolean)

Categorical score

create(name: "category", value: "high", trace_id: "abc123", data_type: :categorical)

Text score (1 to 500 characters)

create(name: "reviewer_notes", value: "Helpful but verbose", trace_id: "abc123", data_type: :text)

Corrected output (conventionally named "output")

create(name: "output", value: "The corrected output", trace_id: "abc123",
       observation_id: "def456", data_type: :correction)

Parameters:

  • name (String) —

    Score name (required)

  • value (Numeric, Integer, String) —

    Score value (type depends on data_type)

  • id (String, nil) (defaults to: nil) —

    Score ID; use a stable value as an idempotency key

  • trace_id (String, nil) (defaults to: nil) —

    Trace ID to associate with the score

  • session_id (String, nil) (defaults to: nil) —

    Session ID to associate with the score

  • observation_id (String, nil) (defaults to: nil) —

    Observation ID to associate with the score

  • comment (String, nil) (defaults to: nil) —

    Optional comment

  • metadata (Hash, nil) (defaults to: nil) —

    Optional metadata hash

  • environment (String, nil) (defaults to: nil) —

    Optional per-score environment override

  • data_type (Symbol) (defaults to: :numeric) —

    Data type (:numeric, :boolean, :categorical, :text, :correction)

  • dataset_run_id (String, nil) (defaults to: nil) —

    Optional dataset run ID to associate with the score

  • config_id (String, nil) (defaults to: nil) —

    Optional score config ID

Raises:

  • (ArgumentError) —

    if validation fails



96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/langfuse/score_client.rb', line 96

def create(name:, value:, id: nil, trace_id: nil, session_id: nil, observation_id: nil, comment: nil,
           metadata: nil, environment: nil, data_type: :numeric, dataset_run_id: nil, config_id: nil)
  return unless config.telemetry_enabled?

  score = build_score_body(
    name: name,
    value: value,
    id: id,
    trace_id: trace_id,
    session_id: session_id,
    observation_id: observation_id,
    comment: comment,
    metadata: metadata,
    environment: environment,
    data_type: data_type,
    dataset_run_id: dataset_run_id,
    config_id: config_id
  )

  return unless enqueue_trace_linked_score?(trace_id)

  enqueue_score_event(build_score_event(score))
rescue StandardError => e
  logger.error("Langfuse score creation failed: #{e.message}")
  raise
end

#create!(name:, value:, id: nil, trace_id: nil, session_id: nil, observation_id: nil, comment: nil, metadata: nil, environment: nil, data_type: :numeric, dataset_run_id: nil, config_id: nil) ⇒ String?

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.

Create a score immediately through the Scores API.

#create is fire-and-forget — it queues the event and reports nothing about whether it was actually delivered, matching how this SDK's tracing already works. That fits scoring inline from a still-open span (see #score_active_observation/#score_active_trace), but not a standalone verdict arriving out-of-band (e.g. user feedback landing in a request unrelated to the turn it's scoring). Pass a stable id when the caller may retry after an ambiguous network failure.

rubocop:disable Metrics/ParameterLists

Examples:

Create a score with an idempotency key

score_client.create!(id: "feedback-abc123", name: "quality", value: 0.85, trace_id: "abc123")

Parameters:

  • name (String) —

    Score name (required)

  • value (Numeric, Integer, String) —

    Score value (type depends on data_type)

  • id (String, nil) (defaults to: nil) —

    Score ID; use a stable value as an idempotency key

  • trace_id (String, nil) (defaults to: nil) —

    Trace ID to associate with the score

  • session_id (String, nil) (defaults to: nil) —

    Session ID to associate with the score

  • observation_id (String, nil) (defaults to: nil) —

    Observation ID to associate with the score

  • comment (String, nil) (defaults to: nil) —

    Optional comment

  • metadata (Hash, nil) (defaults to: nil) —

    Optional metadata hash

  • environment (String, nil) (defaults to: nil) —

    Optional per-score environment override

  • data_type (Symbol) (defaults to: :numeric) —

    Data type (:numeric, :boolean, :categorical, :text, :correction)

  • dataset_run_id (String, nil) (defaults to: nil) —

    Optional dataset run ID to associate with the score

  • config_id (String, nil) (defaults to: nil) —

    Optional score config ID

Returns:

  • (String, nil) —

    ID of the created score, or nil when telemetry is disabled

Raises:

  • (ArgumentError) —

    if validation fails

  • (UnauthorizedError) —

    if authentication fails

  • (ApiError) —

    if the API request fails



154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/langfuse/score_client.rb', line 154

def create!(name:, value:, id: nil, trace_id: nil, session_id: nil, observation_id: nil, comment: nil,
            metadata: nil, environment: nil, data_type: :numeric, dataset_run_id: nil, config_id: nil)
  return unless config.telemetry_enabled?

  score = build_score_body(
    name: name,
    value: value,
    id: id,
    trace_id: trace_id,
    session_id: session_id,
    observation_id: observation_id,
    comment: comment,
    metadata: metadata,
    environment: environment,
    data_type: data_type,
    dataset_run_id: dataset_run_id,
    config_id: config_id
  )

  api_client.create_score(payload: score)
end

#flush ⇒ void

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.

This method returns an undefined value.

Force flush all queued score events

Sends all queued events to the API immediately.



243
244
245
246
247
248
249
250
 
# File 'lib/langfuse/score_client.rb', line 243

def flush
  return unless config.telemetry_enabled?

  @flush_mutex.synchronize { flush_pending_batches }
rescue StandardError => e
  logger.error("Langfuse score flush failed: #{e.message}")
  # Don't raise - silent error handling for batch operations
end

#score_active_observation(name:, value:, comment: nil, metadata: nil, data_type: :numeric) ⇒ void

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.

This method returns an undefined value.

Create a score for the currently active observation (from OTel span)

Extracts observation_id and trace_id from the active OpenTelemetry span.

Examples:

Langfuse.observe("operation") do |obs|
  score_client.score_active_observation(name: "accuracy", value: 0.92)
end

Parameters:

  • name (String) —

    Score name (required)

  • value (Numeric, Integer, String) —

    Score value

  • comment (String, nil) (defaults to: nil) —

    Optional comment

  • metadata (Hash, nil) (defaults to: nil) —

    Optional metadata hash

  • data_type (Symbol) (defaults to: :numeric) —

    Data type (:numeric, :boolean, :categorical, :text, :correction)

Raises:

  • (ArgumentError) —

    if no active span or validation fails



193
194
195
196
197
198
199
200
201
202
203
204
205
206
 
# File 'lib/langfuse/score_client.rb', line 193

def score_active_observation(name:, value:, comment: nil, metadata: nil, data_type: :numeric)
  ids = extract_ids_from_active_span
  raise ArgumentError, "No active OpenTelemetry span found" unless ids[:observation_id]

  create(
    name: name,
    value: value,
    trace_id: ids[:trace_id],
    observation_id: ids[:observation_id],
    comment: comment,
    metadata: metadata,
    data_type: data_type
  )
end

#score_active_trace(name:, value:, comment: nil, metadata: nil, data_type: :numeric) ⇒ void

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.

This method returns an undefined value.

Create a score for the currently active trace (from OTel span)

Extracts trace_id from the active OpenTelemetry span.

Examples:

Langfuse.observe("operation") do |obs|
  score_client.score_active_trace(name: "overall_quality", value: 5)
end

Parameters:

  • name (String) —

    Score name (required)

  • value (Numeric, Integer, String) —

    Score value

  • comment (String, nil) (defaults to: nil) —

    Optional comment

  • metadata (Hash, nil) (defaults to: nil) —

    Optional metadata hash

  • data_type (Symbol) (defaults to: :numeric) —

    Data type (:numeric, :boolean, :categorical, :text, :correction)

Raises:

  • (ArgumentError) —

    if no active span or validation fails



224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/langfuse/score_client.rb', line 224

def score_active_trace(name:, value:, comment: nil, metadata: nil, data_type: :numeric)
  ids = extract_ids_from_active_span
  raise ArgumentError, "No active OpenTelemetry span found" unless ids[:trace_id]

  create(
    name: name,
    value: value,
    trace_id: ids[:trace_id],
    comment: comment,
    metadata: metadata,
    data_type: data_type
  )
end

#shutdown ⇒ void

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.

This method returns an undefined value.

Shutdown the score client and flush remaining events

Stops the flush timer thread and sends any remaining queued events.



257
258
259
260
261
262
263
264
265
 
# File 'lib/langfuse/score_client.rb', line 257

def shutdown
  @shutdown_mutex.synchronize do
    return if @shutdown

    @shutdown = true
    stop_flush_timer
    flush
  end
end