Build custom eval pipelines in Langfuse

Introduction

In production, LLM applications processing tens of thousands of requests per day cannot rely on human reviewers to catch quality regressions — the volume is too high and manual review too inconsistent. Without automated checks, a model update or prompt change can silently degrade output quality for days before anyone notices. By the end of this lesson, you'll be able to design and implement a LangfuseEvalPipeline that applies pluggable scorers — covering factual accuracy, tone compliance, schema adherence, and citation coverage — to every LLM response, then persists those scores to Langfuse traces for trend analysis and drift detection.

Key Terminology

• EvalInput — A dataclass that bundles everything a scorer needs to grade one LLM response: the trace_id that links the result back to a Langfuse trace, the original prompt, the model's response, the model name, arbitrary metadata, and an optional reference_data dict for scorers that require ground-truth context.
• EvalScore — A dataclass representing a single scorer's verdict, carrying a name string that identifies the quality dimension, a value normalized to the range 0.0–1.0, and an optional comment that records the human-readable rationale persisted alongside the score in Langfuse.
• BaseScorer — An abstract base class that defines the pluggable scorer contract; every quality dimension (factual accuracy, tone compliance, schema adherence, citation coverage) is implemented as a concrete subclass that supplies a name property and an async score(eval_input: EvalInput) -> EvalScore method.
• LangfuseEvalPipeline — The orchestrator that fans an EvalInput out to all registered BaseScorer instances concurrently via asyncio.gather, collects the resulting EvalScore objects, and writes each score to the corresponding Langfuse trace with langfuse.score() before calling langfuse.flush().
• Semaphore-bounded concurrency — A backpressure pattern implemented with asyncio.Semaphore(max_concurrent) inside _run_scorer that caps how many scorers execute simultaneously, preventing a burst of evaluated traces from overwhelming downstream APIs even though all scorers are launched together with asyncio.gather.
• Per-scorer timeout — A fault-isolation guard applied with asyncio.wait_for(scorer.score(...), timeout=self.timeout) that converts a slow or hung scorer into a timed-out EvalScore with value=0.0, ensuring one unresponsive grader cannot stall the pipeline or block the Langfuse write path for the remaining scorers.

Concepts

Why Automated Eval Pipelines Replace Human Review at Scale

Human review is the natural first instinct for quality assurance, but it doesn't survive contact with production traffic. At tens of thousands of requests per day, manual sampling is too slow and too inconsistent to catch regressions reliably — the lag between a prompt change and a human verdict is measured in days, not minutes, and the verdict itself varies by reviewer. A silent degradation in factual accuracy or tone compliance can run for days before anyone notices.

An automated eval pipeline inverts this dynamic: it runs against every trace, applies a fixed roster of scorers with a consistent rubric, and writes structured numeric scores into Langfuse so dashboards and alerting can surface drift automatically. The fundamental shift is from periodic, ad-hoc sampling to continuous, exhaustive coverage.

The Pluggable Scorer Abstraction

The pipeline's extensibility rests on the BaseScorer interface. Each scorer encapsulates exactly one quality dimension behind a uniform contract: given an EvalInput, produce an EvalScore. Because every scorer honors this interface, LangfuseEvalPipeline never needs to know which scorers are registered — it fans the input out to all of them and collects results identically regardless of what each scorer does internally.

This separation controls deployment risk. A deterministic scorer like ResponseLengthScorer (see Code Walkthrough) can be registered first to verify that the pipeline wires correctly and that scores land on Langfuse traces before any LLM-backed scorers are introduced. Once the plumbing is confirmed, each new quality dimension — schema adherence, citation coverage — is added by implementing name and score alone, with no changes to the pipeline orchestrator.

Concurrency Control and Fault Isolation

Running scorers sequentially would sum their latencies end-to-end; asyncio.gather solves that by launching all scorer coroutines concurrently. Unbounded concurrency creates its own hazard, however: a spike in evaluated traces could fan out into hundreds of simultaneous downstream API calls. The asyncio.Semaphore(max_concurrent) inside _run_scorer acts as a backpressure valve — scorers that exceed the cap wait for a slot rather than proceeding immediately.

Fault isolation operates through two mechanisms in tandem. asyncio.wait_for enforces a hard per-scorer deadline, converting a hung scorer into a timed-out EvalScore with value=0.0 rather than blocking indefinitely. asyncio.gather(..., return_exceptions=True) ensures that an exception raised by one scorer is captured as an exception object in the results list rather than propagating and cancelling the remaining tasks. The pipeline logs the failure and continues writing all successful scores to Langfuse — one broken grader never prevents the others from persisting their results.

Code Walkthrough

Building on the BaseScorer interface and EvalInput/EvalScore data classes from the Concepts section, the pipeline wires those abstractions into a concrete orchestrator that fans out across all registered scorers concurrently and writes every result back to Langfuse.

The pipeline uses asyncio.gather to run scorers in parallel without blocking the request path. A Semaphore caps concurrency so a burst of requests does not overwhelm downstream APIs, and asyncio.wait_for enforces a per-scorer timeout so a slow grader cannot stall the rest. Each scorer returns an EvalScore with a normalized value between 0.0 and 1.0; the pipeline persists that score to the corresponding Langfuse trace before flushing.

