Langfuse Ruby SDK

gem 'langfuse-ruby'

require "langfuse"

Langfuse.configure do |config|
  config.public_key = ENV.fetch("LANGFUSE_PUBLIC_KEY")
  config.secret_key = ENV.fetch("LANGFUSE_SECRET_KEY")
  config.host = ENV["LANGFUSE_HOST"] || ENV["LANGFUSE_BASE_URL"] || "https://us.cloud.langfuse.com"
  config.ingestion_mode = :otel   # Langfuse v4 — required for real-time OTEL ingestion
end

# Equivalent:
#   Langfuse.new(..., ingestion_mode: :otel)
#   LANGFUSE_INGESTION_MODE=otel

Langfuse.trace("chat-completion", user_id: "user-123", session_id: "sess-456",
               input: { message: "Hello, world!" }) do |trace|
  generation = trace.generation(
    name: "openai-completion",
    model: "gpt-4o",
    input: [{ role: "user", content: "Hello, world!" }],
    model_parameters: { temperature: 0.7, max_tokens: 100 }
  )

  response = call_llm(...)  # your code

  generation.end(
    output: response.content,
    usage_details: { input: 12, output: 18, total: 30 },   # v4 cost model
    cost_details: { input: 0.0001, output: 0.0006, total: 0.0007 }
  )
  generation.score(name: "faithfulness", value: 0.9)

  trace.update(output: response.content)
end  # flush is automatic in the block form

Langfuse.trace("document-qa", user_id: "user-456", input: { query: "What is Ruby?" }) do |trace|
  agent = trace.agent(name: "qa-agent", input: { query: "What is Ruby?" })

  retrieval = agent.retriever(name: "vector-search", input: { query: "What is Ruby?", top_k: 5 })
  retrieval.end(output: { documents: ["Ruby is a programming language..."] })

  gen = agent.generation(
    name: "openai-completion",
    model: "gpt-4o",
    input: [{ role: "user", content: "What is Ruby?" }],
    prompt: Langfuse.get_prompt("qa-system")  # optional prompt link
  )
  gen.end(
    output: "Ruby is a dynamic programming language.",
    usage_details: { input: 50, output: 20, total: 70 }
  )

  agent.end(output: { answer: "Ruby is a dynamic programming language." })
  trace.update(output: { answer: "Ruby is a dynamic programming language." })
end

require 'langfuse'

# Configure once — v4 / OTEL is the recommended ingestion path
Langfuse.configure do |config|
  config.public_key = ENV["LANGFUSE_PUBLIC_KEY"]
  config.secret_key = ENV["LANGFUSE_SECRET_KEY"]
  config.ingestion_mode = :otel
end

# Use block-based tracing - flush happens automatically!
Langfuse.trace("my-trace", user_id: "user-1", input: { message: "Hello" }) do |trace|
  generation = trace.generation(
    name: "openai-chat",
    model: "gpt-4",
    input: [{ role: "user", content: "Hello" }],
    model_parameters: { temperature: 0.7 }
  )

  # Call your LLM
  response = call_openai(...)

  # Record the response
  generation.end(
    output: response.content,
    usage_details: { input: 10, output: 15, total: 25 }  # or usage: response.usage
  )
  trace.update(output: response.content)
end  # Automatic flush here!

# Get and compile a prompt in one call
prompt = Langfuse.get_prompt("greeting-prompt", variables: { name: "Alice" })
# => "Hello Alice! How can I help you today?"

# Get prompt without compilation
prompt_obj = Langfuse.get_prompt("greeting-prompt")
compiled = prompt_obj.compile(name: "Bob")

# If Langfuse fails, a NullTrace is used - your code still runs
Langfuse.trace("my-trace") do |trace|
  # This works even if Langfuse is down
  gen = trace.generation(name: "test", model: "gpt-4", input: "hello")
  gen.end(output: "world")
end

# Get the process-wide, thread-safe singleton client
client = Langfuse.client

# Manual flush (when not using block-based tracing)
Langfuse.flush

