Compare commits

...
36 Commits
Author SHA1 Message Date
Marc Klingen c79c84ea96 chore: release v3.6.2 2025-01-10 14:49:19 +01:00
Hassieb PakzadandGitHub 4634669fcd chore(docker-compose): add ingestion env vars (#4969) 2025-01-10 14:47:25 +01:00
Marc KlingenandGitHub 549e0eb629 chore: daily instead of hourly telemetry (#4968) 2025-01-10 14:10:51 +01:00
Steffen SchmitzandGitHub 80adf63fe6 fix: adjust deletion order for apikey invalidation on project delete (#4964) 2025-01-10 10:24:26 +00:00
Hassieb Pakzad 6c1d7534bb chore: release v3.6.1 2025-01-10 09:15:57 +01:00
Max DeichmannandGitHub 23a4bbd2cb fix: fix trace exports (#4958)
* fix: fix trace exports

* push

* fix: fix trace exports
2025-01-09 19:36:07 +00:00
Steffen SchmitzandGitHub 07d048d46f chore: enable clickhouse cluster mode by default (#4956) 2025-01-09 16:32:55 +00:00
Hassieb PakzadandGitHub bd7226bbb8 fix(pubic-api): observations and traces FINAL (#4951) 2025-01-09 17:41:35 +01:00
Steffen SchmitzandGitHub fe70d92211 chore: set created_at in ClickHouse based on min S3 timestamp (#4955)
* chore: set created_at in ClickHouse based on min S3 timestamp

* chore: fix worker tests
2025-01-09 16:12:23 +00:00
Max DeichmannandGitHub bb3621f31d fix: sessions metrics api performance (#4954) 2025-01-09 17:15:30 +01:00
Steffen SchmitzandGitHub 63225a9fd6 chore: accept projectId list which skips clickhouse reads during ingestion (#4952) 2025-01-09 14:13:23 +00:00
Max DeichmannandGitHub 947da9cc84 fix: sessions table id filter (#4953) 2025-01-09 14:33:09 +01:00
Max DeichmannandGitHub 93a0ab6f3a fix: remove trace tracesession fk (#4946)
* fix: remove trace tracesession fk

* push

* push

* push

* fix: remove trace tracesession fk

* fix: remove trace tracesession fk

* fix: remove trace tracesession fk
2025-01-09 09:50:58 +00:00
Hassieb PakzadandGitHub 68ad079d8c fix(models-table): sorting across pages (#4947) 2025-01-09 10:30:00 +01:00
Max DeichmannandGitHub cf29c6b7e4 feat: backup redis queues to postgres (#4943)
feat: backup redis queuer
2025-01-08 23:15:03 +00:00
Max DeichmannandGitHub e4d2a5ec6c perf: improve session all time search performance (#4926)
* push

* fix: correctly export categorical scores for trace exports

* fixes

* push

* Merge branch 'max/lfe-3345-fix-session-table-all-time-search' of https://github.com/langfuse/langfuse into max/lfe-3345-fix-session-table-all-time-search

* fix

* fixes
2025-01-08 16:54:01 +00:00
Steffen SchmitzandGitHub 35f3092a1c feat: retry 429 errors for eval executions for up to 24h (#4933)
* feat: retry 429 errors for eval executions for up to 24h

* chore: skip type error retries for eval execution
2025-01-08 14:51:36 +00:00
Hassieb PakzadandGitHub 7b222318e1 chore(cost): add langchain-compatible price keys (#4932) 2025-01-08 15:49:21 +01:00
Steffen SchmitzandGitHub d9c077ea56 chore: add BUILD_ID to otel setup and exclude health from tracing (#4930)
* chore: add BUILD_ID to otel setup

* chore: remove comment
2025-01-08 13:17:00 +00:00
Hassieb PakzadandGitHub 1ec7325b85 feat(models): lastUsed column to full lookback (#4931) 2025-01-08 14:21:07 +01:00
Steffen SchmitzandGitHub 96b2127f36 chore: reduce number of parsing iterations for observations and traces (#4927)
* chore: reduce number of parsing iterations for observations and traces

* chore: lint
2025-01-08 12:36:37 +00:00
Max DeichmannandGitHub d3c2b8cf27 fix: correctly export categorical scores for trace exports (#4928)
* fix: correctly export categorical scores for trace exports

* push
2025-01-08 11:31:50 +00:00
Hassieb PakzadandGitHub beeef222e4 feat(models): sort models table by lastUsed (#4925) 2025-01-08 12:06:57 +01:00
steffen911 ee376f9cad chore: release v3.6.0 2025-01-07 14:43:35 +01:00
c8936ec68d feat: add CLICKHOUSE_DB support (#4911)
* take CLICKHOUSE_DB environment variable into account

Fixes: #4888
Signed-off-by: Aleksandar Pesic <peske.nis@gmail.com>

* chore: cleanup env usage

* chore: extend env.prod.example

---------

Signed-off-by: Aleksandar Pesic <peske.nis@gmail.com>
Co-authored-by: Aleksandar Pesic <peske.nis@gmail.com>
2025-01-07 13:18:03 +00:00
Max DeichmannandGitHub 8d8170c62c security: prevent prototype pollution in dataset compare view (#4910) 2025-01-07 12:15:33 +00:00
Max DeichmannandGitHub 72e1a43192 security: upgrade nanoid (#4909) 2025-01-07 11:47:31 +00:00
Hassieb PakzadandGitHub a842625bd1 chore(media): increase rate-limit to use ingestion budget (#4907) 2025-01-07 11:42:07 +01:00
Max DeichmannandGitHub 890a542759 fix: do not error on 403 by model provider for eval executions (#4905)
* fix: do not error on 403 by model provider

* push
2025-01-07 10:23:58 +00:00
Steffen SchmitzandGitHub 1e8af5b821 chore: move sessions, traces, comments router logs to logger (#4902)
* chore: move sessions router logs to logger

* chore: adjust traces and comments
2025-01-07 07:45:44 +00:00
Hassieb PakzadandGitHub d3f3f354db fix(models): allow string tokenizer config (#4898) 2025-01-06 16:24:49 +01:00
steffen911 18a7cb6684 chore: release v3.5.3 2025-01-06 13:43:19 +01:00
Steffen SchmitzandGitHub 2d70ca574e chore: skip existence check for score deletion (#4896)
* chore: skip existence check for score deletion

* chore: lint
2025-01-06 11:03:51 +00:00
Steffen SchmitzandGitHub 1c980c78dd fix: use singular unit in clickhouse datediff for 23.x support (#4894) 2025-01-06 10:35:53 +00:00
Marc KlingenGitHubMax Deichmanndependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>Steffen SchmitzHassieb Pakzad
5170718aad fix(prompt-experiments): do not restrict to evalModelsOnly in order to use anthropic, bedrock, vertex (#4872)
* fix(prompt-experiments): do not restrict to evalModelsOnly in order to use anthropic, bedrock, vertex

* chore(cloud): disable deletion of traces via UI as it caused performance issues (#4873)

chore(cloud): disable deletion of traces via the UI as it caused performance issues

* docs: add link to data model documentation to api reference of ingestion endpoint (#4877)

* fix: use database upserts correctly (#4884)

* fix: use database upserts correctly

* fix: use database upserts correctly

* push

* fix: use database upserts correctly

* fix: use database upserts correctly

* push

* fix: use database upserts correctly

* chore(deps): bump next from 14.2.15 to 14.2.21 (#4887)

Bumps [next](https://github.com/vercel/next.js) from 14.2.15 to 14.2.21.
- [Release notes](https://github.com/vercel/next.js/releases)
- [Changelog](https://github.com/vercel/next.js/blob/canary/release.js)
- [Commits](https://github.com/vercel/next.js/compare/v14.2.15...v14.2.21)

---
updated-dependencies:
- dependency-name: next
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* chore: release v3.5.2

* chore: remove postgres from ingestion merge (#4467)

* chore: remove postgres from ingestion merge

* chore: remove pg tests

* chore: remove postgres tracing

* remove evalModelsOnly

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: Max Deichmann <m.deichmann@tum.de>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Steffen Schmitz <steffen@langfuse.com>
Co-authored-by: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com>
2025-01-06 10:29:57 +00:00
Steffen SchmitzandGitHub a01c27f4cc chore: remove postgres from ingestion merge (#4467)
* chore: remove postgres from ingestion merge

* chore: remove pg tests

* chore: remove postgres tracing
2025-01-06 09:29:26 +00:00
73 changed files with 1384 additions and 888 deletions
+1
View File
@@ -245,6 +245,7 @@ OTEL_SERVICE_NAME="langfuse"
# CLICKHOUSE_URL=
# CLICKHOUSE_USER=
# CLICKHOUSE_PASSWORD=
# CLICKHOUSE_DB=
# Ingestion
# LANGFUSE_INGESTION_QUEUE_DELAY_MS=
+2
View File
@@ -38,6 +38,8 @@ services:
LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: ${LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT:-http://minio:9000}
LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE: ${LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE:-true}
LANGFUSE_S3_MEDIA_UPLOAD_PREFIX: ${LANGFUSE_S3_MEDIA_UPLOAD_PREFIX:-media/}
LANGFUSE_INGESTION_QUEUE_DELAY_MS: ${LANGFUSE_INGESTION_QUEUE_DELAY_MS:-}
LANGFUSE_INGESTION_CLICKHOUSE_WRITE_INTERVAL_MS: ${LANGFUSE_INGESTION_CLICKHOUSE_WRITE_INTERVAL_MS:-}
REDIS_HOST: ${REDIS_HOST:-redis}
REDIS_PORT: ${REDIS_PORT:-6379}
REDIS_AUTH: ${REDIS_AUTH:-myredissecret}
+2 -1
View File
@@ -47,7 +47,8 @@
},
"pnpm": {
"overrides": {
"jsonpath-plus": "10.0.7"
"jsonpath-plus": "10.0.7",
"nanoid": "^3.3.8"
}
}
}
+3 -2
View File
@@ -1,6 +1,6 @@
{
"name": "langfuse",
"version": "3.5.2",
"version": "3.6.2",
"author": "engineering@langfuse.com",
"license": "MIT",
"private": true,
@@ -84,7 +84,8 @@
"packageManager": "pnpm@9.5.0",
"pnpm": {
"overrides": {
"jsonpath-plus": "10.0.7"
"jsonpath-plus": "10.0.7",
"nanoid": "^3.3.8"
}
}
}
+14 -4
View File
@@ -18,21 +18,31 @@ then
exit 1
fi
# Ensure CLICKHOUSE_DB is set
if [ -z "${CLICKHOUSE_DB}" ]; then
export CLICKHOUSE_DB="default"
fi
# Ensure CLICKHOUSE_CLUSTER_NAME is set
if [ -z "${CLICKHOUSE_CLUSTER_NAME}" ]; then
export CLICKHOUSE_CLUSTER_NAME="default"
fi
# Construct the database URL
if [ "$CLICKHOUSE_CLUSTER_ENABLED" == "false" ] ; then
if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
else
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-migrations-table-engine=MergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
fi
# Execute the up command
migrate -source file://clickhouse/migrations/unclustered -database "$DATABASE_URL" down
else
if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
else
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
fi
# Execute the up command
+7 -2
View File
@@ -12,11 +12,16 @@ then
exit 1
fi
# Ensure CLICKHOUSE_DB is set
if [ -z "${CLICKHOUSE_DB}" ]; then
export CLICKHOUSE_DB="default"
fi
# Construct the database URL
if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
else
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-migrations-table-engine=MergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
fi
# Execute the drop command
migrate -source file://clickhouse/migrations -database "$DATABASE_URL" drop
+14 -4
View File
@@ -18,21 +18,31 @@ then
exit 1
fi
# Ensure CLICKHOUSE_DB is set
if [ -z "${CLICKHOUSE_DB}" ]; then
export CLICKHOUSE_DB="default"
fi
# Ensure CLICKHOUSE_CLUSTER_NAME is set
if [ -z "${CLICKHOUSE_CLUSTER_NAME}" ]; then
export CLICKHOUSE_CLUSTER_NAME="default"
fi
# Construct the database URL
if [ "$CLICKHOUSE_CLUSTER_ENABLED" == "false" ] ; then
if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
else
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-migrations-table-engine=MergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
fi
# Execute the up command
migrate -source file://clickhouse/migrations/unclustered -database "$DATABASE_URL" up
else
if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
else
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
fi
# Execute the up command
+2 -1
View File
@@ -121,7 +121,8 @@
},
"pnpm": {
"overrides": {
"jsonpath-plus": "10.0.7"
"jsonpath-plus": "10.0.7",
"nanoid": "^3.3.8"
}
}
}
@@ -479,6 +479,13 @@ export type Prompt = {
tags: Generated<string[]>;
labels: Generated<string[]>;
};
export type QueueBackUp = {
id: string;
project_id: string | null;
queue_name: string;
content: unknown;
created_at: Generated<Timestamp>;
};
export type Score = {
id: string;
timestamp: Generated<Timestamp>;
@@ -628,6 +635,7 @@ export type DB = {
project_memberships: ProjectMembership;
projects: Project;
prompts: Prompt;
queue_backups: QueueBackUp;
score_configs: ScoreConfig;
scores: Score;
Session: Session;
@@ -0,0 +1,13 @@
-- CreateTable
CREATE TABLE "queue_backups" (
"id" TEXT NOT NULL,
"project_id" TEXT,
"queue_name" TEXT NOT NULL,
"content" JSONB NOT NULL,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "queue_backups_pkey" PRIMARY KEY ("id")
);
-- AddForeignKey
ALTER TABLE "traces" ADD CONSTRAINT "traces_session_id_project_id_fkey" FOREIGN KEY ("session_id", "project_id") REFERENCES "trace_sessions"("id", "project_id") ON DELETE RESTRICT ON UPDATE CASCADE;
@@ -0,0 +1,2 @@
-- DropForeignKey
ALTER TABLE "traces" DROP CONSTRAINT "traces_session_id_project_id_fkey";
+23 -14
View File
@@ -272,7 +272,6 @@ model TraceSession {
project Project @relation(fields: [projectId], references: [id], onDelete: Cascade)
bookmarked Boolean @default(false)
public Boolean @default(false)
traces Trace[]
@@id([id, projectId])
@@index([projectId])
@@ -284,25 +283,24 @@ model TraceSession {
// Update TraceView below when making changes to this model!
model Trace {
id String @id @default(cuid())
externalId String? @map("external_id")
timestamp DateTime @default(now())
id String @id @default(cuid())
externalId String? @map("external_id")
timestamp DateTime @default(now())
name String?
userId String? @map("user_id")
userId String? @map("user_id")
metadata Json?
release String?
version String?
projectId String @map("project_id")
project Project @relation(fields: [projectId], references: [id], onDelete: Cascade)
public Boolean @default(false)
bookmarked Boolean @default(false)
tags String[] @default([])
projectId String @map("project_id")
project Project @relation(fields: [projectId], references: [id], onDelete: Cascade)
public Boolean @default(false)
bookmarked Boolean @default(false)
tags String[] @default([])
input Json?
output Json?
sessionId String? @map("session_id")
session TraceSession? @relation(fields: [sessionId, projectId], references: [id, projectId])
createdAt DateTime @default(now()) @map("created_at")
updatedAt DateTime @default(now()) @updatedAt @map("updated_at")
sessionId String? @map("session_id")
createdAt DateTime @default(now()) @map("created_at")
updatedAt DateTime @default(now()) @updatedAt @map("updated_at")
@@index([projectId, timestamp])
@@index([sessionId])
@@ -1017,3 +1015,14 @@ model ObservationMedia {
@@index([projectId, observationId])
@@map("observation_media")
}
model QueueBackUp {
id String @id @default(cuid())
projectId String? @map("project_id")
queueName String @map("queue_name")
content Json
createdAt DateTime @default(now()) @map("created_at")
@@map("queue_backups")
}
+2
View File
@@ -29,6 +29,8 @@ const EnvSchema = z.object({
LANGFUSE_CACHE_PROMPT_ENABLED: z.enum(["true", "false"]).default("false"),
LANGFUSE_CACHE_PROMPT_TTL_SECONDS: z.coerce.number().default(60 * 60),
CLICKHOUSE_URL: z.string().url(),
CLICKHOUSE_CLUSTER_NAME: z.string().default("default"),
CLICKHOUSE_DB: z.string().default("default"),
CLICKHOUSE_USER: z.string(),
CLICKHOUSE_PASSWORD: z.string(),
LANGFUSE_SDK_CI_SYNC_PROCESSING_ENABLED: z
@@ -10,7 +10,7 @@ export const clickhouseClient = (opts?: NodeClickHouseClientConfigOptions) =>
url: env.CLICKHOUSE_URL,
username: env.CLICKHOUSE_USER,
password: env.CLICKHOUSE_PASSWORD,
database: "default",
database: env.CLICKHOUSE_DB,
clickhouse_settings: {
async_insert: 1,
wait_for_async_insert: 1, // if disabled, we won't get errors from clickhouse
+1
View File
@@ -45,6 +45,7 @@ export * from "./logger";
export * from "./queries";
export * from "./repositories";
export * from "./redis/evalExecutionQueue";
export * from "./services/sessions-ui-table-service";
// test utils
export * from "./test-utils";
@@ -418,7 +418,7 @@ export const getObservationLatencies = async (
// Skipping FINAL here, as the quantiles are approximate to begin with.
const query = `
SELECT
quantiles(0.5, 0.9, 0.95, 0.99)(date_diff('milliseconds', o.start_time, o.end_time)) as quantiles,
quantiles(0.5, 0.9, 0.95, 0.99)(date_diff('millisecond', o.start_time, o.end_time)) as quantiles,
name
FROM observations o
${chFilter.find((f) => f.clickhouseTable === "traces") ? "LEFT JOIN traces t ON o.trace_id = t.id AND o.project_id = t.project_id" : ""}
@@ -465,7 +465,7 @@ export const getTracesLatencies = async (
select o.trace_id,
t.name,
o.project_id,
date_diff('milliseconds', min(o.start_time), coalesce(max(o.end_time), max(o.start_time))) as duration
date_diff('millisecond', min(o.start_time), coalesce(max(o.end_time), max(o.start_time))) as duration
FROM traces t
JOIN observations o
ON o.trace_id = t.id AND o.project_id = t.project_id
@@ -521,7 +521,7 @@ export const getModelLatenciesOverTime = async (
SELECT
${selectTimeseriesColumn(groupBy, "o.start_time", "start_time_bucket")},
provided_model_name,
quantiles(0.5, 0.75, 0.9, 0.95, 0.99)(date_diff('milliseconds', o.start_time, o.end_time)) as quantiles
quantiles(0.5, 0.75, 0.9, 0.95, 0.99)(date_diff('millisecond', o.start_time, o.end_time)) as quantiles
FROM observations o
${traceFilter ? "JOIN traces t ON o.trace_id = t.id AND o.project_id = t.project_id" : ""}
WHERE project_id = {projectId: String}
@@ -540,8 +540,8 @@ const getObservationsTableInternal = async <T>(
o.prompt_name as "prompt_name",
o.prompt_version as "prompt_version",
internal_model_id as "internal_model_id",
if(isNull(end_time), NULL, date_diff('milliseconds', start_time, end_time)) as latency,
if(isNull(completion_start_time), NULL, date_diff('milliseconds', start_time, completion_start_time)) as "time_to_first_token"`;
if(isNull(end_time), NULL, date_diff('millisecond', start_time, end_time)) as latency,
if(isNull(completion_start_time), NULL, date_diff('millisecond', start_time, completion_start_time)) as "time_to_first_token"`;
const { projectId, filter, selectIOAndMetadata, limit, offset, orderBy } =
opts;
@@ -1010,7 +1010,7 @@ export const getObservationMetricsForPrompts = async (
end_time,
usage_details,
cost_details,
dateDiff('milliseconds', start_time, end_time) AS latency_ms
dateDiff('millisecond', start_time, end_time) AS latency_ms
FROM observations
FINAL
WHERE (type = 'GENERATION')
@@ -1073,7 +1073,7 @@ export const getLatencyAndTotalCostForObservations = async (
SELECT
id,
cost_details['total'] AS total_cost,
dateDiff('milliseconds', start_time, end_time) AS latency_ms
dateDiff('millisecond', start_time, end_time) AS latency_ms
FROM observations FINAL
WHERE project_id = {projectId: String}
AND id IN ({observationIds: Array(String)})
@@ -1105,7 +1105,7 @@ export const getLatencyAndTotalCostForObservationsByTraces = async (
SELECT
trace_id,
sumMap(cost_details)['total'] AS total_cost,
dateDiff('milliseconds', min(start_time), max(end_time)) AS latency_ms
dateDiff('millisecond', min(start_time), max(end_time)) AS latency_ms
FROM observations FINAL
WHERE project_id = {projectId: String}
AND trace_id IN ({traceIds: Array(String)})
@@ -1225,12 +1225,12 @@ export const getGenerationsForPostHog = async (
o.start_time as start_time,
o.id as id,
o.total_cost as total_cost,
if(isNull(completion_start_time), NULL, date_diff('milliseconds', start_time, completion_start_time)) as time_to_first_token,
if(isNull(completion_start_time), NULL, date_diff('millisecond', start_time, completion_start_time)) as time_to_first_token,
o.usage_details['total'] as input_tokens,
o.usage_details['output'] as output_tokens,
o.cost_details['total'] as total_tokens,
o.project_id as project_id,
if(isNull(end_time), NULL, date_diff('milliseconds', start_time, end_time) / 1000) as latency,
if(isNull(end_time), NULL, date_diff('millisecond', start_time, end_time) / 1000) as latency,
o.provided_model_name as model,
o.level as level,
o.version as version,
@@ -9,7 +9,6 @@ import Decimal from "decimal.js";
import { parseClickhouseUTCDateTimeFormat } from "./clickhouse";
import { ObservationRecordReadType } from "./definitions";
import { parseJsonPrioritised } from "../../utils/json";
import { jsonSchema } from "../../utils/zod";
export const convertObservationToView = (
record: ObservationRecordReadType,
@@ -62,15 +61,22 @@ export const convertObservation = (
? parseClickhouseUTCDateTimeFormat(record.end_time)
: null,
name: record.name ?? null,
metadata: record.metadata,
metadata:
record.metadata &&
Object.fromEntries(
Object.entries(record.metadata ?? {}).map(([key, val]) => [
key,
val && parseJsonPrioritised(val),
]),
),
level: record.level as ObservationLevel,
statusMessage: record.status_message ?? null,
version: record.version ?? null,
input: (record.input
? jsonSchema.parse(parseJsonPrioritised(record.input))
? parseJsonPrioritised(record.input)
: null) as Prisma.JsonValue | null,
output: (record.output
? jsonSchema.parse(parseJsonPrioritised(record.output))
? parseJsonPrioritised(record.output)
: null) as Prisma.JsonValue | null,
modelParameters: record.model_parameters
? JSON.parse(record.model_parameters)
@@ -135,13 +135,16 @@ export const upsertScore = async (score: Partial<ScoreRecordReadType>) => {
});
};
export const getScoresForTraces = async (
projectId: string,
traceIds: string[],
timestamp?: Date,
limit?: number,
offset?: number,
) => {
export type GetScoresForTracesProps = {
projectId: string;
traceIds: string[];
timestamp?: Date;
limit?: number;
offset?: number;
};
export const getScoresForTraces = async (props: GetScoresForTracesProps) => {
const { projectId, traceIds, timestamp, limit, offset } = props;
const query = `
select
*
@@ -16,10 +16,7 @@ import {
} from "../queries/clickhouse-sql/clickhouse-filter";
import { TraceRecordReadType } from "./definitions";
import { tracesTableUiColumnDefinitions } from "../../tableDefinitions/mapTracesTable";
import { OrderByState } from "../../interfaces/orderBy";
import { orderByToClickhouseSql } from "../queries/clickhouse-sql/orderby-factory";
import { UiColumnMapping } from "../../tableDefinitions";
import { sessionCols } from "../../tableDefinitions/mapSessionTable";
import { convertDateToClickhouseDateTime } from "../clickhouse/client";
import { convertClickhouseToDomain } from "./traces_converters";
import { clickhouseSearchCondition } from "../queries/clickhouse-sql/search";
@@ -383,196 +380,6 @@ export const getTracesGroupedByTags = async (props: GroupedTracesQueryProp) => {
return rows;
};
export type SessionDataReturnType = {
session_id: string;
max_timestamp: string;
min_timestamp: string;
trace_ids: string[];
user_ids: string[];
trace_count: number;
trace_tags: string[];
total_observations: number;
duration: number;
session_usage_details: Record<string, number>;
session_cost_details: Record<string, number>;
session_input_cost: string;
session_output_cost: string;
session_total_cost: string;
session_input_usage: string;
session_output_usage: string;
session_total_usage: string;
};
export const getSessionsTableCount = async (props: {
projectId: string;
filter: FilterState;
orderBy?: OrderByState;
limit?: number;
page?: number;
}) => {
const rows = await getSessionsTableGeneric<{ count: string }>({
select: `
count(session_id) as count
`,
projectId: props.projectId,
filter: props.filter,
orderBy: props.orderBy,
limit: props.limit,
page: props.page,
});
return rows.length > 0 ? Number(rows[0].count) : 0;
};
export const getSessionsTable = async (props: {
projectId: string;
filter: FilterState;
orderBy?: OrderByState;
limit?: number;
page?: number;
}) => {
const rows = await getSessionsTableGeneric<SessionDataReturnType>({
select: `
session_id,
max_timestamp,
min_timestamp,
trace_ids,
user_ids,
trace_count,
trace_tags,
total_observations,
duration,
session_usage_details,
session_cost_details,
session_input_cost,
session_output_cost,
session_total_cost,
session_input_usage,
session_output_usage,
session_total_usage
`,
projectId: props.projectId,
filter: props.filter,
orderBy: props.orderBy,
limit: props.limit,
page: props.page,
});
return rows;
};
export type FetchSessionsTableProps = {
select: string;
projectId: string;
filter: FilterState;
searchQuery?: string;
orderBy?: OrderByState;
limit?: number;
page?: number;
};
const getSessionsTableGeneric = async <T>(props: FetchSessionsTableProps) => {
const { select, projectId, filter, orderBy, limit, page } = props;
const { tracesFilter, scoresFilter, observationsFilter } =
getProjectIdDefaultFilter(projectId, { tracesPrefix: "s" });
tracesFilter.push(...createFilterFromFilterState(filter, sessionCols));
const tracesFilterRes = tracesFilter.apply();
const scoresAvgFilterRes = scoresFilter.apply();
const observationsStatsRes = observationsFilter.apply();
const traceTimestampFilter: DateTimeFilter | undefined = tracesFilter.find(
(f) =>
f.field === "min_timestamp" &&
(f.operator === ">=" || f.operator === ">"),
) as DateTimeFilter | undefined;
const singleTraceFilter = traceTimestampFilter
? new FilterList([
new DateTimeFilter({
clickhouseTable: "traces",
field: "timestamp",
operator: traceTimestampFilter.operator,
value: traceTimestampFilter.value,
}),
]).apply()
: undefined;
const query = `
WITH observations_agg AS (
SELECT o.trace_id,
count(*) as obs_count,
min(o.start_time) as min_start_time,
max(o.end_time) as max_end_time,
sumMap(usage_details) as sum_usage_details,
sumMap(cost_details) as sum_cost_details,
anyLast(project_id) as project_id
FROM observations o FINAL
WHERE o.project_id = {projectId: String}
${traceTimestampFilter ? `AND o.start_time >= {observationsStartTime: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}` : ""}
GROUP BY o.trace_id
),
session_data AS (
SELECT
t.session_id,
anyLast(t.project_id) as project_id,
max(t.timestamp) as max_timestamp,
min(t.timestamp) as min_timestamp,
groupArray(t.id) AS trace_ids,
groupUniqArray(t.user_id) AS user_ids,
count(*) as trace_count,
groupUniqArrayArray(t.tags) as trace_tags,
-- Aggregate observations data at session level
sum(o.obs_count) as total_observations,
date_diff('milliseconds', min(min_start_time), max(max_end_time)) as duration,
sumMap(o.sum_usage_details) as session_usage_details,
sumMap(o.sum_cost_details) as session_cost_details,
arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'input') > 0, sumMap(o.sum_cost_details)))) as session_input_cost,
arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, sumMap(o.sum_cost_details)))) as session_output_cost,
sumMap(o.sum_cost_details)['total'] as session_total_cost,
arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'input') > 0, sumMap(o.sum_usage_details)))) as session_input_usage,
arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, sumMap(o.sum_usage_details)))) as session_output_usage,
sumMap(o.sum_usage_details)['total'] as session_total_usage
FROM traces t FINAL
LEFT JOIN observations_agg o
ON t.id = o.trace_id AND t.project_id = o.project_id
WHERE t.session_id IS NOT NULL
AND t.project_id = {projectId: String}
${singleTraceFilter?.query ? ` AND ${singleTraceFilter.query}` : ""}
GROUP BY t.session_id
)
SELECT ${select}
FROM session_data s
WHERE ${tracesFilterRes.query ? tracesFilterRes.query : ""}
${orderByToClickhouseSql(orderBy ?? null, sessionCols)}
${limit !== undefined && page !== undefined ? `LIMIT {limit: Int32} OFFSET {offset: Int32}` : ""}
`;
const obsStartTimeValue = traceTimestampFilter
? convertDateToClickhouseDateTime(traceTimestampFilter.value)
: null;
const res = await queryClickhouse<T>({
query: query,
params: {
projectId,
limit: limit,
offset: limit && page ? limit * page : 0,
...tracesFilterRes.params,
...observationsStatsRes.params,
...scoresAvgFilterRes.params,
...singleTraceFilter?.params,
...(obsStartTimeValue
? { observationsStartTime: obsStartTimeValue }
: {}),
},
});
return res;
};
export const getTracesIdentifierForSession = async (
projectId: string,
sessionId: string,
@@ -831,7 +638,7 @@ export const getTracesForPostHog = async (
o.trace_id,
sum(total_cost) as total_cost,
count(*) as observation_count,
date_diff('milliseconds', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds
date_diff('millisecond', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds
FROM observations o FINAL
WHERE o.project_id = {projectId: String}
AND o.start_time >= {minTimestamp: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}
@@ -3,7 +3,6 @@ import { parseClickhouseUTCDateTimeFormat } from "./clickhouse";
import { TraceRecordReadType } from "./definitions";
import { convertDateToClickhouseDateTime } from "../clickhouse/client";
import { parseJsonPrioritised } from "../../utils/json";
import { jsonSchema } from "../../utils/zod";
export const convertTraceDomainToClickhouse = (
trace: Trace,
@@ -46,12 +45,19 @@ export const convertClickhouseToDomain = (
sessionId: record.session_id ?? null,
public: record.public,
input: (record.input
? jsonSchema.parse(parseJsonPrioritised(record.input))
? parseJsonPrioritised(record.input)
: null) as Prisma.JsonValue | null,
output: (record.output
? jsonSchema.parse(parseJsonPrioritised(record.output))
? parseJsonPrioritised(record.output)
: null) as Prisma.JsonValue | null,
metadata: record.metadata,
metadata:
record.metadata &&
Object.fromEntries(
Object.entries(record.metadata ?? {}).map(([key, val]) => [
key,
val && parseJsonPrioritised(val),
]),
),
createdAt: parseClickhouseUTCDateTimeFormat(record.created_at),
updatedAt: parseClickhouseUTCDateTimeFormat(record.updated_at),
externalId: null,
@@ -30,7 +30,7 @@ export interface StorageService {
download(path: string): Promise<string>;
listFiles(prefix: string): Promise<string[]>;
listFiles(prefix: string): Promise<{ file: string; createdAt: Date }[]>;
getSignedUrl(
fileName: string,
@@ -201,7 +201,9 @@ class AzureBlobStorageService implements StorageService {
}
}
public async listFiles(prefix: string): Promise<string[]> {
public async listFiles(
prefix: string,
): Promise<{ file: string; createdAt: Date }[]> {
try {
await this.createContainerIfNotExists();
@@ -209,7 +211,10 @@ class AzureBlobStorageService implements StorageService {
const files = [];
for await (const blob of result) {
if (blob.name.startsWith(prefix)) {
files.push(blob.name);
files.push({
file: blob.name,
createdAt: blob?.properties?.createdOn ?? new Date(),
});
}
}
return files;
@@ -364,7 +369,9 @@ class S3StorageService implements StorageService {
}
}
public async listFiles(prefix: string): Promise<string[]> {
public async listFiles(
prefix: string,
): Promise<{ file: string; createdAt: Date }[]> {
const listCommand = new ListObjectsV2Command({
Bucket: this.bucketName,
Prefix: prefix,
@@ -373,7 +380,11 @@ class S3StorageService implements StorageService {
try {
const response = await this.client.send(listCommand);
return (
response.Contents?.flatMap((file) => (file.Key ? [file.Key] : [])) ?? []
response.Contents?.flatMap((file) =>
file.Key
? [{ file: file.Key, createdAt: file.LastModified ?? new Date() }]
: [],
) ?? []
);
} catch (err) {
logger.error(`Failed to list files from S3 ${prefix}`, err);
@@ -0,0 +1,307 @@
import { OrderByState } from "../../interfaces/orderBy";
import { sessionCols } from "../../tableDefinitions/mapSessionTable";
import { FilterState } from "../../types";
import { convertDateToClickhouseDateTime } from "../clickhouse/client";
import { DateTimeFilter, FilterList, orderByToClickhouseSql } from "../queries";
import {
getProjectIdDefaultFilter,
createFilterFromFilterState,
} from "../queries/clickhouse-sql/factory";
import {
TRACE_TO_OBSERVATIONS_INTERVAL,
queryClickhouse,
} from "../repositories";
export type SessionDataReturnType = {
session_id: string;
max_timestamp: string;
min_timestamp: string;
trace_ids: string[];
user_ids: string[];
trace_count: number;
trace_tags: string[];
};
export type SessionWithMetricsReturnType = SessionDataReturnType & {
total_observations: number;
duration: number;
session_usage_details: Record<string, number>;
session_cost_details: Record<string, number>;
session_input_cost: string;
session_output_cost: string;
session_total_cost: string;
session_input_usage: string;
session_output_usage: string;
session_total_usage: string;
};
export const getSessionsTableCount = async (props: {
projectId: string;
filter: FilterState;
orderBy?: OrderByState;
limit?: number;
page?: number;
}) => {
const rows = await getSessionsTableGeneric<{ count: string }>({
select: "count",
projectId: props.projectId,
filter: props.filter,
orderBy: props.orderBy,
limit: props.limit,
page: props.page,
});
return rows.length > 0 ? Number(rows[0].count) : 0;
};
export const getSessionsTable = async (props: {
projectId: string;
filter: FilterState;
orderBy?: OrderByState;
limit?: number;
page?: number;
}) => {
const rows = await getSessionsTableGeneric<SessionDataReturnType>({
select: "rows",
projectId: props.projectId,
filter: props.filter,
orderBy: props.orderBy,
limit: props.limit,
page: props.page,
});
return rows.map((row) => ({
...row,
trace_count: Number(row.trace_count),
}));
};
export const getSessionsWithMetrics = async (props: {
projectId: string;
filter: FilterState;
orderBy?: OrderByState;
limit?: number;
page?: number;
}) => {
const rows = await getSessionsTableGeneric<SessionWithMetricsReturnType>({
select: "metrics",
projectId: props.projectId,
filter: props.filter,
orderBy: props.orderBy,
limit: props.limit,
page: props.page,
});
return rows.map((row) => ({
...row,
trace_count: Number(row.trace_count),
total_observations: Number(row.total_observations),
}));
};
export type FetchSessionsTableProps = {
select: "count" | "rows" | "metrics";
projectId: string;
filter: FilterState;
searchQuery?: string;
orderBy?: OrderByState;
limit?: number;
page?: number;
};
const getSessionsTableGeneric = async <T>(props: FetchSessionsTableProps) => {
const { select, projectId, filter, orderBy, limit, page } = props;
let sqlSelect: string;
switch (select) {
case "count":
sqlSelect = "count(session_id) as count";
break;
case "rows":
sqlSelect = `
session_id,
max_timestamp,
min_timestamp,
trace_ids,
user_ids,
trace_count,
trace_tags`;
break;
case "metrics":
sqlSelect = `
session_id,
max_timestamp,
min_timestamp,
trace_ids,
user_ids,
trace_count,
trace_tags,
total_observations,
duration,
session_usage_details,
session_cost_details,
session_input_cost,
session_output_cost,
session_total_cost,
session_input_usage,
session_output_usage,
session_total_usage`;
break;
default:
const exhaustiveCheckDefault: never = select;
throw new Error(`Unknown select type: ${select}`);
}
const { tracesFilter } = getProjectIdDefaultFilter(projectId, {
tracesPrefix: "s",
});
tracesFilter.push(...createFilterFromFilterState(filter, sessionCols));
const tracesFilterRes = tracesFilter.apply();
const traceTimestampFilter: DateTimeFilter | undefined = tracesFilter.find(
(f) =>
f.field === "min_timestamp" &&
(f.operator === ">=" || f.operator === ">"),
) as DateTimeFilter | undefined;
const filters = [];
if (traceTimestampFilter) {
filters.push(
new DateTimeFilter({
clickhouseTable: "traces",
field: "timestamp",
operator: traceTimestampFilter.operator,
value: traceTimestampFilter.value,
}),
);
}
const additionalSingleTraceFilter = tracesFilter.find(
(f) => f.field === "bookmarked" || f.field === "session_id",
);
if (additionalSingleTraceFilter) {
filters.push(additionalSingleTraceFilter);
}
const singleTraceFilter =
filters.length > 0 ? new FilterList(filters).apply() : undefined;
const hasMetricsFilter = tracesFilter.find((f) =>
[
"session_total_cost",
"session_input_cost",
"session_output_cost",
"duration",
"session_total_usage",
"session_output_usage",
"session_input_usage",
].includes(f.field),
);
const selectMetrics = select === "metrics" || hasMetricsFilter;
// We use deduplicated traces and observations CTEs instead of final to be able to use Skip indices in Clickhouse.
const query = `
WITH deduplicated_traces AS (
SELECT * EXCEPT input, output, metadata
FROM traces t
WHERE t.session_id IS NOT NULL
AND t.project_id = {projectId: String}
${singleTraceFilter?.query ? ` AND ${singleTraceFilter.query}` : ""}
ORDER BY event_ts DESC
LIMIT 1 BY id, project_id
),
deduplicated_observations AS (
SELECT *
FROM observations o
WHERE o.project_id = {projectId: String}
${traceTimestampFilter ? `AND o.start_time >= {observationsStartTime: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}` : ""}
AND o.trace_id IN (
SELECT id
FROM deduplicated_traces
)
ORDER BY event_ts DESC
LIMIT 1 BY id, project_id
),
observations_agg AS (
SELECT o.trace_id,
count(*) as obs_count,
min(o.start_time) as min_start_time,
max(o.end_time) as max_end_time,
sumMap(usage_details) as sum_usage_details,
sumMap(cost_details) as sum_cost_details,
anyLast(project_id) as project_id
FROM deduplicated_observations o
WHERE o.project_id = {projectId: String}
${traceTimestampFilter ? `AND o.start_time >= {observationsStartTime: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}` : ""}
GROUP BY o.trace_id
),
session_data AS (
SELECT
t.session_id,
anyLast(t.project_id) as project_id,
max(t.timestamp) as max_timestamp,
min(t.timestamp) as min_timestamp,
groupArray(t.id) AS trace_ids,
groupUniqArray(t.user_id) AS user_ids,
count(*) as trace_count,
groupUniqArrayArray(t.tags) as trace_tags
-- Aggregate observations data at session level
${
selectMetrics
? `
,
sum(o.obs_count) as total_observations,
date_diff('millisecond', min(min_start_time), max(max_end_time)) as duration,
sumMap(o.sum_usage_details) as session_usage_details,
sumMap(o.sum_cost_details) as session_cost_details,
arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'input') > 0, sumMap(o.sum_cost_details)))) as session_input_cost,
arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, sumMap(o.sum_cost_details)))) as session_output_cost,
sumMap(o.sum_cost_details)['total'] as session_total_cost,
arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'input') > 0, sumMap(o.sum_usage_details)))) as session_input_usage,
arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, sumMap(o.sum_usage_details)))) as session_output_usage,
sumMap(o.sum_usage_details)['total'] as session_total_usage`
: ""
}
FROM deduplicated_traces t
${
selectMetrics
? `LEFT JOIN observations_agg o
ON t.id = o.trace_id AND t.project_id = o.project_id`
: ""
}
WHERE t.session_id IS NOT NULL
AND t.project_id = {projectId: String}
${singleTraceFilter?.query ? ` AND ${singleTraceFilter.query}` : ""}
GROUP BY t.session_id
)
SELECT ${sqlSelect}
FROM session_data s
WHERE ${tracesFilterRes.query ? tracesFilterRes.query : ""}
${orderByToClickhouseSql(orderBy ?? null, sessionCols)}
${limit !== undefined && page !== undefined ? `LIMIT {limit: Int32} OFFSET {offset: Int32}` : ""}
`;
const obsStartTimeValue = traceTimestampFilter
? convertDateToClickhouseDateTime(traceTimestampFilter.value)
: null;
const res = await queryClickhouse<T>({
query: query,
params: {
projectId,
limit: limit,
offset: limit && page ? limit * page : 0,
...tracesFilterRes.params,
...singleTraceFilter?.params,
...(obsStartTimeValue
? { observationsStartTime: obsStartTimeValue }
: {}),
},
});
return res;
};
@@ -351,7 +351,7 @@ const getTracesTableGeneric = async <T>(props: FetchTracesTableProps) => {
COUNT(*) AS observation_count,
sumMap(usage_details) as usage_details,
SUM(total_cost) AS total_cost,
date_diff('milliseconds', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds,
date_diff('millisecond', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds,
multiIf(
arrayExists(x -> x = 'ERROR', groupArray(level)), 'ERROR',
arrayExists(x -> x = 'WARNING', groupArray(level)), 'WARNING',
@@ -68,7 +68,7 @@ export const observationsTableUiColumnDefinitions: UiColumnMapping[] = [
uiTableId: "timeToFirstToken",
clickhouseTableName: "observations",
clickhouseSelect:
"if(isNull(completion_start_time), NULL, date_diff('milliseconds', start_time, completion_start_time) / 1000)",
"if(isNull(completion_start_time), NULL, date_diff('millisecond', start_time, completion_start_time) / 1000)",
// If we use the default of Decimal64(12), we cannot filter for more than ~40min due to an overflow
clickhouseTypeOverwrite: "Decimal64(3)",
},
@@ -77,7 +77,7 @@ export const observationsTableUiColumnDefinitions: UiColumnMapping[] = [
uiTableId: "latency",
clickhouseTableName: "observations",
clickhouseSelect:
"if(isNull(end_time), NULL, date_diff('milliseconds', start_time, end_time) / 1000)",
"if(isNull(end_time), NULL, date_diff('millisecond', start_time, end_time) / 1000)",
// If we use the default of Decimal64(12), we cannot filter for more than ~40min due to an overflow
clickhouseTypeOverwrite: "Decimal64(3)",
},
@@ -86,7 +86,7 @@ export const observationsTableUiColumnDefinitions: UiColumnMapping[] = [
uiTableId: "tokensPerSecond",
clickhouseTableName: "observations",
clickhouseSelect:
"(arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, usage_details))) / (date_diff('milliseconds', start_time, end_time) / 1000))",
"(arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, usage_details))) / (date_diff('millisecond', start_time, end_time) / 1000))",
},
{
uiTableName: "Input Cost ($)",
-16
View File
@@ -55,25 +55,9 @@ export const parseJsonPrioritised = (
): z.infer<typeof jsonSchema> | string | undefined => {
try {
const parsedJson = JSON.parse(json);
if (Object.keys(parsedJson).length === 0) {
return parsedJson;
}
const parsedArray = z.array(jsonSchemaNullable).safeParse(parsedJson);
if (parsedArray.success) {
return parsedArray.data;
}
const parsedObject = z.record(jsonSchemaNullable).safeParse(parsedJson);
if (parsedObject.success) {
return parsedObject.data;
}
return jsonSchema.parse(parsedJson);
} catch (error) {
const parsed = jsonSchema.safeParse(json);
return parsed.success ? parsed.data : json;
}
};
+2 -2
View File
@@ -23,18 +23,18 @@ type Json = Root | { [key: string]: JsonNested } | JsonNested[];
// Here, you define the schema recursively
export const jsonSchemaNullable: z.ZodType<JsonNested> = z.lazy(() =>
z.union([
nestedLiteralSchema,
z.array(jsonSchemaNullable),
z.record(jsonSchemaNullable),
nestedLiteralSchema,
]),
);
// Root schema that does not allow nulls at the root level
export const jsonSchema: z.ZodType<Json> = z.lazy(() =>
z.union([
rootLiteralSchema,
z.array(jsonSchemaNullable),
z.record(jsonSchemaNullable),
rootLiteralSchema,
]),
);
+11 -35
View File
@@ -6,6 +6,7 @@ settings:
overrides:
jsonpath-plus: 10.0.7
nanoid: ^3.3.8
importers:
@@ -742,7 +743,7 @@ importers:
version: 7.12.0(eslint@8.57.0)(typescript@5.4.5)
autoprefixer:
specifier: ^10.4.19
version: 10.4.19(postcss@8.4.47)
version: 10.4.19(postcss@8.4.49)
dotenv-cli:
specifier: ^7.4.2
version: 7.4.2
@@ -761,9 +762,6 @@ importers:
node-mocks-http:
specifier: ^1.14.1
version: 1.14.1
postcss:
specifier: ^8.4.47
version: 8.4.47
prettier:
specifier: ^3.3.3
version: 3.3.3
@@ -8976,13 +8974,8 @@ packages:
mz@2.7.0:
resolution: {integrity: sha512-z81GNO7nnYMEhrGh9LeymoE4+Yr0Wn5McHIZMK5cfQCl+NDX08sCZgUc9/6MHni9IWuFLm1Z3HTCXu2z9fN62Q==}
nanoid@3.3.6:
resolution: {integrity: sha512-BGcqMMJuToF7i1rt+2PWSNVnWIkGCU78jBG3RxO/bZlnZPK2Cmi2QaffxGO/2RvWi9sL+FAiRiXMgsyxQ1DIDA==}
engines: {node: ^10 || ^12 || ^13.7 || ^14 || >=15.0.1}
hasBin: true
nanoid@3.3.7:
resolution: {integrity: sha512-eSRppjcPIatRIMC1U6UngP8XFcz8MQWGQdt1MTBQ7NaAmvXDfvNxbvWV3x2y6CdEUciCSsDHDQZbhYaB8QEo2g==}
nanoid@3.3.8:
resolution: {integrity: sha512-WNLf5Sd8oZxOm+TzppcYk8gVOgP+l58xNy58D0nbUnOxOWRWvlcCV4kUF7ltmI6PsrLl/BgKEyS4mqsGChFN0w==}
engines: {node: ^10 || ^12 || ^13.7 || ^14 || >=15.0.1}
hasBin: true
@@ -9518,9 +9511,6 @@ packages:
picocolors@1.0.0:
resolution: {integrity: sha512-1fygroTLlHu66zi26VoTDv8yRgm0Fccecssto+MhsZ0D/DGW2sm8E8AjW7NU5VVTRt5GxbeZ5qBuJr+HyLYkjQ==}
picocolors@1.1.0:
resolution: {integrity: sha512-TQ92mBOW0l3LeMeyLV6mzy/kWr8lkd/hp3mTg7wYK7zJhuBStmGMBG0BdeDZS/dZx1IukaX6Bk11zcln25o1Aw==}
picocolors@1.1.1:
resolution: {integrity: sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA==}
@@ -9612,10 +9602,6 @@ packages:
resolution: {integrity: sha512-PS08Iboia9mts/2ygV3eLpY5ghnUcfLV/EXTOW1E2qYxJKGGBUtNjN76FYHnMs36RmARn41bC0AZmn+rR0OVpQ==}
engines: {node: ^10 || ^12 || >=14}
postcss@8.4.47:
resolution: {integrity: sha512-56rxCq7G/XfB4EkXq9Egn5GCqugWvDFjafDOThIdMBsI15iqPqR5r15TfSr1YPYeEI19YeaXMCbY6u88Y76GLQ==}
engines: {node: ^10 || ^12 || >=14}
postcss@8.4.49:
resolution: {integrity: sha512-OCVPnIObs4N29kxTjzLfUryOkvZEq+pf8jTF0lg8E7uETuWHA+v7j3c/xJmiqpX450191LlmZfUKkXxkTry7nA==}
engines: {node: ^10 || ^12 || >=14}
@@ -11575,7 +11561,7 @@ snapshots:
dependencies:
'@ai-sdk/provider': 0.0.24
eventsource-parser: 1.1.2
nanoid: 3.3.6
nanoid: 3.3.8
secure-json-parse: 2.7.0
optionalDependencies:
zod: 3.23.8
@@ -17720,7 +17706,7 @@ snapshots:
eventsource-parser: 1.1.2
json-schema: 0.4.0
jsondiffpatch: 0.6.0
nanoid: 3.3.6
nanoid: 3.3.8
secure-json-parse: 2.7.0
zod-to-json-schema: 3.23.2(zod@3.23.8)
optionalDependencies:
@@ -17933,14 +17919,14 @@ snapshots:
asynckit@0.4.0: {}
autoprefixer@10.4.19(postcss@8.4.47):
autoprefixer@10.4.19(postcss@8.4.49):
dependencies:
browserslist: 4.23.0
caniuse-lite: 1.0.30001599
fraction.js: 4.3.7
normalize-range: 0.1.2
picocolors: 1.0.0
postcss: 8.4.47
postcss: 8.4.49
postcss-value-parser: 4.2.0
available-typed-arrays@1.0.7:
@@ -22282,9 +22268,7 @@ snapshots:
object-assign: 4.1.1
thenify-all: 1.6.0
nanoid@3.3.6: {}
nanoid@3.3.7: {}
nanoid@3.3.8: {}
natural-compare@1.4.0: {}
@@ -22851,8 +22835,6 @@ snapshots:
picocolors@1.0.0: {}
picocolors@1.1.0: {}
picocolors@1.1.1: {}
picomatch@2.3.1: {}
@@ -22926,19 +22908,13 @@ snapshots:
postcss@8.4.31:
dependencies:
nanoid: 3.3.7
nanoid: 3.3.8
picocolors: 1.1.1
source-map-js: 1.2.1
postcss@8.4.47:
dependencies:
nanoid: 3.3.7
picocolors: 1.1.0
source-map-js: 1.2.1
postcss@8.4.49:
dependencies:
nanoid: 3.3.7
nanoid: 3.3.8
picocolors: 1.1.1
source-map-js: 1.2.1
+3
View File
@@ -95,6 +95,9 @@ ARG BUILDPLATFORM
WORKDIR /app
ARG NEXT_PUBLIC_BUILD_ID
ENV BUILD_ID=$NEXT_PUBLIC_BUILD_ID
ENV NODE_ENV production
# Uncomment the following line in case you want to disable telemetry during runtime.
ENV NEXT_TELEMETRY_DISABLED 1
+3 -3
View File
@@ -1,6 +1,6 @@
{
"name": "web",
"version": "3.5.2",
"version": "3.6.2",
"private": true,
"license": "MIT",
"engines": {
@@ -175,7 +175,6 @@
"jest": "^29.7.0",
"jest-environment-jsdom": "^29.7.0",
"node-mocks-http": "^1.14.1",
"postcss": "^8.4.47",
"prettier": "^3.3.3",
"prettier-plugin-tailwindcss": "^0.6.6",
"tailwindcss": "^3.4.17",
@@ -193,7 +192,8 @@
},
"pnpm": {
"overrides": {
"jsonpath-plus": "10.0.7"
"jsonpath-plus": "10.0.7",
"nanoid": "^3.3.8"
}
}
}
@@ -204,6 +204,7 @@ describe("/api/public/observations API Endpoint", () => {
trace_id: traceId,
name: "generation-name",
start_time: new Date("2021-01-01T00:00:00.000Z").getTime(),
event_ts: new Date("2021-01-01T00:00:00.000Z").getTime(),
end_time: new Date("2021-01-01T00:00:00.000Z").getTime(),
project_id: "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a",
type: "GENERATION",
@@ -214,6 +215,7 @@ describe("/api/public/observations API Endpoint", () => {
trace_id: traceId,
name: "generation-name",
start_time: new Date("2021-02-01T00:00:00.000Z").getTime(),
event_ts: new Date("2021-02-01T00:00:00.000Z").getTime(),
end_time: new Date("2021-02-01T00:00:00.000Z").getTime(),
project_id: "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a",
type: "SPAN",
@@ -223,6 +225,7 @@ describe("/api/public/observations API Endpoint", () => {
trace_id: traceId,
name: "generation-name",
start_time: new Date("2021-03-01T00:00:00.000Z").getTime(),
event_ts: new Date("2021-03-01T00:00:00.000Z").getTime(),
project_id: "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a",
type: "EVENT",
});
@@ -231,6 +234,7 @@ describe("/api/public/observations API Endpoint", () => {
trace_id: traceId,
name: "generation-name",
start_time: new Date("2021-04-01T00:00:00.000Z").getTime(),
event_ts: new Date("2021-04-01T00:00:00.000Z").getTime(),
end_time: new Date("2021-04-01T00:00:00.000Z").getTime(),
project_id: "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a",
type: "GENERATION",
@@ -0,0 +1,192 @@
import { v4 } from "uuid";
import { prisma } from "@langfuse/shared/src/db";
import {
createObservation,
createObservationsCh,
createOrgProjectAndApiKey,
createTracesCh,
getSessionsWithMetrics,
} from "@langfuse/shared/src/server";
import { createTrace, getSessionsTable } from "@langfuse/shared/src/server";
describe("trpc.sessions", () => {
describe("GET sessions.all", () => {
it("should GET all session", async () => {
const { projectId } = await createOrgProjectAndApiKey();
const sessionId = v4();
await prisma.traceSession.create({
data: {
id: sessionId,
projectId: projectId,
},
});
const traces = [
createTrace({ session_id: sessionId, project_id: projectId }),
createTrace({ session_id: sessionId, project_id: projectId }),
];
await createTracesCh(traces);
const uiSessions = await getSessionsTable({
projectId: projectId,
filter: [],
orderBy: null,
limit: 10000,
page: 0,
});
expect(uiSessions.length).toBe(1);
expect(uiSessions[0].session_id).toBe(sessionId);
expect(uiSessions[0].trace_count).toBe(2);
expect(uiSessions[0].trace_tags).toEqual(["doe", "john"]);
});
});
it("should GET all session filtered by trace attribute only", async () => {
const { projectId } = await createOrgProjectAndApiKey();
const sessionId = v4();
await prisma.traceSession.create({
data: {
id: sessionId,
projectId: projectId,
},
});
const traces = [
createTrace({
session_id: sessionId,
project_id: projectId,
user_id: "user1",
}),
createTrace({
session_id: sessionId,
project_id: projectId,
user_id: undefined,
}),
];
await createTracesCh(traces);
const uiSessions = await getSessionsTable({
projectId: projectId,
filter: [
{
column: "userIds",
type: "stringOptions",
operator: "any of",
value: ["user1"],
},
],
orderBy: null,
limit: 10000,
page: 0,
});
expect(uiSessions.length).toBe(1);
expect(uiSessions[0].session_id).toBe(sessionId);
expect(uiSessions[0].trace_count).toBe(2);
expect(uiSessions[0].trace_tags).toEqual(["doe", "john"]);
expect(uiSessions[0].user_ids).toEqual(["user1"]);
});
it("should GET metrics for a list of sessions", async () => {
const { projectId } = await createOrgProjectAndApiKey();
const sessionId1 = v4();
const sessionId2 = v4();
await prisma.traceSession.createMany({
data: [
{
id: sessionId1,
projectId: projectId,
},
{
id: sessionId2,
projectId: projectId,
},
],
});
const traces = [
createTrace({
session_id: sessionId1,
project_id: projectId,
user_id: "user1",
}),
createTrace({
session_id: sessionId1,
project_id: projectId,
user_id: "user2",
}),
createTrace({
session_id: sessionId2,
project_id: projectId,
user_id: "user3",
}),
];
const observations = traces.flatMap((trace) => [
createObservation({
trace_id: trace.id,
project_id: projectId,
}),
createObservation({
trace_id: trace.id,
project_id: projectId,
}),
]);
await createObservationsCh(observations);
await createTracesCh(traces);
const sessions = await getSessionsWithMetrics({
projectId: projectId,
filter: [
{
column: "id",
type: "stringOptions",
operator: "any of",
value: [sessionId1, sessionId2],
},
],
});
expect(sessions.length).toBe(2);
// Session 1 checks
const session1 = sessions.find((s) => s.session_id === sessionId1);
expect(session1).toBeDefined();
expect(session1?.trace_count).toBe(2);
expect(session1?.user_ids).toEqual(
expect.arrayContaining(["user1", "user2"]),
);
expect(session1?.trace_tags).toEqual(["doe", "john"]);
expect(session1?.total_observations).toEqual(4);
expect(Number(session1?.session_input_cost)).toBeGreaterThan(0);
expect(Number(session1?.session_output_cost)).toBeGreaterThan(0);
expect(Number(session1?.session_total_cost)).toBeGreaterThan(0);
expect(Number(session1?.session_input_usage)).toBeGreaterThan(0);
expect(Number(session1?.session_output_usage)).toBeGreaterThan(0);
expect(Number(session1?.session_total_usage)).toBeGreaterThan(0);
// Session 2 checks
const session2 = sessions.find((s) => s.session_id === sessionId2);
expect(session2).toBeDefined();
expect(session2?.trace_count).toBe(1);
expect(session2?.user_ids).toEqual(["user3"]);
expect(session2?.trace_tags).toEqual(["doe", "john"]);
expect(session2?.total_observations).toEqual(2);
expect(Number(session2?.session_input_cost)).toBeGreaterThan(0);
expect(Number(session2?.session_output_cost)).toBeGreaterThan(0);
expect(Number(session2?.session_total_cost)).toBeGreaterThan(0);
expect(Number(session2?.session_input_usage)).toBeGreaterThan(0);
expect(Number(session2?.session_output_usage)).toBeGreaterThan(0);
expect(Number(session2?.session_total_usage)).toBeGreaterThan(0);
});
});
+29 -1
View File
@@ -1,4 +1,4 @@
import { paginationZod } from "@langfuse/shared";
import { paginationZod, parseJsonPrioritised } from "@langfuse/shared";
import { ZodError } from "zod";
// Create test cases
@@ -24,3 +24,31 @@ describe("Pagination Zod Schema", () => {
expect(() => paginationZod.limit.parse("abc")).toThrowError(ZodError);
});
});
describe("parseJsonPrioritised", () => {
it.each([
["test", "test"], // Raw string
['{"hello": "world"}', { hello: "world" }], // Simple object
["[1, 2, 3]", [1, 2, 3]], // Array
['{"nested": {"key": "value"}}', { nested: { key: "value" } }], // Nested object
["[]", []], // Empty array
["{}", {}], // Empty object
['"simple string"', "simple string"], // Quoted string
["42", 42], // Number
["true", true], // Boolean true
["false", false], // Boolean false
["null", "null"], // Null
["", ""], // Empty string
["invalid{json}", "invalid{json}"], // Invalid JSON string
['[null, 123, "abc"]', [null, 123, "abc"]], // Mixed array
[
'{"array": [1, 2], "nested": {"key": "value"}}',
{ array: [1, 2], nested: { key: "value" } },
], // Complex object
])(
"should parse input correctly (%s, %s)",
(input: string, expectedOutput: any) => {
expect(parseJsonPrioritised(input)).toEqual(expectedOutput);
},
);
});
+4 -13
View File
@@ -34,15 +34,12 @@ export type ModelParamsContext = {
modelParamsDescription?: string;
};
export const ModelParameters: React.FC<
ModelParamsContext & { evalModelsOnly: boolean }
> = ({
export const ModelParameters: React.FC<ModelParamsContext> = ({
modelParams,
availableProviders,
availableModels,
updateModelParamValue,
setModelParamEnabled,
evalModelsOnly,
formDisabled = false,
modelParamsDescription,
}) => {
@@ -55,11 +52,8 @@ export const ModelParameters: React.FC<
<p className="font-semibold">Model</p>
{availableProviders.length === 0 ? (
<>
<p className="text-xs">
No LLM API key set in project.{" "}
{evalModelsOnly && "For evals, only OpenAI models are supported."}
</p>
<CreateLLMApiKeyDialog evalModelsOnly={evalModelsOnly} />
<p className="text-xs">No LLM API key set in project. </p>
<CreateLLMApiKeyDialog />
</>
) : (
<div className="space-y-4">
@@ -70,7 +64,6 @@ export const ModelParameters: React.FC<
value={modelParams.provider.value}
options={availableProviders}
updateModelParam={updateModelParamValue}
evalModelsOnly={evalModelsOnly}
/>
<ModelParamsSelect
title="Model name"
@@ -149,7 +142,6 @@ type ModelParamsSelectProps = {
updateModelParam: ModelParamsContext["updateModelParamValue"];
disabled?: boolean;
modelParamsDescription?: string;
evalModelsOnly?: boolean;
};
const ModelParamsSelect = ({
title,
@@ -159,7 +151,6 @@ const ModelParamsSelect = ({
updateModelParam,
disabled,
modelParamsDescription,
evalModelsOnly,
}: ModelParamsSelectProps) => {
return (
<div className="space-y-2">
@@ -191,7 +182,7 @@ const ModelParamsSelect = ({
</SelectItem>
))}
<SelectSeparator />
<CreateLLMApiKeyDialog evalModelsOnly={evalModelsOnly} />
<CreateLLMApiKeyDialog />
</SelectContent>
</Select>
{modelParamsDescription ? (
@@ -34,6 +34,7 @@ export type ModelTableRow = {
prices?: Record<string, number>;
tokenizerId?: string;
config?: Prisma.JsonValue;
lastUsed?: Date | null;
serverResponse: GetModelResult;
};
@@ -51,6 +52,7 @@ const modelConfigDescriptions = {
"Some tokenizers require additional configuration (e.g. openai tiktoken). See docs for details.",
maintainer:
"Maintainer of the model. Langfuse managed models can be cloned, user managed models can be edited and deleted. To supersede a Langfuse managed model, set the custom model name to the Langfuse model name.",
lastUsed: "Start time of the latest generation using this model",
} as const;
export default function ModelTable({ projectId }: { projectId: string }) {
@@ -198,6 +200,20 @@ export default function ModelTable({ projectId }: { projectId: string }) {
) : null;
},
},
{
accessorKey: "lastUsed",
id: "lastUsed",
header: "Last used",
headerTooltip: {
description: modelConfigDescriptions.lastUsed,
},
enableHiding: true,
size: 120,
cell: ({ row }) => {
const value: Date | null | undefined = row.getValue("lastUsed");
return value?.toLocaleString() ?? "";
},
},
{
accessorKey: "actions",
header: "Actions",
@@ -246,6 +262,7 @@ export default function ModelTable({ projectId }: { projectId: string }) {
prices: model.prices,
tokenizerId: model.tokenizerId ?? undefined,
config: model.tokenizerConfig,
lastUsed: model.lastUsed,
serverResponse: model,
};
};
+1 -1
View File
@@ -1 +1 @@
export const VERSION = "v3.5.2";
export const VERSION = "v3.6.2";
@@ -545,7 +545,6 @@ export const InnerEvalTemplateForm = (props: {
modelParamsDescription:
"Select a model which supports function calling.",
}}
evalModelsOnly
formDisabled={!props.isEditing}
/>
</div>
@@ -683,10 +683,7 @@ export const CreateExperimentsForm = ({
availableProviders,
updateModelParamValue: updateModelParamValue,
setModelParamEnabled,
modelParamsDescription:
"Select a model which supports function calling.",
}}
evalModelsOnly
/>
</Card>
{form.formState.errors.modelConfig && (
@@ -14,7 +14,7 @@ export default function Playground() {
<div className="max-h-full min-h-0 basis-1/4 pr-2">
<div className="grid h-full grid-rows-[minmax(20dvh,max-content),minmax(20dvh,auto)] overflow-auto">
<div className="mb-4 max-h-[80dvh] min-h-[20dvh] overflow-y-auto">
<ModelParameters {...playgroundContext} evalModelsOnly={false} />
<ModelParameters {...playgroundContext} />
</div>
<div className="min-h-[20dvh]">
<Variables />
+9 -5
View File
@@ -9,6 +9,7 @@ export const env = createEnv({
server: {
DATABASE_URL: z.string().url(),
NODE_ENV: z.enum(["development", "test", "production"]),
BUILD_ID: z.string().optional(),
NEXTAUTH_SECRET:
process.env.NODE_ENV === "production"
? z.string().min(1)
@@ -150,9 +151,11 @@ export const env = createEnv({
// clickhouse
CLICKHOUSE_URL: z.string().url(),
CLICKHOUSE_CLUSTER_NAME: z.string().default("default"),
CLICKHOUSE_DB: z.string().default("default"),
CLICKHOUSE_USER: z.string(),
CLICKHOUSE_PASSWORD: z.string(),
CLICKHOUSE_CLUSTER_ENABLED: z.enum(["true", "false"]).default("false"),
CLICKHOUSE_CLUSTER_ENABLED: z.enum(["true", "false"]).default("true"),
// EE ui customization
LANGFUSE_UI_API_HOST: z.string().optional(),
@@ -301,6 +304,7 @@ export const env = createEnv({
NEXT_PUBLIC_DEMO_ORG_ID: process.env.NEXT_PUBLIC_DEMO_ORG_ID,
DATABASE_URL: process.env.DATABASE_URL,
NODE_ENV: process.env.NODE_ENV,
BUILD_ID: process.env.BUILD_ID,
NEXT_PUBLIC_BUILD_ID: process.env.NEXT_PUBLIC_BUILD_ID,
NEXTAUTH_SECRET: process.env.NEXTAUTH_SECRET,
NEXTAUTH_COOKIE_DOMAIN: process.env.NEXTAUTH_COOKIE_DOMAIN,
@@ -378,12 +382,10 @@ export const env = createEnv({
AUTH_CUSTOM_ISSUER: process.env.AUTH_CUSTOM_ISSUER,
AUTH_CUSTOM_NAME: process.env.AUTH_CUSTOM_NAME,
AUTH_CUSTOM_SCOPE: process.env.AUTH_CUSTOM_SCOPE,
AUTH_CUSTOM_CLIENT_AUTH_METHOD:
process.env.AUTH_CUSTOM_CLIENT_AUTH_METHOD,
AUTH_CUSTOM_CLIENT_AUTH_METHOD: process.env.AUTH_CUSTOM_CLIENT_AUTH_METHOD,
AUTH_CUSTOM_ALLOW_ACCOUNT_LINKING:
process.env.AUTH_CUSTOM_ALLOW_ACCOUNT_LINKING,
AUTH_IGNORE_ACCOUNT_FIELDS:
process.env.AUTH_IGNORE_ACCOUNT_FIELDS,
AUTH_IGNORE_ACCOUNT_FIELDS: process.env.AUTH_IGNORE_ACCOUNT_FIELDS,
AUTH_DOMAINS_WITH_SSO_ENFORCEMENT:
process.env.AUTH_DOMAINS_WITH_SSO_ENFORCEMENT,
AUTH_DISABLE_USERNAME_PASSWORD: process.env.AUTH_DISABLE_USERNAME_PASSWORD,
@@ -445,6 +447,8 @@ export const env = createEnv({
NEXT_PUBLIC_CRISP_WEBSITE_ID: process.env.NEXT_PUBLIC_CRISP_WEBSITE_ID,
// clickhouse
CLICKHOUSE_URL: process.env.CLICKHOUSE_URL,
CLICKHOUSE_CLUSTER_NAME: process.env.CLICKHOUSE_CLUSTER_NAME,
CLICKHOUSE_DB: process.env.CLICKHOUSE_DB,
CLICKHOUSE_USER: process.env.CLICKHOUSE_USER,
CLICKHOUSE_PASSWORD: process.env.CLICKHOUSE_PASSWORD,
CLICKHOUSE_CLUSTER_ENABLED: process.env.CLICKHOUSE_CLUSTER_ENABLED,
@@ -26,7 +26,7 @@ import { DatasetCompareRunPeekView } from "@/src/features/datasets/components/Da
import { useClickhouse } from "@/src/components/layouts/ClickhouseAdminToggle";
import { getQueryKey } from "@trpc/react-query";
import { useQueryClient } from "@tanstack/react-query";
import _ from "lodash";
export type RunMetrics = {
id: string;
scores: ScoreAggregate;
@@ -182,7 +182,7 @@ export function DatasetCompareRunsTable(props: {
({ datasetItemId, trace, observation, scores }) => {
if (!itemsAcc[datasetItemId]) itemsAcc[datasetItemId] = {};
itemsAcc[datasetItemId][runId] = {
_.set(itemsAcc[datasetItemId], runId, {
id: runId,
traceId: trace?.id ?? "",
observationId: observation?.id ?? undefined,
@@ -196,7 +196,7 @@ export function DatasetCompareRunsTable(props: {
: usdFormatter(trace?.totalCost)) ?? undefined,
},
scores,
};
});
},
);
+6 -6
View File
@@ -176,7 +176,7 @@ export const createTempTableInClickhouse = async (
clickhouseSession: string,
) => {
const query = `
CREATE TABLE IF NOT EXISTS ${tableName} ${env.CLICKHOUSE_CLUSTER_ENABLED === "true" ? "ON CLUSTER default" : ""}
CREATE TABLE IF NOT EXISTS ${tableName} ${env.CLICKHOUSE_CLUSTER_ENABLED === "true" ? "ON CLUSTER " + env.CLICKHOUSE_CLUSTER_NAME : ""}
(
project_id String,
run_id String,
@@ -289,7 +289,7 @@ const getObservationLatencyAndCostForDataset = async (
const query = `
WITH agg AS (
SELECT
dateDiff('milliseconds', start_time, end_time) AS latency_ms,
dateDiff('millisecond', start_time, end_time) AS latency_ms,
total_cost AS cost,
run_id
FROM observations AS o
@@ -342,7 +342,7 @@ const getTraceLatencyAndCostForDataset = async (
SELECT
o.trace_id,
run_id,
dateDiff('milliseconds', min(start_time), max(end_time)) AS latency_ms,
dateDiff('millisecond', min(start_time), max(end_time)) AS latency_ms,
sum(total_cost) AS cost
FROM observations o JOIN ${tableName} tmp
ON tmp.project_id = o.project_id
@@ -531,10 +531,10 @@ export const getRunItemsByRunIdOrItemId = async (
) => {
const [traceScores, observationAggregates, traceAggregate] =
await Promise.all([
getScoresForTraces(
getScoresForTraces({
projectId,
runItems.map((ri) => ri.traceId),
),
traceIds: runItems.map((ri) => ri.traceId),
}),
getLatencyAndTotalCostForObservations(
projectId,
runItems
+12 -3
View File
@@ -17,11 +17,13 @@ export const GetModelResultSchema = z.object({
projectId: z.string().nullable(),
modelName: z.string(),
matchPattern: z.string(),
tokenizerConfig: z
.record(z.union([z.string(), z.coerce.number()]))
.nullable(),
tokenizerConfig: z.union([
z.record(z.union([z.string(), z.coerce.number()])).nullable(),
z.string(),
]),
tokenizerId: TokenizerSchema,
prices: PriceMapSchema,
lastUsed: z.date().nullish(),
});
export type GetModelResult = z.infer<typeof GetModelResultSchema>;
@@ -75,3 +77,10 @@ export enum PriceUnit {
Per1KUnits = "per 1K units",
Per1MUnits = "per 1M units",
}
export const ModelLastUsedQueryResult = z.array(
z.object({
modelId: z.string(),
lastUsed: z.coerce.date(),
}),
);
@@ -132,18 +132,18 @@ export const projectsRouter = createTRPCRouter({
action: "delete",
});
// Delete API keys from DB first
// API keys need to be deleted from cache. Otherwise, they will still be valid.
await new ApiAuthService(ctx.prisma, redis).invalidateProjectApiKeys(
input.projectId,
);
// Delete API keys from DB
await ctx.prisma.apiKey.deleteMany({
where: {
projectId: input.projectId,
},
});
// API keys need to be deleted from cache. Otherwise, they will still be valid.
await new ApiAuthService(ctx.prisma, redis).invalidateProjectApiKeys(
input.projectId,
);
await ctx.prisma.project.update({
where: {
id: input.projectId,
@@ -14,11 +14,7 @@ import useProjectIdFromURL from "@/src/hooks/useProjectIdFromURL";
import { useUiCustomization } from "@/src/ee/features/ui-customization/useUiCustomization";
import { CreateLLMApiKeyForm } from "@/src/features/public-api/components/CreateLLMApiKeyForm";
export function CreateLLMApiKeyDialog({
evalModelsOnly,
}: {
evalModelsOnly?: boolean;
}) {
export function CreateLLMApiKeyDialog() {
const projectId = useProjectIdFromURL();
const [open, setOpen] = useState(false);
const hasAccess = useHasProjectAccess({
@@ -49,7 +45,6 @@ export function CreateLLMApiKeyDialog({
{open && (
<CreateLLMApiKeyForm
projectId={projectId}
evalModelsOnly={evalModelsOnly}
onSuccess={() => setOpen(false)}
customization={uiCustomization}
/>
@@ -66,12 +66,10 @@ const formSchema = z
export function CreateLLMApiKeyForm({
projectId,
evalModelsOnly,
onSuccess,
customization,
}: {
projectId?: string;
evalModelsOnly?: boolean;
onSuccess: () => void;
customization: ReturnType<typeof useUiCustomization>;
}) {
@@ -248,18 +246,11 @@ export function CreateLLMApiKeyForm({
</SelectTrigger>
</FormControl>
<SelectContent>
{Object.values(LLMAdapter)
.filter(
(provider) =>
!evalModelsOnly ||
provider === LLMAdapter.OpenAI ||
provider === LLMAdapter.Azure,
)
.map((provider) => (
<SelectItem value={provider} key={provider}>
{provider}
</SelectItem>
))}
{Object.values(LLMAdapter).map((provider) => (
<SelectItem value={provider} key={provider}>
{provider}
</SelectItem>
))}
</SelectContent>
</Select>
<FormMessage />
@@ -61,7 +61,7 @@ export const generateObservationsForPublicApi = async (props: QueryType) => {
WHERE o.project_id = {projectId: String}
${traceFilter ? `AND t.project_id = {projectId: String}` : ""}
AND ${appliedFilter.query}
ORDER BY start_time desc
ORDER BY event_ts desc
LIMIT 1 by id, project_id
${props.limit !== undefined && props.page !== undefined ? `LIMIT {limit: Int32} OFFSET {offset: Int32}` : ""}
`;
+1 -1
View File
@@ -58,7 +58,7 @@ export const generateTracesForPublicApi = async (
trace_id,
project_id,
sum(total_cost) as total_cost,
date_diff('milliseconds', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds,
date_diff('millisecond', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds,
groupArray(id) as observation_ids
FROM observations FINAL
WHERE project_id = {projectId: String}
+3 -3
View File
@@ -10,11 +10,11 @@ import {
} from "@langfuse/shared/src/server";
import { env } from "@/src/env.mjs";
// Interval between jobs in milliseconds
const JOB_INTERVAL_MINUTES = Prisma.raw("60");
// Interval between jobs in minutes
const JOB_INTERVAL_MINUTES = Prisma.raw("24 * 60"); // 24 hours
// Timeout for job in minutes, if job is not finished in this time, it will be retried
const JOB_TIMEOUT_MINUTES = Prisma.raw("10");
const JOB_TIMEOUT_MINUTES = Prisma.raw("10"); // 10 minutes
export async function telemetry() {
try {
+9
View File
@@ -26,6 +26,7 @@ if (!process.env.VERCEL && process.env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION) {
const sdk = new NodeSDK({
resource: new Resource({
"service.name": env.OTEL_SERVICE_NAME,
"service.version": env.BUILD_ID,
}),
traceExporter: new OTLPTraceExporter({
url: `${env.OTEL_EXPORTER_OTLP_ENDPOINT}/v1/traces`,
@@ -34,6 +35,14 @@ if (!process.env.VERCEL && process.env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION) {
new IORedisInstrumentation(),
new HttpInstrumentation({
requireParentforOutgoingSpans: true,
ignoreIncomingRequestHook: (req) => {
// Ignore health checks
return [
"/api/public/health",
"/api/public/ready",
"/api/health",
].some((path) => req.url?.includes(path));
},
ignoreOutgoingRequestHook: (req) => {
return req.host === "127.0.0.1";
},
+104 -9
View File
@@ -2,11 +2,22 @@ import { type NextApiRequest, type NextApiResponse } from "next";
import { z } from "zod";
import { logger, QueueName, getQueue } from "@langfuse/shared/src/server";
import { env } from "@/src/env.mjs";
import { prisma } from "@langfuse/shared/src/db";
/*
This API route is used by Langfuse Cloud to retry failed bullmq jobs.
*/
const BullStatus = z.enum([
"completed",
"failed",
"active",
"delayed",
"prioritized",
"paused",
"wait",
]);
const ManageBullBody = z.discriminatedUnion("action", [
z.object({
action: z.literal("retry"),
@@ -15,15 +26,18 @@ const ManageBullBody = z.discriminatedUnion("action", [
z.object({
action: z.literal("remove"),
queueNames: z.array(z.string()),
bullStatus: z.enum([
"completed",
"failed",
"active",
"delayed",
"prioritized",
"paused",
"wait",
]),
bullStatus: BullStatus,
}),
z.object({
action: z.literal("backup"),
queueName: z.string(),
bullStatus: BullStatus,
numberOfEvents: z.number(),
}),
z.object({
action: z.literal("restore"),
queueName: z.string(),
numberOfEvents: z.number(),
}),
]);
@@ -162,6 +176,21 @@ export default async function handler(
return res.status(200).json({ message: "Retried all jobs" });
}
if (req.method === "POST" && body.data.action === "backup") {
await backUpEvents(
body.data.queueName as QueueName,
body.data.numberOfEvents,
body.data.bullStatus,
);
}
if (req.method === "POST" && body.data.action === "restore") {
await restoreEvents(
body.data.queueName as QueueName,
body.data.numberOfEvents,
);
}
// return not implemented error
res.status(404).json({ error: "Action does not exist" });
} catch (e) {
@@ -169,3 +198,69 @@ export default async function handler(
res.status(500).json({ error: e });
}
}
const backUpEvents = async (
queueName: QueueName,
numberOfEvents: number,
bullStatus: z.infer<typeof BullStatus>,
) => {
const queue = getQueue(queueName);
let processedEvents = 0;
const batchSize = 1000;
while (processedEvents < numberOfEvents) {
const remainingEvents = numberOfEvents - processedEvents;
const currentBatchSize = Math.min(batchSize, remainingEvents);
const events = await queue?.getJobs(
[bullStatus],
0,
currentBatchSize,
true,
);
if (!events || events.length === 0) {
break;
}
await prisma.queueBackUp.createMany({
data: events.map((event) => ({
queueName,
content: event,
projectId: event.data.projectId ?? undefined,
createdAt: new Date(),
})),
});
// remove events from the queue but might throw in case if the job is already processing
await Promise.all(
events.map(async (event) => {
try {
await event.remove();
} catch (error) {
logger.error(`Failed to remove event ${event.id}:`, error);
}
}),
);
processedEvents += events.length;
}
};
const restoreEvents = async (queueName: QueueName, numberOfEvents: number) => {
const queue = getQueue(queueName);
const queueBackUp = await prisma.queueBackUp.findMany({
where: {
queueName,
},
orderBy: {
createdAt: "desc",
},
take: numberOfEvents,
});
await queue?.addBulk(
queueBackUp.map((event) => ({ name: queueName, data: event.content })),
);
};
@@ -73,6 +73,7 @@ export default withMiddlewares({
}),
bodySchema: PatchMediaBodySchema,
responseSchema: z.void(),
rateLimitResource: "ingestion",
fn: async ({ query, body, auth }) => {
if (auth.scope.accessLevel !== "all") throw new ForbiddenError();
+1
View File
@@ -26,6 +26,7 @@ export default withMiddlewares({
bodySchema: GetMediaUploadUrlQuerySchema,
responseSchema: GetMediaUploadUrlResponseSchema,
successStatusCode: 201,
rateLimitResource: "ingestion",
fn: async ({ body, auth }) => {
if (auth.scope.accessLevel !== "all") throw new ForbiddenError();
+9 -27
View File
@@ -1,4 +1,3 @@
import { env } from "@/src/env.mjs";
import { createAuthedAPIRoute } from "@/src/features/public-api/server/createAuthedAPIRoute";
import { withMiddlewares } from "@/src/features/public-api/server/withMiddlewares";
import { measureAndReturnApi } from "@/src/server/utils/checkClickhouseAccess";
@@ -65,32 +64,15 @@ export default withMiddlewares({
fn: async ({ query, auth }) => {
const { scoreId } = query;
const score = await prisma.score.findUnique({
select: {
id: true,
},
where: {
id: scoreId,
projectId: auth.scope.projectId,
},
});
if (!score) {
throw new LangfuseNotFoundError(
"Score not found within authorized project",
);
}
if (env.CLICKHOUSE_URL) {
await deleteScore(auth.scope.projectId, scoreId);
}
await prisma.score.delete({
where: {
id: scoreId,
projectId: auth.scope.projectId,
},
});
await Promise.all([
prisma.score.deleteMany({
where: {
id: scoreId,
projectId: auth.scope.projectId,
},
}),
deleteScore(auth.scope.projectId, scoreId),
]);
return { message: "Score deleted successfully" };
},
@@ -22,28 +22,9 @@ export default withMiddlewares({
operation: "scores.countAll",
user: null,
pgExecution: async () => {
const session = await prisma.traceSession.findUnique({
where: {
id_projectId: {
id: sessionId,
projectId: auth.scope.projectId,
},
},
select: {
id: true,
createdAt: true,
projectId: true,
traces: true,
},
});
if (!session) {
throw new LangfuseNotFoundError(
"Session not found within authorized project",
);
}
return session;
throw new LangfuseNotFoundError(
"Session not found within authorized project",
);
},
clickhouseExecution: async () => {
const session = await prisma.traceSession.findUnique({
+5 -5
View File
@@ -91,11 +91,11 @@ export default withMiddlewares({
trace?.timestamp,
true,
),
getScoresForTraces(
auth.scope.projectId,
[traceId],
trace?.timestamp,
),
getScoresForTraces({
projectId: auth.scope.projectId,
traceIds: [traceId],
timestamp: trace?.timestamp,
}),
]);
const uniqueModels: string[] = Array.from(
+17 -45
View File
@@ -11,7 +11,10 @@ import { auditLog } from "@/src/features/audit-logs/auditLog";
import { TRPCError } from "@trpc/server";
import { validateCommentReferenceObject } from "@/src/features/comments/validateCommentReferenceObject";
import { measureAndReturnApi } from "@/src/server/utils/checkClickhouseAccess";
import { getTracesIdentifierForSession } from "@langfuse/shared/src/server";
import {
getTracesIdentifierForSession,
logger,
} from "@langfuse/shared/src/server";
export const commentsRouter = createTRPCRouter({
create: protectedProjectProcedure
@@ -56,7 +59,7 @@ export const commentsRouter = createTRPCRouter({
return comment;
} catch (error) {
console.error(error);
logger.error("Failed to call comments.create", error);
if (error instanceof TRPCError) {
throw error;
}
@@ -115,7 +118,7 @@ export const commentsRouter = createTRPCRouter({
before: comment,
});
} catch (error) {
console.error(error);
logger.error("Failed to call comments.delete", error);
if (error instanceof TRPCError) {
throw error;
}
@@ -172,7 +175,7 @@ export const commentsRouter = createTRPCRouter({
return comments;
} catch (error) {
console.error(error);
logger.error("Failed to call comments.getByObjectId", error);
if (error instanceof TRPCError) {
throw error;
}
@@ -207,7 +210,7 @@ export const commentsRouter = createTRPCRouter({
});
return new Map([[input.objectId, commentCount]]);
} catch (error) {
console.error(error);
logger.error("Failed to call comments.getCountByObjectId", error);
if (error instanceof TRPCError) {
throw error;
}
@@ -251,7 +254,7 @@ export const commentsRouter = createTRPCRouter({
]),
);
} catch (error) {
console.error(error);
logger.error("Failed to call comments.getCountByObjectType", error);
if (error instanceof TRPCError) {
throw error;
}
@@ -276,44 +279,10 @@ export const commentsRouter = createTRPCRouter({
operation: "comments.getTraceCommentCountsBySessionId",
user: ctx.session.user ?? undefined,
pgExecution: async () => {
const session = await ctx.prisma.traceSession.findFirst({
where: {
id: input.sessionId,
projectId: input.projectId,
},
include: {
traces: {
select: {
id: true,
},
},
},
throw new TRPCError({
code: "NOT_FOUND",
message: "Session not found in project",
});
if (!session) {
throw new TRPCError({
code: "NOT_FOUND",
message: "Session not found in project",
});
}
const allTraceCommentCounts = await ctx.prisma.$queryRaw<
Array<{ objectId: string; count: bigint }>
>`
SELECT object_id as "objectId", COUNT(*) as count
FROM comments
WHERE project_id = ${input.projectId}
AND object_type = 'TRACE'
GROUP BY object_id
`;
const traceIds = new Set(
session.traces.map((t: { id: string }) => t.id),
);
return new Map(
allTraceCommentCounts
.filter((c) => traceIds.has(c.objectId))
.map(({ objectId, count }) => [objectId, Number(count)]),
);
},
clickhouseExecution: async () => {
const clickhouseTraces = await getTracesIdentifierForSession(
@@ -339,8 +308,11 @@ export const commentsRouter = createTRPCRouter({
);
},
});
} catch (e) {
console.error(e);
} catch (error) {
logger.error(
"Failed to call comments.getTraceCommentCountBySessionId",
error,
);
throw new TRPCError({
code: "INTERNAL_SERVER_ERROR",
message: "Unable to get trace comment counts by session id",
+68 -4
View File
@@ -5,6 +5,7 @@ import { auditLog } from "@/src/features/audit-logs/auditLog";
import { isValidPostgresRegex } from "@/src/features/models/server/isValidPostgresRegex";
import {
GetModelResultSchema,
ModelLastUsedQueryResult,
UpsertModelSchema,
} from "@/src/features/models/validation";
import { throwIfNoProjectAccess } from "@/src/features/rbac/utils/checkProjectAccess";
@@ -13,6 +14,7 @@ import {
protectedProjectProcedure,
} from "@/src/server/api/trpc";
import { ModelUsageUnit, paginationZod } from "@langfuse/shared";
import { queryClickhouse } from "@langfuse/shared/src/server";
import { TRPCError } from "@trpc/server";
const ModelAllOptions = z.object({
@@ -20,6 +22,18 @@ const ModelAllOptions = z.object({
...paginationZod,
});
const paginateArray = <T>(params: {
limit: number;
page: number;
data: Array<T>;
}): Array<T> => {
const { data, limit, page } = params;
const startIndex = limit * page;
const endIndex = startIndex + limit;
return data.slice(startIndex, endIndex);
};
export const modelRouter = createTRPCRouter({
getById: protectedProjectProcedure
.input(z.object({ projectId: z.string(), modelId: z.string() }))
@@ -68,6 +82,8 @@ export const modelRouter = createTRPCRouter({
getAll: protectedProjectProcedure
.input(ModelAllOptions)
.query(async ({ input, ctx }) => {
const { projectId, page, limit } = input;
const [allModelsQueryResult, totalCountQuery] = await Promise.all([
// All models
ctx.prisma.$queryRaw`
@@ -93,12 +109,11 @@ export const modelRouter = createTRPCRouter({
models m
WHERE
project_id IS NULL
OR project_id = ${input.projectId}
OR project_id = ${projectId}
ORDER BY
project_id,
model_name,
m.created_at DESC NULLS LAST
LIMIT ${input.limit} OFFSET ${input.page * input.limit};
`,
// Total count
@@ -110,7 +125,7 @@ export const modelRouter = createTRPCRouter({
SELECT COUNT(DISTINCT (project_id, model_name))
FROM models
WHERE project_id IS NULL
OR project_id = '${input.projectId}';
OR project_id = ${projectId};
`,
]);
@@ -119,8 +134,57 @@ export const modelRouter = createTRPCRouter({
.parse(allModelsQueryResult);
const totalCount = z.coerce.number().parse(totalCountQuery[0].count);
// Check whether model was used in last month
const lastUsedQuery = `
SELECT
internal_model_id as modelId,
MAX(start_time) as lastUsed
FROM
observations
WHERE
project_id = {projectId: String}
AND internal_model_id IS NOT NULL
GROUP BY
internal_model_id
`;
const lastUsedQueryResult = ModelLastUsedQueryResult.safeParse(
await queryClickhouse({
query: lastUsedQuery,
params: { projectId },
}),
);
if (lastUsedQueryResult.success) {
const lastUsedMap = new Map(
lastUsedQueryResult.data.map((res) => [res.modelId, res.lastUsed]),
);
const sortedModels = allModels
.map((m) => ({ ...m, lastUsed: lastUsedMap.get(m.id) }))
.sort((a, b) => {
const lastUsedA = lastUsedMap.get(a.id);
const lastUsedB = lastUsedMap.get(b.id);
if (lastUsedA && lastUsedB) {
return lastUsedB.getTime() - lastUsedA.getTime();
} else if (lastUsedA) {
return -1;
} else if (lastUsedB) {
return 1;
}
return 0;
});
return {
models: paginateArray({ data: sortedModels, page, limit }),
totalCount,
};
}
return {
models: allModels,
models: paginateArray({ data: allModels, page, limit }),
totalCount,
};
}),
+66 -85
View File
@@ -32,6 +32,8 @@ import {
getCostForTraces,
getTracesGroupedByUsers,
getPublicSessionsFilter,
logger,
getSessionsWithMetrics,
} from "@langfuse/shared/src/server";
const SessionFilterOptions = z.object({
@@ -108,14 +110,7 @@ export const sessionRouter = createTRPCRouter({
sessions: sessions.map((s) => ({
id: s.session_id,
userIds: s.user_ids,
countTraces: s.trace_ids.length,
sessionDuration: Number(s.duration) / 1000,
inputCost: new Decimal(s.session_input_cost),
outputCost: new Decimal(s.session_output_cost),
totalCost: new Decimal(s.session_total_cost),
promptTokens: Number(s.session_input_usage),
completionTokens: Number(s.session_output_usage),
totalTokens: Number(s.session_total_usage),
countTraces: s.trace_count,
traceTags: s.trace_tags,
createdAt: new Date(s.min_timestamp),
bookmarked:
@@ -129,7 +124,7 @@ export const sessionRouter = createTRPCRouter({
},
});
} catch (e) {
console.error(e);
logger.error("Unable to call sessions.all", e);
throw new TRPCError({
code: "INTERNAL_SERVER_ERROR",
message: "unable to get sessions",
@@ -196,7 +191,7 @@ export const sessionRouter = createTRPCRouter({
},
});
} catch (e) {
console.error(e);
logger.error("Error in sessions.countAll", e);
throw new TRPCError({
code: "INTERNAL_SERVER_ERROR",
message: "unable to get session count",
@@ -273,11 +268,59 @@ export const sessionRouter = createTRPCRouter({
}));
},
clickhouseExecution: async () => {
return []; // initial endpoint returns all data from CH.
const finalFilter = await getPublicSessionsFilter(input.projectId, [
{
column: "id",
type: "stringOptions",
operator: "any of",
value: input.sessionIds,
},
]);
const sessions = await getSessionsWithMetrics({
projectId: input.projectId,
filter: finalFilter,
});
const prismaSessionInfo = await ctx.prisma.traceSession.findMany({
where: {
id: {
in: sessions.map((s) => s.session_id),
},
projectId: input.projectId,
},
select: {
id: true,
bookmarked: true,
public: true,
},
});
return sessions.map((s) => ({
id: s.session_id,
userIds: s.user_ids,
countTraces: s.trace_count,
traceTags: s.trace_tags,
createdAt: new Date(s.min_timestamp),
bookmarked:
prismaSessionInfo.find((p) => p.id === s.session_id)
?.bookmarked ?? false,
public:
prismaSessionInfo.find((p) => p.id === s.session_id)?.public ??
false,
trace_count: Number(s.trace_count),
total_observations: Number(s.total_observations),
sessionDuration: Number(s.duration) / 1000,
inputCost: new Decimal(s.session_input_cost),
outputCost: new Decimal(s.session_output_cost),
totalCost: new Decimal(s.session_total_cost),
promptTokens: Number(s.session_input_usage),
completionTokens: Number(s.session_output_usage),
totalTokens: Number(s.session_total_usage),
}));
},
});
} catch (e) {
console.error(e);
logger.error("Error in sessions.metrics", e);
throw new TRPCError({
code: "INTERNAL_SERVER_ERROR",
message: "unable to get session metrics",
@@ -391,7 +434,7 @@ export const sessionRouter = createTRPCRouter({
},
});
} catch (e) {
console.error(e);
logger.error("Unable to get sessions.filterOptions", e);
throw new TRPCError({
code: "INTERNAL_SERVER_ERROR",
message: "unable to get session filter options",
@@ -413,75 +456,10 @@ export const sessionRouter = createTRPCRouter({
operation: "sessions.byId",
user: ctx.session.user ?? undefined,
pgExecution: async () => {
const session = await ctx.prisma.traceSession.findFirst({
where: {
id: input.sessionId,
projectId: input.projectId,
},
include: {
traces: {
orderBy: {
timestamp: "asc",
},
select: {
id: true,
userId: true,
name: true,
timestamp: true,
},
},
},
throw new TRPCError({
code: "NOT_FOUND",
message: "Session not found in project",
});
if (!session) {
throw new TRPCError({
code: "NOT_FOUND",
message: "Session not found in project",
});
}
const totalCostQuery = Prisma.sql`
SELECT
SUM(COALESCE(o."calculated_total_cost", 0)) AS "totalCost"
FROM observations_view o
JOIN traces t ON t.id = o.trace_id
WHERE
t."session_id" = ${input.sessionId}
AND t."project_id" = ${input.projectId}
`;
const [scores, costData] = await Promise.all([
ctx.prisma.score.findMany({
where: {
traceId: {
in: session.traces.map((t) => t.id),
},
projectId: input.projectId,
},
}),
// costData
ctx.prisma.$queryRaw<Array<{ totalCost: number }>>(
totalCostQuery,
),
]);
const validatedScores = filterAndValidateDbScoreList(
scores,
traceException,
);
return {
...session,
traces: session.traces.map((t) => ({
...t,
scores: validatedScores.filter((s) => s.traceId === t.id),
})),
totalCost: costData[0].totalCost ?? 0,
users: [
...new Set(
session.traces.map((t) => t.userId).filter((t) => t !== null),
),
],
};
},
clickhouseExecution: async () => {
const postgresSession = await ctx.prisma.traceSession.findFirst({
@@ -514,7 +492,10 @@ export const sessionRouter = createTRPCRouter({
const [scores, costs] = await Promise.all([
Promise.all(
chunks.map((chunk) =>
getScoresForTraces(input.projectId, chunk),
getScoresForTraces({
projectId: input.projectId,
traceIds: chunk,
}),
),
).then((results) => results.flat()),
Promise.all(
@@ -549,7 +530,7 @@ export const sessionRouter = createTRPCRouter({
},
});
} catch (e) {
console.error(e);
logger.error("Unable to get sessions.byId", e);
throw new TRPCError({
code: "INTERNAL_SERVER_ERROR",
message: "unable to get session",
@@ -593,7 +574,7 @@ export const sessionRouter = createTRPCRouter({
});
return session;
} catch (error) {
console.error(error);
logger.error("Unable to call sessions.bookmark", error);
if (
error instanceof Prisma.PrismaClientKnownRequestError &&
error.code === "P2025" // Record to update not found
@@ -643,7 +624,7 @@ export const sessionRouter = createTRPCRouter({
},
});
} catch (e) {
console.error(e);
logger.error("Unable to call sessions.publish", e);
throw new TRPCError({
code: "INTERNAL_SERVER_ERROR",
message: "unable to publish session",
+14 -17
View File
@@ -291,13 +291,12 @@ export const traceRouter = createTRPCRouter({
],
});
const scores = await getScoresForTraces(
ctx.session.projectId,
res.map((r) => r.id),
undefined,
1000,
0,
);
const scores = await getScoresForTraces({
projectId: ctx.session.projectId,
traceIds: res.map((r) => r.id),
limit: 1000,
offset: 0,
});
const validatedScores = filterAndValidateDbScoreList(
scores,
@@ -576,13 +575,11 @@ export const traceRouter = createTRPCRouter({
input.projectId,
input.timestamp ?? undefined,
),
getScoresForTraces(
input.projectId,
[input.traceId],
input.timestamp ?? undefined,
undefined,
undefined,
),
getScoresForTraces({
projectId: input.projectId,
traceIds: [input.traceId],
timestamp: input.timestamp ?? undefined,
}),
]);
if (!trace) {
@@ -769,7 +766,7 @@ export const traceRouter = createTRPCRouter({
return trace;
} catch (error) {
console.error(error);
logger.error("Failed to call traces.bookmark", error);
if (
error instanceof Prisma.PrismaClientKnownRequestError &&
error.code === "P2025" // Record to update not found
@@ -839,7 +836,7 @@ export const traceRouter = createTRPCRouter({
return clickhouseTrace;
}
} catch (error) {
console.error(error);
logger.error("Failed to call traces.publish", error);
if (
error instanceof Prisma.PrismaClientKnownRequestError &&
error.code === "P2025" // Record to update not found
@@ -907,7 +904,7 @@ export const traceRouter = createTRPCRouter({
await upsertTrace(convertTraceDomainToClickhouse(clickhouseTrace));
}
} catch (error) {
console.error(error);
logger.error("Failed to call traces.updateTags", error);
throw new TRPCError({
code: "INTERNAL_SERVER_ERROR",
});
+3
View File
@@ -54,6 +54,9 @@ RUN apk add --no-cache dumb-init
WORKDIR /app
ARG NEXT_PUBLIC_BUILD_ID
ENV BUILD_ID=$NEXT_PUBLIC_BUILD_ID
ENV NODE_ENV production
ENV DOCKER_BUILD 0
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "worker",
"version": "3.5.2",
"version": "3.6.2",
"description": "",
"license": "MIT",
"private": true,
+14 -1
View File
@@ -380,7 +380,17 @@ describe("batch export test suite", () => {
value: 123,
});
await createScoresCh([score]);
const qualitativeScore = createScore({
project_id: projectId,
trace_id: traces[0].id,
observation_id: generations[0].id,
name: "qualitative_test",
value: undefined,
string_value: "This is some qualitative text",
data_type: "CATEGORICAL",
});
await createScoresCh([score, qualitativeScore]);
await createObservationsCh(generations);
const stream = await getDatabaseReadStream({
@@ -405,10 +415,13 @@ describe("batch export test suite", () => {
id: traces[0].id,
latency: 1,
test: [score.value],
qualitative_test: ["This is some qualitative text"],
}),
expect.objectContaining({
id: traces[1].id,
latency: 2.123,
test: null,
qualitative_test: null,
}),
]),
);
+1 -1
View File
@@ -1 +1 @@
export const VERSION = "v3.5.2";
export const VERSION = "v3.6.2";
+48 -15
View File
@@ -8,6 +8,7 @@
"prices": {
"input": 2.5e-6,
"input_cached_tokens": 1.25e-6,
"input_cache_read": 1.25e-6,
"output": 10e-6
},
"tokenizer_config": {
@@ -844,7 +845,9 @@
"output": 15e-6,
"output_tokens": 15e-6,
"cache_creation_input_tokens": 3.75e-6,
"cache_read_input_tokens": 0.3e-6
"input_cache_creation": 3.75e-6,
"cache_read_input_tokens": 0.3e-6,
"input_cache_read": 0.3e-6
},
"tokenizer_config": null,
"tokenizer_id": "claude"
@@ -1010,7 +1013,9 @@
"output": 15e-6,
"output_tokens": 15e-6,
"cache_creation_input_tokens": 3.75e-6,
"cache_read_input_tokens": 0.3e-6
"input_cache_creation": 3.75e-6,
"cache_read_input_tokens": 0.3e-6,
"input_cache_read": 0.3e-6
},
"tokenizer_config": null,
"tokenizer_id": "claude"
@@ -1024,7 +1029,8 @@
"prices": {
"input": 0.15e-6,
"output": 0.6e-6,
"input_cached_tokens": 0.075e-6
"input_cached_tokens": 0.075e-6,
"input_cache_read": 0.075e-6
},
"tokenizer_config": {
"tokensPerName": 1,
@@ -1042,6 +1048,7 @@
"prices": {
"input": 0.15e-6,
"input_cached_tokens": 0.075e-6,
"input_cache_read": 0.075e-6,
"output": 0.6e-6
},
"tokenizer_config": {
@@ -1060,6 +1067,7 @@
"prices": {
"input": 2.5e-6,
"input_cached_tokens": 1.25e-6,
"input_cache_read": 1.25e-6,
"output": 10e-6
},
"tokenizer_config": {
@@ -1078,8 +1086,10 @@
"prices": {
"input": 15e-6,
"input_cached_tokens": 7.5e-6,
"input_cache_read": 7.5e-6,
"output": 60e-6,
"output_reasoning_tokens": 60e-6
"output_reasoning_tokens": 60e-6,
"output_reasoning": 60e-6
},
"tokenizer_config": null,
"tokenizer_id": null
@@ -1093,8 +1103,10 @@
"prices": {
"input": 15e-6,
"input_cached_tokens": 7.5e-6,
"input_cache_read": 7.5e-6,
"output": 60e-6,
"output_reasoning_tokens": 60e-6
"output_reasoning_tokens": 60e-6,
"output_reasoning": 60e-6
},
"tokenizer_config": null,
"tokenizer_id": null
@@ -1108,8 +1120,10 @@
"prices": {
"input": 3e-6,
"input_cached_tokens": 1.5e-6,
"input_cache_read": 1.5e-6,
"output": 12e-6,
"output_reasoning_tokens": 12e-6
"output_reasoning_tokens": 12e-6,
"output_reasoning": 12e-6
},
"tokenizer_config": null,
"tokenizer_id": null
@@ -1123,8 +1137,10 @@
"prices": {
"input": 3e-6,
"input_cached_tokens": 1.5e-6,
"input_cache_read": 1.5e-6,
"output": 12e-6,
"output_reasoning_tokens": 12e-6
"output_reasoning_tokens": 12e-6,
"output_reasoning": 12e-6
},
"tokenizer_config": null,
"tokenizer_id": null
@@ -1141,7 +1157,9 @@
"output": 15e-6,
"output_tokens": 15e-6,
"cache_creation_input_tokens": 3.75e-6,
"cache_read_input_tokens": 0.3e-6
"input_cache_creation": 3.75e-6,
"cache_read_input_tokens": 0.3e-6,
"input_cache_read": 0.3e-6
},
"tokenizer_config": null,
"tokenizer_id": "claude"
@@ -1158,7 +1176,9 @@
"output": 15e-6,
"output_tokens": 15e-6,
"cache_creation_input_tokens": 3.75e-6,
"cache_read_input_tokens": 0.3e-6
"input_cache_creation": 3.75e-6,
"cache_read_input_tokens": 0.3e-6,
"input_cache_read": 0.3e-6
},
"tokenizer_config": null,
"tokenizer_id": "claude"
@@ -1175,7 +1195,9 @@
"output": 4e-6,
"output_tokens": 4e-6,
"cache_creation_input_tokens": 1e-6,
"cache_read_input_tokens": 0.08e-6
"input_cache_creation": 1e-6,
"cache_read_input_tokens": 0.08e-6,
"input_cache_read": 0.08e-6
},
"tokenizer_config": null,
"tokenizer_id": "claude"
@@ -1192,7 +1214,9 @@
"output": 4e-6,
"output_tokens": 4e-6,
"cache_creation_input_tokens": 1e-6,
"cache_read_input_tokens": 0.08e-6
"input_cache_creation": 1e-6,
"cache_read_input_tokens": 0.08e-6,
"input_cache_read": 0.08e-6
},
"tokenizer_config": null,
"tokenizer_id": "claude"
@@ -1223,6 +1247,7 @@
"prices": {
"input": 2.5e-6,
"input_cached_tokens": 1.25e-6,
"input_cache_read": 1.25e-6,
"output": 10e-6
},
"tokenizer_config": {
@@ -1242,7 +1267,9 @@
"input_text_tokens": 2.5e-6,
"output_text_tokens": 10e-6,
"input_audio_tokens": 100e-6,
"output_audio_tokens": 200e-6
"input_audio": 100e-6,
"output_audio_tokens": 200e-6,
"output_audio": 200e-6
},
"tokenizer_config": null,
"tokenizer_id": null
@@ -1257,7 +1284,9 @@
"input_text_tokens": 2.5e-6,
"output_text_tokens": 10e-6,
"input_audio_tokens": 100e-6,
"output_audio_tokens": 200e-6
"input_audio": 100e-6,
"output_audio_tokens": 200e-6,
"output_audio": 200e-6
},
"tokenizer_config": null,
"tokenizer_id": null
@@ -1273,8 +1302,10 @@
"input_cached_text_tokens": 2.5e-6,
"output_text_tokens": 20e-6,
"input_audio_tokens": 100e-6,
"input_audio": 100e-6,
"input_cached_audio_tokens": 20e-6,
"output_audio_tokens": 200e-6
"output_audio_tokens": 200e-6,
"output_audio": 200e-6
},
"tokenizer_config": null,
"tokenizer_id": null
@@ -1290,8 +1321,10 @@
"input_cached_text_tokens": 2.5e-6,
"output_text_tokens": 20e-6,
"input_audio_tokens": 100e-6,
"input_audio": 100e-6,
"input_cached_audio_tokens": 20e-6,
"output_audio_tokens": 200e-6
"output_audio_tokens": 200e-6,
"output_audio": 200e-6
},
"tokenizer_config": null,
"tokenizer_id": null
+8
View File
@@ -2,6 +2,7 @@ import { removeEmptyEnvVariables } from "@langfuse/shared";
import { z } from "zod";
const EnvSchema = z.object({
BUILD_ID: z.string().optional(),
NODE_ENV: z
.enum(["development", "test", "production"])
.default("development"),
@@ -82,6 +83,8 @@ const EnvSchema = z.object({
CLICKHOUSE_URL: z.string().url(),
CLICKHOUSE_USER: z.string(),
CLICKHOUSE_CLUSTER_NAME: z.string().default("default"),
CLICKHOUSE_DB: z.string().default("default"),
CLICKHOUSE_PASSWORD: z.string(),
LANGFUSE_LEGACY_INGESTION_WORKER_CONCURRENCY: z.coerce
@@ -107,6 +110,11 @@ const EnvSchema = z.object({
// TODO: Remove for go-live
LANGFUSE_RETURN_FROM_CLICKHOUSE: z.enum(["true", "false"]).default("true"),
// Skip the read from ClickHouse within the Ingestion pipeline for the given
// project ids. Applicable for projects that were created after the S3 write
// was activated and which don't rely on historic updates.
LANGFUSE_SKIP_INGESTION_CLICKHOUSE_READ_PROJECT_IDS: z.string().default(""),
// Otel
OTEL_EXPORTER_OTLP_ENDPOINT: z.string().default("http://localhost:4318"),
OTEL_SERVICE_NAME: z.string().default("worker"),
@@ -24,7 +24,6 @@ import {
parseTraceAllFilters,
FullObservationsWithScores,
getPublicSessionsFilter,
getSessionsTable,
getScoresForObservations,
getObservationsTableWithModelData,
getDistinctScoreNames,
@@ -33,6 +32,7 @@ import {
getScoresForTraces,
logger,
getTracesByIds,
getSessionsWithMetrics,
} from "@langfuse/shared/src/server";
import { env } from "../../env";
import { BatchExportSessionsRow, BatchExportTracesRow } from "./types";
@@ -182,7 +182,7 @@ export const getDatabaseReadStream = async ({
projectId,
finalFilter ?? [],
);
const sessions = await getSessionsTable({
const sessions = await getSessionsWithMetrics({
projectId: projectId,
filter: sessionsFilter,
orderBy: orderBy,
@@ -382,10 +382,11 @@ export const getDatabaseReadStream = async ({
),
]);
const scores = await getScoresForTraces(
const scores = await getScoresForTraces({
projectId,
traces.map((t) => t.id),
);
traceIds: traces.map((t) => t.id),
});
chunk = traces.map((t) => {
const metric = metrics.find((m) => m.id === t.id);
const filteredScores = scores.filter((s) => s.traceId === t.id);
@@ -408,7 +409,15 @@ export const getDatabaseReadStream = async ({
completionTokens: metric?.completionTokens,
totalTokens: metric?.totalTokens,
},
inputCost: metric?.calculatedInputCost,
outputCost: metric?.calculatedOutputCost,
totalCost: metric?.calculatedTotalCost,
level: metric?.level,
observationCount: Number(metric?.observationCount),
scores: outputScores,
inputTokens: metric?.promptTokens,
outputTokens: metric?.completionTokens,
totalTokens: metric?.totalTokens,
};
});
} else {
@@ -604,7 +613,8 @@ function prepareScoresForOutput(
(acc, score) => {
// If this score name already exists in acc, use its existing type
const existingValues = acc[score.name];
const newValue = score.value ?? score.stringValue;
const newValue =
score.dataType === "NUMERIC" ? score.value : score.stringValue;
if (!newValue) return acc;
if (!existingValues) {
+7
View File
@@ -25,6 +25,7 @@ dd.init({
const sdk = new NodeSDK({
resource: new Resource({
"service.name": env.OTEL_SERVICE_NAME,
"service.version": env.BUILD_ID,
}),
traceExporter: new OTLPTraceExporter({
url: `${env.OTEL_EXPORTER_OTLP_ENDPOINT}/v1/traces`,
@@ -33,6 +34,12 @@ const sdk = new NodeSDK({
new IORedisInstrumentation(),
new HttpInstrumentation({
requireParentforOutgoingSpans: true,
ignoreIncomingRequestHook: (req) => {
// Ignore health checks
return ["/api/public/health", "/api/public/ready", "/api/health"].some(
(path) => req.url?.includes(path),
);
},
ignoreOutgoingRequestHook: (req) => {
return req.host === "127.0.0.1";
},
+61 -13
View File
@@ -1,5 +1,5 @@
import { Job } from "bullmq";
import { BaseError } from "@langfuse/shared";
import { ApiError, BaseError } from "@langfuse/shared";
import { kyselyPrisma } from "@langfuse/shared/src/db";
import { sql } from "kysely";
import {
@@ -7,8 +7,11 @@ import {
TQueueJobTypes,
logger,
traceException,
EvalExecutionQueue,
QueueJobs,
} from "@langfuse/shared/src/server";
import { createEvalJobs, evaluate } from "../ee/evaluation/evalService";
import { randomUUID } from "crypto";
export const evalJobTraceCreatorQueueProcessor = async (
job: Job<TQueueJobTypes[QueueName.TraceUpsert]>,
@@ -50,6 +53,51 @@ export const evalJobExecutorQueueProcessor = async (
await evaluate({ event: job.data.payload });
return true;
} catch (e) {
// If the job fails with a 429, we want to retry it unless it's older than 24h.
if (e instanceof ApiError && e.httpCode === 429) {
try {
// Check if the job execution is older than 24h
const jobExecution = await kyselyPrisma.$kysely
.selectFrom("job_executions")
.select("created_at")
.where("id", "=", job.data.payload.jobExecutionId)
.where("project_id", "=", job.data.payload.projectId)
.executeTakeFirstOrThrow();
if (
// Do nothing if job execution is older than 24h
jobExecution.created_at < new Date(Date.now() - 24 * 60 * 60 * 1000)
) {
logger.info(
`Job ${job.data.payload.jobExecutionId} is rate limited for more than 24h. Stop retrying.`,
);
} else {
// Add the job into the queue with a random delay between 1 and 10min and return
const delay = Math.floor(Math.random() * 9 + 1) * 60 * 1000;
logger.info(
`Job ${job.data.payload.jobExecutionId} is rate limited. Retrying in ${delay}ms.`,
);
await EvalExecutionQueue.getInstance()?.add(
QueueName.EvaluationExecution,
{
name: QueueJobs.EvaluationExecution,
id: randomUUID(),
timestamp: new Date(),
payload: job.data.payload,
},
{
delay,
},
);
return;
}
} catch (innerErr) {
logger.error(
`Failed to handle 429 retry for ${job.data.payload.jobExecutionId}. Continuing regular processing.`,
innerErr,
);
}
}
const displayError =
e instanceof BaseError ? e.message : "An internal error occurred";
@@ -64,22 +112,22 @@ export const evalJobExecutorQueueProcessor = async (
// do not log expected errors (api failures + missing api keys not provided by the user)
if (
!(e instanceof BaseError && e.message.includes("API key for provider")) &&
!(
e instanceof BaseError &&
(e instanceof BaseError && e.message.includes("API key for provider")) || // api key not provided
(e instanceof ApiError && e.httpCode >= 400 && e.httpCode < 500) || // do not error and retry on 4xx errors. They are visible to the user in the UI but do not alert us.
(e instanceof ApiError && e.message.includes("TypeError")) || // Zod parsing the response failed. User should update prompt to consistently return expected output structure.
(e instanceof BaseError &&
e.message.includes(
"Please ensure the mapped data exists and consider extending the job delay.",
)
)
)) // Trace not found.
) {
traceException(e);
logger.error(
`Failed Evaluation_Execution job for id ${job.data.payload.jobExecutionId}`,
e,
);
throw e;
return;
}
return;
traceException(e);
logger.error(
`Failed Evaluation_Execution job for id ${job.data.payload.jobExecutionId}`,
e,
);
throw e;
}
};
+8 -2
View File
@@ -97,10 +97,15 @@ export const ingestionQueueProcessorBuilder = (
`${env.LANGFUSE_S3_EVENT_UPLOAD_PREFIX}${job.data.payload.authCheck.scope.projectId}/${getClickhouseEntityType(job.data.payload.data.type)}/${job.data.payload.data.eventBodyId}/`,
);
const firstS3WriteTime =
eventFiles
.map((fileRef) => fileRef.createdAt)
.sort()
.shift() ?? new Date();
const events: IngestionEventType[] = (
await Promise.all(
eventFiles.map(async (key) => {
const file = await s3Client.download(key);
eventFiles.map(async (fileRef) => {
const file = await s3Client.download(fileRef.file);
const parsedFile = JSON.parse(file);
return Array.isArray(parsedFile) ? parsedFile : [parsedFile];
}),
@@ -126,6 +131,7 @@ export const ingestionQueueProcessorBuilder = (
getClickhouseEntityType(events[0].type),
job.data.payload.authCheck.scope.projectId,
job.data.payload.data.eventBodyId,
firstS3WriteTime,
events,
);
} catch (e) {
+101 -192
View File
@@ -9,9 +9,6 @@ import {
convertObservationReadToInsert,
convertScoreReadToInsert,
convertTraceReadToInsert,
convertPostgresObservationToInsert,
convertPostgresScoreToInsert,
convertPostgresTraceToInsert,
eventTypes,
findModel,
IngestionEventType,
@@ -42,6 +39,7 @@ import { tokenCount } from "../../features/tokenisation/usage";
import { ClickhouseWriter, TableName } from "../ClickhouseWriter";
import { convertJsonSchemaToRecord, overwriteObject } from "./utils";
import { randomUUID } from "crypto";
import { env } from "../../env";
type InsertRecord =
| TraceRecordInsertType
@@ -86,6 +84,7 @@ export class IngestionService {
eventType: IngestionEntityTypes,
projectId: string,
eventBodyId: string,
createdAtTimestamp: Date,
events: IngestionEventType[],
): Promise<void> {
logger.debug(
@@ -97,18 +96,21 @@ export class IngestionService {
return await this.processTraceEventList({
projectId,
entityId: eventBodyId,
createdAtTimestamp,
traceEventList: events as TraceEventType[],
});
case "observation":
return await this.processObservationEventList({
projectId,
entityId: eventBodyId,
createdAtTimestamp,
observationEventList: events as ObservationEvent[],
});
case "score": {
return await this.processScoreEventList({
projectId,
entityId: eventBodyId,
createdAtTimestamp,
scoreEventList: events as ScoreEventType[],
});
}
@@ -118,9 +120,10 @@ export class IngestionService {
private async processScoreEventList(params: {
projectId: string;
entityId: string;
createdAtTimestamp: Date;
scoreEventList: ScoreEventType[];
}) {
const { projectId, entityId, scoreEventList } = params;
const { projectId, entityId, createdAtTimestamp, scoreEventList } = params;
if (scoreEventList.length === 0) return;
const timeSortedEvents =
@@ -135,59 +138,47 @@ export class IngestionService {
minTimestamp === Infinity
? undefined
: convertDateToClickhouseDateTime(new Date(minTimestamp));
const [postgresScoreRecord, clickhouseScoreRecord, scoreRecords] =
await Promise.all([
this.getPostgresRecord({
projectId,
entityId,
table: TableName.Scores,
}),
this.getClickhouseRecord({
projectId,
entityId,
table: TableName.Scores,
additionalFilters: {
whereCondition: timestamp
? " AND timestamp >= {timestamp: DateTime64(3)} "
: "",
params: { timestamp },
},
}),
Promise.all(
timeSortedEvents.map(async (scoreEvent) => {
const validatedScore = await validateAndInflateScore({
body: scoreEvent.body,
scoreId: entityId,
projectId,
});
const [clickhouseScoreRecord, scoreRecords] = await Promise.all([
this.getClickhouseRecord({
projectId,
entityId,
table: TableName.Scores,
additionalFilters: {
whereCondition: timestamp
? " AND timestamp >= {timestamp: DateTime64(3)} "
: "",
params: { timestamp },
},
}),
Promise.all(
timeSortedEvents.map(async (scoreEvent) => {
const validatedScore = await validateAndInflateScore({
body: scoreEvent.body,
scoreId: entityId,
projectId,
});
return {
id: entityId,
project_id: projectId,
timestamp: this.getMillisecondTimestamp(scoreEvent.timestamp),
name: validatedScore.name,
value: validatedScore.value,
source: validatedScore.source,
trace_id: validatedScore.traceId,
data_type: validatedScore.dataType,
observation_id: validatedScore.observationId,
comment: validatedScore.comment,
string_value: validatedScore.stringValue,
created_at: Date.now(),
updated_at: Date.now(),
event_ts: new Date(scoreEvent.timestamp).getTime(),
is_deleted: 0,
};
}),
),
]);
return {
id: entityId,
project_id: projectId,
timestamp: this.getMillisecondTimestamp(scoreEvent.timestamp),
name: validatedScore.name,
value: validatedScore.value,
source: validatedScore.source,
trace_id: validatedScore.traceId,
data_type: validatedScore.dataType,
observation_id: validatedScore.observationId,
comment: validatedScore.comment,
string_value: validatedScore.stringValue,
created_at: Date.now(),
updated_at: Date.now(),
event_ts: new Date(scoreEvent.timestamp).getTime(),
is_deleted: 0,
};
}),
),
]);
if (postgresScoreRecord) {
recordIncrement("langfuse.ingestion.lookup.hit", 1, {
store: "postgres",
object: "score",
});
}
if (clickhouseScoreRecord) {
recordIncrement("langfuse.ingestion.lookup.hit", 1, {
store: "clickhouse",
@@ -198,9 +189,10 @@ export class IngestionService {
const finalScoreRecord: ScoreRecordInsertType =
await this.mergeScoreRecords({
clickhouseScoreRecord,
postgresScoreRecord,
scoreRecords,
});
finalScoreRecord.created_at =
clickhouseScoreRecord?.created_at ?? createdAtTimestamp.getTime();
this.clickHouseWriter.addToQueue(TableName.Scores, finalScoreRecord);
}
@@ -208,9 +200,10 @@ export class IngestionService {
private async processTraceEventList(params: {
projectId: string;
entityId: string;
createdAtTimestamp: Date;
traceEventList: TraceEventType[];
}) {
const { projectId, entityId, traceEventList } = params;
const { projectId, entityId, createdAtTimestamp, traceEventList } = params;
if (traceEventList.length === 0) return;
const timeSortedEvents =
@@ -231,31 +224,18 @@ export class IngestionService {
minTimestamp === Infinity
? undefined
: convertDateToClickhouseDateTime(new Date(minTimestamp));
const [postgresTraceRecord, clickhouseTraceRecord] = await Promise.all([
this.getPostgresRecord({
projectId,
entityId,
table: TableName.Traces,
}),
this.getClickhouseRecord({
projectId,
entityId,
table: TableName.Traces,
additionalFilters: {
whereCondition: timestamp
? " AND timestamp >= {timestamp: DateTime64(3)} "
: "",
params: { timestamp },
},
}),
]);
const clickhouseTraceRecord = await this.getClickhouseRecord({
projectId,
entityId,
table: TableName.Traces,
additionalFilters: {
whereCondition: timestamp
? " AND timestamp >= {timestamp: DateTime64(3)} "
: "",
params: { timestamp },
},
});
if (postgresTraceRecord) {
recordIncrement("langfuse.ingestion.lookup.hit", 1, {
store: "postgres",
object: "trace",
});
}
if (clickhouseTraceRecord) {
recordIncrement("langfuse.ingestion.lookup.hit", 1, {
store: "clickhouse",
@@ -265,9 +245,10 @@ export class IngestionService {
const finalTraceRecord = await this.mergeTraceRecords({
clickhouseTraceRecord,
postgresTraceRecord,
traceRecords,
});
finalTraceRecord.created_at =
clickhouseTraceRecord?.created_at ?? createdAtTimestamp.getTime();
// If the trace has a sessionId, we upsert the corresponding session into Postgres.
if (finalTraceRecord.session_id) {
@@ -321,9 +302,11 @@ export class IngestionService {
private async processObservationEventList(params: {
projectId: string;
entityId: string;
createdAtTimestamp: Date;
observationEventList: ObservationEvent[];
}) {
const { projectId, entityId, observationEventList } = params;
const { projectId, entityId, createdAtTimestamp, observationEventList } =
params;
if (observationEventList.length === 0) return;
const timeSortedEvents =
@@ -340,34 +323,22 @@ export class IngestionService {
? undefined
: convertDateToClickhouseDateTime(new Date(minStartTime));
const [postgresObservationRecord, clickhouseObservationRecord, prompt] =
await Promise.all([
this.getPostgresRecord({
projectId,
entityId,
table: TableName.Observations,
}),
this.getClickhouseRecord({
projectId,
entityId,
table: TableName.Observations,
additionalFilters: {
whereCondition: `AND type = {type: String} ${startTime ? "AND start_time >= {startTime: DateTime64(3)} " : ""}`,
params: {
type,
startTime,
},
const [clickhouseObservationRecord, prompt] = await Promise.all([
this.getClickhouseRecord({
projectId,
entityId,
table: TableName.Observations,
additionalFilters: {
whereCondition: `AND type = {type: String} ${startTime ? "AND start_time >= {startTime: DateTime64(3)} " : ""}`,
params: {
type,
startTime,
},
}),
this.getPrompt(projectId, observationEventList),
]);
},
}),
this.getPrompt(projectId, observationEventList),
]);
if (postgresObservationRecord) {
recordIncrement("langfuse.ingestion.lookup.hit", 1, {
store: "postgres",
object: "observation",
});
}
if (clickhouseObservationRecord) {
recordIncrement("langfuse.ingestion.lookup.hit", 1, {
store: "clickhouse",
@@ -385,9 +356,10 @@ export class IngestionService {
const finalObservationRecord = await this.mergeObservationRecords({
projectId,
observationRecords,
postgresObservationRecord,
clickhouseObservationRecord,
});
finalObservationRecord.created_at =
clickhouseObservationRecord?.created_at ?? createdAtTimestamp.getTime();
// Backward compat: create wrapper trace for SDK < 2.0.0 events that do not have a traceId
if (!finalObservationRecord.trace_id) {
@@ -417,17 +389,14 @@ export class IngestionService {
private async mergeScoreRecords(params: {
scoreRecords: ScoreRecordInsertType[];
postgresScoreRecord?: ScoreRecordInsertType | null;
clickhouseScoreRecord?: ScoreRecordInsertType | null;
}): Promise<ScoreRecordInsertType> {
const { scoreRecords, postgresScoreRecord, clickhouseScoreRecord } = params;
const { scoreRecords, clickhouseScoreRecord } = params;
// Set clickhouse first as this is the baseline for immutable fields
const recordsToMerge = [
clickhouseScoreRecord,
postgresScoreRecord,
...scoreRecords,
].filter(Boolean) as ScoreRecordInsertType[];
const recordsToMerge = [clickhouseScoreRecord, ...scoreRecords].filter(
Boolean,
) as ScoreRecordInsertType[];
const mergedRecord = this.mergeRecords(
recordsToMerge,
@@ -439,17 +408,14 @@ export class IngestionService {
private async mergeTraceRecords(params: {
traceRecords: TraceRecordInsertType[];
postgresTraceRecord?: TraceRecordInsertType | null;
clickhouseTraceRecord?: TraceRecordInsertType | null;
}): Promise<TraceRecordInsertType> {
const { traceRecords, postgresTraceRecord, clickhouseTraceRecord } = params;
const { traceRecords, clickhouseTraceRecord } = params;
// Set clickhouse first as this is the baseline for immutable fields
const recordsToMerge = [
clickhouseTraceRecord,
postgresTraceRecord,
...traceRecords,
].filter(Boolean) as TraceRecordInsertType[];
const recordsToMerge = [clickhouseTraceRecord, ...traceRecords].filter(
Boolean,
) as TraceRecordInsertType[];
const mergedRecord = this.mergeRecords(
recordsToMerge,
@@ -462,20 +428,14 @@ export class IngestionService {
private async mergeObservationRecords(params: {
projectId: string;
observationRecords: ObservationRecordInsertType[];
postgresObservationRecord?: ObservationRecordInsertType | null;
clickhouseObservationRecord?: ObservationRecordInsertType | null;
}): Promise<ObservationRecordInsertType> {
const {
projectId,
observationRecords,
postgresObservationRecord,
clickhouseObservationRecord,
} = params;
const { projectId, observationRecords, clickhouseObservationRecord } =
params;
// Set clickhouse first as this is the baseline for immutable fields
const recordsToMerge = [
clickhouseObservationRecord,
postgresObservationRecord,
...observationRecords,
].filter(Boolean) as ObservationRecordInsertType[];
@@ -500,14 +460,6 @@ export class IngestionService {
observationRecord: parsedObservationRecord,
});
if (
"usage_details" in generationUsage &&
Object.keys(generationUsage.usage_details).length === 0
) {
generationUsage.usage_details =
postgresObservationRecord?.usage_details ?? {};
}
return {
...parsedObservationRecord,
...generationUsage,
@@ -805,6 +757,15 @@ export class IngestionService {
params: Record<string, unknown>;
};
}) {
if (
env.LANGFUSE_SKIP_INGESTION_CLICKHOUSE_READ_PROJECT_IDS &&
env.LANGFUSE_SKIP_INGESTION_CLICKHOUSE_READ_PROJECT_IDS.split(
",",
).includes(params.projectId)
) {
return null;
}
const recordParser = {
traces: traceRecordReadSchema,
scores: scoreRecordReadSchema,
@@ -844,58 +805,6 @@ export class IngestionService {
);
}
private async getPostgresRecord(params: {
projectId: string;
entityId: string;
table: TableName.Traces;
}): Promise<TraceRecordInsertType | null>;
private async getPostgresRecord(params: {
projectId: string;
entityId: string;
table: TableName.Scores;
}): Promise<ScoreRecordInsertType | null>;
private async getPostgresRecord(params: {
projectId: string;
entityId: string;
table: TableName.Observations;
}): Promise<ObservationRecordInsertType | null>;
private async getPostgresRecord(params: {
projectId: string;
entityId: string;
table: TableName;
}) {
const recordParser = {
traces: convertPostgresTraceToInsert,
scores: convertPostgresScoreToInsert,
observations: convertPostgresObservationToInsert,
};
const { projectId, entityId, table } = params;
const query =
table === TableName.Observations
? Prisma.sql`
SELECT o.*,
o."modelParameters" as model_parameters,
p.name as prompt_name,
p.version as prompt_version
FROM observations o
LEFT JOIN prompts p ON o.prompt_id = p.id
WHERE o.project_id = ${projectId}
AND o.id = ${entityId}
LIMIT 1;`
: Prisma.sql`
SELECT *
FROM ${Prisma.raw(table)}
WHERE project_id = ${projectId}
AND id = ${entityId}
LIMIT 1;`;
const result =
await this.prisma.$queryRaw<Array<Record<string, unknown>>>(query);
return result.length === 0 ? null : recordParser[table](result[0]);
}
private mapTraceEventsToRecords(params: {
traceEventList: TraceEventType[];
projectId: string;
@@ -73,6 +73,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(timestamp),
traceEventList: eventList,
});
await clickhouseWriter.flushAll(true);
@@ -442,21 +443,25 @@ describe("Ingestion end-to-end tests", () => {
ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: spanId,
createdAtTimestamp: new Date(),
observationEventList: spanEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: generationEventList,
}),
ingestionService.processScoreEventList({
projectId,
entityId: scoreId,
createdAtTimestamp: new Date(),
scoreEventList,
}),
]);
@@ -778,11 +783,13 @@ describe("Ingestion end-to-end tests", () => {
ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: generationEventList,
}),
]);
@@ -918,26 +925,31 @@ describe("Ingestion end-to-end tests", () => {
ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: spanId,
createdAtTimestamp: new Date(),
observationEventList: spanEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: generationEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: eventId,
createdAtTimestamp: new Date(),
observationEventList: eventEventList,
}),
ingestionService.processScoreEventList({
projectId,
entityId: scoreId,
createdAtTimestamp: new Date(),
scoreEventList,
}),
]);
@@ -1009,6 +1021,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList: traceEventList1,
});
@@ -1032,6 +1045,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList: traceEventList2,
});
@@ -1088,6 +1102,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
});
@@ -1135,6 +1150,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
});
@@ -1190,6 +1206,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processScoreEventList({
projectId,
entityId: scoreId,
createdAtTimestamp: new Date(),
scoreEventList,
});
@@ -1202,68 +1219,6 @@ describe("Ingestion end-to-end tests", () => {
expect(score.project_id).toBe("7a88fb47-b4e2-43b8-a06c-a5ce950dc53a");
});
it("should merge observations from postgres and event list", async () => {
const traceId = randomUUID();
const observationId = randomUUID();
const latestEvent = new Date();
const oldEvent = new Date(latestEvent).setSeconds(
latestEvent.getSeconds() - 1,
);
await prisma.observation.create({
data: {
id: observationId,
type: "GENERATION",
traceId,
name: "generation-name",
input: { key: "value" },
output: "should be overwritten",
model: "gpt-3.5",
projectId,
startTime: new Date(oldEvent),
completionTokens: 5,
// Validates that numbers are parsed correctly. Since there is no usage, no effect on result
calculatedTotalCost: "0.273330000000000000000000000000",
modelParameters: { hello: "world" },
},
});
const observationEventList: ObservationEvent[] = [
{
id: randomUUID(),
type: "generation-create",
timestamp: new Date().toISOString(),
body: {
id: observationId,
traceId: traceId,
output: "overwritten",
usage: undefined,
},
},
];
await ingestionService.processObservationEventList({
projectId,
entityId: observationId,
observationEventList,
});
await clickhouseWriter.flushAll(true);
const observation = await getClickhouseRecord(
TableName.Observations,
observationId,
);
expect(observation.name).toBe("generation-name");
expect(observation.input).toBe(JSON.stringify({ key: "value" }));
expect(observation.output).toBe("overwritten");
expect(observation.model_parameters).toBe('{"hello":"world"}');
expect(observation.usage_details.output).toBe(5);
expect(observation.project_id).toBe("7a88fb47-b4e2-43b8-a06c-a5ce950dc53a");
});
it("should merge observations and set negative tokens and cost to null", async () => {
await prisma.model.create({
data: {
@@ -1374,6 +1329,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processObservationEventList({
projectId,
entityId: observationId,
createdAtTimestamp: new Date(),
observationEventList,
});
@@ -1513,6 +1469,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processObservationEventList({
projectId,
entityId: observationId,
createdAtTimestamp: new Date(),
observationEventList,
});
@@ -1566,6 +1523,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processObservationEventList({
projectId,
entityId: observationId,
createdAtTimestamp: new Date(),
observationEventList: observationEventList1,
});
await clickhouseWriter.flushAll(true);
@@ -1587,6 +1545,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processObservationEventList({
projectId,
entityId: observationId,
createdAtTimestamp: new Date(),
observationEventList: observationEventList2,
});
await clickhouseWriter.flushAll(true);
@@ -1636,6 +1595,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: generationEventList,
});
@@ -1697,11 +1657,13 @@ describe("Ingestion end-to-end tests", () => {
ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: generationEventList1,
}),
]);
@@ -1742,6 +1704,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: generationEventList2,
});
@@ -1833,11 +1796,13 @@ describe("Ingestion end-to-end tests", () => {
ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: generationEventList,
}),
]);
@@ -1933,11 +1898,13 @@ describe("Ingestion end-to-end tests", () => {
ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: generationEventList,
}),
]);
@@ -2001,6 +1968,7 @@ describe("Ingestion end-to-end tests", () => {
await ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
});
@@ -2114,11 +2082,13 @@ describe("Ingestion end-to-end tests", () => {
ingestionService.processTraceEventList({
projectId,
entityId: traceId,
createdAtTimestamp: new Date(),
traceEventList,
}),
ingestionService.processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: generationEventList,
}),
]);
@@ -427,6 +427,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -507,6 +508,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -585,6 +587,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -665,6 +668,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -748,6 +752,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -837,6 +842,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -921,6 +927,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -1002,6 +1009,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -1088,6 +1096,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -1159,6 +1168,7 @@ describe("Token Cost Calculation", () => {
await (mockIngestionService as any).processObservationEventList({
projectId,
entityId: generationId,
createdAtTimestamp: new Date(),
observationEventList: events,
});
expect(mockAddToClickhouseWriter).toHaveBeenCalled();