Module: Langfuse

Defined in:
lib/langfuse.rb,
lib/langfuse.rb,
lib/langfuse/types.rb,
lib/langfuse/client.rb,
lib/langfuse/config.rb,
lib/langfuse/version.rb,
lib/langfuse/api_client.rb,
lib/langfuse/evaluation.rb,
lib/langfuse/otel_setup.rb,
lib/langfuse/item_result.rb,
lib/langfuse/propagation.rb,
lib/langfuse/cache_warmer.rb,
lib/langfuse/observations.rb,
lib/langfuse/prompt_cache.rb,
lib/langfuse/score_client.rb,
lib/langfuse/dataset_client.rb,
lib/langfuse/span_processor.rb,
lib/langfuse/experiment_item.rb,
lib/langfuse/otel_attributes.rb,
lib/langfuse/timestamp_parser.rb,
lib/langfuse/traced_execution.rb,
lib/langfuse/experiment_result.rb,
lib/langfuse/experiment_runner.rb,
lib/langfuse/chat_prompt_client.rb,
lib/langfuse/text_prompt_client.rb,
lib/langfuse/dataset_item_client.rb,
lib/langfuse/rails_cache_adapter.rb,
lib/langfuse/stale_while_revalidate.rb

Overview

rubocop:disable Metrics/ModuleLength

Defined Under Namespace

Modules: OtelAttributes, OtelSetup, Propagation, StaleWhileRevalidate, TimestampParser, TracedExecution, Types Classes: Agent, ApiClient, ApiError, BaseObservation, CacheWarmer, CacheWarmingError, Chain, ChatPromptClient, Client, Config, ConfigurationError, DatasetClient, DatasetItemClient, Embedding, Error, Evaluation, Evaluator, Event, ExperimentItem, ExperimentResult, ExperimentRunner, Generation, Guardrail, ItemResult, NotFoundError, PromptCache, RailsCacheAdapter, Retriever, ScoreClient, Span, SpanProcessor, TextPromptClient, Tool, UnauthorizedError

Constant Summary collapse

FLUSH_TIMEOUT =

Default timeout (in seconds) for flushing traces during experiment runs.

5
OBSERVATION_TYPE_REGISTRY =

Registry mapping observation type strings to their wrapper classes

{
  OBSERVATION_TYPES[:generation] => Generation,
  OBSERVATION_TYPES[:embedding] => Embedding,
  OBSERVATION_TYPES[:event] => Event,
  OBSERVATION_TYPES[:agent] => Agent,
  OBSERVATION_TYPES[:tool] => Tool,
  OBSERVATION_TYPES[:chain] => Chain,
  OBSERVATION_TYPES[:retriever] => Retriever,
  OBSERVATION_TYPES[:evaluator] => Evaluator,
  OBSERVATION_TYPES[:guardrail] => Guardrail,
  OBSERVATION_TYPES[:span] => Span
}.freeze
VERSION = 
"0.4.0"
OBSERVATION_TYPES =

Observation type constants

{
  span: "span",
  generation: "generation",
  embedding: "embedding",
  event: "event",
  agent: "agent",
  tool: "tool",
  chain: "chain",
  retriever: "retriever",
  evaluator: "evaluator",
  guardrail: "guardrail"
}.freeze

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.configuration ⇒ Config

Returns the global configuration object

Returns:

  • (Config) —

    the global configuration



76
77
78
 
# File 'lib/langfuse.rb', line 76

def configuration
  @configuration ||= Config.new
end

Class Method Details

.client ⇒ Client

Returns the global singleton client

Returns:

  • (Client) —

    the global client instance



102
103
104
 
# File 'lib/langfuse.rb', line 102

def client
  @client ||= Client.new(configuration)
end

.configure {|Config| ... } ⇒ Config

Configure Langfuse globally

Examples:

Langfuse.configure do |config|
  config.public_key = ENV['LANGFUSE_PUBLIC_KEY']
  config.secret_key = ENV['LANGFUSE_SECRET_KEY']
end

Yields:

  • (Config) —

    the configuration object

Returns:

  • (Config) —

    the configured configuration



90
91
92
93
94
95
96
97
 
# File 'lib/langfuse.rb', line 90

def configure
  yield(configuration)

  # Auto-initialize OpenTelemetry
  OtelSetup.setup(configuration)

  configuration
end

.create_score(name:, value:, trace_id: nil, observation_id: nil, comment: nil, metadata: nil, data_type: :numeric) ⇒ void

This method returns an undefined value.

Create a score event and queue it for batching

rubocop:disable Metrics/ParameterLists

Examples:

Numeric score

Langfuse.create_score(name: "quality", value: 0.85, trace_id: "abc123")

Boolean score

Langfuse.create_score(name: "passed", value: true, trace_id: "abc123", data_type: :boolean)

Categorical score

Langfuse.create_score(name: "category", value: "high", trace_id: "abc123", data_type: :categorical)

