Module: PlatformSdk::Observability::Langfuse::NotificationSubscriber

Defined in:
lib/platform_sdk/observability/langfuse/notification_subscriber.rb

Overview

Subscribes to llm_call.platform_sdk ActiveSupport::Notifications events and forwards them to Recorder.record_generation.

Apps instrument their LLM calls with the convention:

ActiveSupport::Notifications.instrument(
'llm_call.platform_sdk',
name: 'chat_response',
model: 'claude-3-5-sonnet',
input: prompt_messages,
output: response_text,
usage: { input_tokens: 120, output_tokens: 84 }
) { call_the_llm }

Inside the block, the payload Hash can be mutated to enrich it with values only known after the call completes. (Instrumenter#instrument passes the same Hash to the block and reads it again after yield; this isn't documented in the Rails Guide but is relied on across the ecosystem.)

Two error-passing paths:

  1. Direct instrument-with-a-block callers that let the block raise: Rails populates :exception_object automatically and this subscriber records a failure observation from it.

  2. Caller-rescues-and-rethrows patterns (e.g. RubyLLMAdapter.fire, course-builder's LlmErrorHandling#with_llm_error_handling): set payload[:error] = exception before firing. Rails only auto-sets :exception_object for the block form, so adapter-style callers that instrument without a block need to populate :error.

format_output prefers :error when both are set so the caller-provided value wins.

Idempotent: install! is safe to call repeatedly — only the first call registers a subscriber.

Constant Summary collapse

EVENT_NAME = 
LLM_CALL_EVENT
DEFAULT_NAME = 
'llm_call'

Class Method Summary collapse

Class Method Details

.install! ⇒ Object 



54
55
56
57
58
59
60
61
62
63
 
# File 'lib/platform_sdk/observability/langfuse/notification_subscriber.rb', line 54

def install!
  @mutex.synchronize do
    return if @installed

    @subscriber = ActiveSupport::Notifications.subscribe(EVENT_NAME) do |_name, _start, _finish, _id, payload|
      handle_event(payload)
    end
    @installed = true
  end
end

.installed? ⇒ Boolean

Returns:

  • (Boolean) 


73
74
75
 
# File 'lib/platform_sdk/observability/langfuse/notification_subscriber.rb', line 73

def installed?
  @installed
end

.reset! ⇒ Object 



65
66
67
68
69
70
71
 
# File 'lib/platform_sdk/observability/langfuse/notification_subscriber.rb', line 65

def reset!
  @mutex.synchronize do
    ActiveSupport::Notifications.unsubscribe(@subscriber) if @subscriber
    @subscriber = nil
    @installed = false
  end
end