Skip to content

[AAI-235] Add OTLP v4 header and v2 reads #95

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kxzk wants to merge 4 commits into
base: main
Choose a base branch
from
+842 −4
Open

[AAI-235] Add OTLP v4 header and v2 reads #95

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c415dea
[AAI-235] Add OTLP v4 header and v2 reads
kxzk Jul 13, 2026
c4a2d4d
[AAI-235] Treat blank strings as absent in bounded-read guard
kxzk Jul 13, 2026
9c52210
[AAI-235] Treat false as absent in bounded-read guard
kxzk Jul 13, 2026
91ba75d
fix(read-api): reject invalid observation bounds
kxzk Jul 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
162 changes: 162 additions & 0 deletions docs/API_REFERENCE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -909,6 +909,112 @@ trace = client.get_trace("trace-uuid-123")
puts trace["name"]
```

### `Client#list_observations`

List observation rows with cursor-based pagination and field selection.

> **Cloud-only.** Delegates to `GET /api/public/v2/observations`, which is only
> available on Langfuse Cloud. There is no fallback to legacy endpoints because
> their response and pagination semantics differ. Returns observation rows, not
> reconstructed trace objects — group rows by `traceId` when you need trace
> activity.

**Signature:**

```ruby
list_observations(from_start_time: nil, to_start_time: nil, trace_id: nil, **filters) # => Hash
```

**Parameters:**

| Parameter | Type | Required | Description |
| ----------------------- | --------------------- | -------- | ---------------------------------------------------------------------------------------------------- |
| `from_start_time` | Time, String | Yes* | Inclusive lower bound on observation start time. *Required unless `trace_id` is provided. |
| `to_start_time` | Time, String | Yes* | Exclusive upper bound on observation start time. *Required unless `trace_id` is provided. |
| `trace_id` | String | No | Filter by trace ID |
| `fields` | String | No | Comma-separated field groups: `core`, `basic`, `time`, `io`, `metadata`, `model`, `usage`, `prompt`, `metrics`, `trace_context` |
| `cursor` | String | No | Cursor from the previous response's `meta` for the next page |
| `limit` | Integer | No | Items per page (max 1000, default 50) |
| `filter` | String | No | JSON string with structured filter conditions (takes precedence over individual filters) |
| `name` | String | No | Filter by observation name |
| `user_id` | String | No | Filter by user ID |
| `type` | String | No | Filter by observation type (e.g. `"GENERATION"`, `"SPAN"`) |
| `level` | String | No | Filter by level (e.g. `"DEFAULT"`, `"ERROR"`) |
| `parent_observation_id` | String | No | Filter by parent observation ID |
| `environment` | String | No | Filter by environment |
| `version` | String | No | Filter by observation version |
| `expand_metadata` | String | No | Comma-separated metadata keys to return non-truncated |

**Returns:** `Hash` with `"data"` (observation rows) and `"meta"` (pagination cursor)

**Raises:**

- `ArgumentError` if the read is unbounded (no `trace_id` and missing start-time bounds)
- `UnauthorizedError` if authentication fails
- `ApiError` for other API errors (including non-Cloud deployments)

**Examples:**

```ruby
# Bounded read of recent generations
page = client.list_observations(
from_start_time: Time.now - 3600,
to_start_time: Time.now,
type: "GENERATION",
fields: "core,basic,usage"
)
page["data"].each { |obs| puts obs["id"] }

# Next page via cursor
next_page = client.list_observations(
from_start_time: Time.now - 3600,
to_start_time: Time.now,
cursor: page.dig("meta", "cursor")
)

# All observations for one trace
rows = client.list_observations(trace_id: "trace-uuid-123")["data"]
```

### `Client#query_metrics`

Query aggregate metrics.

> **Cloud-only.** Delegates to `GET /api/public/v2/metrics`. Supports the
> `observations`, `scores-numeric`, and `scores-categorical` views.

**Signature:**

```ruby
query_metrics(query:) # => Hash
```

**Parameters:**

| Parameter | Type | Required | Description |
| --------- | ------------ | -------- | ------------------------------------------------------------------------------------------ |
| `query` | Hash, String | Yes | Metrics query. A Hash is JSON-encoded; a pre-encoded JSON String is passed through as-is. |

**Returns:** `Hash` of metrics results

**Raises:**

- `ArgumentError` if query is neither a Hash nor a String
- `UnauthorizedError` if authentication fails
- `ApiError` for other API errors (including non-Cloud deployments)

**Example:**

```ruby
result = client.query_metrics(query: {
view: "observations",
metrics: [{ measure: "count", aggregation: "count" }],
dimensions: [{ field: "name" }],
fromTimestamp: "2026-07-01T00:00:00Z",
toTimestamp: "2026-07-02T00:00:00Z"
})
```

## Scoring

### `Client#create_score`
Expand Down Expand Up @@ -1012,6 +1118,62 @@ Langfuse.flush_scores

See [SCORING.md](SCORING.md) for complete guide.

### `Client#list_scores`

List scores with polymorphic values via the v3 scores API.

**Signature:**

```ruby
list_scores(**filters) # => Hash
```

**Parameters:**