Parameters:

  • name (String) —

    Score name (required)

  • value (Numeric, Integer, String) —

    Score value (type depends on data_type)

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

    Trace 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

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

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

Raises:

  • (ArgumentError) —

    if validation fails



210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/langfuse.rb', line 210

def create_score(name:, value:, trace_id: nil, observation_id: nil, comment: nil, metadata: nil,
                 data_type: :numeric)
  client.create_score(
    name: name,
    value: value,
    trace_id: trace_id,
    observation_id: observation_id,
    comment: comment,
    metadata: metadata,
    data_type: data_type
  )
end

.flush_scores ⇒ void

This method returns an undefined value.

Force flush all queued score events

Sends all queued score events to the API immediately.

Examples:

Langfuse.flush_scores


284
285
286
 
# File 'lib/langfuse.rb', line 284

def flush_scores
  client.flush_scores if @client
end

.force_flush(timeout: 30) ⇒ void

This method returns an undefined value.

Force flush all pending traces

Parameters:

  • timeout (Integer) (defaults to: 30) —

    Timeout in seconds



126
127
128
 
# File 'lib/langfuse.rb', line 126

def force_flush(timeout: 30)
  OtelSetup.force_flush(timeout: timeout)
end

.observe(name, attrs = {}, as_type: :span, **kwargs) {|observation| ... } ⇒ BaseObservation, Object

User-facing convenience method for creating root observations

Examples:

Block-based API (auto-ends)

Langfuse.observe("operation") do |obs|
  result = perform_operation
  obs.update(output: result)
end

Stateful API (manual end)

obs = Langfuse.observe("operation", input: { data: "test" })
obs.update(output: { result: "success" })
obs.end

Parameters:

  • name (String) —

    Descriptive name for the observation

  • attrs (Hash) (defaults to: {}) —

    Observation attributes (optional positional or keyword)

  • as_type (Symbol, String) (defaults to: :span) —

    Observation type (:span, :generation, :event, etc.)

  • kwargs (Hash) —

    Additional keyword arguments merged into observation attributes (e.g., input:, output:, metadata:)

Yields:

  • (observation) —

    Optional block that receives the observation object

Yield Parameters:

Returns:

  • (BaseObservation, Object) —

    The observation (or block return value if block given)



377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
 
# File 'lib/langfuse.rb', line 377

def observe(name, attrs = {}, as_type: :span, **kwargs, &block)
  # Merge positional attrs and keyword kwargs
  merged_attrs = attrs.to_h.merge(kwargs)
  observation = start_observation(name, merged_attrs, as_type: as_type)

  if block
    # Block-based API: auto-ends when block completes
    # Set context and execute block
    current_context = OpenTelemetry::Context.current
    result = OpenTelemetry::Context.with_current(
      OpenTelemetry::Trace.context_with_span(observation.otel_span, parent_context: current_context)
    ) do
      block.call(observation)
    end
    # Only end if not already ended (events auto-end in start_observation)
    observation.end unless as_type.to_s == OBSERVATION_TYPES[:event]
    result
  else
    # Stateful API - return observation
    # Events already auto-ended in start_observation
    observation
  end
end

.propagate_attributes(user_id: nil, session_id: nil, metadata: nil, version: nil, tags: nil, as_baggage: false) { ... } ⇒ Object

Propagate trace-level attributes to all spans created within this context.

This method sets attributes on the currently active span AND automatically propagates them to all new child spans created within the block. This is the recommended way to set trace-level attributes like user_id, session_id, and metadata dimensions that should be consistently applied across all observations in a trace.

IMPORTANT: Call this as early as possible within your trace/workflow. Only the currently active span and spans created after entering this context will have these attributes. Pre-existing spans will NOT be retroactively updated.

Examples:

Basic usage

Langfuse.propagate_attributes(user_id: "user_123", session_id: "session_abc") do
  Langfuse.observe("operation") do |span|
    # Current span has user_id and session_id
    span.start_observation("child") do |child|
      # Child span inherits user_id and session_id
    end
  end
end

With metadata and tags

Langfuse.propagate_attributes(
  user_id: "user_123",
  metadata: { environment: "production", region: "us-east" },
  tags: ["api", "v2"]
) do
  # All spans inherit these attributes
end

Cross-service propagation

Langfuse.propagate_attributes(
  user_id: "user_123",
  as_baggage: true
) do
  # Attributes propagate via HTTP headers
end

Parameters:

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

    User identifier (≤200 characters)

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

    Session identifier (≤200 characters)

  • metadata (Hash<String, String>, nil) (defaults to: nil) —

    Additional metadata (all values ≤200 characters)

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

    Version identifier (≤200 characters)

  • tags (Array<String>, nil) (defaults to: nil) —

    List of tags (each ≤200 characters)

  • as_baggage (Boolean) (defaults to: false) —

    If true, propagates via OpenTelemetry baggage for cross-service propagation

Yields:

  • Block within which attributes are propagated

Returns:

  • (Object) —

    The result of the block



176
177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/langfuse.rb', line 176