# Shutdown the client (idempotent; also runs via at_exit when shutdown_on_exit is true)
Langfuse.shutdown

# Reset the singleton (useful for testing; prefer Langfuse.new for isolated clients)
Langfuse.reset!

trace = client.trace(name: "document-qa")

# Create a span for document retrieval
retrieval_span = trace.span(
  name: "document-retrieval",
  input: { query: "What is machine learning?" }
)

# Add a generation for embedding
embedding_gen = retrieval_span.generation(
  name: "embedding-generation",
  model: "text-embedding-ada-002",
  input: "What is machine learning?",
  output: [0.1, 0.2, 0.3], # embedding vector
  usage: { prompt_tokens: 5, total_tokens: 5 }
)

# End the retrieval span
retrieval_span.end(
  output: { documents: ["ML is...", "Machine learning involves..."] }
)

# Create a span for answer generation
answer_span = trace.span(
  name: "answer-generation",
  input: { 
    query: "What is machine learning?",
    context: ["ML is...", "Machine learning involves..."]
  }
)

# Add LLM generation
llm_gen = answer_span.generation(
  name: "openai-completion",
  model: "gpt-3.5-turbo",
  input: [
    { role: "system", content: "Answer based on context" },
    { role: "user", content: "What is machine learning?" }
  ]
)

llm_gen.end(
  output: { answer: "Machine learning is a subset of AI..." },
  usage_details: { input: 50, output: 30, total: 80 }
)
answer_span.end(output: { answer: "Machine learning is a subset of AI..." })

trace = client.trace(name: "support-agent", user_id: "u1")

agent = trace.agent(name: "planner", input: { question: "Reset my password" })
tool  = agent.tool(name: "lookup-user", input: { email: "a@example.com" })
tool.end(output: { user_id: "u1" })

guard = agent.guardrail(name: "content-filter", input: { text: "Reset my password" })
guard.end(output: { blocked: false })

gen = agent.generation(name: "reply", model: "gpt-4o", input: [...])
gen.end(output: "I can help with that.", usage_details: { input: 40, output: 12, total: 52 })
agent.end(output: { reply: "I can help with that." })

# Create events from trace
event = trace.event(
  name: "user_action",
  input: { action: "login", user_id: "123" },
  output: { success: true },
  metadata: { ip: "192.168.1.1" }
)

# Create events from spans or generations
validation_event = span.event(
  name: "validation_check",
  input: { rules: ["required", "format"] },
  output: { valid: true, warnings: [] }
)

# Direct event creation
event = client.event(
  trace_id: trace.id,
  name: "audit_log",
  input: { operation: "data_export" },
  output: { status: "completed" },
  level: "INFO"
)

# Get a prompt
prompt = client.get_prompt("chat-prompt", version: 1)

# Prompt names with special characters are automatically URL-encoded
prompt = client.get_prompt("EXEMPLE/my-prompt")  # Works correctly!

# Compile prompt with variables
compiled = prompt.compile(
  user_name: "Alice",
  topic: "machine learning"
)

puts compiled
# Output: "Hello Alice! Let's discuss machine learning today."

# Create a text prompt
text_prompt = client.create_prompt(
  name: "greeting-prompt",
  prompt: "Hello {{user_name}}! How can I help you with {{topic}} today?",
  labels: ["greeting", "customer-service"],
  config: { temperature: 0.7 }
)

# Create a chat prompt
chat_prompt = client.create_prompt(
  name: "chat-prompt",
  prompt: [
    { role: "system", content: "You are a helpful assistant specialized in {{domain}}." },
    { role: "user", content: "{{user_message}}" }
  ],
  labels: ["chat", "assistant"]
)

# Create prompt templates for reuse
template = Langfuse::PromptTemplate.from_template(
  "Translate the following text to {{language}}: {{text}}"
)

translated = template.format(
  language: "Spanish",
  text: "Hello, world!"
)

# Chat prompt templates
chat_template = Langfuse::ChatPromptTemplate.from_messages([
  { role: "system", content: "You are a {{role}} assistant." },
  { role: "user", content: "{{user_input}}" }
])

