Compare commits

...
38 Commits
Author SHA1 Message Date
Max Deichmann 9957c51d66 chore: release v3.53.0 2025-04-28 12:19:51 +02:00
0f06a066df feat(scores): support session level scores (#6367)
* chore: make changes to `scores` data structure

* fixup: migrations

* chore: update CH score types and schemas

* feat: support session level scores

* chore: add mock session level scores

* feat(session-scores): create v2 APIs; extract score and trace api service

* chore: clean up todos

* eslint

* eslint

* chore: reorder migrations after rebase

* chore: order unclustered migrations after rebase

* chore: fixes after rebase

* chore

* revert changes to .json file

* chore

* chore: move session_id column

* typo

* chore: migration adjustment

* chore: styling adjustments

* chore: materialize index

* chore: fix types after rebase

* chore: fix types after rebase

* chore: feedback

* chore: refactor scores types file structure

* chore: fix lint

* chore: add api types

* chore: rename

* fix: wait for clickhouse mutations to succeed (#6629)

* fix: wait for clickhouse mutations to succeed

* push

* fix: wait for clickhouse mutations to succeed

---------

Co-authored-by: Max Deichmann <m.deichmann@tum.de>
2025-04-28 09:51:22 +00:00
marliessophieandGitHub ff8ccf1d61 chore(ui): add support for input_text and output_text in markdown rendering (#6625) 2025-04-28 08:39:15 +00:00
Max DeichmannandGitHub eeff4a4f68 feat: add dlx retry queue and service (#6621)
* feat: add dlx retry queue and service

* feat: add dlx retry queue and service

* feat: add dlx retry queue and service

* feat: add dlx retry queue and service

* feat: add dlx retry queue and service
2025-04-27 10:15:24 +00:00
Hassieb PakzadandGitHub 876d43e317 perf(api-traces): increase observations payload limit to 10mb (#6615)
* perf(api-traces): increase observations payload limit to 10mb

* add test
2025-04-25 18:47:05 +02:00
Hassieb PakzadandGitHub 3da7f3c8b2 perf(trace-api): increase observations payload limit to 6mb (#6614) 2025-04-25 15:11:48 +00:00
Hassieb PakzadandGitHub 6e5d0cee7b fix(api-prompts): use middleware to send response when updating labels (#6611)
* fix(api-prompts): use middleware to send response when updating labels

* fix lint error
2025-04-25 14:57:45 +00:00
Steffen SchmitzandGitHub 9f1fafd1bb build: prepare deployment for HIPAA compatibility (#6613) 2025-04-25 14:26:04 +00:00
Steffen SchmitzandGitHub a2cbcbc585 Revert "perf: race byId queries for traces and observations (#6555)" (#6612)
* Revert "perf: race byId queries for traces and observations (#6555)"

This reverts commit 591fb98a

* chore: compare on toDate starttime
2025-04-25 13:29:25 +00:00
Steffen SchmitzandGitHub e7aa0fbe08 feat: add support for score name filters on dashboard (#6608) 2025-04-25 09:52:23 +00:00
Hassieb PakzadandGitHub 30e608244a test(trace-api): fix observations too large test (#6607) 2025-04-25 11:32:59 +02:00
Hassieb PakzadandGitHub 3ec5476c04 perf(api-traces): increase byId size limit to 5MB (#6604) 2025-04-25 10:11:01 +02:00
Max DeichmannandGitHub 8c46c79e4f perf: reduce optimistic lookback for byid queries (#6593)
* perf: reduce optimistic lookback for byid queries

* perf: reduce optimistic lookback for byid queries
2025-04-24 15:58:13 +00:00
Hassieb PakzadandGitHub 231ee5873c perf(public-api-traces): raise early error if observations are too large (#6592) 2025-04-24 14:25:01 +00:00
Steffen SchmitzandGitHub 7a74f7da9d perf: refactor clickhouseClient creation to singletons (#6584)
* perf: refactor clickhouseClient creation to singletons

* chore: align remaining client usages

* chore: update test usage
2025-04-24 14:03:53 +00:00
Steffen SchmitzandGitHub 82bf0baab7 perf: return only on API if no trace found (#6587) 2025-04-24 14:29:26 +02:00
Steffen SchmitzandGitHub 591fb98afb perf: race byId queries for traces and observations (#6555)
* perf: race byId queries for traces and observations

* chore: update tests
2025-04-24 10:02:04 +00:00
marliessophieandGitHub e919893c2b fix(ui): add deprecated col mapping on traces table for cached filter states (#6583)
* fix(ui): add deprecated col mapping on traces table for cached filter states

* chore: improve comments

* chore: fix

* chore: push
2025-04-24 07:01:54 +00:00
Max DeichmannandGitHub 237076bf92 chore: improve user identification in observability (#6575) 2025-04-23 20:11:27 +00:00
Steffen SchmitzandGitHub c7b09e7843 perf: include timestamp on trace retrieval on sessions and datasetrunitemstable (#6549) 2025-04-23 16:03:12 +00:00
marliessophieandGitHub 277e7b7501 feat(scores): support categorical filters across tables (#6484)
* fixup(scores): add categorical scores filter to traces table

* chore: support categorical scores filter on traces table

* chore: throw error if categories options filter is used in postgres

* chore: eslint

* feat: support categorical scores filter on observation scores

* chore: adjust typing

* style: filter builder

* chore: rename function
2025-04-23 07:30:43 +00:00
a01d12c064 fix: use source entity environment for annotation scores (#6545)
* fix: use source entity environment for annotation scores

* chore: resolve typo

* perf: remove redundant trace fetch for traces.byIdWithObservationsAndScores

* Revert "perf: remove redundant trace fetch for traces.byIdWithObservationsAndScores"

This reverts commit 544d241f0ddeccc9e1b7c1cd8f33a63d7253ae6c.

* fix: include `environment` in `handleCommentUpdate` call

---------

Co-authored-by: Marlies Mayerhofer <74332854+marliessophie@users.noreply.github.com>
2025-04-22 16:14:36 +00:00
Max DeichmannGitHubellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com>Steffen Schmitz
e1ae79e0d1 perf: soft delete event logs clickhouse (#6533)
* chore: mograte event_log table to new blob storage table

* push

* push

* push

* fix

* fix

* Update packages/shared/clickhouse/migrations/clustered/00011_add_blob_storage_file_log.up.sql

Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com>

* Update worker/src/backgroundMigrations/migrateEventLogToBlobStorageRefTable.ts

Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com>

* push

* push

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* push

* fix

* fix

* fix

* push

* fix

* fix

* fix

* fix

* fix

* fix

* push

* push

* Update packages/shared/src/server/data-deletion/ingestionFileDeletion.ts

Co-authored-by: Steffen Schmitz <steffen@langfuse.com>

* push

* fix

* fix

* fix

* push

---------

Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com>
Co-authored-by: Steffen Schmitz <steffen@langfuse.com>
2025-04-22 13:43:10 +00:00
marliessophieandGitHub 9804739ff1 chore(observations): show model name in table column (#6552) 2025-04-22 12:39:17 +00:00
Steffen SchmitzandGitHub 2d1f104a1c perf: remove redundant trace fetch for traces.byIdWithObservationsAndScores (#6548)
* perf: remove redundant trace fetch for traces.byIdWithObservationsAndScores

* chore: include fromTimestamp in enforceTraceAccess util
2025-04-22 11:38:31 +00:00
Max DeichmannandGitHub 2a871d1e75 chore: migrate event_log table to new blob storage table (#6534) 2025-04-22 12:24:55 +02:00
marliessophieandGitHub 4c749298e6 chore(prompts): throw 400 if prompt name type mismatched (#6547)
chore(prompts): throw 400 if prompt name type mismatched
2025-04-22 10:17:52 +00:00
marliessophieandGitHub b207712bcd chore(model-costs): support alias for all models (#6542)
* chore(model-costs): support alias for all models

* chore: remove from playground
2025-04-22 09:29:46 +00:00
Hassieb PakzadandGitHub 472b053c44 fix(llm-keys): drop top_p from test call (#6544) 2025-04-22 09:22:26 +00:00
steffen911 5b77e5835b chore: release v3.52.0 2025-04-22 10:00:35 +02:00
Steffen SchmitzandGitHub 657bb63c3e chore: make custom dashboards available for self-hosters (#6541) 2025-04-22 07:33:56 +00:00
Steffen SchmitzandGitHub b101e7c558 feat: allow optional metadata param on org / project CRUD API (#6524)
* feat: allow optional metadata param on org / project CRUD API

* chore: drop debugger statements

* chore: remove more debuggers

* chore: update tests

* chore: add metadata to fern spec
2025-04-22 07:30:02 +00:00
Steffen SchmitzandGitHub bac6b13145 fix: handle data retention on clustered clickhouse deployments (#6538) 2025-04-22 07:27:09 +00:00
Steffen SchmitzandGitHub 6817cdab0a chore: confirm pagination behaviour for scores-api (#6537)
* chore: confirm pagination behaviour for scores-api

* chore: use correct logger for score errors

* chore: inline error
2025-04-22 07:22:31 +00:00
Steffen SchmitzandGitHub 2d4708921c feat(dashboard): allow filtering by observation name (#6511) 2025-04-18 10:12:44 +00:00
marliessophieandGitHub 756b2f7501 chore(ui-forms): ensure name uniqueness enforced on form-level also (#6503)
* chore(ui-forms): ensure name uniqueness enforced on form-level also

* chore: nits
2025-04-17 14:13:01 +00:00
marliessophieandGitHub 2a685edbfa fix(tables): rename usage column to tokens column (#6498) 2025-04-17 11:58:12 +00:00
Steffen SchmitzandGitHub c1c22a9b9b fix: parse negative integer signs on otel timestamps (#6497) 2025-04-17 11:23:55 +00:00
218 changed files with 6148 additions and 2283 deletions
+2 -1
View File
@@ -21,6 +21,7 @@ on:
- staging
- prod-eu
- prod-us
- prod-hipaa
required: true
concurrency:
@@ -78,7 +79,7 @@ jobs:
return `["staging"]`
}
if (context.ref === "refs/heads/production") {
return `["prod-eu", "prod-us"]`
return `["prod-eu", "prod-us", "prod-hipaa"]`
}
}
return "[]"
+8 -2
View File
@@ -14,12 +14,13 @@ types:
CreateScoreRequest:
properties:
id: optional<string>
traceId: string
traceId: optional<string>
sessionId: optional<string>
observationId: optional<string>
name: string
value:
type: CreateScoreValue
docs: The value of the score. Must be passed as string for categorical scores, and numeric for boolean and numeric scores. Boolean score values must equal either 1 or 0 (true or false)
observationId: optional<string>
comment: optional<string>
metadata: optional<unknown>
dataType:
@@ -65,6 +66,10 @@ types:
dataType: "BOOLEAN"
configId: "1234-5678-90ab-cdef"
traceId: "cdef-1234-5678-90ab"
- value:
name: "contextrelevant"
value: "not relevant"
sessionId: "abyt-1234-5678-80ab"
BaseScore:
properties:
id: string
@@ -105,6 +110,7 @@ types:
stringValue:
type: string
docs: The string representation of the score value. If no config is linked, can be any string. Otherwise, must map to a config category
# Question: where is this used, if not needed remove BaseScore too
Score:
discriminant: "dataType"
union:
@@ -24,6 +24,9 @@ service:
body:
properties:
name: string
metadata:
type: optional<map<string, unknown>>
docs: Optional metadata for the organization
response: Organization
errors:
- commons.Error
@@ -53,6 +56,9 @@ service:
body:
properties:
name: string
metadata:
type: optional<map<string, unknown>>
docs: Optional metadata for the organization
response: Organization
errors:
- commons.Error
@@ -128,6 +134,9 @@ types:
id: string
name: string
createdAt: datetime
metadata:
type: map<string, unknown>
docs: Metadata for the organization
DeleteOrganizationResponse:
docs: Response for successful organization deletion
+62 -2
View File
@@ -75,7 +75,7 @@ types:
type: list<ObservationsView>
docs: List of observations
scores:
type: list<Score>
type: list<ScoreV1>
docs: List of scores
Session:
properties:
@@ -240,7 +240,7 @@ types:
properties:
value: double
label: string
BaseScore:
BaseScoreV1:
properties:
id: string
traceId: string
@@ -262,6 +262,66 @@ types:
environment:
type: optional<string>
docs: The environment from which this score originated. Can be any lowercase alphanumeric string with hyphens and underscores that does not start with 'langfuse'.
NumericScoreV1:
extends: BaseScoreV1
properties:
value:
type: double
docs: The numeric value of the score
BooleanScoreV1:
extends: BaseScoreV1
properties:
value:
type: double
docs: The numeric value of the score. Equals 1 for "True" and 0 for "False"
stringValue:
type: string
docs: The string representation of the score value. Is inferred from the numeric value and equals "True" or "False"
CategoricalScoreV1:
extends: BaseScoreV1
properties:
value:
type: optional<double>
docs: Only defined if a config is linked. Represents the numeric category mapping of the stringValue
stringValue:
type: string
docs: The string representation of the score value. If no config is linked, can be any string. Otherwise, must map to a config category
ScoreV1:
discriminant: "dataType"
union:
NUMERIC:
type: NumericScoreV1
docs: "Score with NUMERIC data type"
CATEGORICAL:
type: CategoricalScoreV1
docs: "Score with CATEGORICAL data type"
BOOLEAN:
type: BooleanScoreV1
docs: "Score with BOOLEAN data type"
BaseScore:
properties:
id: string
traceId: optional<string>
sessionId: optional<string>
observationId: optional<string>
name: string
source: ScoreSource
timestamp: datetime
createdAt: datetime
updatedAt: datetime
authorUserId: optional<string>
comment: optional<string>
metadata: optional<unknown>
configId:
type: optional<string>
docs: Reference a score config on a score. When set, config and score name must be equal and value must comply to optionally defined numerical range
queueId:
type: optional<string>
docs: Reference an annotation queue on a score. Populated if the score was initially created in an annotation queue.
environment:
type: optional<string>
docs: The environment from which this score originated. Can be any lowercase alphanumeric string with hyphens and underscores that does not start with 'langfuse'.
NumericScore:
extends: BaseScore
properties:
+7 -2
View File
@@ -263,13 +263,14 @@ types:
ScoreBody:
properties:
id: optional<string>
traceId: string
traceId: optional<string>
sessionId: optional<string>
observationId: optional<string>
name: string
environment: optional<string>
value:
type: commons.CreateScoreValue
docs: The value of the score. Must be passed as string for categorical scores, and numeric for boolean and numeric scores. Boolean score values must equal either 1 or 0 (true or false)
observationId: optional<string>
comment: optional<string>
metadata: optional<unknown>
dataType:
@@ -315,6 +316,10 @@ types:
dataType: "BOOLEAN"
configId: "1234-5678-90ab-cdef"
traceId: "cdef-1234-5678-90ab"
- value:
name: "contextrelevant"
value: "not relevant"
sessionId: "abyt-1234-5678-80ab"
BaseEvent:
properties:
+9
View File
@@ -21,6 +21,9 @@ service:
body:
properties:
name: string
metadata:
type: optional<map<string, unknown>>
docs: Optional metadata for the project
retention:
type: integer
docs: Number of days to retain data. Must be 0 or at least 7 days. Requires data-retention entitlement for non-zero values. Optional.
@@ -37,6 +40,9 @@ service:
body:
properties:
name: string
metadata:
type: optional<map<string, unknown>>
docs: Optional metadata for the project
retention:
type: integer
docs: Number of days to retain data. Must be 0 or at least 7 days. Requires data-retention entitlement for non-zero values. Optional.
@@ -94,6 +100,9 @@ types:
properties:
id: string
name: string
metadata:
type: map<string, unknown>
docs: Metadata for the project
retentionDays:
type: integer
docs: Number of days to retain data. Null or 0 means no retention. Omitted if no retention is configured.
+112
View File
@@ -0,0 +1,112 @@
# yaml-language-server: $schema=https://raw.githubusercontent.com/fern-api/fern/main/fern.schema.json
imports:
pagination: ./utils/pagination.yml
commons: ./commons.yml
service:
auth: true
base-path: /api/public/v2
endpoints:
get:
docs: Get a list of scores (supports both trace and session scores)
method: GET
path: /scores
request:
name: GetScoresRequest
query-parameters:
page:
type: optional<integer>
docs: Page number, starts at 1.
limit:
type: optional<integer>
docs: Limit of items per page. If you encounter api issues due to too large page sizes, try to reduce the limit.
userId:
type: optional<string>
docs: Retrieve only scores with this userId associated to the trace.
name:
type: optional<string>
docs: Retrieve only scores with this name.
fromTimestamp:
type: optional<datetime>
docs: Optional filter to only include scores created on or after a certain datetime (ISO 8601)
toTimestamp:
type: optional<datetime>
docs: Optional filter to only include scores created before a certain datetime (ISO 8601)
environment:
type: optional<string>
allow-multiple: true
docs: Optional filter for scores where the environment is one of the provided values.
source:
type: optional<commons.ScoreSource>
docs: Retrieve only scores from a specific source.
operator:
type: optional<string>
docs: Retrieve only scores with <operator> value.
value:
type: optional<double>
docs: Retrieve only scores with <operator> value.
scoreIds:
type: optional<string>
docs: Comma-separated list of score IDs to limit the results to.
configId:
type: optional<string>
docs: Retrieve only scores with a specific configId.
queueId:
type: optional<string>
docs: Retrieve only scores with a specific annotation queueId.
dataType:
type: optional<commons.ScoreDataType>
docs: Retrieve only scores with a specific dataType.
traceTags:
type: optional<string>
allow-multiple: true
docs: Only scores linked to traces that include all of these tags will be returned.
response: GetScoresResponse
get-by-id:
docs: Get a score (supports both trace and session scores)
method: GET
path: /scores/{scoreId}
path-parameters:
scoreId:
type: string
docs: The unique langfuse identifier of a score
response: commons.Score
types:
GetScoresResponseTraceData:
properties:
userId:
type: optional<string>
docs: The user ID associated with the trace referenced by score
tags:
type: optional<list<string>>
docs: A list of tags associated with the trace referenced by score
environment:
type: optional<string>
docs: The environment of the trace referenced by score
GetScoresResponseDataNumeric:
extends: commons.NumericScore
properties:
trace: optional<GetScoresResponseTraceData>
GetScoresResponseDataCategorical:
extends: commons.CategoricalScore
properties:
trace: optional<GetScoresResponseTraceData>
GetScoresResponseDataBoolean:
extends: commons.BooleanScore
properties:
trace: optional<GetScoresResponseTraceData>
GetScoresResponseData:
discriminant: dataType
union:
NUMERIC: GetScoresResponseDataNumeric
CATEGORICAL: GetScoresResponseDataCategorical
BOOLEAN: GetScoresResponseDataBoolean
GetScoresResponse:
properties:
data: list<GetScoresResponseData>
meta: pagination.MetaResponse
+5 -106
View File
@@ -7,77 +7,13 @@ service:
base-path: /api/public
endpoints:
create:
docs: Create a score
docs: Create a score (supports both trace and session scores)
method: POST
path: /scores
request: CreateScoreRequest
response: CreateScoreResponse
get:
docs: Get a list of scores
method: GET
path: /scores
request:
name: GetScoresRequest
query-parameters:
page:
type: optional<integer>
docs: Page number, starts at 1.
limit:
type: optional<integer>
docs: Limit of items per page. If you encounter api issues due to too large page sizes, try to reduce the limit.
userId:
type: optional<string>
docs: Retrieve only scores with this userId associated to the trace.
name:
type: optional<string>
docs: Retrieve only scores with this name.
fromTimestamp:
type: optional<datetime>
docs: Optional filter to only include scores created on or after a certain datetime (ISO 8601)
toTimestamp:
type: optional<datetime>
docs: Optional filter to only include scores created before a certain datetime (ISO 8601)
environment:
type: optional<string>
allow-multiple: true
docs: Optional filter for scores where the environment is one of the provided values.
source:
type: optional<commons.ScoreSource>
docs: Retrieve only scores from a specific source.
operator:
type: optional<string>
docs: Retrieve only scores with <operator> value.
value:
type: optional<double>
docs: Retrieve only scores with <operator> value.
scoreIds:
type: optional<string>
docs: Comma-separated list of score IDs to limit the results to.
configId:
type: optional<string>
docs: Retrieve only scores with a specific configId.
queueId:
type: optional<string>
docs: Retrieve only scores with a specific annotation queueId.
dataType:
type: optional<commons.ScoreDataType>
docs: Retrieve only scores with a specific dataType.
traceTags:
type: optional<string>
allow-multiple: true
docs: Only scores linked to traces that include all of these tags will be returned.
response: GetScoresResponse
get-by-id:
docs: Get a score
method: GET
path: /scores/{scoreId}
path-parameters:
scoreId:
type: string
docs: The unique langfuse identifier of a score
response: commons.Score
delete:
docs: Delete a score
docs: Delete a score (supports both trace and session scores)
method: DELETE
path: /scores/{scoreId}
path-parameters:
@@ -88,12 +24,13 @@ types:
CreateScoreRequest:
properties:
id: optional<string>
traceId: string
traceId: optional<string>
sessionId: optional<string>
observationId: optional<string>
name: string
value:
type: commons.CreateScoreValue
docs: The value of the score. Must be passed as string for categorical scores, and numeric for boolean and numeric scores. Boolean score values must equal either 1 or 0 (true or false)
observationId: optional<string>
comment: optional<string>
metadata: optional<unknown>
environment:
@@ -149,41 +86,3 @@ types:
id:
type: string
docs: The id of the created object in Langfuse
GetScoresResponseTraceData:
properties:
userId:
type: optional<string>
docs: The user ID associated with the trace referenced by score
tags:
type: optional<list<string>>
docs: A list of tags associated with the trace referenced by score
environment:
type: optional<string>
docs: The environment of the trace referenced by score
GetScoresResponseDataNumeric:
extends: commons.NumericScore
properties:
trace: GetScoresResponseTraceData
GetScoresResponseDataCategorical:
extends: commons.CategoricalScore
properties:
trace: GetScoresResponseTraceData
GetScoresResponseDataBoolean:
extends: commons.BooleanScore
properties:
trace: GetScoresResponseTraceData
GetScoresResponseData:
discriminant: dataType
union:
NUMERIC: GetScoresResponseDataNumeric
CATEGORICAL: GetScoresResponseDataCategorical
BOOLEAN: GetScoresResponseDataBoolean
GetScoresResponse:
properties:
data: list<GetScoresResponseData>
meta: pagination.MetaResponse
+3 -1
View File
@@ -1,6 +1,6 @@
{
"name": "langfuse",
"version": "3.51.2",
"version": "3.53.0",
"author": "engineering@langfuse.com",
"license": "MIT",
"private": true,
@@ -23,6 +23,8 @@
"build": "turbo run build",
"start": "turbo run start",
"dev": "turbo run dev",
"dev:worker": "turbo run dev --filter=worker",
"dev:web": "turbo run dev --filter=web",
"lint": "turbo run lint",
"test": "turbo run test",
"release": "dotenv -e ../.env -- release-it",
@@ -0,0 +1 @@
DROP TABLE blob_storage_file_log ON CLUSTER default;
@@ -0,0 +1,21 @@
CREATE TABLE blob_storage_file_log ON CLUSTER default (
`id` String,
`project_id` String,
`entity_type` String,
`entity_id` String,
`event_id` String,
`bucket_name` String,
`bucket_path` String,
`created_at` DateTime64(3) DEFAULT now(),
`updated_at` DateTime64(3) DEFAULT now(),
event_ts DateTime64(3),
is_deleted UInt8,
) ENGINE = ReplicatedReplacingMergeTree(event_ts, is_deleted)
ORDER BY (
project_id,
entity_type,
entity_id,
event_id
);
@@ -0,0 +1 @@
ALTER TABLE scores ON CLUSTER default DROP COLUMN IF EXISTS session_id SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores ON CLUSTER default ADD COLUMN session_id Nullable(String) AFTER trace_id SETTINGS mutations_sync = 2;
@@ -0,0 +1,2 @@
ALTER TABLE scores ON CLUSTER default ADD INDEX IF NOT EXISTS idx_project_trace_observation (project_id, trace_id, observation_id) TYPE bloom_filter(0.001) GRANULARITY 1 SETTINGS mutations_sync = 2;
ALTER TABLE scores ON CLUSTER default MATERIALIZE INDEX IF EXISTS idx_project_trace_observation SETTINGS mutations_sync = 2;
@@ -0,0 +1,2 @@
ALTER TABLE scores ON CLUSTER default DROP INDEX IF EXISTS idx_project_trace_observation SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores ON CLUSTER default MODIFY COLUMN trace_id Nullable(String) SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores ON CLUSTER default MODIFY COLUMN trace_id Nullable(String) SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores ON CLUSTER default DROP INDEX IF EXISTS idx_project_trace_observation SETTINGS mutations_sync = 2;
@@ -0,0 +1,2 @@
ALTER TABLE scores ON CLUSTER default ADD INDEX IF NOT EXISTS idx_project_trace_observation (project_id, trace_id, observation_id) TYPE bloom_filter(0.001) GRANULARITY 1 SETTINGS mutations_sync = 2;
ALTER TABLE scores ON CLUSTER default MATERIALIZE INDEX IF EXISTS idx_project_trace_observation SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores ON CLUSTER default DROP INDEX IF EXISTS idx_project_session SETTINGS mutations_sync = 2;
@@ -0,0 +1,2 @@
ALTER TABLE scores ON CLUSTER default ADD INDEX IF NOT EXISTS idx_project_session (project_id, session_id) TYPE bloom_filter(0.001) GRANULARITY 1 SETTINGS mutations_sync = 2;
ALTER TABLE scores ON CLUSTER default MATERIALIZE INDEX IF EXISTS idx_project_session SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
DROP TABLE blob_storage_file_log;
@@ -0,0 +1,22 @@
CREATE TABLE blob_storage_file_log
(
`id` String,
`project_id` String,
`entity_type` String,
`entity_id` String,
`event_id` String,
`bucket_name` String,
`bucket_path` String,
`created_at` DateTime64(3) DEFAULT now(),
`updated_at` DateTime64(3) DEFAULT now(),
event_ts DateTime64(3),
is_deleted UInt8,
) ENGINE = ReplacingMergeTree(event_ts, is_deleted)
ORDER BY (
project_id,
entity_type,
entity_id,
event_id
);
@@ -0,0 +1 @@
ALTER TABLE scores DROP COLUMN IF EXISTS session_id SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores ADD COLUMN session_id Nullable(String) AFTER trace_id SETTINGS mutations_sync = 2;
@@ -0,0 +1,2 @@
ALTER TABLE scores ADD INDEX IF NOT EXISTS idx_project_trace_observation (project_id, trace_id, observation_id) TYPE bloom_filter(0.001) GRANULARITY 1 SETTINGS mutations_sync = 2;
ALTER TABLE scores MATERIALIZE INDEX IF EXISTS idx_project_trace_observation SETTINGS mutations_sync = 2;
@@ -0,0 +1,2 @@
ALTER TABLE scores DROP INDEX IF EXISTS idx_project_trace_observation SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores MODIFY COLUMN trace_id Nullable(String) SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores MODIFY COLUMN trace_id Nullable(String) SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores DROP INDEX IF EXISTS idx_project_trace_observation SETTINGS mutations_sync = 2;
@@ -0,0 +1,2 @@
ALTER TABLE scores ADD INDEX IF NOT EXISTS idx_project_trace_observation (project_id, trace_id, observation_id) TYPE bloom_filter(0.001) GRANULARITY 1 SETTINGS mutations_sync = 2;
ALTER TABLE scores MATERIALIZE INDEX IF EXISTS idx_project_trace_observation SETTINGS mutations_sync = 2;
@@ -0,0 +1 @@
ALTER TABLE scores DROP INDEX IF EXISTS idx_project_session SETTINGS mutations_sync = 2;
@@ -0,0 +1,2 @@
ALTER TABLE scores ADD INDEX IF NOT EXISTS idx_project_session (project_id, session_id) TYPE bloom_filter(0.001) GRANULARITY 1 SETTINGS mutations_sync = 2;
ALTER TABLE scores MATERIALIZE INDEX IF EXISTS idx_project_session SETTINGS mutations_sync = 2;
@@ -531,6 +531,7 @@ export type Organization = {
created_at: Generated<Timestamp>;
updated_at: Generated<Timestamp>;
cloud_config: unknown | null;
metadata: unknown | null;
};
export type OrganizationMembership = {
id: string;
@@ -564,6 +565,7 @@ export type Project = {
deleted_at: Timestamp | null;
name: string;
retention_days: number | null;
metadata: unknown | null;
};
export type ProjectMembership = {
org_membership_id: string;
@@ -0,0 +1,5 @@
-- AlterTable
ALTER TABLE "organizations" ADD COLUMN "metadata" JSONB;
-- AlterTable
ALTER TABLE "projects" ADD COLUMN "metadata" JSONB;
@@ -0,0 +1,2 @@
INSERT INTO background_migrations (id, name, script, args)
VALUES ('c19b91d9-f9a2-468b-8209-95578f970c5b', '20250417_1737_migrate_event_log_to_blob_storage', 'migrateEventLogToBlobStorageRefTable', '{}');
+2
View File
@@ -105,6 +105,7 @@ model Organization {
createdAt DateTime @default(now()) @map("created_at")
updatedAt DateTime @default(now()) @updatedAt @map("updated_at")
cloudConfig Json? @map("cloud_config") // Langfuse Cloud, for zod schema see @/src/features/organizations/utils/cloudConfigSchema
metadata Json?
organizationMemberships OrganizationMembership[]
projects Project[]
MembershipInvitation MembershipInvitation[]
@@ -121,6 +122,7 @@ model Project {
deletedAt DateTime? @map("deleted_at")
name String
retentionDays Int? @map("retention_days")
metadata Json?
projectMembers ProjectMembership[]
organization Organization @relation(fields: [orgId], references: [id], onUpdate: Cascade, onDelete: Cascade)
apiKeys ApiKey[]
+42 -2
View File
@@ -141,8 +141,6 @@ export const prepareClickhouse = async (
FROM numbers(${observationsPerProject});
`;
console.log(observationsQuery);
const scoresQuery = `
INSERT INTO scores
SELECT toString(number) AS id,
@@ -150,6 +148,7 @@ export const prepareClickhouse = async (
'${projectId}' AS project_id,
'default' AS environment,
toString(floor(randUniform(0, ${tracesPerProject}))) AS trace_id,
NULL AS session_id,
if(
rand() > 0.9,
toString(floor(randUniform(0, ${observationsPerProject}))),
@@ -200,6 +199,47 @@ export const prepareClickhouse = async (
project_id: string;
}>();
const sessionsToScore = sessionData
.filter(() => Math.random() < 0.5)
.slice(0, Math.min(500, sessionData.length));
if (sessionsToScore.length > 0) {
// Generate session scores query with specific session IDs
const sessionScoresQuery = `
INSERT INTO scores
SELECT
concat('session-', toString(number)) AS id,
toDateTime(now() - randUniform(0, ${opts.numberOfDays} * 24 * 60 * 60)) AS timestamp,
'${projectId}' AS project_id,
'default' AS environment,
NULL AS trace_id,
arrayElement(['${sessionsToScore.map((s) => s.session_id).join("','")}'], 1 + (number % ${sessionsToScore.length})) AS session_id,
NULL AS observation_id,
concat('session_quality_', toString(rand() % 10)) AS name,
randUniform(0, 100) AS value,
'API' AS source,
'Session-level assessment score' AS comment,
map('key', 'value') AS metadata,
toString(rand() % 100) AS author_user_id,
toString(rand() % 100) AS config_id,
if(rand() < 0.33, 'NUMERIC', if(rand() < 0.5, 'CATEGORICAL', 'BOOLEAN')) AS data_type,
toString(rand() % 100) AS string_value,
NULL AS queue_id,
timestamp AS created_at,
timestamp AS updated_at,
timestamp AS event_ts,
0 AS is_deleted
FROM numbers(${sessionsToScore.length})
`;
await clickhouseClient().command({
query: sessionScoresQuery,
clickhouse_settings: {
wait_end_of_query: 1,
},
});
}
const idProjectIdCombinations = sessionData.map((session) => ({
id: session.session_id,
projectId: session.project_id,
+2 -1
View File
@@ -21,8 +21,9 @@ export const ScoreSchema = z.object({
source: ScoreSourceDomain,
authorUserId: z.string().nullable(),
comment: z.string().nullable(),
traceId: z.string().nullable(),
sessionId: z.string().nullable(),
metadata: MetadataDomain,
traceId: z.string(),
observationId: z.string().nullable(),
configId: z.string().nullable(),
stringValue: z.string().nullable(),
+12
View File
@@ -63,6 +63,15 @@ const EnvSchema = z.object({
LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: z
.enum(["true", "false"])
.default("false"),
LANGFUSE_S3_MEDIA_UPLOAD_BUCKET: z.string().optional(),
LANGFUSE_S3_MEDIA_UPLOAD_PREFIX: z.string().default(""),
LANGFUSE_S3_MEDIA_UPLOAD_REGION: z.string().optional(),
LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: z.string().optional(),
LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID: z.string().optional(),
LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY: z.string().optional(),
LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE: z
.enum(["true", "false"])
.default("false"),
LANGFUSE_USE_AZURE_BLOB: z.enum(["true", "false"]).default("false"),
LANGFUSE_USE_GOOGLE_CLOUD_STORAGE: z.enum(["true", "false"]).default("false"),
LANGFUSE_GOOGLE_CLOUD_STORAGE_CREDENTIALS: z.string().optional(),
@@ -78,6 +87,9 @@ const EnvSchema = z.object({
LANGFUSE_CUSTOM_SSO_EMAIL_CLAIM: z.string().default("email"),
LANGFUSE_CUSTOM_SSO_NAME_CLAIM: z.string().default("name"),
LANGFUSE_CUSTOM_SSO_SUB_CLAIM: z.string().default("sub"),
LANGFUSE_API_TRACE_OBSERVATIONS_SIZE_LIMIT_BYTES: z.coerce
.number()
.default(10e6), // 10MB
});
export const env: z.infer<typeof EnvSchema> =
@@ -25,13 +25,33 @@ const BooleanData = z.object({
dataType: z.literal("BOOLEAN"),
});
const ScoreTargetTrace = z.object({
type: z.literal("trace"),
traceId: z.string(),
observationId: z.string().optional(),
});
const ScoreTargetSession = z.object({
type: z.literal("session"),
sessionId: z.string(),
});
// Your existing ScoreTarget remains the same, but can now use these components
const ScoreTarget = z.discriminatedUnion("type", [
ScoreTargetTrace,
ScoreTargetSession,
]);
export type ScoreTargetTrace = z.infer<typeof ScoreTargetTrace>;
export type ScoreTargetSession = z.infer<typeof ScoreTargetSession>;
export type ScoreTarget = z.infer<typeof ScoreTarget>;
const CreateAnnotationScoreBase = z.object({
name: z.string(),
projectId: z.string(),
environment: z.string().default("default"),
traceId: z.string(),
scoreTarget: ScoreTarget,
configId: z.string().optional(),
observationId: z.string().optional(),
comment: z.string().nullish(),
queueId: z.string().nullish(),
});
+1 -2
View File
@@ -1,3 +1,2 @@
export * from "./scoreTypes";
export * from "./types";
export * from "./interfaces";
export * from "./scoreConfigTypes";
@@ -0,0 +1,68 @@
# Score Interfaces
This directory contains all type definitions, schemas, and validation logic for Langfuse scores.
## Structure
```
interfaces/
├── api/ # API-specific schemas and validations
│ ├── v1/ # Legacy API types (trace-focused)
│ │ ├── endpoints.ts # Endpoint types and schemas
│ │ ├── schemas.ts # Type definitions
│ │ └── validation.ts # Validation logic
│ ├── v2/ # Current API types (supports traces, sessions)
│ │ ├── endpoints.ts # Endpoint types and schemas
│ │ ├── schemas.ts # Type definitions
│ │ └── validation.ts # Validation logic
│ └── shared.ts # Common schemas used across API versions
├── application/ # Internal application logic
│ └── validation.ts # Validation functions for application layers
├── ingestion/ # Types for data ingestion
│ └── validation.ts # Validation for ingestion endpoints
├── ui/ # Simplified types for UI components
│ └── types.ts # UI-specific type definitions
└── index.ts # Exports all interfaces
```
## API Versioning
We have added a new v2 api and will continue to support the v1 api for the foreseeable future.
POST and DELETE APIs will continue to support both trace and session level scores across v1 and v2.
For GET APIs:
- **V1 API**: Requires `traceId`, ONLY supports trace-level scores
- **V2 API**: Makes `traceId` optional, adds `sessionId` support, supports BOTH trace and session level scores
- Both versions maintain compatibility for existing clients
## Usage Guide
### When building API endpoints:
- **New endpoints**: Use types from `api/v2/schemas.ts`
- **Legacy compatibility**: Use types from `api/v1/schemas.ts`
- **Validation**: Use validators from the corresponding `validation.ts` files
### When building UI components:
Use the simplified types from `ui/types.ts` which are optimized for frontend use:
```typescript
import { ScoreSimplified, LastUserScore } from "../interfaces/ui/types";
```
### When working with internal application logic:
Use the validation functions from `application/validation.ts`:
```typescript
import {
validateDbScore,
filterAndValidateDbScoreList,
} from "../interfaces/application/validation";
```
## Type Flow
Client → `PostScoresBody` → Validation → Database → `ScoreDomain` → API response (`APIScoreV2`) → UI (`ScoreSimplified` or `LastUserScore`)
@@ -0,0 +1,6 @@
export * from "./v1/schemas";
export * from "./v1/validation";
export * from "./v1/endpoints";
export * from "./v2/schemas";
export * from "./v2/validation";
export * from "./v2/endpoints";
@@ -0,0 +1,139 @@
import { z } from "zod";
import { jsonSchema, publicApiPaginationZod } from "../../../../utils/zod";
import { stringDateTime } from "../../../../utils/typeChecks";
import { applyScoreValidation } from "../../../../utils/scores";
import { PostScoreBodyFoundationSchema } from "../shared";
const operators = ["<", ">", "<=", ">=", "!=", "="] as const;
const ScoreDataType = ["NUMERIC", "CATEGORICAL", "BOOLEAN"] as const;
const ScoreSource = ["API", "EVAL", "ANNOTATION"] as const;
/**
* Objects
*/
export const NumericData = z.object({
value: z.number(),
stringValue: z.undefined().nullish(),
dataType: z.literal("NUMERIC"),
});
export const CategoricalData = z.object({
value: z.number().nullish(),
stringValue: z.string(),
dataType: z.literal("CATEGORICAL"),
});
export const BooleanData = z.object({
value: z.number(),
stringValue: z.string(),
dataType: z.literal("BOOLEAN"),
});
/**
* Foundation schema for all score types, needs to be extended with entity score may be associated with. Note there are two API versions, where v1 allows only trace and observation scores, while v2 additionally allows session scores
* @see {@link ScoreFoundationSchemaV1}, {@link ScoreFoundationSchemaV2}
*
* Must also be extended with score data specific schema (numeric, categorical, boolean)
* @see {@link NumericData}, {@link CategoricalData}, {@link BooleanData}
*/
export const ScoreFoundationSchema = z.object({
id: z.string(),
timestamp: z.coerce.date(),
projectId: z.string(),
environment: z.string().default("default"),
name: z.string(),
source: z.enum(ScoreSource),
authorUserId: z.string().nullish(),
comment: z.string().nullish(),
metadata: jsonSchema.nullish(),
configId: z.string().nullish(),
createdAt: z.coerce.date(),
updatedAt: z.coerce.date(),
queueId: z.string().nullish(),
});
/**
* Endpoints
*/
// GET /scores/{scoreId}
export const GetScoreQuery = z.object({
scoreId: z.string(),
});
// GET /scores
export const GetScoresQuery = z.object({
...publicApiPaginationZod,
userId: z.string().nullish(),
dataType: z.enum(ScoreDataType).nullish(),
configId: z.string().nullish(),
queueId: z.string().nullish(),
traceTags: z.union([z.array(z.string()), z.string()]).nullish(),
environment: z.union([z.array(z.string()), z.string()]).nullish(),
name: z.string().nullish(),
fromTimestamp: stringDateTime,
toTimestamp: stringDateTime,
source: z.enum(ScoreSource).nullish(),
value: z.coerce.number().nullish(),
operator: z.enum(operators).nullish(),
scoreIds: z
.string()
.transform((str) => str.split(",").map((id) => id.trim())) // Split the comma-separated string
.refine((arr) => arr.every((id) => typeof id === "string"), {
message: "Each score ID must be a string",
})
.nullish(),
});
// POST /scores
// Please note that the POST /scores endpoint supports both trace and session level scores across v1 and v2.
/**
* PostScoresBody is copied for the ingestion API as `ScoreBody`. Please copy any changes here in `packages/shared/src/features/ingestion/types.ts`
*/
export const PostScoresBody = applyScoreValidation(
z.discriminatedUnion("dataType", [
PostScoreBodyFoundationSchema.merge(
z.object({
value: z.number(),
dataType: z.literal("NUMERIC"),
configId: z.string().nullish(),
}),
),
PostScoreBodyFoundationSchema.merge(
z.object({
value: z.string(),
dataType: z.literal("CATEGORICAL"),
configId: z.string().nullish(),
}),
),
PostScoreBodyFoundationSchema.merge(
z.object({
value: z.number().refine((value) => value === 0 || value === 1, {
message:
"Value must be a number equal to either 0 or 1 for data type BOOLEAN",
}),
dataType: z.literal("BOOLEAN"),
configId: z.string().nullish(),
}),
),
PostScoreBodyFoundationSchema.merge(
z.object({
value: z.union([z.string(), z.number()]),
dataType: z.undefined(),
configId: z.string().nullish(),
}),
),
]),
);
export const PostScoresResponse = z.object({ id: z.string() });
// DELETE /scores/{scoreId}
// Please note that the DELETE /scores/{scoreId} endpoint supports both trace and session level scores across v1 and v2.
export const DeleteScoreQuery = z.object({
scoreId: z.string(),
});
export const DeleteScoreResponse = z.object({
message: z.string(),
});
@@ -0,0 +1,44 @@
import { z } from "zod";
import { paginationMetaResponseZod } from "../../../../../utils/zod";
import {
DeleteScoreQuery,
DeleteScoreResponse,
GetScoreQuery,
GetScoresQuery,
PostScoresBody,
PostScoresResponse,
} from "../shared";
import { APIScoreSchemaV1 } from "./schemas";
// GET /scores/{scoreId}
export const GetScoreQueryV1 = GetScoreQuery;
export const GetScoreResponseV1 = APIScoreSchemaV1;
// DELETE /scores/{scoreId}
export const DeleteScoreQueryV1 = DeleteScoreQuery;
export const DeleteScoreResponseV1 = DeleteScoreResponse;
// GET /scores
export const GetScoresQueryV1 = GetScoresQuery;
// GetScoreResponseDataV1 is only used for response of GET /scores list endpoint
export const GetScoreResponseDataV1 = z.intersection(
APIScoreSchemaV1,
z.object({
trace: z.object({
userId: z.string().nullish(),
tags: z.array(z.string()).nullish(),
environment: z.string().nullish(),
}),
}),
);
export const GetScoresResponseV1 = z.object({
data: z.array(GetScoreResponseDataV1),
meta: paginationMetaResponseZod,
});
// POST /scores
export const PostScoresBodyV1 = PostScoresBody;
export const PostScoresResponseV1 = PostScoresResponse;
@@ -0,0 +1,26 @@
import { z } from "zod";
import {
CategoricalData,
NumericData,
BooleanData,
ScoreFoundationSchema,
} from "../shared";
/**
* Foundation schema for scores API v1 i.e. trace and observation scores ONLY
*
* Must be extended with score data specific schema (numeric, categorical, boolean)
* @see {@link NumericData}, {@link CategoricalData}, {@link BooleanData}
*/
const ScoreFoundationSchemaV1 = ScoreFoundationSchema.extend({
traceId: z.string(),
observationId: z.string().nullish(),
});
export const APIScoreSchemaV1 = z.discriminatedUnion("dataType", [
ScoreFoundationSchemaV1.merge(NumericData),
ScoreFoundationSchemaV1.merge(CategoricalData),
ScoreFoundationSchemaV1.merge(BooleanData),
]);
export type APIScoreV1 = z.infer<typeof APIScoreSchemaV1>;
@@ -0,0 +1,29 @@
import z from "zod";
import { GetScoreResponseDataV1 } from "./endpoints";
/**
* @deprecated
* Use `filterAndValidateV2GetScoreList` instead. This function is only used for the legacy v1 API where scores were only associated with traces.
* Use this function when pulling a list of scores from the database before returning to the public API to ensure type safety.
* All scores are expected to pass the validation. If a score fails validation, it will be logged to Otel.
* @param scores
* @returns list of validated scores with trace information
*/
export const filterAndValidateV1GetScoreList = (
scores: unknown[],
// eslint-disable-next-line no-unused-vars
onParseError?: (error: z.ZodError) => void,
): z.infer<typeof GetScoreResponseDataV1>[] =>
scores.reduce(
(acc: z.infer<typeof GetScoreResponseDataV1>[], ts) => {
const result = GetScoreResponseDataV1.safeParse(ts);
if (result.success) {
acc.push(result.data);
} else {
console.error(`Score parsing error ${JSON.stringify(result.error)}`);
onParseError?.(result.error);
}
return acc;
},
[] as z.infer<typeof GetScoreResponseDataV1>[],
);
@@ -0,0 +1,28 @@
import z from "zod";
import { paginationMetaResponseZod } from "../../../../../utils/zod";
import { GetScoreQuery, GetScoresQuery } from "../shared";
import { APIScoreSchemaV2 } from "./schemas";
// GET /scores/{scoreId} v2
export const GetScoreQueryV2 = GetScoreQuery;
export const GetScoreResponseV2 = APIScoreSchemaV2;
// GET /scores v2
export const GetScoresQueryV2 = GetScoresQuery;
export const GetScoreResponseDataV2 = z.intersection(
APIScoreSchemaV2,
z.object({
trace: z
.object({
userId: z.string().nullish(),
tags: z.array(z.string()).nullish(),
environment: z.string().nullish(),
})
.nullish(),
}),
);
export const GetScoresResponseV2 = z.object({
data: z.array(GetScoreResponseDataV2),
meta: paginationMetaResponseZod,
});
@@ -0,0 +1,27 @@
import { z } from "zod";
import {
CategoricalData,
NumericData,
BooleanData,
ScoreFoundationSchema,
} from "../shared";
/**
* Foundation schema for scores API v2 i.e. trace, observation AND session scores
*
* Must be extended with score data specific schema (numeric, categorical, boolean)
* @see {@link NumericData}, {@link CategoricalData}, {@link BooleanData}
*/
const ScoreFoundationSchemaV2 = ScoreFoundationSchema.extend({
traceId: z.string().nullish(),
observationId: z.string().nullish(),
sessionId: z.string().nullish(),
});
export const APIScoreSchemaV2 = z.discriminatedUnion("dataType", [
ScoreFoundationSchemaV2.merge(NumericData),
ScoreFoundationSchemaV2.merge(CategoricalData),
ScoreFoundationSchemaV2.merge(BooleanData),
]);
export type APIScoreV2 = z.infer<typeof APIScoreSchemaV2>;
@@ -0,0 +1,27 @@
import z from "zod";
import { GetScoreResponseDataV2 } from "./endpoints";
/**
* Use this function when pulling a list of scores from the database before returning to the public API to ensure type safety.
* All scores are expected to pass the validation. If a score fails validation, it will be logged to Otel.
* @param scores
* @returns list of validated scores with optional trace information in case of trace scores
*/
export const filterAndValidateV2GetScoreList = (
scores: unknown[],
// eslint-disable-next-line no-unused-vars
onParseError?: (error: z.ZodError) => void,
): z.infer<typeof GetScoreResponseDataV2>[] =>
scores.reduce(
(acc: z.infer<typeof GetScoreResponseDataV2>[], ts) => {
const result = GetScoreResponseDataV2.safeParse(ts);
if (result.success) {
acc.push(result.data);
} else {
console.error("Score parsing error: ", result.error);
onParseError?.(result.error);
}
return acc;
},
[] as z.infer<typeof GetScoreResponseDataV2>[],
);
@@ -0,0 +1,94 @@
import z from "zod";
import { APIScoreSchemaV2, APIScoreV2 } from "../api/v2/schemas";
import { APIScoreSchemaV1, APIScoreV1 } from "../api/v1/schemas";
import { ScoreDomain } from "../../../../domain";
type ValidatedAPIScore<IncludeHasMetadata extends boolean> = APIScoreV2 & {
hasMetadata: IncludeHasMetadata extends true ? boolean : never;
};
type InputScore = ScoreDomain & { hasMetadata?: boolean };
/**
* Use this function when pulling a single score from the database before using in the application to ensure type safety.
* The score is expected to pass the validation. If a score fails validation, an error will be thrown.
* @param score
* @returns validated score
* @throws error if score fails validation
*/
export const validateDbScore = (score: ScoreDomain): APIScoreV2 =>
APIScoreSchemaV2.parse(score);
/**
* Use this function when pulling a list of scores from the database before using in the application to ensure type safety.
* All scores are expected to pass the validation. If a score fails validation, it will be logged to Otel.
* @param scores
* @returns list of validated scores
*/
export const filterAndValidateDbScoreList = <
IncludeHasMetadata extends boolean,
>({
scores,
includeHasMetadata = false as IncludeHasMetadata,
onParseError,
}: {
scores: InputScore[];
includeHasMetadata?: IncludeHasMetadata;
// eslint-disable-next-line no-unused-vars
onParseError?: (error: z.ZodError) => void;
}): ValidatedAPIScore<IncludeHasMetadata>[] => {
return scores.reduce((acc, ts) => {
const result = APIScoreSchemaV2.safeParse(ts);
if (result.success) {
const score = { ...result.data };
if (includeHasMetadata) {
Object.assign(score, { hasMetadata: ts.hasMetadata ?? false });
}
acc.push(score as ValidatedAPIScore<IncludeHasMetadata>);
} else {
console.error("Score parsing error: ", result.error);
onParseError?.(result.error);
}
return acc;
}, [] as ValidatedAPIScore<IncludeHasMetadata>[]);
};
type ValidatedAPITraceScore<IncludeHasMetadata extends boolean> = APIScoreV1 & {
hasMetadata: IncludeHasMetadata extends true ? boolean : never;
};
/**
* @deprecated
* Use `filterAndValidateDbScoreList` instead. This function is only used for the legacy v1 API where scores were only associated with traces.
* Use this function when pulling a list of scores from the database before using in the application to ensure type safety.
* All scores are expected to pass the validation. If a score fails validation, it will be logged to Otel.
* @param scores
* @returns list of validated scores
*/
export const filterAndValidateDbTraceScoreList = <
IncludeHasMetadata extends boolean,
>({
scores,
includeHasMetadata = false as IncludeHasMetadata,
onParseError,
}: {
scores: InputScore[];
includeHasMetadata?: IncludeHasMetadata;
// eslint-disable-next-line no-unused-vars
onParseError?: (error: z.ZodError) => void;
}): ValidatedAPITraceScore<IncludeHasMetadata>[] => {
return scores.reduce((acc, ts) => {
const result = APIScoreSchemaV1.safeParse(ts);
if (result.success) {
const score = { ...result.data };
if (includeHasMetadata) {
Object.assign(score, { hasMetadata: ts.hasMetadata ?? false });
}
acc.push(score as ValidatedAPITraceScore<IncludeHasMetadata>);
} else {
console.error("Score parsing error: ", result.error);
onParseError?.(result.error);
}
return acc;
}, [] as ValidatedAPITraceScore<IncludeHasMetadata>[]);
};
@@ -0,0 +1,4 @@
export * from "./api";
export * from "./application/validation";
export * from "./ingestion/validation";
export * from "./ui/types";
@@ -0,0 +1,78 @@
import z from "zod";
import { applyScoreValidation } from "../../../../utils/scores";
import { PostScoreBodyFoundationSchema } from "../shared";
import { isPresent } from "../../../../utils/typeChecks";
import { Category as ConfigCategory } from "../../scoreConfigTypes";
export const ScoreBodyWithoutConfig = applyScoreValidation(
z.discriminatedUnion("dataType", [
PostScoreBodyFoundationSchema.merge(
z.object({
value: z.number(),
dataType: z.literal("NUMERIC"),
}),
),
PostScoreBodyFoundationSchema.merge(
z.object({
value: z.string(),
dataType: z.literal("CATEGORICAL"),
}),
),
PostScoreBodyFoundationSchema.merge(
z.object({
value: z.number().refine((val) => val === 0 || val === 1, {
message: "Value must be either 0 or 1",
}),
dataType: z.literal("BOOLEAN"),
}),
),
]),
);
const ScorePropsAgainstConfigNumeric = z
.object({
value: z.number(),
maxValue: z.number().optional(),
minValue: z.number().optional(),
dataType: z.literal("NUMERIC"),
})
.superRefine((data, ctx) => {
if (isPresent(data.maxValue) && data.value >= data.maxValue) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: `Value exceeds maximum value of ${data.maxValue} defined in config`,
});
}
if (isPresent(data.minValue) && data.value <= data.minValue) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: `Value is below minimum value of ${data.minValue} defined in config`,
});
}
});
const ScorePropsAgainstConfigCategorical = z
.object({
value: z.string(),
categories: z.array(ConfigCategory),
dataType: z.literal("CATEGORICAL"),
})
.superRefine((data, ctx) => {
if (!data.categories.some(({ label }) => label === data.value)) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: `Value ${data.value} does not map to a valid category. Pass a valid category value.`,
});
}
});
export const ScorePropsAgainstConfig = z.union([
ScorePropsAgainstConfigNumeric,
ScorePropsAgainstConfigCategorical,
z.object({
value: z.number().refine((val) => val === 0 || val === 1, {
message: "Value must be either 0 or 1",
}),
dataType: z.literal("BOOLEAN"),
}),
]);
@@ -0,0 +1,19 @@
import { jsonSchema } from "../../../utils/zod";
import z from "zod";
import { NonEmptyString } from "../../../utils/zod";
/**
* Foundation schema for all score types. Used for ingestion and public API. Supports trace, observation and session scores.
* Needs to be extended with with score data specific schema (numeric, categorical, boolean)
* @see {@link NumericData}, {@link CategoricalData}, {@link BooleanData}
*/
export const PostScoreBodyFoundationSchema = z.object({
id: z.string().nullish(),
name: NonEmptyString,
traceId: z.string().nullish(),
sessionId: z.string().nullish(),
observationId: z.string().nullish(),
comment: z.string().nullish(),
metadata: jsonSchema.nullish(),
environment: z.string().default("default"),
});
@@ -1,5 +1,5 @@
import { ScoreDataType } from "@prisma/client";
import { ScoreSourceType, type MetadataDomain } from "../../domain";
import { MetadataDomain, ScoreSourceType } from "../../../../domain";
export type BaseAggregate = {
comment?: string | null;
@@ -1,328 +0,0 @@
import { z } from "zod";
import { isPresent, stringDateTime } from "../../utils/typeChecks";
import {
jsonSchema,
NonEmptyString,
paginationMetaResponseZod,
publicApiPaginationZod,
} from "../../utils/zod";
import { Category as ConfigCategory } from "./scoreConfigTypes";
import { ScoreDomain } from "../../domain";
/**
* Types to use across codebase
*/
export type APIScore = z.infer<typeof APIScoreSchema>;
/**
* Helpers
*/
const ScoreSource = ["API", "EVAL", "ANNOTATION"] as const;
const ScoreDataType = ["NUMERIC", "CATEGORICAL", "BOOLEAN"] as const;
const operators = ["<", ">", "<=", ">=", "!=", "="] as const;
const NumericData = z.object({
value: z.number(),
stringValue: z.undefined().nullish(),
dataType: z.literal("NUMERIC"),
});
const CategoricalData = z.object({
value: z.number().nullish(),
stringValue: z.string(),
dataType: z.literal("CATEGORICAL"),
});
const BooleanData = z.object({
value: z.number(),
stringValue: z.string(),
dataType: z.literal("BOOLEAN"),
});
const ScoreBase = z.object({
id: z.string(),
timestamp: z.coerce.date(),
projectId: z.string(),
environment: z.string().default("default"),
name: z.string(),
source: z.enum(ScoreSource),
authorUserId: z.string().nullish(),
comment: z.string().nullish(),
metadata: jsonSchema.nullish(),
traceId: z.string(),
observationId: z.string().nullish(),
configId: z.string().nullish(),
createdAt: z.coerce.date(),
updatedAt: z.coerce.date(),
queueId: z.string().nullish(),
});
const BaseScoreBody = z.object({
id: z.string().nullish(),
name: NonEmptyString,
traceId: z.string(),
observationId: z.string().nullish(),
comment: z.string().nullish(),
metadata: jsonSchema.nullish(),
environment: z.string().default("default"),
});
/**
* Objects
*/
export const APIScoreSchema = z.discriminatedUnion("dataType", [
ScoreBase.merge(NumericData),
ScoreBase.merge(CategoricalData),
ScoreBase.merge(BooleanData),
]);
/**
* Validation objects
*/
export const ScoreBodyWithoutConfig = z.discriminatedUnion("dataType", [
BaseScoreBody.merge(
z.object({
value: z.number(),
dataType: z.literal("NUMERIC"),
}),
),
BaseScoreBody.merge(
z.object({
value: z.string(),
dataType: z.literal("CATEGORICAL"),
}),
),
BaseScoreBody.merge(
z.object({
value: z.number().refine((val) => val === 0 || val === 1, {
message: "Value must be either 0 or 1",
}),
dataType: z.literal("BOOLEAN"),
}),
),
]);
const ScorePropsAgainstConfigNumeric = z
.object({
value: z.number(),
maxValue: z.number().optional(),
minValue: z.number().optional(),
dataType: z.literal("NUMERIC"),
})
.superRefine((data, ctx) => {
if (isPresent(data.maxValue) && data.value >= data.maxValue) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: `Value exceeds maximum value of ${data.maxValue} defined in config`,
});
}
if (isPresent(data.minValue) && data.value <= data.minValue) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: `Value is below minimum value of ${data.minValue} defined in config`,
});
}
});
const ScorePropsAgainstConfigCategorical = z
.object({
value: z.string(),
categories: z.array(ConfigCategory),
dataType: z.literal("CATEGORICAL"),
})
.superRefine((data, ctx) => {
if (!data.categories.some(({ label }) => label === data.value)) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: `Value ${data.value} does not map to a valid category. Pass a valid category value.`,
});
}
});
export const ScorePropsAgainstConfig = z.union([
ScorePropsAgainstConfigNumeric,
ScorePropsAgainstConfigCategorical,
z.object({
value: z.number().refine((val) => val === 0 || val === 1, {
message: "Value must be either 0 or 1",
}),
dataType: z.literal("BOOLEAN"),
}),
]);
/**
* Transformations
*/
type ValidatedAPIScore<IncludeHasMetadata extends boolean> = APIScore & {
hasMetadata: IncludeHasMetadata extends true ? boolean : never;
};
type InputScore = ScoreDomain & { hasMetadata?: boolean };
/**
* Use this function when pulling a list of scores from the database before using in the application to ensure type safety.
* All scores are expected to pass the validation. If a score fails validation, it will be logged to Otel.
* @param scores
* @returns list of validated scores
*/
export const filterAndValidateDbScoreList = <
IncludeHasMetadata extends boolean,
>({
scores,
includeHasMetadata = false as IncludeHasMetadata,
onParseError,
}: {
scores: InputScore[];
includeHasMetadata?: IncludeHasMetadata;
// eslint-disable-next-line no-unused-vars
onParseError?: (error: z.ZodError) => void;
}): ValidatedAPIScore<IncludeHasMetadata>[] => {
return scores.reduce((acc, ts) => {
const result = APIScoreSchema.safeParse(ts);
if (result.success) {
const score = { ...result.data };
if (includeHasMetadata) {
Object.assign(score, { hasMetadata: ts.hasMetadata ?? false });
}
acc.push(score as ValidatedAPIScore<IncludeHasMetadata>);
} else {
console.error("Score parsing error: ", result.error);
onParseError?.(result.error);
}
return acc;
}, [] as ValidatedAPIScore<IncludeHasMetadata>[]);
};
/**
* Use this function when pulling a single score from the database before using in the application to ensure type safety.
* The score is expected to pass the validation. If a score fails validation, an error will be thrown.
* @param score
* @returns validated score
* @throws error if score fails validation
*/
export const validateDbScore = (score: ScoreDomain): APIScore =>
APIScoreSchema.parse(score);
/**
* Endpoints
*/
// POST /scores
/**
* PostScoresBody is copied for the ingestion API as `ScoreBody`. Please copy any changes here in `packages/shared/src/features/ingestion/types.ts`
*/
export const PostScoresBody = z.discriminatedUnion("dataType", [
BaseScoreBody.merge(
z.object({
value: z.number(),
dataType: z.literal("NUMERIC"),
configId: z.string().nullish(),
}),
),
BaseScoreBody.merge(
z.object({
value: z.string(),
dataType: z.literal("CATEGORICAL"),
configId: z.string().nullish(),
}),
),
BaseScoreBody.merge(
z.object({
value: z.number().refine((value) => value === 0 || value === 1, {
message:
"Value must be a number equal to either 0 or 1 for data type BOOLEAN",
}),
dataType: z.literal("BOOLEAN"),
configId: z.string().nullish(),
}),
),
BaseScoreBody.merge(
z.object({
value: z.union([z.string(), z.number()]),
dataType: z.undefined(),
configId: z.string().nullish(),
}),
),
]);
export const PostScoresResponse = z.object({ id: z.string() });
// GET /scores
export const GetScoresQuery = z.object({
...publicApiPaginationZod,
userId: z.string().nullish(),
dataType: z.enum(ScoreDataType).nullish(),
configId: z.string().nullish(),
queueId: z.string().nullish(),
traceTags: z.union([z.array(z.string()), z.string()]).nullish(),
environment: z.union([z.array(z.string()), z.string()]).nullish(),
name: z.string().nullish(),
fromTimestamp: stringDateTime,
toTimestamp: stringDateTime,
source: z.enum(ScoreSource).nullish(),
value: z.coerce.number().nullish(),
operator: z.enum(operators).nullish(),
scoreIds: z
.string()
.transform((str) => str.split(",").map((id) => id.trim())) // Split the comma-separated string
.refine((arr) => arr.every((id) => typeof id === "string"), {
message: "Each score ID must be a string",
})
.nullish(),
});
// LegacyGetScoreResponseDataV1 is only used for response of GET /scores list endpoint
const LegacyGetScoreResponseDataV1 = z.intersection(
APIScoreSchema,
z.object({
trace: z.object({
userId: z.string().nullish(),
tags: z.array(z.string()).nullish(),
environment: z.string().nullish(),
}),
}),
);
export const GetScoresResponse = z.object({
data: z.array(LegacyGetScoreResponseDataV1),
meta: paginationMetaResponseZod,
});
export const legacyFilterAndValidateV1GetScoreList = (
scores: unknown[],
// eslint-disable-next-line no-unused-vars
onParseError?: (error: z.ZodError) => void,
): z.infer<typeof LegacyGetScoreResponseDataV1>[] =>
scores.reduce(
(acc: z.infer<typeof LegacyGetScoreResponseDataV1>[], ts) => {
const result = LegacyGetScoreResponseDataV1.safeParse(ts);
if (result.success) {
acc.push(result.data);
} else {
console.error("Score parsing error: ", result.error);
onParseError?.(result.error);
}
return acc;
},
[] as z.infer<typeof LegacyGetScoreResponseDataV1>[],
);
// GET /scores/{scoreId}
export const GetScoreQuery = z.object({
scoreId: z.string(),
});
export const GetScoreResponse = APIScoreSchema;
// DELETE /scores/{scoreId}
export const DeleteScoreQuery = z.object({
scoreId: z.string(),
});
export const DeleteScoreResponse = z.object({
message: z.string(),
});
+1
View File
@@ -15,6 +15,7 @@ export * from "./utils/objects";
export * from "./utils/typeChecks";
export * from "./features/entitlements/plans";
export * from "./interfaces/rate-limits";
export * from "./tableDefinitions/typeHelpers";
// llm api
export * from "./server/llm/types";
@@ -4,6 +4,7 @@ export const filterOperators = {
datetime: [">", "<", ">=", "<="],
string: ["=", "contains", "does not contain", "starts with", "ends with"],
stringOptions: ["any of", "none of"],
categoryOptions: ["any of", "none of"],
arrayOptions: ["any of", "none of", "all of"],
number: ["=", ">", "<", ">=", "<="],
stringObject: [
@@ -75,11 +76,19 @@ export const nullFilter = z.object({
operator: z.enum(filterOperators.null),
value: z.literal(""),
});
export const categoryOptionsFilter = z.object({
type: z.literal("categoryOptions"),
column: z.string(),
key: z.string(),
operator: z.enum(filterOperators.categoryOptions),
value: z.array(z.string()),
});
export const singleFilter = z.discriminatedUnion("type", [
timeFilter,
stringFilter,
numberFilter,
stringOptionsFilter,
categoryOptionsFilter,
arrayOptionsFilter,
stringObjectFilter,
numberObjectFilter,
+33 -19
View File
@@ -1,8 +1,10 @@
import { ObservationLevelType } from "./domain";
import {
type OptionsDefinition,
type SingleValueOption,
type ColumnDefinition,
MultiValueOption,
} from "./tableDefinitions";
import { formatColumnOptions } from "./tableDefinitions/typeHelpers";
// to be used server side
export const observationsTableCols: ColumnDefinition[] = [
@@ -154,8 +156,8 @@ export const observationsTableCols: ColumnDefinition[] = [
nullable: true,
},
{
name: "Usage",
id: "usage",
name: "Tokens",
id: "tokens",
type: "number",
internal: 'o."total_tokens"',
},
@@ -166,11 +168,19 @@ export const observationsTableCols: ColumnDefinition[] = [
internal: 'o."metadata"',
},
{
name: "Scores",
name: "Scores (numeric)",
id: "scores_avg",
type: "numberObject",
internal: "scores_avg",
},
{
name: "Scores (categorical)",
id: "score_categories",
type: "categoryOptions",
internal: "score_categories",
options: [], // to be added at runtime
nullable: true,
},
{
name: "Version",
id: "version",
@@ -205,14 +215,15 @@ export const observationsTableCols: ColumnDefinition[] = [
// to be used client side, insert options for use in filter-builder
// allows for undefined options, to offer filters while options are still loading
export type ObservationOptions = {
model: Array<OptionsDefinition>;
modelId: Array<OptionsDefinition>;
name: Array<OptionsDefinition>;
traceName: Array<OptionsDefinition>;
model: Array<SingleValueOption>;
modelId: Array<SingleValueOption>;
name: Array<SingleValueOption>;
traceName: Array<SingleValueOption>;
scores_avg: Array<string>;
promptName: Array<OptionsDefinition>;
tags: Array<OptionsDefinition>;
type: Array<OptionsDefinition>;
score_categories: Array<MultiValueOption>;
promptName: Array<SingleValueOption>;
tags: Array<SingleValueOption>;
type: Array<SingleValueOption>;
};
export function observationsTableColsWithOptions(
@@ -220,28 +231,31 @@ export function observationsTableColsWithOptions(
): ColumnDefinition[] {
return observationsTableCols.map((col) => {
if (col.id === "model") {
return { ...col, options: options?.model ?? [] };
return formatColumnOptions(col, options?.model ?? []);
}
if (col.id === "modelId") {
return { ...col, options: options?.modelId ?? [] };
return formatColumnOptions(col, options?.modelId ?? []);
}
if (col.id === "name") {
return { ...col, options: options?.name ?? [] };
return formatColumnOptions(col, options?.name ?? []);
}
if (col.id === "traceName") {
return { ...col, options: options?.traceName ?? [] };
return formatColumnOptions(col, options?.traceName ?? []);
}
if (col.id === "scores_avg") {
return { ...col, keyOptions: options?.scores_avg ?? [] };
return formatColumnOptions(col, options?.scores_avg ?? []);
}
if (col.id === "score_categories") {
return formatColumnOptions(col, options?.score_categories ?? []);
}
if (col.id === "promptName") {
return { ...col, options: options?.promptName ?? [] };
return formatColumnOptions(col, options?.promptName ?? []);
}
if (col.id === "tags") {
return { ...col, options: options?.tags ?? [] };
return formatColumnOptions(col, options?.tags ?? []);
}
if (col.id === "type") {
return { ...col, options: options?.type ?? [] };
return formatColumnOptions(col, options?.type ?? []);
}
return col;
});
+102 -32
View File
@@ -6,42 +6,112 @@ import { propagation, context } from "@opentelemetry/api";
export type ClickhouseClientType = ReturnType<typeof createClient>;
export const clickhouseClient = (
params: {
tags?: Record<string, string>;
opts?: NodeClickHouseClientConfigOptions;
} = {},
) => {
const headers = params.opts?.http_headers ?? {};
const activeSpan = getCurrentSpan();
if (activeSpan) {
propagation.inject(context.active(), headers);
/**
* ClickHouseClientManager provides a singleton pattern for managing ClickHouse clients.
* It creates and reuses clients based on their configuration to avoid creating
* a new connection for each query.
*/
export class ClickHouseClientManager {
private static instance: ClickHouseClientManager;
private clientMap: Map<string, ClickhouseClientType> = new Map();
/**
* Private constructor to enforce singleton pattern
*/
private constructor() {}
/**
* Get the singleton instance of the ClickHouseClientManager
*/
public static getInstance(): ClickHouseClientManager {
if (!ClickHouseClientManager.instance) {
ClickHouseClientManager.instance = new ClickHouseClientManager();
}
return ClickHouseClientManager.instance;
}
const cloudOptions: Record<string, unknown> = {};
if (
["STAGING", "EU", "US"].includes(
env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION ?? "",
)
) {
cloudOptions.input_format_json_throw_on_bad_escape_sequence = 0;
/**
* Generate a consistent hash key for client configurations
* @param opts Client parameters
* @returns String hash key
*/
private generateClientSettingsKey(
opts: NodeClickHouseClientConfigOptions,
): string {
const keyParams = {
url: env.CLICKHOUSE_URL,
username: env.CLICKHOUSE_USER,
password: env.CLICKHOUSE_PASSWORD,
database: env.CLICKHOUSE_DB,
http_headers: opts?.http_headers,
settings: opts?.clickhouse_settings,
// Include any other relevant config options
};
return JSON.stringify(keyParams);
}
return createClient({
...params.opts,
url: env.CLICKHOUSE_URL,
username: env.CLICKHOUSE_USER,
password: env.CLICKHOUSE_PASSWORD,
database: env.CLICKHOUSE_DB,
http_headers: headers,
clickhouse_settings: {
...cloudOptions,
...(params.opts?.clickhouse_settings ?? {}),
log_comment: JSON.stringify(params.tags ?? {}),
async_insert: 1,
wait_for_async_insert: 1, // if disabled, we won't get errors from clickhouse
},
});
/**
* Get or create a client based on the provided parameters
* @param opts Client configuration parameters
* @returns ClickHouse client instance
*/
public getClient(
opts: NodeClickHouseClientConfigOptions,
): ClickhouseClientType {
const key = this.generateClientSettingsKey(opts);
if (!this.clientMap.has(key)) {
const headers = opts?.http_headers ?? {};
const activeSpan = getCurrentSpan();
if (activeSpan) {
propagation.inject(context.active(), headers);
}
const cloudOptions: Record<string, unknown> = {};
if (
["STAGING", "EU", "US", "HIPAA"].includes(
env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION ?? "",
)
) {
cloudOptions.input_format_json_throw_on_bad_escape_sequence = 0;
}
const client = createClient({
...opts,
url: env.CLICKHOUSE_URL,
username: env.CLICKHOUSE_USER,
password: env.CLICKHOUSE_PASSWORD,
database: env.CLICKHOUSE_DB,
http_headers: headers,
clickhouse_settings: {
...cloudOptions,
...opts.clickhouse_settings,
async_insert: 1,
wait_for_async_insert: 1, // if disabled, we won't get errors from clickhouse
},
});
this.clientMap.set(key, client);
}
return this.clientMap.get(key)!;
}
/**
* Close all client connections - useful for application shutdown
*/
public closeAllConnections(): Promise<void[]> {
const closePromises = Array.from(this.clientMap.values()).map((client) =>
client.close(),
);
this.clientMap.clear();
return Promise.all(closePromises);
}
}
export const clickhouseClient = (opts?: NodeClickHouseClientConfigOptions) => {
return ClickHouseClientManager.getInstance().getClient(opts ?? {});
};
/**
@@ -0,0 +1,108 @@
import {
BlobStorageFileRefRecordReadType,
getBlobStorageByProjectId,
getBlobStorageByProjectIdAndEntityIds,
getBlobStorageByProjectIdAndTraceIds,
getBlobStorageByProjectIdBeforeDate,
logger,
} from "..";
import { env } from "../../env";
import { clickhouseClient } from "../clickhouse/client";
import { getS3EventStorageClient } from "../s3";
export const deleteIngestionEventsFromS3AndClickhouseForScores = async (p: {
projectId: string;
scoreIds: string[];
}) => {
const stream = getBlobStorageByProjectIdAndEntityIds(
p.projectId,
"score",
p.scoreIds,
);
return removeIngestionEventsFromS3AndDeleteClickhouseRefs({
projectId: p.projectId,
stream,
});
};
export const removeIngestionEventsFromS3AndDeleteClickhouseRefsForTraces =
async (p: { projectId: string; traceIds: string[] }) => {
const stream = getBlobStorageByProjectIdAndTraceIds(
p.projectId,
p.traceIds,
);
return removeIngestionEventsFromS3AndDeleteClickhouseRefs({
projectId: p.projectId,
stream: stream,
});
};
export const removeIngestionEventsFromS3AndDeleteClickhouseRefsForProject = (
projectId: string,
cutOffDate: Date | undefined,
) => {
const stream = cutOffDate
? getBlobStorageByProjectIdBeforeDate(projectId, cutOffDate)
: getBlobStorageByProjectId(projectId);
return removeIngestionEventsFromS3AndDeleteClickhouseRefs({
projectId: projectId,
stream: stream,
});
};
async function removeIngestionEventsFromS3AndDeleteClickhouseRefs(p: {
projectId: string;
stream: AsyncGenerator<BlobStorageFileRefRecordReadType>;
}) {
const { projectId, stream } = p;
let batch = 0;
let blobStorageRefs: BlobStorageFileRefRecordReadType[] = [];
const eventStorageClient = getS3EventStorageClient(
env.LANGFUSE_S3_EVENT_UPLOAD_BUCKET,
);
for await (const eventLog of stream) {
blobStorageRefs.push(eventLog);
if (blobStorageRefs.length > 500) {
// Delete the current batch and reset the list
await eventStorageClient.deleteFiles(
blobStorageRefs.map((r) => r.bucket_path),
);
// soft delete the blob storage references in clickhouse
await softDeleteInClickhouse(blobStorageRefs);
batch++;
logger.info(
`Deleted batch ${batch} of size ${blobStorageRefs.length} for ${projectId} of deleting s3 refs`,
);
blobStorageRefs = [];
}
}
// Delete any remaining files
await eventStorageClient.deleteFiles(
blobStorageRefs.map((r) => r.bucket_path),
);
await softDeleteInClickhouse(blobStorageRefs);
logger.info(
`Deleted batch ${batch} of size ${blobStorageRefs.length} for ${projectId} of deleting s3 refs`,
);
}
async function softDeleteInClickhouse(
blobStorageRefs: BlobStorageFileRefRecordReadType[],
) {
await clickhouseClient().insert({
table: "blob_storage_file_log",
values: blobStorageRefs.map((e) => ({
...e,
is_deleted: "1",
event_ts: new Date().getTime(),
updated_at: new Date().getTime(),
})),
format: "JSONEachRow",
});
}
@@ -109,6 +109,10 @@ export function tableColumnsToSqlFilter(
case "boolean":
valuePrisma = Prisma.sql`${filter.value}`;
break;
case "categoryOptions":
// LFE-4815: Support category options in postgres
logger.warn("Category options not supported in postgres yet");
throw new Error("Category options not supported in postgres yet");
case "null":
valuePrisma = Prisma.sql``;
break;
+4
View File
@@ -41,6 +41,7 @@ export * from "./redis/dataRetentionProcessingQueue";
export * from "./redis/coreDataS3ExportQueue";
export * from "./redis/meteringDataPostgresExportQueue";
export * from "./redis/experimentCreateQueue";
export * from "./redis/dlxRetryQueue";
export * from "./auth/types";
export * from "./queues";
export * from "./orderByToPrisma";
@@ -54,5 +55,8 @@ export * from "./services/sessions-ui-table-service";
export * from "./services/datasets-ui-table-service";
export * from "./services/DashboardService";
export * from "./data-deletion/ingestionFileDeletion";
export * from "./s3";
// test utils
export * from "./test-utils";
+37 -33
View File
@@ -4,6 +4,7 @@ import { z } from "zod";
import { NonEmptyString, jsonSchema } from "../../utils/zod";
import { ModelUsageUnit } from "../../constants";
import { ScoreSourceType } from "../../domain";
import { applyScoreValidation } from "../../utils/scores";
export const idSchema = z
.string()
@@ -316,7 +317,8 @@ export const UpdateGenerationBody = UpdateSpanBody.extend({
const BaseScoreBody = z.object({
id: idSchema.nullish(),
name: NonEmptyString,
traceId: z.string(),
traceId: z.string().nullish(),
sessionId: z.string().nullish(),
environment: EnvironmentName,
observationId: z.string().nullish(),
comment: z.string().nullish(),
@@ -329,39 +331,41 @@ const BaseScoreBody = z.object({
/**
* ScoreBody exactly mirrors `PostScoresBody` in the public API. Please refer there for source of truth.
*/
export const ScoreBody = z.discriminatedUnion("dataType", [
BaseScoreBody.merge(
z.object({
value: z.number(),
dataType: z.literal("NUMERIC"),
configId: z.string().nullish(),
}),
),
BaseScoreBody.merge(
z.object({
value: z.string(),
dataType: z.literal("CATEGORICAL"),
configId: z.string().nullish(),
}),
),
BaseScoreBody.merge(
z.object({
value: z.number().refine((value) => value === 0 || value === 1, {
message:
"Value must be a number equal to either 0 or 1 for data type BOOLEAN",
export const ScoreBody = applyScoreValidation(
z.discriminatedUnion("dataType", [
BaseScoreBody.merge(
z.object({
value: z.number(),
dataType: z.literal("NUMERIC"),
configId: z.string().nullish(),
}),
dataType: z.literal("BOOLEAN"),
configId: z.string().nullish(),
}),
),
BaseScoreBody.merge(
z.object({
value: z.union([z.string(), z.number()]),
dataType: z.undefined(),
configId: z.string().nullish(),
}),
),
]);
),
BaseScoreBody.merge(
z.object({
value: z.string(),
dataType: z.literal("CATEGORICAL"),
configId: z.string().nullish(),
}),
),
BaseScoreBody.merge(
z.object({
value: z.number().refine((value) => value === 0 || value === 1, {
message:
"Value must be a number equal to either 0 or 1 for data type BOOLEAN",
}),
dataType: z.literal("BOOLEAN"),
configId: z.string().nullish(),
}),
),
BaseScoreBody.merge(
z.object({
value: z.union([z.string(), z.number()]),
dataType: z.undefined(),
configId: z.string().nullish(),
}),
),
]),
);
// LEGACY, only required for backwards compatibility
export const LegacySpanPostSchema = z.object({
@@ -142,7 +142,13 @@ export const traceException = (
};
export const addUserToSpan = (
attributes: { userId?: string; projectId?: string; email?: string },
attributes: {
userId?: string;
projectId?: string;
email?: string;
orgId?: string;
plan?: string;
},
span?: opentelemetry.Span,
) => {
const activeSpan = span ?? getCurrentSpan();
@@ -155,6 +161,8 @@ export const addUserToSpan = (
attributes.email && activeSpan.setAttribute("user.email", attributes.email);
attributes.projectId &&
activeSpan.setAttribute("project.id", attributes.projectId);
attributes.orgId && activeSpan.setAttribute("org.id", attributes.orgId);
attributes.plan && activeSpan.setAttribute("org.plan", attributes.plan);
};
export const getTracer = (name: string) => opentelemetry.trace.getTracer(name);
+7 -3
View File
@@ -253,11 +253,13 @@ export const ExperimentMetadataSchema = z
export type ExperimentMetadata = z.infer<typeof ExperimentMetadataSchema>;
// NOTE: Update docs page when changing this! https://langfuse.com/docs/playground#openai-playground--anthropic-playground
// WARNING: The first entry in the array is chosen as the default model to add LLM API keys. Make sure it supports top_p, max_tokens and temperature.
// WARNING: The first entry in the array is chosen as the default model to add LLM API keys
export const openAIModels = [
"gpt-4.1",
"gpt-4.1-2025-04-14",
"gpt-4.1-mini",
"gpt-4.1-mini-2025-04-14",
"gpt-4.1-nano",
"gpt-4.1-nano-2025-04-14",
"o3",
"o3-2025-04-16",
@@ -293,7 +295,7 @@ export const openAIModels = [
export type OpenAIModel = (typeof openAIModels)[number];
// NOTE: Update docs page when changing this! https://langfuse.com/docs/playground#openai-playground--anthropic-playground
// WARNING: The first entry in the array is chosen as the default model to add LLM API keys. Make sure it supports top_p, max_tokens and temperature.
// WARNING: The first entry in the array is chosen as the default model to add LLM API keys
export const anthropicModels = [
"claude-3-7-sonnet-20250219",
"claude-3-5-sonnet-20241022",
@@ -307,10 +309,11 @@ export const anthropicModels = [
"claude-instant-1.2",
] as const;
// WARNING: The first entry in the array is chosen as the default model to add LLM API keys. Make sure it supports top_p, max_tokens and temperature.
// WARNING: The first entry in the array is chosen as the default model to add LLM API keys
export const vertexAIModels = [
"gemini-2.5-pro-exp-03-25",
"gemini-2.0-pro-exp-02-05",
"gemini-2.0-flash",
"gemini-2.0-flash-001",
"gemini-2.0-flash-lite-preview-02-05",
"gemini-2.0-flash-exp",
@@ -323,6 +326,7 @@ export const vertexAIModels = [
export const googleAIStudioModels = [
"gemini-2.5-pro-exp-03-25",
"gemini-2.0-flash",
"gemini-2.0-flash-lite-preview",
"gemini-2.0-flash-lite-preview-02-05",
"gemini-2.0-flash-thinking-exp-01-21",
"gemini-1.5-pro",
@@ -168,6 +168,59 @@ export class StringOptionsFilter implements Filter {
}
}
export class CategoryOptionsFilter implements Filter {
public clickhouseTable: string;
public field: string;
public key: string;
public values: string[];
public operator: (typeof filterOperators.categoryOptions)[number];
protected tablePrefix?: string;
constructor(opts: {
clickhouseTable: string;
field: string;
operator: (typeof filterOperators.categoryOptions)[number];
key: string;
values: string[];
tablePrefix?: string;
}) {
this.clickhouseTable = opts.clickhouseTable;
this.field = opts.field;
this.key = opts.key;
this.values = opts.values;
this.operator = opts.operator;
this.tablePrefix = opts.tablePrefix;
}
apply(): ClickhouseFilter {
const uid = clickhouseCompliantRandomCharacters();
const varName = `categoryOptionsFilter${uid}`;
// Flatten the hierarchical structure into array of "parent:child" strings for improved query performance
const flattenedValues: string[] = [];
this.values.forEach((child) => {
flattenedValues.push(`${this.key}:${child}`);
});
const fieldRef = `${this.tablePrefix ? this.tablePrefix + "." : ""}${this.field}`;
switch (this.operator) {
case "any of":
return {
query: `hasAny(${fieldRef}, {${varName}: Array(String)})`,
params: { [varName]: flattenedValues },
};
case "none of":
return {
query: `NOT hasAny(${fieldRef}, {${varName}: Array(String)})`,
params: { [varName]: flattenedValues },
};
default:
throw new Error(`Unsupported operator: ${this.operator}`);
}
}
}
// stringObject filter is used when we want to filter on a key value pair in a clickhouse map.
// As we use the MAP form clickhouse, we can only filter efficiently on the first level of a json obj.
export class StringObjectFilter implements Filter {
@@ -8,6 +8,7 @@ import {
StringFilter,
DateTimeFilter,
StringOptionsFilter,
CategoryOptionsFilter,
FilterList,
NumberFilter,
ArrayOptionsFilter,
@@ -60,6 +61,15 @@ export const createFilterFromFilterState = (
values: frontEndFilter.value,
tablePrefix: column.queryPrefix,
});
case "categoryOptions":
return new CategoryOptionsFilter({
clickhouseTable: column.clickhouseTableName,
field: column.clickhouseSelect,
operator: frontEndFilter.operator,
key: frontEndFilter.key,
values: frontEndFilter.value,
tablePrefix: column.queryPrefix,
});
case "number":
return new NumberFilter({
clickhouseTable: column.clickhouseTableName,
@@ -111,6 +121,7 @@ export const createFilterFromFilterState = (
tablePrefix: column.queryPrefix,
});
default:
// eslint-disable-next-line no-case-declarations
const exhaustiveCheck: never = frontEndFilter;
logger.error(`Invalid filter type: ${JSON.stringify(exhaustiveCheck)}`);
throw new QueryBuilderError(`Invalid filter type`);
@@ -8,6 +8,7 @@ export {
StringFilter,
DateTimeFilter,
StringOptionsFilter,
CategoryOptionsFilter,
NumberFilter,
ArrayOptionsFilter,
BooleanFilter,
+15
View File
@@ -123,6 +123,10 @@ export const CreateEvalQueueEventSchema = DatasetRunItemUpsertEventSchema.and(
),
);
export const DeadLetterRetryQueueEventSchema = z.object({
timestamp: z.date(),
});
export type CreateEvalQueueEventType = z.infer<
typeof CreateEvalQueueEventSchema
>;
@@ -151,6 +155,9 @@ export type BatchActionProcessingEventType = z.infer<
export type BlobStorageIntegrationProcessingEventType = z.infer<
typeof BlobStorageIntegrationProcessingEventSchema
>;
export type DeadLetterRetryQueueEventType = z.infer<
typeof DeadLetterRetryQueueEventSchema
>;
export enum QueueName {
TraceUpsert = "trace-upsert", // Ingestion pipeline adds events on each Trace upsert
@@ -174,6 +181,7 @@ export enum QueueName {
BatchActionQueue = "batch-action-queue",
CreateEvalQueue = "create-eval-queue",
ScoreDelete = "score-delete",
DeadLetterRetryQueue = "dead-letter-retry-queue",
}
export enum QueueJobs {
@@ -198,6 +206,7 @@ export enum QueueJobs {
BatchActionProcessingJob = "batch-action-processing-job",
CreateEvalJob = "create-eval-job",
ScoreDelete = "score-delete",
DeadLetterRetryJob = "dead-letter-retry-job",
}
export type TQueueJobTypes = {
@@ -291,4 +300,10 @@ export type TQueueJobTypes = {
payload: BlobStorageIntegrationProcessingEventType;
name: QueueJobs.BlobStorageIntegrationProcessingJob;
};
[QueueName.DeadLetterRetryQueue]: {
timestamp: Date;
id: string;
payload: DeadLetterRetryQueueEventType;
name: QueueJobs.DeadLetterRetryJob;
};
};
@@ -0,0 +1,55 @@
import { Queue } from "bullmq";
import { QueueName, QueueJobs } from "../queues";
import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
import { logger } from "../logger";
export class DeadLetterRetryQueue {
private static instance: Queue | null = null;
public static getInstance(): Queue | null {
if (DeadLetterRetryQueue.instance) {
return DeadLetterRetryQueue.instance;
}
const newRedis = createNewRedisInstance({
enableOfflineQueue: false,
...redisQueueRetryOptions,
});
DeadLetterRetryQueue.instance = newRedis
? new Queue(QueueName.DeadLetterRetryQueue, {
connection: newRedis,
defaultJobOptions: {
removeOnComplete: true,
removeOnFail: 100,
attempts: 5,
backoff: {
type: "exponential",
delay: 5000,
},
},
})
: null;
DeadLetterRetryQueue.instance?.on("error", (err) => {
logger.error("DeadLetterRetryQueue error", err);
});
if (DeadLetterRetryQueue.instance) {
logger.debug("Scheduling jobs for DeadLetterRetryQueue");
DeadLetterRetryQueue.instance
.add(
QueueJobs.DeadLetterRetryJob,
{ timestamp: new Date() },
{
repeat: { pattern: "0 */10 * * * *" }, // every 10 minutes (with seconds precision)
},
)
.catch((err) => {
logger.error("Error adding DeadLetterRetryQueue schedule", err);
});
}
return DeadLetterRetryQueue.instance;
}
}
@@ -20,6 +20,7 @@ import { DataRetentionProcessingQueue } from "./dataRetentionProcessingQueue";
import { BatchActionQueue } from "./batchActionQueue";
import { CreateEvalQueue } from "./createEvalQueue";
import { ScoreDeleteQueue } from "./scoreDelete";
import { DeadLetterRetryQueue } from "./dlxRetryQueue";
export function getQueue(queueName: QueueName): Queue | null {
switch (queueName) {
@@ -65,6 +66,8 @@ export function getQueue(queueName: QueueName): Queue | null {
return CreateEvalQueue.getInstance();
case QueueName.ScoreDelete:
return ScoreDeleteQueue.getInstance();
case QueueName.DeadLetterRetryQueue:
return DeadLetterRetryQueue.getInstance();
default:
const exhaustiveCheckDefault: never = queueName;
throw new Error(`Queue ${queueName} not found`);
@@ -3,23 +3,23 @@ import {
queryClickhouse,
queryClickhouseStream,
} from "./clickhouse";
import { EventLogRecordReadType } from "./definitions";
import { BlobStorageFileRefRecordReadType } from "./definitions";
import { convertDateToClickhouseDateTime } from "../clickhouse/client";
export const getEventLogByProjectAndEntityId = async (
export const getBlobStorageByProjectAndEntityId = async (
projectId: string,
entityType: string,
entityId: string,
): Promise<EventLogRecordReadType[]> => {
): Promise<BlobStorageFileRefRecordReadType[]> => {
const query = `
select *
from event_log
from blob_storage_file_log FINAL
where project_id = {projectId: String}
and entity_type = {entityType: String}
and entity_id = {entityId: String}
`;
return queryClickhouse<EventLogRecordReadType>({
return queryClickhouse<BlobStorageFileRefRecordReadType>({
query,
params: {
projectId,
@@ -34,16 +34,16 @@ export const getEventLogByProjectAndEntityId = async (
});
};
export const getEventLogByProjectId = (
export const getBlobStorageByProjectId = (
projectId: string,
): AsyncGenerator<EventLogRecordReadType> => {
): AsyncGenerator<BlobStorageFileRefRecordReadType> => {
const query = `
select *
from event_log
from blob_storage_file_log FINAL
where project_id = {projectId: String}
`;
return queryClickhouseStream<EventLogRecordReadType>({
return queryClickhouseStream<BlobStorageFileRefRecordReadType>({
query,
params: {
projectId,
@@ -56,18 +56,18 @@ export const getEventLogByProjectId = (
});
};
export const getEventLogByProjectIdBeforeDate = (
export const getBlobStorageByProjectIdBeforeDate = (
projectId: string,
beforeDate: Date,
): AsyncGenerator<EventLogRecordReadType> => {
): AsyncGenerator<BlobStorageFileRefRecordReadType> => {
const query = `
select *
from event_log
from blob_storage_file_log FINAL
where project_id = {projectId: String}
and created_at <= {beforeDate: DateTime64(3)}
`;
return queryClickhouseStream<EventLogRecordReadType>({
return queryClickhouseStream<BlobStorageFileRefRecordReadType>({
query,
params: {
projectId,
@@ -81,20 +81,20 @@ export const getEventLogByProjectIdBeforeDate = (
});
};
export const getEventLogByProjectIdAndEntityIds = (
export const getBlobStorageByProjectIdAndEntityIds = (
projectId: string,
entityType: "observation" | "trace" | "score",
entityIds: string[],
): AsyncGenerator<EventLogRecordReadType> => {
): AsyncGenerator<BlobStorageFileRefRecordReadType> => {
const query = `
select *
from event_log
from blob_storage_file_log FINAL
where project_id = {projectId: String}
and entity_type = {entityType: String}
and entity_id in ({entityIds: Array(String)})
`;
return queryClickhouseStream<EventLogRecordReadType>({
return queryClickhouseStream<BlobStorageFileRefRecordReadType>({
query,
params: {
projectId,
@@ -112,10 +112,10 @@ export const getEventLogByProjectIdAndEntityIds = (
});
};
export const getEventLogByProjectIdAndTraceIds = (
export const getBlobStorageByProjectIdAndTraceIds = (
projectId: string,
traceIds: string[],
): AsyncGenerator<EventLogRecordReadType> => {
): AsyncGenerator<BlobStorageFileRefRecordReadType> => {
const query = `
with filtered_traces as (
select distinct
@@ -155,13 +155,13 @@ export const getEventLogByProjectIdAndTraceIds = (
-- We use a semi join because we only use the 'filtered_events' as a filter.
-- There is no need to build the cartesian product (i.e. the combination) between the event log and the events.
select el.*
from event_log el
from blob_storage_file_log el FINAL
left semi join filtered_events fe
on el.project_id = fe.project_id and el.entity_id = fe.entity_id and el.entity_type = fe.entity_type
where el.project_id = {projectId: String}
`;
return queryClickhouseStream<EventLogRecordReadType>({
return queryClickhouseStream<BlobStorageFileRefRecordReadType>({
query,
params: {
projectId,
@@ -178,83 +178,76 @@ export const getEventLogByProjectIdAndTraceIds = (
});
};
/**
* Deletes event log records by projectId and the _eventLog_.id
* @param projectId - Project ID
* @param ids - ID record of the event log table to be deleted
*/
export const deleteEventLogByProjectIdAndIds = async (
projectId: string,
ids: string[],
): Promise<void> => {
const query = `
delete from event_log
where project_id = {projectId: String}
and id in ({ids: Array(String)});
// this function is only used for the background migration from event_log to blob_storage_file_log
export const insertIntoS3RefsTableFromEventLog = async (
limit: number,
offset: number,
) => {
const query = `
INSERT INTO blob_storage_file_log
SELECT
id,
project_id,
entity_type,
entity_id,
event_id,
bucket_name,
bucket_path,
created_at,
updated_at,
created_at AS event_ts,
0 AS is_deleted
FROM event_log
ORDER BY (project_id, entity_type, entity_id, bucket_path) DESC
LIMIT {limit: Int32}
OFFSET {offset: Int32}
`;
await commandClickhouse({
query,
params: {
projectId,
ids,
},
clickhouseConfigs: {
request_timeout: 300_000, // 5 minutes
limit,
offset,
},
tags: {
feature: "eventLog",
kind: "delete",
projectId,
feature: "backgroundMigration",
kind: "list",
},
});
};
export const deleteEventLogByProjectId = async (
projectId: string,
): Promise<void> => {
export const getLastEventLogPrimaryKey = async () => {
const query = `
DELETE FROM event_log
WHERE project_id = {projectId: String};
SELECT project_id, entity_type, entity_id, bucket_path
FROM event_log
ORDER BY (project_id, entity_type, entity_id, bucket_path) ASC
LIMIT 1
`;
await commandClickhouse({
query: query,
params: {
projectId,
},
clickhouseConfigs: {
request_timeout: 120_000, // 2 minutes
},
tags: {
feature: "eventLog",
kind: "delete",
projectId,
},
});
const result = await queryClickhouse<{
project_id: string;
entity_type: string;
entity_id: string;
bucket_path: string;
}>({ query });
return result.shift();
};
export const deleteEventLogByProjectIdBeforeDate = async (
projectId: string,
beforeDate: Date,
): Promise<void> => {
export const findS3RefsByPrimaryKey = async (primaryKey: {
project_id: string;
entity_type: string;
entity_id: string;
bucket_path: string;
}) => {
const query = `
DELETE FROM event_log
WHERE project_id = {projectId: String}
AND created_at <= {beforeDate: DateTime64(3)};
SELECT *
FROM blob_storage_file_log
WHERE project_id = {project_id: String}
AND entity_type = {entity_type: String}
AND entity_id = {entity_id: String}
AND bucket_path = {bucket_path: String}
`;
await commandClickhouse({
query: query,
params: {
projectId,
beforeDate: convertDateToClickhouseDateTime(beforeDate),
},
clickhouseConfigs: {
request_timeout: 120_000, // 2 minutes
},
tags: {
feature: "eventLog",
kind: "delete",
projectId,
},
return queryClickhouse<BlobStorageFileRefRecordReadType>({
query,
params: primaryKey,
});
};
@@ -57,10 +57,8 @@ export async function upsertClickhouse<
// Write new file directly to ClickHouse. We don't use the ClickHouse writer here as we expect more limited traffic
// and are not worried that much about latency.
await clickhouseClient({
tags: opts.tags,
}).insert({
table: "event_log",
await clickhouseClient().insert({
table: "blob_storage_file_log",
values: [
{
id: randomUUID(),
@@ -70,9 +68,14 @@ export async function upsertClickhouse<
event_id: eventId,
bucket_name: env.LANGFUSE_S3_EVENT_UPLOAD_BUCKET,
bucket_path: bucketPath,
event_ts: convertDateToClickhouseDateTime(new Date()),
is_deleted: 0,
},
],
format: "JSONEachRow",
clickhouse_settings: {
log_comment: JSON.stringify(opts.tags ?? {}),
},
});
return getS3StorageServiceClient(
@@ -88,13 +91,16 @@ export async function upsertClickhouse<
}),
);
const res = await clickhouseClient({ tags: opts.tags }).insert({
const res = await clickhouseClient().insert({
table: opts.table,
values: opts.records.map((record) => ({
...record,
event_ts: convertDateToClickhouseDateTime(new Date()),
})),
format: "JSONEachRow",
clickhouse_settings: {
log_comment: JSON.stringify(opts.tags ?? {}),
},
});
// same logic as for prisma. we want to see queries in development
if (env.NODE_ENV === "development") {
@@ -139,13 +145,13 @@ export async function* queryClickhouseStream<T>(opts: {
// https://opentelemetry.io/docs/specs/semconv/database/database-spans/
span.setAttribute("ch.query.text", opts.query);
const res = await clickhouseClient({
tags: opts.tags,
opts: opts.clickhouseConfigs,
}).query({
const res = await clickhouseClient(opts.clickhouseConfigs).query({
query: opts.query,
format: "JSONEachRow",
query_params: opts.params,
clickhouse_settings: {
log_comment: JSON.stringify(opts.tags ?? {}),
},
});
// same logic as for prisma. we want to see queries in development
if (env.NODE_ENV === "development") {
@@ -195,13 +201,13 @@ export async function queryClickhouse<T>(opts: {
// https://opentelemetry.io/docs/specs/semconv/database/database-spans/
span.setAttribute("ch.query.text", opts.query);
const res = await clickhouseClient({
tags: opts.tags,
opts: opts.clickhouseConfigs,
}).query({
const res = await clickhouseClient(opts.clickhouseConfigs).query({
query: opts.query,
format: "JSONEachRow",
query_params: opts.params,
clickhouse_settings: {
log_comment: JSON.stringify(opts.tags ?? {}),
},
});
// same logic as for prisma. we want to see queries in development
if (env.NODE_ENV === "development") {
@@ -241,12 +247,12 @@ export async function commandClickhouse(opts: {
return await instrumentAsync({ name: "clickhouse-command" }, async (span) => {
// https://opentelemetry.io/docs/specs/semconv/database/database-spans/
span.setAttribute("ch.query.text", opts.query);
const res = await clickhouseClient({
tags: opts.tags,
opts: opts.clickhouseConfigs,
}).command({
const res = await clickhouseClient(opts.clickhouseConfigs).command({
query: opts.query,
query_params: opts.params,
clickhouse_settings: {
log_comment: JSON.stringify(opts.tags ?? {}),
},
});
// same logic as for prisma. we want to see queries in development
if (env.NODE_ENV === "development") {
@@ -125,7 +125,8 @@ export type TraceRecordInsertType = z.infer<typeof traceRecordInsertSchema>;
export const scoreRecordBaseSchema = z.object({
id: z.string(),
project_id: z.string(),
trace_id: z.string(),
trace_id: z.string().nullish(),
session_id: z.string().nullish(),
observation_id: z.string().nullish(),
environment: z.string().default("default"),
name: z.string(),
@@ -157,7 +158,7 @@ export const scoreRecordInsertSchema = scoreRecordBaseSchema.extend({
});
export type ScoreRecordInsertType = z.infer<typeof scoreRecordInsertSchema>;
export const eventLogRecordBaseSchema = z.object({
export const blobStorageFileLogRecordBaseSchema = z.object({
id: z.string(),
project_id: z.string(),
entity_type: z.string(),
@@ -167,18 +168,25 @@ export const eventLogRecordBaseSchema = z.object({
event_id: z.string().nullable(),
bucket_name: z.string(),
bucket_path: z.string(),
is_deleted: z.number(),
});
export const eventLogRecordReadSchema = eventLogRecordBaseSchema.extend({
created_at: clickhouseStringDateSchema,
updated_at: clickhouseStringDateSchema,
});
export type EventLogRecordReadType = z.infer<typeof eventLogRecordReadSchema>;
export const eventLogRecordInsertSchema = eventLogRecordBaseSchema.extend({
created_at: z.number(),
updated_at: z.number(),
});
export type EventLogRecordInsertType = z.infer<
typeof eventLogRecordInsertSchema
export const blobStorageFileRefRecordReadSchema =
blobStorageFileLogRecordBaseSchema.extend({
created_at: clickhouseStringDateSchema,
updated_at: clickhouseStringDateSchema,
event_ts: clickhouseStringDateSchema,
});
export type BlobStorageFileRefRecordReadType = z.infer<
typeof blobStorageFileRefRecordReadSchema
>;
export const blobStorageFileLogRecordInsertSchema =
blobStorageFileLogRecordBaseSchema.extend({
created_at: z.number(),
updated_at: z.number(),
event_ts: z.number(),
});
export type BlobStorageFileLogInsertType = z.infer<
typeof blobStorageFileLogRecordInsertSchema
>;
export const convertTraceReadToInsert = (
@@ -345,6 +353,7 @@ export const convertPostgresScoreToInsert = (
timestamp: score.timestamp?.getTime(),
project_id: score.project_id,
trace_id: score.trace_id,
session_id: null,
observation_id: score.observation_id,
environment: score.environment,
name: score.name,
@@ -9,5 +9,6 @@ export * from "./observations_converters";
export * from "./clickhouse";
export * from "./constants";
export * from "./trace-sessions";
export * from "./eventLog";
export * from "./scores-utils";
export * from "./blobStorageLog";
export * from "./environments";
@@ -171,6 +171,37 @@ export const getObservationsForTrace = async (
},
});
// Large number of observations in trace with large input / output / metadata will lead to
// high CPU and memory consumption in the convertObservation step, where parsing occurs
// Thus, limit the size of the payload to 5MB, follows NextJS response size limitation:
// https://nextjs.org/docs/messages/api-routes-response-size-limit
// See also LFE-4882 for more details
let payloadSize = 0;
for (const observation of records) {
for (const key of ["input", "output"] as const) {
const value = observation[key];
if (value && typeof value === "string") {
payloadSize += value.length;
}
}
const metadataValues = Object.values(observation["metadata"]);
metadataValues.forEach((value) => {
if (value && typeof value === "string") {
payloadSize += value.length;
}
});
if (payloadSize >= env.LANGFUSE_API_TRACE_OBSERVATIONS_SIZE_LIMIT_BYTES) {
const errorMessage = `Observations in trace are too large: ${(payloadSize / 1e6).toFixed(2)}MB exceeds limit of ${(env.LANGFUSE_API_TRACE_OBSERVATIONS_SIZE_LIMIT_BYTES / 1e6).toFixed(2)}MB`;
throw new Error(errorMessage);
}
}
return records.map(convertObservation);
};
@@ -356,7 +387,7 @@ const getObservationByIdInternal = async (
FROM observations
WHERE id = {id: String}
AND project_id = {projectId: String}
${startTime ? `AND start_time = {startTime: DateTime64(3)}` : ""}
${startTime ? `AND toDate(start_time) = toDate({startTime: DateTime64(3)})` : ""}
ORDER BY event_ts desc
LIMIT 1 by id, project_id`;
return await queryClickhouse<ObservationRecordReadType>({
@@ -552,8 +583,8 @@ const getObservationsTableInternal = async <T>(
.includes(f.column),
);
const hasScoresFilter = filter.some(
(f) => f.column === "Scores" || f.column === "scores",
const hasScoresFilter = filter.some((f) =>
f.column.toLowerCase().includes("scores"),
);
const orderByTraces = opts.orderBy
@@ -598,17 +629,28 @@ const getObservationsTableInternal = async <T>(
const search = clickhouseSearchCondition(opts.searchQuery);
const scoresCte = `WITH scores_avg AS (
const scoresCte = `WITH scores_agg AS (
SELECT
trace_id,
observation_id,
groupArray(tuple(name, avg_value)) AS "scores_avg"
-- For numeric scores, use tuples of (name, avg_value)
groupArrayIf(
tuple(name, avg_value),
data_type IN ('NUMERIC', 'BOOLEAN')
) AS scores_avg,
-- For categorical scores, use name:value format for improved query performance
groupArrayIf(
concat(name, ':', string_value),
data_type = 'CATEGORICAL' AND notEmpty(string_value)
) AS score_categories
FROM (
SELECT
trace_id,
observation_id,
name,
avg(value) avg_value,
string_value,
data_type,
comment
FROM
scores final
@@ -617,6 +659,8 @@ const getObservationsTableInternal = async <T>(
trace_id,
observation_id,
name,
string_value,
data_type,
comment
ORDER BY
trace_id
@@ -652,7 +696,7 @@ const getObservationsTableInternal = async <T>(
${selectString}
FROM observations o
${traceTableFilter.length > 0 || orderByTraces || search.query ? "LEFT JOIN traces t FINAL ON t.id = o.trace_id AND t.project_id = o.project_id" : ""}
${hasScoresFilter ? `LEFT JOIN scores_avg AS s_avg ON s_avg.trace_id = o.trace_id and s_avg.observation_id = o.id` : ""}
${hasScoresFilter ? `LEFT JOIN scores_agg AS s ON s.trace_id = o.trace_id and s.observation_id = o.id` : ""}
WHERE ${appliedObservationsFilter.query}
${timeFilter && (traceTableFilter.length > 0 || orderByTraces) ? `AND t.timestamp > {tracesTimestampFilter: DateTime64(3)} - ${OBSERVATIONS_TO_TRACE_INTERVAL}` : ""}
@@ -998,18 +1042,18 @@ export const deleteObservationsByProjectId = async (projectId: string) => {
export const deleteObservationsOlderThanDays = async (
projectId: string,
days: number,
beforeDate: Date,
) => {
const query = `
DELETE FROM observations
WHERE project_id = {projectId: String}
AND start_time < now() - INTERVAL {numDays: Int} DAYS;
AND start_time < {cutoffDate: DateTime64(3)};
`;
await commandClickhouse({
query: query,
params: {
projectId,
numDays: days,
cutoffDate: convertDateToClickhouseDateTime(beforeDate),
},
clickhouseConfigs: {
request_timeout: 120_000, // 2 minutes
@@ -0,0 +1,93 @@
import { ScoreSourceType } from "../../domain";
import { queryClickhouse } from "./clickhouse";
import { ScoreRecordReadType } from "./definitions";
import { convertToScore } from "./scores_converters";
/**
* @internal
* Internal utility function for getting scores by ID.
* Do not use directly - use ScoresApiService or repository functions instead.
*/
export const _handleGetScoreById = async ({
projectId,
scoreId,
source,
scoreScope,
}: {
projectId: string;
scoreId: string;
source?: ScoreSourceType;
scoreScope: "traces_only" | "all";
}) => {
const query = `
SELECT *
FROM scores s
WHERE s.project_id = {projectId: String}
AND s.id = {scoreId: String}
${source ? `AND s.source = {source: String}` : ""}
${scoreScope === "traces_only" ? "AND s.session_id IS NULL" : ""}
ORDER BY s.event_ts DESC
LIMIT 1 BY s.id, s.project_id
LIMIT 1
`;
const rows = await queryClickhouse<ScoreRecordReadType>({
query,
params: {
projectId,
scoreId,
...(source !== undefined ? { source } : {}),
},
tags: {
feature: "tracing",
type: "score",
kind: "byId",
projectId,
},
});
return rows.map(convertToScore).shift();
};
/**
* @internal
* Internal utility function for getting scores by ID.
* Do not use directly - use ScoresApiService or repository functions instead.
*/
export const _handleGetScoresByIds = async ({
projectId,
scoreId,
source,
scoreScope,
}: {
projectId: string;
scoreId: string[];
source?: ScoreSourceType;
scoreScope: "traces_only" | "all";
}) => {
const query = `
SELECT *
FROM scores s
WHERE s.project_id = {projectId: String}
AND s.id IN ({scoreId: Array(String)})
${source ? `AND s.source = {source: String}` : ""}
${scoreScope === "traces_only" ? "AND s.session_id IS NULL" : ""}
ORDER BY s.event_ts DESC
LIMIT 1 BY s.id, s.project_id
`;
const rows = await queryClickhouse<ScoreRecordReadType>({
query,
params: {
projectId,
scoreId,
...(source !== undefined ? { source } : {}),
},
tags: {
feature: "tracing",
type: "score",
kind: "byId",
projectId,
},
});
return rows.map(convertToScore);
};
+190 -101
View File
@@ -27,19 +27,22 @@ import { SCORE_TO_TRACE_OBSERVATIONS_INTERVAL } from "./constants";
import { convertDateToClickhouseDateTime } from "../clickhouse/client";
import { ScoreRecordReadType } from "./definitions";
import { env } from "../../env";
import { _handleGetScoreById, _handleGetScoresByIds } from "./scores-utils";
import { parseMetadataCHRecordToDomain } from "../utils/metadata_conversion";
import { ClickHouseClientConfigOptions } from "@clickhouse/client";
export const searchExistingAnnotationScore = async (
projectId: string,
traceId: string,
observationId: string | null,
traceId: string | null,
sessionId: string | null,
name: string | undefined,
configId: string | undefined,
) => {
if (!name && !configId) {
throw new Error("Either name or configId (or both) must be provided.");
}
const query = `
SELECT *
FROM scores s
@@ -76,37 +79,21 @@ export const searchExistingAnnotationScore = async (
return rows.map((row) => convertToScore(row)).shift();
};
export const getScoreById = async (
projectId: string,
scoreId: string,
source?: ScoreSourceType,
) => {
const query = `
SELECT *
FROM scores s
WHERE s.project_id = {projectId: String}
AND s.id = {scoreId: String}
${source ? `AND s.source = {source: String}` : ""}
ORDER BY s.event_ts DESC
LIMIT 1 BY s.id, s.project_id
LIMIT 1
`;
const rows = await queryClickhouse<ScoreRecordReadType>({
query,
params: {
projectId,
scoreId,
...(source !== undefined ? { source } : {}),
},
tags: {
feature: "tracing",
type: "score",
kind: "byId",
projectId,
},
export const getScoreById = async ({
projectId,
scoreId,
source,
}: {
projectId: string;
scoreId: string;
source?: ScoreSourceType;
}) => {
return _handleGetScoreById({
projectId,
scoreId,
source,
scoreScope: "all",
});
return rows.map(convertToScore).shift();
};
export const getScoresByIds = async (
@@ -114,31 +101,12 @@ export const getScoresByIds = async (
scoreId: string[],
source?: ScoreSourceType,
) => {
const query = `
SELECT *
FROM scores s
WHERE s.project_id = {projectId: String}
AND s.id IN ({scoreId: Array(String)})
${source ? `AND s.source = {source: String}` : ""}
ORDER BY s.event_ts DESC
LIMIT 1 BY s.id, s.project_id
`;
const rows = await queryClickhouse<ScoreRecordReadType>({
query,
params: {
projectId,
scoreId,
...(source !== undefined ? { source } : {}),
},
tags: {
feature: "tracing",
type: "score",
kind: "byId",
projectId,
},
return _handleGetScoresByIds({
projectId,
scoreId,
source,
scoreScope: "all",
});
return rows.map(convertToScore);
};
/**
@@ -176,6 +144,82 @@ export type GetScoresForTracesProps<
includeHasMetadata?: IncludeHasMetadata;
};
type GetScoresForSessionsProps<
ExcludeMetadata extends boolean,
IncludeHasMetadata extends boolean,
> = {
projectId: string;
sessionIds: string[];
limit?: number;
offset?: number;
clickhouseConfigs?: ClickHouseClientConfigOptions;
excludeMetadata?: ExcludeMetadata;
includeHasMetadata?: IncludeHasMetadata;
};
const formatMetadataSelect = (
excludeMetadata: boolean,
includeHasMetadata: boolean,
) => {
return [
!excludeMetadata ? "*" : "* EXCEPT (metadata)",
includeHasMetadata
? "length(mapKeys(s.metadata)) > 0 AS has_metadata"
: null,
]
.filter((s) => s != null)
.join(", ");
};
export const getScoresForSessions = async <
ExcludeMetadata extends boolean,
IncludeHasMetadata extends boolean,
>(
props: GetScoresForSessionsProps<ExcludeMetadata, IncludeHasMetadata>,
) => {
const {
projectId,
sessionIds,
limit,
offset,
clickhouseConfigs,
excludeMetadata = false,
includeHasMetadata = false,
} = props;
const select = formatMetadataSelect(excludeMetadata, includeHasMetadata);
const query = `
select
${select}
from scores s
WHERE s.project_id = {projectId: String}
AND s.session_id IN ({sessionIds: Array(String)})
ORDER BY s.event_ts DESC
LIMIT 1 BY s.id, s.project_id
${limit && offset ? `limit {limit: Int32} offset {offset: Int32}` : ""}
`;
const rows = await queryClickhouse<ScoreRecordReadType>({
query: query,
params: {
projectId,
sessionIds,
limit,
offset,
},
tags: {
feature: "sessions",
type: "score",
kind: "list",
projectId,
},
clickhouseConfigs,
});
return rows.map(convertToScore);
};
// Used in multiple places, including the public API, hence the non-default exclusion of metadata via excludeMetadata flag
export const getScoresForTraces = async <
ExcludeMetadata extends boolean,
@@ -194,14 +238,7 @@ export const getScoresForTraces = async <
includeHasMetadata = false,
} = props;
const select = [
!excludeMetadata ? "*" : "* EXCEPT (metadata)",
includeHasMetadata
? "length(mapKeys(s.metadata)) > 0 AS has_metadata"
: null,
]
.filter((s) => s != null)
.join(", ");
const select = formatMetadataSelect(excludeMetadata, includeHasMetadata);
const query = `
select
@@ -389,7 +426,7 @@ export const getScoresGroupedByNameSourceType = async (
}));
};
export const getScoresGroupedByName = async (
export const getNumericScoresGroupedByName = async (
projectId: string,
timestampFilter?: FilterState,
) => {
@@ -441,6 +478,58 @@ export const getScoresGroupedByName = async (
return rows;
};
export const getCategoricalScoresGroupedByName = async (
projectId: string,
timestampFilter?: FilterState,
) => {
const chFilter = timestampFilter
? createFilterFromFilterState(timestampFilter, [
{
uiTableName: "Timestamp",
uiTableId: "timestamp",
clickhouseTableName: "scores",
clickhouseSelect: "timestamp",
},
])
: undefined;
const timestampFilterRes = chFilter
? new FilterList(chFilter).apply()
: undefined;
const query = `
SELECT
name AS label,
groupArray(DISTINCT string_value) AS values
FROM scores s
WHERE s.project_id = {projectId: String}
AND s.data_type = 'CATEGORICAL'
${timestampFilterRes?.query ? `AND ${timestampFilterRes.query}` : ""}
GROUP BY name
ORDER BY count() DESC
LIMIT 1000;
`;
const rows = await queryClickhouse<{
label: string;
values: string[];
}>({
query: query,
params: {
projectId: projectId,
...(timestampFilterRes ? timestampFilterRes.params : {}),
},
tags: {
feature: "tracing",
type: "score",
kind: "list",
projectId,
},
});
return rows;
};
export const getScoresUiCount = async (props: {
projectId: string;
filter: FilterState;
@@ -495,8 +584,9 @@ export async function getScoresUiTable<
source: string;
data_type: string;
comment: string | null;
trace_id: string | null;
session_id: string | null;
metadata: ExcludeMetadata extends true ? never : Record<string, string>;
trace_id: string;
observation_id: string | null;
author_user_id: string | null;
user_id: string | null;
@@ -520,39 +610,38 @@ export async function getScoresUiTable<
...rest,
});
return rows.map((row) => {
return {
projectId: row.project_id,
environment: row.environment,
authorUserId: row.author_user_id,
traceId: row.trace_id,
observationId: row.observation_id,
traceUserId: row.user_id,
traceName: row.trace_name,
traceTags: row.trace_tags,
configId: row.config_id,
queueId: row.queue_id,
createdAt: parseClickhouseUTCDateTimeFormat(row.created_at),
updatedAt: parseClickhouseUTCDateTimeFormat(row.updated_at),
stringValue: row.string_value,
comment: row.comment,
dataType: row.data_type as ScoreDataType,
source: row.source as ScoreSourceType,
name: row.name,
value: row.value,
timestamp: parseClickhouseUTCDateTimeFormat(row.timestamp),
id: row.id,
metadata: (excludeMetadata
? undefined
: (parseMetadataCHRecordToDomain(row.metadata ?? {}) ??
{})) as ExcludeMetadata extends true
? never
: NonNullable<ReturnType<typeof parseMetadataCHRecordToDomain>>,
hasMetadata: (includeHasMetadataFlag
? !!row.has_metadata
: undefined) as IncludeHasMetadata extends true ? boolean : never,
};
});
return rows.map((row) => ({
projectId: row.project_id,
environment: row.environment,
authorUserId: row.author_user_id,
traceId: row.trace_id,
sessionId: row.session_id,
observationId: row.observation_id,
traceUserId: row.user_id,
traceName: row.trace_name,
traceTags: row.trace_tags,
configId: row.config_id,
queueId: row.queue_id,
createdAt: parseClickhouseUTCDateTimeFormat(row.created_at),
updatedAt: parseClickhouseUTCDateTimeFormat(row.updated_at),
stringValue: row.string_value,
comment: row.comment,
dataType: row.data_type as ScoreDataType,
source: row.source as ScoreSourceType,
name: row.name,
value: row.value,
timestamp: parseClickhouseUTCDateTimeFormat(row.timestamp),
id: row.id,
metadata: (excludeMetadata
? undefined
: (parseMetadataCHRecordToDomain(row.metadata ?? {}) ??
{})) as ExcludeMetadata extends true
? never
: NonNullable<ReturnType<typeof parseMetadataCHRecordToDomain>>,
hasMetadata: (includeHasMetadataFlag
? !!row.has_metadata
: undefined) as IncludeHasMetadata extends true ? boolean : never,
}));
}
const getScoresUiGeneric = async <T>(props: {
@@ -775,18 +864,18 @@ export const deleteScoresByProjectId = async (projectId: string) => {
export const deleteScoresOlderThanDays = async (
projectId: string,
days: number,
beforeDate: Date,
) => {
const query = `
DELETE FROM scores
WHERE project_id = {projectId: String}
AND timestamp < now() - INTERVAL {numDays: Int} DAYS;
AND timestamp < {cutoffDate: DateTime64(3)};
`;
await commandClickhouse({
query: query,
params: {
projectId,
numDays: days,
cutoffDate: convertDateToClickhouseDateTime(beforeDate),
},
clickhouseConfigs: {
request_timeout: 120_000, // 2 minutes
@@ -19,7 +19,8 @@ export const convertToScore = (row: ScoreRecordReadType): ScoreDomain => {
timestamp: new Date(row.timestamp),
projectId: row.project_id,
environment: row.environment,
traceId: row.trace_id,
traceId: row.trace_id ?? null,
sessionId: row.session_id ?? null,
observationId: row.observation_id ?? null,
name: row.name,
value: row.value ?? null,
@@ -302,17 +302,31 @@ export const getTraceCountOfProjectsSinceCreationDate = async ({
return Number(rows[0]?.count ?? 0);
};
export const getTraceById = async (
traceId: string,
projectId: string,
timestamp?: Date,
) => {
/**
* Retrieves a trace record by its ID and associated project ID, with optional filtering by timestamp range.
* If no timestamp filters are provided, runs two queries in parallel:
* 1. One with a 7-day fromTimestamp filter (typically faster)
* 2. One without any timestamp filters (complete but slower)
* Returns the first non-empty result.
*/
export const getTraceById = async ({
traceId,
projectId,
timestamp,
fromTimestamp,
}: {
traceId: string;
projectId: string;
timestamp?: Date;
fromTimestamp?: Date;
}) => {
const query = `
SELECT *
FROM traces
WHERE id = {traceId: String}
AND project_id = {projectId: String}
${timestamp ? `AND toDate(timestamp) = toDate({timestamp: DateTime64(3)})` : ""}
${fromTimestamp ? `AND timestamp >= {fromTimestamp: DateTime64(3)}` : ""}
ORDER BY event_ts DESC
LIMIT 1
`;
@@ -325,6 +339,9 @@ export const getTraceById = async (
...(timestamp
? { timestamp: convertDateToClickhouseDateTime(timestamp) }
: {}),
...(fromTimestamp
? { fromTimestamp: convertDateToClickhouseDateTime(fromTimestamp) }
: {}),
},
tags: {
feature: "tracing",
@@ -502,7 +519,8 @@ export const getTracesIdentifierForSession = async (
user_id,
name,
timestamp,
project_id
project_id,
environment
FROM traces
WHERE (project_id = {projectId: String})
AND (session_id = {sessionId: String})
@@ -515,6 +533,7 @@ export const getTracesIdentifierForSession = async (
user_id: string;
name: string;
timestamp: string;
environment: string;
}>({
query: query,
params: {
@@ -534,6 +553,7 @@ export const getTracesIdentifierForSession = async (
userId: row.user_id,
name: row.name,
timestamp: parseClickhouseUTCDateTimeFormat(row.timestamp),
environment: row.environment,
}));
};
@@ -563,18 +583,18 @@ export const deleteTraces = async (projectId: string, traceIds: string[]) => {
export const deleteTracesOlderThanDays = async (
projectId: string,
days: number,
beforeDate: Date,
) => {
const query = `
DELETE FROM traces
WHERE project_id = {projectId: String}
AND timestamp < now() - INTERVAL {numDays: Int} DAYS;
AND timestamp < {cutoffDate: DateTime64(3)};
`;
await commandClickhouse({
query: query,
params: {
projectId,
numDays: days,
cutoffDate: convertDateToClickhouseDateTime(beforeDate),
},
clickhouseConfigs: {
request_timeout: 120_000, // 2 minutes
@@ -987,3 +1007,32 @@ export const getTracesByIdsForAnyProject = async (traceIds: string[]) => {
projectId: record.project_id,
}));
};
export const traceWithSessionIdExists = async (
projectId: string,
sessionId: string,
) => {
const query = `
SELECT id, project_id
FROM traces
WHERE session_id = {sessionId: String}
AND project_id = {projectId: String}
LIMIT 1
`;
const result = await queryClickhouse<{ id: string; project_id: string }>({
query,
params: {
sessionId,
projectId,
},
tags: {
feature: "tracing",
type: "trace",
kind: "exists",
projectId,
},
});
return result.length > 0;
};
+36
View File
@@ -0,0 +1,36 @@
import { env } from "../../env";
import {
StorageService,
StorageServiceFactory,
} from "../services/StorageService";
let s3MediaStorageClient: StorageService;
let s3EventStorageClient: StorageService;
export const getS3MediaStorageClient = (bucketName: string): StorageService => {
if (!s3MediaStorageClient) {
s3MediaStorageClient = StorageServiceFactory.getInstance({
bucketName,
accessKeyId: env.LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID,
secretAccessKey: env.LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY,
endpoint: env.LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT,
region: env.LANGFUSE_S3_MEDIA_UPLOAD_REGION,
forcePathStyle: env.LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE === "true",
});
}
return s3MediaStorageClient;
};
export const getS3EventStorageClient = (bucketName: string): StorageService => {
if (!s3EventStorageClient) {
s3EventStorageClient = StorageServiceFactory.getInstance({
bucketName,
accessKeyId: env.LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID,
secretAccessKey: env.LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY,
endpoint: env.LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT,
region: env.LANGFUSE_S3_EVENT_UPLOAD_REGION,
forcePathStyle: env.LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE === "true",
});
}
return s3EventStorageClient;
};
@@ -9,6 +9,7 @@ const langfuseUrls = {
US: "https://us.cloud.langfuse.com",
EU: "https://cloud.langfuse.com",
STAGING: "https://staging.langfuse.com",
HIPAA: "https://hipaa.cloud.langfuse.com",
};
type SendMembershipInvitationParams = {
@@ -36,7 +37,7 @@ export const sendMembershipInvitationEmail = async ({
}: SendMembershipInvitationParams) => {
if (!env.EMAIL_FROM_ADDRESS || !env.SMTP_CONNECTION_URL) {
logger.error(
"Missing environment variables for sending membership invitation email."
"Missing environment variables for sending membership invitation email.",
);
return;
}
@@ -44,6 +45,7 @@ export const sendMembershipInvitationEmail = async ({
const getAuthURL = () =>
env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION === "US" ||
env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION === "EU" ||
env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION === "HIPAA" ||
env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION === "STAGING"
? langfuseUrls[env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION]
: env.NEXTAUTH_URL;
@@ -51,7 +53,7 @@ export const sendMembershipInvitationEmail = async ({
const authUrl = getAuthURL();
if (!authUrl) {
logger.error(
"Missing NEXTAUTH_URL or NEXT_PUBLIC_LANGFUSE_CLOUD_REGION environment variable."
"Missing NEXTAUTH_URL or NEXT_PUBLIC_LANGFUSE_CLOUD_REGION environment variable.",
);
return;
}
@@ -68,7 +70,7 @@ export const sendMembershipInvitationEmail = async ({
inviteLink: authUrl,
emailFromAddress: env.EMAIL_FROM_ADDRESS,
langfuseCloudRegion: env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION,
})
}),
);
await mailer.sendMail({
@@ -272,6 +272,7 @@ const getTracesTableGeneric = async <T>(props: FetchTracesTableProps) => {
os.debug_count as debug_count,
os.observation_count as observation_count,
s.scores_avg as scores_avg,
s.score_categories as score_categories,
t.public as public`;
break;
case "rows":
@@ -429,19 +430,35 @@ const getTracesTableGeneric = async <T>(props: FetchTracesTableProps) => {
SELECT
project_id,
trace_id,
groupArray(tuple(name, avg_value)) AS "scores_avg"
-- For numeric scores, use tuples of (name, avg_value)
groupArrayIf(
tuple(name, avg_value),
data_type IN ('NUMERIC', 'BOOLEAN')
) AS scores_avg,
-- For categorical scores, use name:value format for improved query performance
groupArrayIf(
concat(name, ':', string_value),
data_type = 'CATEGORICAL' AND notEmpty(string_value)
) AS score_categories
FROM (
SELECT project_id,
trace_id,
name,
avg(value) avg_value
SELECT
project_id,
trace_id,
name,
data_type,
string_value,
avg(value) as avg_value
FROM scores s FINAL
WHERE project_id = {projectId: String}
${timeStampFilter ? `AND s.timestamp >= {traceTimestamp: DateTime64(3)} - ${SCORE_TO_TRACE_OBSERVATIONS_INTERVAL}` : ""}
${scoresFilterRes ? `AND ${scoresFilterRes.query}` : ""}
GROUP BY project_id,
trace_id,
name
WHERE
project_id = {projectId: String}
${timeStampFilter ? `AND s.timestamp >= {traceTimestamp: DateTime64(3)} - ${SCORE_TO_TRACE_OBSERVATIONS_INTERVAL}` : ""}
${scoresFilterRes ? `AND ${scoresFilterRes.query}` : ""}
GROUP BY
project_id,
trace_id,
name,
data_type,
string_value
) tmp
GROUP BY project_id, trace_id
)
@@ -74,7 +74,7 @@ export const createObservation = (
};
};
export const createScore = (
export const createTraceScore = (
score: Partial<ScoreRecordInsertType>,
): ScoreRecordInsertType => {
return {
@@ -95,5 +95,31 @@ export const createScore = (
event_ts: Date.now(),
is_deleted: 0,
...score,
session_id: null,
};
};
export const createSessionScore = (
score: Partial<ScoreRecordInsertType>,
): ScoreRecordInsertType => {
return {
id: v4(),
project_id: v4(),
session_id: v4(),
environment: "default",
name: "test-session-score" + v4(),
timestamp: Date.now(),
value: 100.5,
source: "API",
comment: "comment",
metadata: { "test-key": "test-value" },
data_type: "NUMERIC" as const,
created_at: Date.now(),
updated_at: Date.now(),
event_ts: Date.now(),
is_deleted: 0,
...score,
observation_id: null,
trace_id: null,
};
};
@@ -176,8 +176,8 @@ export const observationsTableUiColumnDefinitions: UiColumnMappings = [
clickhouseTypeOverwrite: "Decimal64(3)",
},
{
uiTableName: "Usage",
uiTableId: "usage",
uiTableName: "Tokens",
uiTableId: "tokens",
clickhouseTableName: "observations",
clickhouseSelect:
"if(mapExists((k, v) -> (k = 'total'), usage_details), usage_details['total'], NULL)",
@@ -189,11 +189,25 @@ export const observationsTableUiColumnDefinitions: UiColumnMappings = [
clickhouseTableName: "observations",
clickhouseSelect: 'o."metadata"',
},
// Scores column duplicated to allow renaming column name. Will be removed once session storage cache is outdated
// Column names are cached in user sessions - changing them breaks existing filters
{
uiTableName: "Scores",
uiTableId: "scores",
clickhouseTableName: "observations",
clickhouseSelect: "s_avg.scores_avg",
clickhouseTableName: "scores",
clickhouseSelect: "s.scores_avg",
},
{
uiTableName: "Scores (numeric)",
uiTableId: "scores",
clickhouseTableName: "scores",
clickhouseSelect: "s.scores_avg",
},
{
uiTableName: "Scores (categorical)",
uiTableId: "scores",
clickhouseTableName: "scores",
clickhouseSelect: "s.score_categories",
},
{
uiTableName: "Version",
@@ -123,19 +123,33 @@ export const tracesTableUiColumnDefinitions: UiColumnMappings = [
clickhouseTypeOverwrite: "Decimal64(3)",
},
{
uiTableName: "Usage",
uiTableId: "usage",
uiTableName: "Tokens",
uiTableId: "tokens",
clickhouseTableName: "observations",
clickhouseSelect:
"if(mapExists((k, v) -> (k = 'total'), usage_details), usage_details['total'], NULL)",
clickhouseTypeOverwrite: "Decimal64(3)",
},
// Scores column duplicated to allow renaming column name. Will be removed once session storage cache is outdated
// Column names are cached in user sessions - changing them breaks existing filters
{
uiTableName: "Scores",
uiTableId: "scores",
clickhouseTableName: "scores",
clickhouseSelect: "s.scores_avg",
},
{
uiTableName: "Scores (numeric)",
uiTableId: "scores",
clickhouseTableName: "scores",
clickhouseSelect: "s.scores_avg",
},
{
uiTableName: "Scores (categorical)",
uiTableId: "scores",
clickhouseTableName: "scores",
clickhouseSelect: "s.score_categories",
},
{
uiTableName: "Latency (s)",
uiTableId: "latency",
@@ -1,7 +1,8 @@
import {
type OptionsDefinition,
type ColumnDefinition,
type SingleValueOption,
} from "../tableDefinitions/types";
import { formatColumnOptions } from "./typeHelpers";
export const sessionsViewCols: ColumnDefinition[] = [
{ name: "⭐️", id: "bookmarked", type: "boolean", internal: "s.bookmarked" },
@@ -89,19 +90,19 @@ export const sessionsViewCols: ColumnDefinition[] = [
];
export type SessionOptions = {
userIds: Array<OptionsDefinition>;
tags: Array<OptionsDefinition>;
userIds: Array<SingleValueOption>;
tags: Array<SingleValueOption>;
};
export function sessionsTableColsWithOptions(
options?: SessionOptions
options?: SessionOptions,
): ColumnDefinition[] {
return sessionsViewCols.map((col) => {
if (col.id === "userIds") {
return { ...col, options: options?.userIds ?? [] };
return formatColumnOptions(col, options?.userIds ?? []);
}
if (col.id === "tags") {
return { ...col, options: options?.tags ?? [] };
return formatColumnOptions(col, options?.tags ?? []);
}
return col;
});
@@ -1,4 +1,10 @@
import { ColumnDefinition, ObservationLevelType, OptionsDefinition } from "..";
import {
type ColumnDefinition,
type MultiValueOption,
type ObservationLevelType,
type SingleValueOption,
} from "..";
import { formatColumnOptions } from "./typeHelpers";
export const tracesOnlyCols: ColumnDefinition[] = [
{
@@ -131,18 +137,26 @@ export const tracesTableCols: ColumnDefinition[] = [
nullable: true,
},
{
name: "Usage",
id: "usage",
name: "Tokens",
id: "tokens",
type: "number",
internal: 'generation_metrics."totalTokens"',
nullable: true,
},
{
name: "Scores",
name: "Scores (numeric)",
id: "scores_avg",
type: "numberObject",
internal: "scores_avg",
},
{
name: "Scores (categorical)",
id: "score_categories",
type: "categoryOptions",
internal: "score_categories",
options: [], // to be filled in at runtime
nullable: true,
},
{
name: "Latency (s)",
id: "latency",
@@ -187,12 +201,13 @@ export const evalTraceTableCols: ColumnDefinition[] = tracesOnlyCols;
export const evalDatasetFormFilterCols: ColumnDefinition[] = datasetOnlyCols;
export type TraceOptions = {
scores_avg?: Array<string>;
name?: Array<OptionsDefinition>;
tags?: Array<OptionsDefinition>;
environment?: Array<OptionsDefinition>;
score_categories?: Array<MultiValueOption>;
name?: Array<SingleValueOption>;
tags?: Array<SingleValueOption>;
environment?: Array<SingleValueOption>;
};
export type DatasetOptions = {
datasetId: Array<OptionsDefinition>;
datasetId: Array<SingleValueOption>;
};
// Used only for dataset evaluator, not on dataset table
@@ -202,7 +217,7 @@ export function datasetFormFilterColsWithOptions(
): ColumnDefinition[] {
return cols.map((col) => {
if (col.id === "datasetId") {
return { ...col, options: options?.datasetId ?? [] };
return formatColumnOptions(col, options?.datasetId ?? []);
}
return col;
});
@@ -214,16 +229,19 @@ export function tracesTableColsWithOptions(
): ColumnDefinition[] {
return cols.map((col) => {
if (col.id === "scores_avg") {
return { ...col, keyOptions: options?.scores_avg ?? [] };
return formatColumnOptions(col, options?.scores_avg ?? []);
}
if (col.id === "name") {
return { ...col, options: options?.name ?? [] };
return formatColumnOptions(col, options?.name ?? []);
}
if (col.id === "tags") {
return { ...col, options: options?.tags ?? [] };
return formatColumnOptions(col, options?.tags ?? []);
}
if (col.id === "environment") {
return { ...col, options: options?.environment ?? [] };
return formatColumnOptions(col, options?.environment ?? []);
}
if (col.id === "score_categories") {
return formatColumnOptions(col, options?.score_categories ?? []);
}
return col;
});
@@ -0,0 +1,21 @@
import { ColumnDefinition, MultiValueOption, SingleValueOption } from "./types";
// Generic helper function that infers the right option type based on column type
export function formatColumnOptions<T extends ColumnDefinition>(
col: T,
newOptions: T extends { type: "categoryOptions" }
? MultiValueOption[]
: T extends { type: "stringOptions" | "arrayOptions" }
? SingleValueOption[]
: T extends { type: "numberObject" | "stringObject" }
? string[]
: never,
): T {
// For numberObject type, set keyOptions instead of options
if (col.type === "numberObject" || col.type === "stringObject") {
return { ...col, keyOptions: newOptions };
}
// For other types, set options
return { ...col, options: newOptions };
}
+18 -3
View File
@@ -9,12 +9,19 @@ export type UiColumnMapping = Readonly<{
queryPrefix?: string;
}>;
export type OptionsDefinition = {
export type SingleValueOption = {
value: string;
count?: number;
displayValue?: string; // FIX: Temporary workaround: Used to display a different value than the actual value since multiSelect doesn't support key-value pairs
};
export type MultiValueOption = {
label: string;
values: string[];
};
export type OptionsDefinition = SingleValueOption | MultiValueOption;
export type ColumnDefinition =
| {
name: string;
@@ -27,7 +34,7 @@ export type ColumnDefinition =
name: string;
id: string;
type: "stringOptions";
options: Array<OptionsDefinition>;
options: Array<SingleValueOption>;
internal: string;
nullable?: boolean;
}
@@ -35,7 +42,7 @@ export type ColumnDefinition =
name: string;
id: string;
type: "arrayOptions";
options: Array<OptionsDefinition>;
options: Array<SingleValueOption>;
internal: string;
nullable?: boolean;
}
@@ -46,6 +53,14 @@ export type ColumnDefinition =
internal: string;
keyOptions?: Array<string>;
nullable?: boolean;
}
| {
name: string;
id: string;
type: "categoryOptions";
options: Array<MultiValueOption>;
internal: string;
nullable?: boolean;
};
export const tableNames = [
+22
View File
@@ -0,0 +1,22 @@
import z from "zod";
export const applyScoreValidation = <T extends z.ZodType<any, any, any>>(
schema: T,
) => {
return schema.refine(
(data) => {
const hasTraceId = !!data.traceId;
const hasSessionId = !!data.sessionId;
return (
(hasTraceId && !hasSessionId) ||
(hasSessionId && !hasTraceId && !data.observationId)
);
},
{
message:
"Either provide traceId (with optional observationId) or sessionId, but not both. ObservationId requires traceId.",
path: ["traceId", "sessionId", "observationId"],
},
);
};
+10
View File
@@ -22,6 +22,16 @@
"persistent": true,
"dependsOn": ["db:generate", "@langfuse/shared#build"]
},
"dev:worker": {
"cache": false,
"persistent": true,
"dependsOn": ["db:generate", "@langfuse/shared#build"]
},
"dev:web": {
"cache": false,
"persistent": true,
"dependsOn": ["db:generate", "@langfuse/shared#build"]
},
"db:generate": {
"cache": false,
"dependsOn": ["^db:generate"]
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "web",
"version": "3.51.2",
"version": "3.53.0",
"private": true,
"license": "MIT",
"engines": {
+7 -5
View File
@@ -56,7 +56,13 @@ components:
nullable: true
traceId:
type: string
example: cdef-1234-5678-90ab
nullable: true
sessionId:
type: string
nullable: true
observationId:
type: string
nullable: true
name:
type: string
example: novelty
@@ -66,9 +72,6 @@ components:
The value of the score. Must be passed as string for categorical
scores, and numeric for boolean and numeric scores. Boolean score
values must equal either 1 or 0 (true or false)
observationId:
type: string
nullable: true
comment:
type: string
nullable: true
@@ -90,7 +93,6 @@ components:
config category. Numeric scores might be constrained by the score
config's max and min values
required:
- traceId
- name
- value
BaseScore:
+299 -130
View File
@@ -2256,6 +2256,11 @@ paths:
properties:
name:
type: string
metadata:
type: object
additionalProperties: true
nullable: true
description: Optional metadata for the project
retention:
type: integer
description: >-
@@ -2320,6 +2325,11 @@ paths:
properties:
name:
type: string
metadata:
type: object
additionalProperties: true
nullable: true
description: Optional metadata for the project
retention:
type: integer
description: >-
@@ -3308,58 +3318,12 @@ paths:
schema: {}
security:
- BasicAuth: []
/api/public/scores:
post:
description: Create a score
operationId: score_create
tags:
- Score
parameters: []
responses:
'200':
description: ''
content:
application/json:
schema:
$ref: '#/components/schemas/CreateScoreResponse'
'400':
description: ''
content:
application/json:
schema: {}
'401':
description: ''
content:
application/json:
schema: {}
'403':
description: ''
content:
application/json:
schema: {}
'404':
description: ''
content:
application/json:
schema: {}
'405':
description: ''
content:
application/json:
schema: {}
security:
- BasicAuth: []
requestBody:
required: true
content:
application/json:
schema:
$ref: '#/components/schemas/CreateScoreRequest'
/api/public/v2/scores:
get:
description: Get a list of scores
operationId: score_get
description: Get a list of scores (supports both trace and session scores)
operationId: scoreV2_get
tags:
- Score
- ScoreV2
parameters:
- name: page
in: query
@@ -3517,12 +3481,12 @@ paths:
schema: {}
security:
- BasicAuth: []
/api/public/scores/{scoreId}:
/api/public/v2/scores/{scoreId}:
get:
description: Get a score
operationId: score_get-by-id
description: Get a score (supports both trace and session scores)
operationId: scoreV2_get-by-id
tags:
- Score
- ScoreV2
parameters:
- name: scoreId
in: path
@@ -3564,8 +3528,56 @@ paths:
schema: {}
security:
- BasicAuth: []
/api/public/scores:
post:
description: Create a score (supports both trace and session scores)
operationId: score_create
tags:
- Score
parameters: []
responses:
'200':
description: ''
content:
application/json:
schema:
$ref: '#/components/schemas/CreateScoreResponse'
'400':
description: ''
content:
application/json:
schema: {}
'401':
description: ''
content:
application/json:
schema: {}
'403':
description: ''
content:
application/json:
schema: {}
'404':
description: ''
content:
application/json:
schema: {}
'405':
description: ''
content:
application/json:
schema: {}
security:
- BasicAuth: []
requestBody:
required: true
content:
application/json:
schema:
$ref: '#/components/schemas/CreateScoreRequest'
/api/public/scores/{scoreId}:
delete:
description: Delete a score
description: Delete a score (supports both trace and session scores)
operationId: score_delete
tags:
- Score
@@ -4333,7 +4345,7 @@ components:
scores:
type: array
items:
$ref: '#/components/schemas/Score'
$ref: '#/components/schemas/ScoreV1'
description: List of scores
required:
- htmlPath
@@ -4647,8 +4659,8 @@ components:
required:
- value
- label
BaseScore:
title: BaseScore
BaseScoreV1:
title: BaseScoreV1
type: object
properties:
id:
@@ -4707,6 +4719,157 @@ components:
- timestamp
- createdAt
- updatedAt
NumericScoreV1:
title: NumericScoreV1
type: object
properties:
value:
type: number
format: double
description: The numeric value of the score
required:
- value
allOf:
- $ref: '#/components/schemas/BaseScoreV1'
BooleanScoreV1:
title: BooleanScoreV1
type: object
properties:
value:
type: number
format: double
description: >-
The numeric value of the score. Equals 1 for "True" and 0 for
"False"
stringValue:
type: string
description: >-
The string representation of the score value. Is inferred from the
numeric value and equals "True" or "False"
required:
- value
- stringValue
allOf:
- $ref: '#/components/schemas/BaseScoreV1'
CategoricalScoreV1:
title: CategoricalScoreV1
type: object
properties:
value:
type: number
format: double
nullable: true
description: >-
Only defined if a config is linked. Represents the numeric category
mapping of the stringValue
stringValue:
type: string
description: >-
The string representation of the score value. If no config is
linked, can be any string. Otherwise, must map to a config category
required:
- stringValue
allOf:
- $ref: '#/components/schemas/BaseScoreV1'
ScoreV1:
title: ScoreV1
oneOf:
- type: object
allOf:
- type: object
properties:
dataType:
type: string
enum:
- NUMERIC
- $ref: '#/components/schemas/NumericScoreV1'
required:
- dataType
- type: object
allOf:
- type: object
properties:
dataType:
type: string
enum:
- CATEGORICAL
- $ref: '#/components/schemas/CategoricalScoreV1'
required:
- dataType
- type: object
allOf:
- type: object
properties:
dataType:
type: string
enum:
- BOOLEAN
- $ref: '#/components/schemas/BooleanScoreV1'
required:
- dataType
BaseScore:
title: BaseScore
type: object
properties:
id:
type: string
traceId:
type: string
nullable: true
sessionId:
type: string
nullable: true
observationId:
type: string
nullable: true
name:
type: string
source:
$ref: '#/components/schemas/ScoreSource'
timestamp:
type: string
format: date-time
createdAt:
type: string
format: date-time
updatedAt:
type: string
format: date-time
authorUserId:
type: string
nullable: true
comment:
type: string
nullable: true
metadata:
nullable: true
configId:
type: string
nullable: true
description: >-
Reference a score config on a score. When set, config and score name
must be equal and value must comply to optionally defined numerical
range
queueId:
type: string
nullable: true
description: >-
Reference an annotation queue on a score. Populated if the score was
initially created in an annotation queue.
environment:
type: string
nullable: true
description: >-
The environment from which this score originated. Can be any
lowercase alphanumeric string with hyphens and underscores that does
not start with 'langfuse'.
required:
- id
- name
- source
- timestamp
- createdAt
- updatedAt
NumericScore:
title: NumericScore
type: object
@@ -5652,7 +5815,13 @@ components:
nullable: true
traceId:
type: string
example: cdef-1234-5678-90ab
nullable: true
sessionId:
type: string
nullable: true
observationId:
type: string
nullable: true
name:
type: string
example: novelty
@@ -5665,9 +5834,6 @@ components:
The value of the score. Must be passed as string for categorical
scores, and numeric for boolean and numeric scores. Boolean score
values must equal either 1 or 0 (true or false)
observationId:
type: string
nullable: true
comment:
type: string
nullable: true
@@ -5689,7 +5855,6 @@ components:
config category. Numeric scores might be constrained by the score
config's max and min values
required:
- traceId
- name
- value
BaseEvent:
@@ -6282,6 +6447,10 @@ components:
type: string
name:
type: string
metadata:
type: object
additionalProperties: true
description: Metadata for the project
retentionDays:
type: integer
description: >-
@@ -6290,6 +6459,7 @@ components:
required:
- id
- name
- metadata
- retentionDays
ProjectDeletionResponse:
title: ProjectDeletionResponse
@@ -6952,67 +7122,6 @@ components:
required:
- name
- dataType
CreateScoreRequest:
title: CreateScoreRequest
type: object
properties:
id:
type: string
nullable: true
traceId:
type: string
example: cdef-1234-5678-90ab
name:
type: string
example: novelty
value:
$ref: '#/components/schemas/CreateScoreValue'
description: >-
The value of the score. Must be passed as string for categorical
scores, and numeric for boolean and numeric scores. Boolean score
values must equal either 1 or 0 (true or false)
observationId:
type: string
nullable: true
comment:
type: string
nullable: true
metadata:
nullable: true
environment:
type: string
nullable: true
description: >-
The environment of the score. Can be any lowercase alphanumeric
string with hyphens and underscores that does not start with
'langfuse'.
dataType:
$ref: '#/components/schemas/ScoreDataType'
nullable: true
description: >-
The data type of the score. When passing a configId this field is
inferred. Otherwise, this field must be passed or will default to
numeric.
configId:
type: string
nullable: true
description: >-
Reference a score config on a score. The unique langfuse identifier
of a score config. When passing this field, the dataType and
stringValue fields are automatically populated.
required:
- traceId
- name
- value
CreateScoreResponse:
title: CreateScoreResponse
type: object
properties:
id:
type: string
description: The id of the created object in Langfuse
required:
- id
GetScoresResponseTraceData:
title: GetScoresResponseTraceData
type: object
@@ -7037,8 +7146,7 @@ components:
properties:
trace:
$ref: '#/components/schemas/GetScoresResponseTraceData'
required:
- trace
nullable: true
allOf:
- $ref: '#/components/schemas/NumericScore'
GetScoresResponseDataCategorical:
@@ -7047,8 +7155,7 @@ components:
properties:
trace:
$ref: '#/components/schemas/GetScoresResponseTraceData'
required:
- trace
nullable: true
allOf:
- $ref: '#/components/schemas/CategoricalScore'
GetScoresResponseDataBoolean:
@@ -7057,8 +7164,7 @@ components:
properties:
trace:
$ref: '#/components/schemas/GetScoresResponseTraceData'
required:
- trace
nullable: true
allOf:
- $ref: '#/components/schemas/BooleanScore'
GetScoresResponseData:
@@ -7110,6 +7216,69 @@ components:
required:
- data
- meta
CreateScoreRequest:
title: CreateScoreRequest
type: object
properties:
id:
type: string
nullable: true
traceId:
type: string
nullable: true
sessionId:
type: string
nullable: true
observationId:
type: string
nullable: true
name:
type: string
example: novelty
value:
$ref: '#/components/schemas/CreateScoreValue'
description: >-
The value of the score. Must be passed as string for categorical
scores, and numeric for boolean and numeric scores. Boolean score
values must equal either 1 or 0 (true or false)
comment:
type: string
nullable: true
metadata:
nullable: true
environment:
type: string
nullable: true
description: >-
The environment of the score. Can be any lowercase alphanumeric
string with hyphens and underscores that does not start with
'langfuse'.
dataType:
$ref: '#/components/schemas/ScoreDataType'
nullable: true
description: >-
The data type of the score. When passing a configId this field is
inferred. Otherwise, this field must be passed or will default to
numeric.
configId:
type: string
nullable: true
description: >-
Reference a score config on a score. The unique langfuse identifier
of a score config. When passing this field, the dataType and
stringValue fields are automatically populated.
required:
- name
- value
CreateScoreResponse:
title: CreateScoreResponse
type: object
properties:
id:
type: string
description: The id of the created object in Langfuse
required:
- id
PaginatedSessions:
title: PaginatedSessions
type: object
@@ -126,6 +126,11 @@ paths:
properties:
name:
type: string
metadata:
type: object
additionalProperties: true
nullable: true
description: Optional metadata for the organization
required:
- name
/api/admin/organizations/{organizationId}:
@@ -224,6 +229,11 @@ paths:
properties:
name:
type: string
metadata:
type: object
additionalProperties: true
nullable: true
description: Optional metadata for the organization
required:
- name
delete:
@@ -441,10 +451,15 @@ components:
createdAt:
type: string
format: date-time
metadata:
type: object
additionalProperties: true
description: Metadata for the organization
required:
- id
- name
- createdAt
- metadata
DeleteOrganizationResponse:
title: DeleteOrganizationResponse
type: object
@@ -80,7 +80,7 @@
"auth": null,
"body": {
"mode": "raw",
"raw": "{\n \"name\": \"example\"\n}",
"raw": "{\n \"name\": \"example\",\n \"metadata\": {\n \"example\": \"UNKNOWN\"\n }\n}",
"options": {
"raw": {
"language": "json"
@@ -152,7 +152,7 @@
"auth": null,
"body": {
"mode": "raw",
"raw": "{\n \"name\": \"example\"\n}",
"raw": "{\n \"name\": \"example\",\n \"metadata\": {\n \"example\": \"UNKNOWN\"\n }\n}",
"options": {
"raw": {
"language": "json"
+50 -41
View File
@@ -1723,7 +1723,7 @@
"auth": null,
"body": {
"mode": "raw",
"raw": "{\n \"name\": \"example\",\n \"retention\": 0\n}",
"raw": "{\n \"name\": \"example\",\n \"metadata\": {\n \"example\": \"UNKNOWN\"\n },\n \"retention\": 0\n}",
"options": {
"raw": {
"language": "json"
@@ -1763,7 +1763,7 @@
"auth": null,
"body": {
"mode": "raw",
"raw": "{\n \"name\": \"example\",\n \"retention\": 0\n}",
"raw": "{\n \"name\": \"example\",\n \"metadata\": {\n \"example\": \"UNKNOWN\"\n },\n \"retention\": 0\n}",
"options": {
"raw": {
"language": "json"
@@ -2460,54 +2460,22 @@
{
"_type": "container",
"description": null,
"name": "Score",
"name": "Score V 2",
"item": [
{
"_type": "endpoint",
"name": "Create",
"request": {
"description": "Create a score",
"url": {
"raw": "{{baseUrl}}/api/public/scores",
"host": [
"{{baseUrl}}"
],
"path": [
"api",
"public",
"scores"
],
"query": [],
"variable": []
},
"header": [],
"method": "POST",
"auth": null,
"body": {
"mode": "raw",
"raw": "{\n \"name\": \"novelty\",\n \"value\": 0.9,\n \"traceId\": \"cdef-1234-5678-90ab\"\n}",
"options": {
"raw": {
"language": "json"
}
}
}
},
"response": []
},
{
"_type": "endpoint",
"name": "Get",
"request": {
"description": "Get a list of scores",
"description": "Get a list of scores (supports both trace and session scores)",
"url": {
"raw": "{{baseUrl}}/api/public/scores?page=&limit=&userId=&name=&fromTimestamp=&toTimestamp=&environment=&source=&operator=&value=&scoreIds=&configId=&queueId=&dataType=&traceTags=",
"raw": "{{baseUrl}}/api/public/v2/scores?page=&limit=&userId=&name=&fromTimestamp=&toTimestamp=&environment=&source=&operator=&value=&scoreIds=&configId=&queueId=&dataType=&traceTags=",
"host": [
"{{baseUrl}}"
],
"path": [
"api",
"public",
"v2",
"scores"
],
"query": [
@@ -2600,15 +2568,16 @@
"_type": "endpoint",
"name": "Get By Id",
"request": {
"description": "Get a score",
"description": "Get a score (supports both trace and session scores)",
"url": {
"raw": "{{baseUrl}}/api/public/scores/:scoreId",
"raw": "{{baseUrl}}/api/public/v2/scores/:scoreId",
"host": [
"{{baseUrl}}"
],
"path": [
"api",
"public",
"v2",
"scores",
":scoreId"
],
@@ -2627,12 +2596,52 @@
"body": null
},
"response": []
}
]
},
{
"_type": "container",
"description": null,
"name": "Score",
"item": [
{
"_type": "endpoint",
"name": "Create",
"request": {
"description": "Create a score (supports both trace and session scores)",
"url": {
"raw": "{{baseUrl}}/api/public/scores",
"host": [
"{{baseUrl}}"
],
"path": [
"api",
"public",
"scores"
],
"query": [],
"variable": []
},
"header": [],
"method": "POST",
"auth": null,
"body": {
"mode": "raw",
"raw": "{\n \"name\": \"novelty\",\n \"value\": 0.9,\n \"traceId\": \"cdef-1234-5678-90ab\"\n}",
"options": {
"raw": {
"language": "json"
}
}
}
},
"response": []
},
{
"_type": "endpoint",
"name": "Delete",
"request": {
"description": "Delete a score",
"description": "Delete a score (supports both trace and session scores)",
"url": {
"raw": "{{baseUrl}}/api/public/scores/:scoreId",
"host": [

Some files were not shown because too many files have changed in this diff Show More