| Parameter | Type | Required | Description |
| ---------------- | ------------ | -------- | ------------------------------------------------------------------------------------------------ |
| `limit` | Integer | No | Items per page (max 100, default 50) |
| `cursor` | String | No | Cursor from the previous response's `meta` for the next page |
| `fields` | String | No | Comma-separated field groups in addition to core: `details`, `subject`, `annotation` |
| `id` | String | No | Comma-separated score IDs |
| `name` | String | No | Comma-separated score names |
| `source` | String | No | Comma-separated sources (e.g. `API`, `ANNOTATION`, `EVAL`) |
| `data_type` | String | No | Comma-separated data types: `NUMERIC`, `BOOLEAN`, `CATEGORICAL`, `TEXT`, `CORRECTION` |
| `environment` | String | No | Comma-separated environments |
| `config_id` | String | No | Comma-separated score config IDs |
| `queue_id` | String | No | Comma-separated annotation queue IDs |
| `author_user_id` | String | No | Comma-separated author user IDs |
| `value` | String | No | Comma-separated exact values (requires a single `NUMERIC`, `BOOLEAN`, or `CATEGORICAL` data type) |
| `value_min` | Numeric | No | Inclusive lower bound (requires `data_type: "NUMERIC"`) |
| `value_max` | Numeric | No | Inclusive upper bound (requires `data_type: "NUMERIC"`) |
| `trace_id` | String | No | Comma-separated trace IDs (mutually exclusive with `session_id`, `experiment_id`) |
| `session_id` | String | No | Comma-separated session IDs |
| `observation_id` | String | No | Comma-separated observation IDs (requires `trace_id`) |
| `experiment_id` | String | No | Comma-separated dataset run (experiment) IDs |
| `from_timestamp` | Time, String | No | Inclusive lower bound on score timestamp |
| `to_timestamp` | Time, String | No | Exclusive upper bound on score timestamp |

**Returns:** `Hash` with `"data"` (score rows) and `"meta"` (pagination cursor). Score
values are polymorphic by `dataType`: `NUMERIC` scores return numbers, `BOOLEAN`
scores return booleans, and `CATEGORICAL`, `TEXT`, and `CORRECTION` scores return
strings.

**Raises:**

- `UnauthorizedError` if authentication fails
- `ApiError` for other API errors

**Examples:**

```ruby
# Scores for a trace
page = client.list_scores(trace_id: "trace-uuid-123")

# Corrections with subject details
corrections = client.list_scores(data_type: "CORRECTION", fields: "subject,details")
corrections["data"].each { |score| puts score["value"] }
```

## Datasets

### `Client#create_dataset`
Expand Down
1 change: 1 addition & 0 deletions lib/langfuse.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ class UnauthorizedError < ApiError; end
require_relative "langfuse/prompt_cache_coordinator"
require_relative "langfuse/cache_warmer"
require_relative "langfuse/prompt_cache_events"
require_relative "langfuse/read_api"
require_relative "langfuse/api_client"
require_relative "langfuse/span_filter"
require_relative "langfuse/sampling"
Expand Down
2 changes: 2 additions & 0 deletions lib/langfuse/api_client.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
require "uri"
require_relative "prompt_fetch_result"
require_relative "prompt_cache_coordinator"
require_relative "read_api"

module Langfuse
# HTTP client for Langfuse API
Expand All @@ -25,6 +26,7 @@ module Langfuse
#
class ApiClient # rubocop:disable Metrics/ClassLength
include PromptCacheEvents
include ReadApi

# @return [String] Langfuse public API key
attr_reader :public_key
Expand Down
6 changes: 6 additions & 0 deletions lib/langfuse/client.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ class Client
# @!method validate_prompt_cache_backend!
# @!method list_traces(**options)
# @!method get_trace(id)
# @!method list_observations(**options)
# @!method query_metrics(query:)
# @!method list_scores(**options)
# @!method list_datasets(page: nil, limit: nil)
# @!method get_dataset_run(dataset_name:, run_name:)
# @!method create_dataset_run_item(**)
Expand All @@ -57,6 +60,9 @@ class Client
:validate_prompt_cache_backend!,
:list_traces,
:get_trace,
:list_observations,
:query_metrics,
:list_scores,
:list_datasets,
:get_dataset_run,
:create_dataset_run_item
Expand Down
17 changes: 13 additions & 4 deletions lib/langfuse/otel_setup.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,7 @@ def build_tracer_provider(config)
def build_exporter(config)
OpenTelemetry::Exporter::OTLP::Exporter.new(
endpoint: "#{config.base_url}/api/public/otel/v1/traces",
headers: build_headers(config.public_key, config.secret_key),
headers: build_headers(config),
compression: "gzip"
)
end
Expand Down Expand Up @@ -170,10 +170,19 @@ def blank?(value)
value.nil? || value.empty?
end

def build_headers(public_key, secret_key)
credentials = "#{public_key}:#{secret_key}"
# Langfuse requires `x-langfuse-ingestion-version: 4` on direct OTLP
# exports for real-time Fast Preview processing; without it, data can be
# delayed on the new UI and v2 read surfaces.
def build_headers(config)
credentials = "#{config.public_key}:#{config.secret_key}"
encoded = Base64.strict_encode64(credentials)
{ "Authorization" => "Basic #{encoded}" }
{
"Authorization" => "Basic #{encoded}",
"x-langfuse-ingestion-version" => "4",
"x-langfuse-sdk-name" => "ruby",
"x-langfuse-sdk-version" => Langfuse::VERSION,
"x-langfuse-public-key" => config.public_key
}
end

def build_sampler(sample_rate)
Expand Down
Loading
Loading