Class: LlmCostTracker::Tracker

Inherits:
Object
  • Object
show all
Defined in:
lib/llm_cost_tracker/tracker.rb

Constant Summary collapse

EVENT_NAME = 
"llm_request.llm_cost_tracker"

Class Method Summary collapse

Class Method Details

.enforce_budget! ⇒ Object 



10
11
12
 
# File 'lib/llm_cost_tracker/tracker.rb', line 10

def enforce_budget!
  Budget.enforce!
end

.record(provider:, model:, input_tokens:, output_tokens:, metadata: {}, latency_ms: nil) ⇒ LlmCostTracker::Event

Build, notify, persist, and budget-check a single LLM usage event.

Parameters:

  • provider (String) —

    Provider name.

  • model (String) —

    Model identifier.

  • input_tokens (Integer) —

    Input token count.

  • output_tokens (Integer) —

    Output token count.

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

    Attribution tags plus provider-specific usage metadata.

  • latency_ms (Integer, nil) (defaults to: nil) —

    Optional latency in milliseconds.

Returns:



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/llm_cost_tracker/tracker.rb', line 23

def record(provider:, model:, input_tokens:, output_tokens:, metadata: {}, latency_ms: nil)
  usage = EventMetadata.usage_data(input_tokens, output_tokens, metadata)

  cost_data = Pricing.cost_for(
    model: model,
    input_tokens: usage[:input_tokens],
    output_tokens: usage[:output_tokens],
    cached_input_tokens: usage[:cached_input_tokens],
    cache_read_input_tokens: usage[:cache_read_input_tokens],
    cache_creation_input_tokens: usage[:cache_creation_input_tokens]
  )

  UnknownPricing.handle!(model) unless cost_data

  event = Event.new(
    provider: provider,
    model: model,
    input_tokens: usage[:input_tokens],
    output_tokens: usage[:output_tokens],
    total_tokens: usage[:total_tokens],
    cost: cost_data,
    tags: LlmCostTracker.configuration.default_tags.merge(EventMetadata.tags(metadata)).freeze,
    latency_ms: normalized_latency_ms(latency_ms),
    tracked_at: Time.now.utc
  )

  # Emit ActiveSupport::Notifications event
  ActiveSupport::Notifications.instrument(EVENT_NAME, event.to_h)

  # Store based on backend
  stored = store(event)
  Budget.check!(event) unless stored == false

  event
end