Code snippetpython
1from langfuse import Langfuse
2from dataclasses import dataclass
3from typing import List, Optional, Dict, Any
4from abc import ABC, abstractmethod
5import asyncio
6import logging
7
8logger = logging.getLogger(__name__)
9
10@dataclass
11class EvalInput:
12    trace_id: str
13    prompt: str
14    response: str
15    model: str
16    metadata: Dict[str, Any]
17    reference_data: Optional[Dict[str, Any]] = None
18
19@dataclass
20class EvalScore:
21    name: str
22    value: float          # normalized 0.0 to 1.0
23    comment: Optional[str] = None
24
25class BaseScorer(ABC):
26    @property
27    @abstractmethod
28    def name(self) -> str: ...
29
30    @abstractmethod
31    async def score(self, eval_input: EvalInput) -> EvalScore: ...
32
33class LangfuseEvalPipeline:
34    def __init__(
35        self,
36        langfuse: Langfuse,
37        scorers: List[BaseScorer],
38        max_concurrent: int = 10,
39        timeout_seconds: float = 30.0,
40    ):
41        self.langfuse = langfuse
42        self.scorers = scorers
43        self.semaphore = asyncio.Semaphore(max_concurrent)
44        self.timeout = timeout_seconds
45
46    async def evaluate(self, eval_input: EvalInput) -> List[EvalScore]:
47        tasks = [self._run_scorer(s, eval_input) for s in self.scorers]
48        results = await asyncio.gather(*tasks, return_exceptions=True)
49        scores = []
50        for result in results:
51            if isinstance(result, EvalScore):
52                scores.append(result)
53                self.langfuse.score(
54                    trace_id=eval_input.trace_id,
55                    name=result.name,
56                    value=result.value,
57                    comment=result.comment,
58                )
59            elif isinstance(result, Exception):
60                logger.error("Scorer failed: %s", result)
61        self.langfuse.flush()
62        return scores
63
64    async def _run_scorer(self, scorer: BaseScorer, eval_input: EvalInput) -> EvalScore:
65        async with self.semaphore:
66            try:
67                return await asyncio.wait_for(scorer.score(eval_input), timeout=self.timeout)
68            except asyncio.TimeoutError:
69                return EvalScore(name=scorer.name, value=0.0,
70                                 comment=f"Timed out after {self.timeout}s")

With the pipeline class defined, registering a new quality dimension requires only implementing name and score. The following deterministic scorer checks response length and is a good first scorer to register when verifying the pipeline end-to-end before adding LLM-based scorers:

Code snippetpython
1class ResponseLengthScorer(BaseScorer):
2    @property
3    def name(self) -> str:
4        return "response_length"
5
6    async def score(self, eval_input: EvalInput) -> EvalScore:
7        word_count = len(eval_input.response.split())
8        if 50 <= word_count <= 300:
9            value = 1.0
10        elif word_count < 50:
11            value = word_count / 50.0
12        else:
13            value = max(0.0, 1.0 - (word_count - 300) / 300.0)
14        return EvalScore(name=self.name, value=round(value, 3),
15                         comment=f"{word_count} words")

Verify by opening your Langfuse project dashboard after running pipeline.evaluate(eval_input) against a sample trace — you should see response_length (and any other registered scorer names) appear as score entries on the trace detail page, each carrying a value between 0.0 and 1.0.

Do's and Don'ts

Having walked through the material above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do pass return_exceptions=True to asyncio.gather — without it, a single scorer raising an unhandled exception cancels every in-flight coroutine in evaluate(); with it, exceptions surface as values in the results list so the pipeline logs the failure, skips that scorer, and still persists EvalScore entries from every scorer that succeeded.
2. ✓Do size max_concurrent on the asyncio.Semaphore relative to your downstream API rate limits — asyncio.gather fans out all scorer coroutines simultaneously, so an unbounded pipeline hitting LLM-based grading APIs during a request burst will exhaust per-minute quota and trigger cascading rate-limit errors across every active scorer at once.
3. ✓Do verify the pipeline end-to-end with a deterministic scorer like ResponseLengthScorer before registering LLM-based scorers — a deterministic scorer has no network dependency, so seeing its name and 0.0–1.0 value appear on the Langfuse trace detail page confirms that langfuse.score() and langfuse.flush() are reaching the backend before you add graders that can fail for unrelated reasons.

Don'ts

1. ✗Don't omit asyncio.wait_for around scorer.score() in _run_scorer — a slow or hung LLM-based grader will hold its semaphore slot for the entire lifetime of the call, starving every other concurrent scorer of the concurrency budget and eventually freezing the entire evaluate() call until the pipeline stops processing new traces.
2. ✗Don't call langfuse.score() inside individual BaseScorer implementations — keeping all Langfuse persistence in evaluate() is what makes scorers stateless and testable in isolation; a scorer that directly calls the Langfuse client cannot be unit-tested offline, cannot be reused outside the pipeline, and silently double-writes scores if the pipeline also persists the returned EvalScore.
3. ✗Don't return EvalScore.value outside the 0.0–1.0 range from any BaseScorer — the pipeline writes values directly to Langfuse traces where they drive drift-detection thresholds and trend charts; scorers on different scales produce aggregates the Langfuse dashboard cannot normalize, making cross-scorer comparisons meaningless and threshold alerts fire incorrectly.