Evaluation Queue Architecture#

Langfuse runs evaluations through two parallel BullMQ pipelines, both backed by Redis and hosted in the worker service:

Both pipelines ultimately call the same executeLLMAsJudgeEvaluation() core but differ in trigger path, data transport, and variable extraction. A third queue — secondary-evaluation-execution-queue — isolates high-throughput projects from the primary trace-based queue 1.

TraceUpsertQueue / DatasetRunItemUpsert / ui-create-eval
  └─► createEvalJobs() ──────────────────────────────► evaluation-execution-queue
                                                         secondary-evaluation-execution-queue (high-throughput)

OTel ingestion path only
  └─► scheduleObservationEvals() ──────────────────► llm-as-a-judge-execution-queue

Queue Names, Event Schemas & Key Files#

Queue names and Zod event schemas are centralized in packages/shared/src/server/queues.ts:

The observationS3Path in the LLMAsJudgeExecution payload is the key architectural difference: observation evals upload the full observation object to S3 at scheduling time and pass the path through the queue, while trace evals carry only metadata and fetch data on-demand during execution.

Key source files:

Sharding#

All three eval queues support horizontal sharding. Shard count is set independently via environment variables (default: 1 each) 3:

Shard naming appends the index suffix starting at shard 1: evaluation-execution-queue (shard 0), evaluation-execution-queue-1 (shard 1), etc. 4

Shard selection at enqueue time 5:

• When Redis cluster is enabled (REDIS_CLUSTER_ENABLED=true) and a shardingKey is provided, getShardIndex() computes SHA-256(shardingKey).slice(0, 8) → integer → % shardCount.
• Otherwise, defaults to shard 0.

The sharding key used when enqueuing trace eval jobs is {projectId}-{jobExecutionId} 6. The same format is used when redirecting to the secondary queue 7.

Worker registration: WorkerManager calls Queue.getShardNames() and registers one BullMQ worker per shard 8. Sharded queues (including eval queues) use extended stall settings — lockDuration: 60 000 ms, stalledInterval: 120 000 ms, maxStalledCount: 3 — to tolerate slow LLM calls 8.

Secondary Queue for High-Throughput Projects#

The secondary-evaluation-execution-queue isolates specific projects from the primary trace-based queue to prevent high-throughput projects from blocking others 9.

Routing is done at execution time, not at job creation. The primary queue processor (evalQueue.ts) checks:

1. Whether LANGFUSE_SECONDARY_EVAL_EXECUTION_QUEUE_ENABLED_PROJECT_IDS is set (comma-separated project IDs).
2. Whether the incoming job's projectId matches an entry in that list.
3. If matched, the job is re-enqueued on SecondaryEvalExecutionQueue with the same {projectId}-{jobExecutionId} sharding key and the primary job is discarded.

The secondary queue has its own independent shard count and worker pool, allowing it to be scaled separately.

Trace-Based vs Observation-Based Pipelines#

The OTel path's otelIngestionQueueProcessor calls fetchObservationEvalConfigs per batch and then scheduleObservationEvals() per observation. The legacy IngestionService only publishes to TraceUpsertQueue and never calls observation eval scheduling — this is confirmed intentional behavior.

scheduleObservationEvals() steps 10:

1. Filter configs by in-memory filter evaluation + sampling.
2. Upload the full observation payload to S3 once (not once per config).
3. Per matching config: generate deterministic jobExecutionId from configId:observationId, upsert JobExecution row, enqueue on LLMAsJudgeExecutionQueue with delay: 0.

Experiment configs (EXPERIMENT target) additionally require the observation to be the experiment root span before firing 11.

Variable Extraction & Multi-Source Mapping#

For trace-based evals, extractVariablesFromTracingData() resolves template variables from three source types:

mapping.objectName is required for any observation-type variable and identifies which named observation to extract from within a trace. It forms the per-function cache key ({projectId}:{traceId}:{objectName}) to avoid redundant ClickHouse hits within a single execution 12.

selectedColumnId picks which field to return (input, output, metadata, expected_output). An optional jsonSelector applies a JSONPath expression to the selected value for nested extraction 13.

Available object types and their supported columns are defined in packages/shared/src/features/evals/types.ts (availableTraceEvalVariables, availableDatasetEvalVariables).

Observation evals use a simpler path: the observation payload is already available via the S3 download at execution time, so observationVariableMapping requires no langfuseObject or objectName — the executor reads directly from the pre-fetched object 14.