def propagate_attributes(user_id: nil, session_id: nil, metadata: nil, version: nil, tags: nil,
                         as_baggage: false, &)
  Propagation.propagate_attributes(
    user_id: user_id,
    session_id: session_id,
    metadata: metadata,
    version: version,
    tags: tags,
    as_baggage: as_baggage,
    &
  )
end

.reset! ⇒ void

This method returns an undefined value.

Reset global configuration and client (useful for testing)



291
292
293
294
295
296
297
298
299
300
 
# File 'lib/langfuse.rb', line 291

def reset!
  client.shutdown if @client
  OtelSetup.shutdown(timeout: 5) if OtelSetup.initialized?
  @configuration = nil
  @client = nil
rescue StandardError
  # Ignore shutdown errors during reset (e.g., in tests)
  @configuration = nil
  @client = nil
end

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

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|
  Langfuse.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)

Raises:

  • (ArgumentError) —

    if no active span or validation fails



240
241
242
243
244
245
246
247
248
 
# File 'lib/langfuse.rb', line 240

def score_active_observation(name:, value:, comment: nil, metadata: nil, data_type: :numeric)
  client.score_active_observation(
    name: name,
    value: value,
    comment: comment,
    metadata: metadata,
    data_type: data_type
  )
end

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

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|
  Langfuse.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)

Raises:

  • (ArgumentError) —

    if no active span or validation fails



266
267
268
269
270
271
272
273
274
 
# File 'lib/langfuse.rb', line 266

def score_active_trace(name:, value:, comment: nil, metadata: nil, data_type: :numeric)
  client.score_active_trace(
    name: name,
    value: value,
    comment: comment,
    metadata: metadata,
    data_type: data_type
  )
end

.shutdown(timeout: 30) ⇒ void

This method returns an undefined value.

Shutdown Langfuse and flush any pending traces and scores

Call this when shutting down your application to ensure all traces and scores are sent to Langfuse.

Examples:

In a Rails initializer or shutdown hook

at_exit { Langfuse.shutdown }

Parameters:

  • timeout (Integer) (defaults to: 30) —

    Timeout in seconds



117
118
119
120
 
# File 'lib/langfuse.rb', line 117

def shutdown(timeout: 30)
  client.shutdown if @client
  OtelSetup.shutdown(timeout: timeout)
end

.start_observation(name, attrs = {}, as_type: :span, parent_span_context: nil, start_time: nil, skip_validation: false) ⇒ BaseObservation

Creates a new observation (root or child)

This is the module-level factory method that creates observations of any type. It can create root observations (when parent_span_context is nil) or child observations (when parent_span_context is provided).

Examples:

Create root span

span = Langfuse.start_observation("root-operation", { input: {...} })

Create child generation

child = Langfuse.start_observation("llm-call", { model: "gpt-4" },
                                    as_type: :generation,
                                    parent_span_context: parent.otel_span.context)

Parameters:

  • name (String) —

    Descriptive name for the observation

  • attrs (Hash, Types::SpanAttributes, Types::GenerationAttributes, nil) (defaults to: {}) —

    Observation attributes

  • as_type (Symbol, String) (defaults to: :span) —

    Observation type (:span, :generation, :event, etc.)

  • parent_span_context (OpenTelemetry::Trace::SpanContext, nil) (defaults to: nil) —

    Parent span context for child observations

  • start_time (Time, Integer, nil) (defaults to: nil) —

    Optional start time (Time object or Unix timestamp in nanoseconds)

  • skip_validation (Boolean) (defaults to: false) —

    Skip validation (for internal use). Defaults to false.

Returns:

  • (BaseObservation) —

    The observation wrapper (Span, Generation, or Event)

Raises:

  • (ArgumentError) —

    if an invalid observation type is provided



324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
 
# File 'lib/langfuse.rb', line 324

def start_observation(name, attrs = {}, as_type: :span, parent_span_context: nil, start_time: nil,
                      skip_validation: false)
  type_str = as_type.to_s

  unless skip_validation || valid_observation_type?(as_type)
    valid_types = OBSERVATION_TYPES.values.sort.join(", ")
    raise ArgumentError, "Invalid observation type: #{type_str}. Valid types: #{valid_types}"
  end

  otel_tracer = otel_tracer()
  otel_span = create_otel_span(
    name: name,
    start_time: start_time,
    parent_span_context: parent_span_context,
    otel_tracer: otel_tracer
  )

  # Serialize attributes
  # Only set attributes if span is still recording (should always be true here, but guard for safety)
  if otel_span.recording?
    otel_attrs = OtelAttributes.create_observation_attributes(type_str, attrs.to_h)
    otel_attrs.each { |key, value| otel_span.set_attribute(key, value) }
  end

  # Wrap in appropriate class
  observation = wrap_otel_span(otel_span, type_str, otel_tracer, attributes: attrs)

  # Events auto-end immediately when created
  observation.end if type_str == OBSERVATION_TYPES[:event]

  observation
end