messages = chat_template.format(
  role: "helpful",
  user_input: "What is Ruby?"
)

# Exact match evaluator
exact_match = Langfuse::Evaluators::ExactMatchEvaluator.new

result = exact_match.evaluate(
  input: "What is 2+2?",
  output: "4",
  expected: "4"
)
# => { name: "exact_match", value: 1, comment: "Exact match" }

# Similarity evaluator
similarity = Langfuse::Evaluators::SimilarityEvaluator.new

result = similarity.evaluate(
  input: "What is AI?",
  output: "Artificial Intelligence is...",
  expected: "AI is artificial intelligence..."
)
# => { name: "similarity", value: 0.85, comment: "Similarity: 85%" }

# Length evaluator
length = Langfuse::Evaluators::LengthEvaluator.new(min_length: 10, max_length: 100)

result = length.evaluate(
  input: "Explain AI",
  output: "AI is a field of computer science that focuses on creating intelligent machines."
)
# => { name: "length", value: 1, comment: "Length 80 within range" }

# Add scores to traces or observations
trace = client.trace(name: "qa-session")

# Score the entire trace
trace.score(
  name: "user-satisfaction",
  value: 0.9,
  comment: "User was very satisfied"
)

# Score specific generations
generation = trace.generation(
  name: "answer-generation",
  model: "gpt-3.5-turbo",
  output: { content: "The answer is 42." }
)

generation.score(
  name: "accuracy",
  value: 0.8,
  comment: "Mostly accurate answer"
)

generation.score(
  name: "helpfulness",
  value: 0.95,
  comment: "Very helpful response"
)

# Tag all events with a tracing environment (also via LANGFUSE_TRACING_ENVIRONMENT)
client = Langfuse.new(
  public_key: "pk-lf-...",
  secret_key: "sk-lf-...",
  environment: "production"
)

# Sample a fraction of traces deterministically (also via LANGFUSE_SAMPLE_RATE)
# All events of a trace share the same keep/drop decision.
sampled_client = Langfuse.new(
  public_key: "pk-lf-...",
  secret_key: "sk-lf-...",
  sample_rate: 0.1
)

# Mask sensitive fields before sending (applied to input/output/metadata)
masked_client = Langfuse.new(
  public_key: "pk-lf-...",
  secret_key: "sk-lf-...",
  mask: ->(value) { value.to_s.gsub(/\b\d{16}\b/, "***CARD***") }
)

client = Langfuse.new(
  public_key: "pk-lf-...",
  secret_key: "sk-lf-...",
  flush_at: 50,           # flush after 50 events
  flush_interval: 10,     # or every 10 seconds
  max_queue_size: 20_000, # drop the oldest events beyond this many queued
  shutdown_on_exit: true  # flush on process exit (default)
)

# Trace-level score
trace.score(name: "accuracy", value: 0.9, comment: "good")

# Observation-level score (trace_id is set automatically on Span/Generation)
generation.score(name: "faithfulness", value: 0.8, data_type: "NUMERIC")

# Session-level score
client.score(name: "csat", value: 5, session_id: "sess-1", data_type: "NUMERIC")

# Dataset-run score with metadata and config link
client.score(
  name: "hallucination",
  value: 0.2,
  dataset_run_id: "run-1",
  trace_id: "trace-1",
  metadata: { evaluator: "llm-judge" },
  config_id: "cfg-abc",
  data_type: "NUMERIC"
)

# Categorical string value
client.score(name: "label", value: "good", trace_id: "t1", data_type: "CATEGORICAL")

# New v4 usage model (arbitrary keys, e.g. cache tokens)
gen = trace.generation(
  name: "chat",
  model: "gpt-4o",
  usage_details: { input: 100, output: 50, cache_read: 30 },
  cost_details: { input: 0.001, output: 0.003, total: 0.004 }
)
gen.end(output: "response")

# Link a generation to a prompt version (accepts a Langfuse::Prompt or a hash)
prompt = Langfuse.get_prompt("chat-prompt")
gen = trace.generation(name: "chat", model: "gpt-4o", prompt: prompt)
# or: prompt: { name: "chat-prompt", version: 3 }

