Module: Legion::LLM::API::Namespaces::Anthropic::Messages::CountTokens

Extended by:
Legion::Logging::Helper, Sinatra::Extension
Defined in:
lib/legion/llm/api/namespaces/anthropic/messages/count_tokens.rb

Class Method Summary collapse

Class Method Details

.count_tokens_result(messages:, model:, system:, tools:) ⇒ Object

SSOT v3 §20 (count_tokens row): choose the EXACT lane first, then use that lane's callable tokenizer. model is a required, caller-supplied value — never a default. When the SSOT registry publishes a count_tokens lane for the requested model, the exact selected callable's tokenizer is authoritative; otherwise (cold registry, no count_tokens lane, or a callable without a tokenizer) fall back to the provider-neutral TokenEstimation.



52
53
54
55
56
57
 
# File 'lib/legion/llm/api/namespaces/anthropic/messages/count_tokens.rb', line 52

def self.count_tokens_result(messages:, model:, system:, tools:)
  lane_count = count_tokens_via_selected_lane(messages: messages, model: model, system: system, tools: tools)
  return lane_count if lane_count

  Legion::LLM::TokenEstimation.estimate(messages: messages, model: model, system: system, tools: tools)
end

.count_tokens_via_selected_lane(messages:, model:, system:, tools:) ⇒ Object 



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
 
# File 'lib/legion/llm/api/namespaces/anthropic/messages/count_tokens.rb', line 59

def self.count_tokens_via_selected_lane(messages:, model:, system:, tools:)
  snapshot = Legion::Extensions::Llm::Inventory::Registry.snapshot
  return nil unless snapshot.generation.positive?

  request = Legion::LLM::Inference::Request.build(
    messages: messages,
    system:   system,
    routing:  { model: model },
    tools:    tools || []
  )
  requirements = Legion::LLM::Router::RequestRequirements.build(
    request:                request,
    operation:              :count_tokens,
    required_capabilities:  Legion::LLM::Router::RequiredCapabilities.call(request: request, operation: :count_tokens),
    estimated_input_bound:  0,
    required_output_tokens: 0
  )
  session = Legion::LLM::Inference::RoutingSession.new(request: request, requirements: requirements)
  attempt = session.next_attempt(snapshot: snapshot)
  return nil if attempt.is_a?(Legion::Extensions::Llm::Routing::Rejection)

  dispatch = Legion::LLM::Call::SelectionDispatch.call(
    attempt_context: attempt,
    arguments:       { messages: messages }
  )
  return nil unless dispatch.success?

  normalize_token_count(dispatch.value)
rescue ::NoMethodError, ::ArgumentError, ::NotImplementedError
  # Programming errors are never swallowed — re-raise.
  raise
rescue StandardError => e
  # Non-programming errors during lane selection/tokenizer dispatch fall
  # back to provider-neutral estimation.
  log.warn "[llm][ns][anthropic][count_tokens] action=lane_tokenizer_fallback error=#{e.class}: #{e.message}"
  nil
end

.normalize_token_count(value) ⇒ Object 



97
98
99
100
101
102
 
# File 'lib/legion/llm/api/namespaces/anthropic/messages/count_tokens.rb', line 97

def self.normalize_token_count(value)
  return value if value.is_a?(Hash) && (value.key?(:input_tokens) || value.key?('input_tokens'))
  return { input_tokens: value } if value.is_a?(Integer)

  nil
end