begin
  client = Langfuse.new(
    public_key: "invalid-key",
    secret_key: "invalid-secret"
  )
  
  trace = client.trace(name: "test")
  client.flush
rescue Langfuse::AuthenticationError => e
  puts "Authentication failed: #{e.message}"
rescue Langfuse::RateLimitError => e
  puts "Rate limit exceeded: #{e.message}"
rescue Langfuse::NetworkError => e
  puts "Network error: #{e.message}"
rescue Langfuse::APIError => e
  puts "API error: #{e.message}"
end

client = Langfuse.new(
  public_key: "pk-lf-...",
  secret_key: "sk-lf-...",
  host: "https://us.cloud.langfuse.com", # US default; EU: cloud.langfuse.com
  ingestion_mode: :otel, # Langfuse v4 (default remains :legacy)
  debug: true,          # Enable debug logging (or LANGFUSE_DEBUG=true)
  timeout: 30,          # Request timeout in seconds
  retries: 3,           # Number of retry attempts
  flush_interval: 30,   # Event flush interval in seconds (default: 5)
  flush_at: 50,         # Flush once this many events are queued (default: 15)
  max_queue_size: 10_000, # Drop the oldest events beyond this many queued (default: 10_000)
  auto_flush: true,     # Enable automatic flushing (default: true)
  environment: "prod",  # Tracing environment (or LANGFUSE_TRACING_ENVIRONMENT)
  sample_rate: 0.5,     # Keep 50% of traces deterministically (or LANGFUSE_SAMPLE_RATE)
  mask: ->(v) { v },    # Callable applied to input/output/metadata
  shutdown_on_exit: true, # Flush pending events on process exit (default: true)
  http_adapter: :net_http_persistent # Faraday adapter (default: Faraday's default)
)

# Gemfile
gem "faraday-net_http_persistent"

Langfuse.configure do |config|
  config.http_adapter = :net_http_persistent
end

# Enable automatic flushing (default)
client = Langfuse.new(
  public_key: "pk-lf-...",
  secret_key: "sk-lf-...",
  auto_flush: true,
  flush_interval: 5  # Flush every 5 seconds
)

# Disable automatic flushing for manual control
client = Langfuse.new(
  public_key: "pk-lf-...",
  secret_key: "sk-lf-...",
  auto_flush: false
)

# Manual flush when auto_flush is disabled
client.flush

Langfuse.configure do |config|
  config.auto_flush = false  # Disable auto flush globally
  config.flush_interval = 10
end

# Example: Batch processing with manual flush
client = Langfuse.new(auto_flush: false)

# Process many items
1000.times do |i|
  trace = client.trace(name: "batch-item-#{i}")
  # ... process item
end

# Flush all events at once
client.flush

# Ensure all events are flushed before shutdown
client.shutdown

# config/initializers/langfuse.rb
Langfuse.configure do |config|
  config.public_key = Rails.application.credentials.langfuse_public_key
  config.secret_key = Rails.application.credentials.langfuse_secret_key
  config.ingestion_mode = :otel
  config.debug = Rails.env.development?
end

# In your controller or service
class ChatController < ApplicationController
  def create
    @client = Langfuse.new
    
    trace = @client.trace(
      name: "chat-request",
      user_id: current_user.id,
      session_id: session.id,
      input: params[:message],
      metadata: { 
        controller: self.class.name,
        action: action_name,
        ip: request.remote_ip
      }
    )
    
    # Your LLM logic here
    response = generate_response(params[:message])

    render json: { response: response }
  end
end

class LLMProcessingJob < ApplicationJob
  def perform(user_id, message)
    client = Langfuse.new

    trace = client.trace(
      name: "background-llm-processing",
      user_id: user_id,
      input: { message: message },
      metadata: { job_class: self.class.name }
    )

    # Process with LLM
    result = process_with_llm(message)

    # Ensure events are flushed
    client.flush
  end
end