Experiments in CI/CD

from langfuse import RunnerContext
from langfuse.api import DatasetItem


# Define a task that calls your agent with each dataset item.
def my_task(item: DatasetItem, **kwargs):
    ...


def experiment(context: RunnerContext):
    return context.run_experiment(
        name="PR gate",
        task=my_task,
    )

import type { ExperimentTaskParams, RunnerContext } from "@langfuse/client";

// Define a task that calls your agent with each dataset item.
async function myTask(item: ExperimentTaskParams) {
  // ...
}

export async function experiment(context: RunnerContext) {
  return await context.runExperiment({
    name: "PR gate",
    task: myTask,
  });
}

from langfuse import Evaluation, RegressionError, RunnerContext


THRESHOLD = 0.95


def experiment(context: RunnerContext):
    result = context.run_experiment(
        name="PR gate: support agent",
        task=answer_support_question,
        evaluators=[exact_match],
        run_evaluators=[avg_accuracy],
    )

    accuracy = next(
        (
            evaluation.value
            for evaluation in result.run_evaluations
            if evaluation.name == "avg_accuracy"
        ),
        None,
    )

    if not isinstance(accuracy, (int, float)) or accuracy < THRESHOLD:
        raise RegressionError(
            # Attach the result so the action can include scores in the PR comment and `result_json` output.
            result=result,
            metric="avg_accuracy",
            value=float(accuracy) if isinstance(accuracy, (int, float)) else 0.0,
            threshold=THRESHOLD,
        )

    return result


def answer_support_question(item, **kwargs):
    # Replace this stub with your application logic.
    return item.input["question"]


def exact_match(*, output, expected_output, **kwargs):
    passed = output.strip() == (expected_output or "").strip()
    return Evaluation(
        name="exact_match",
        value=1.0 if passed else 0.0,
        comment="match" if passed else "mismatch",
    )


def avg_accuracy(*, item_results, **kwargs):
    scores = [
        evaluation.value
        for item in item_results
        for evaluation in item.evaluations
        if evaluation.name == "exact_match" and isinstance(evaluation.value, (int, float))
    ]
    return Evaluation(name="avg_accuracy", value=sum(scores) / len(scores) if scores else 0.0)

import {
  RegressionError,
  type Evaluation,
  type ExperimentTaskParams,
  type RunnerContext,
} from "@langfuse/client";

const THRESHOLD = 0.95;

export async function experiment(context: RunnerContext) {
  const result = await context.runExperiment({
    name: "PR gate: support agent",
    task: answerSupportQuestion,
    evaluators: [exactMatch],
    runEvaluators: [avgAccuracy],
  });

  const accuracy = result.runEvaluations.find(
    (evaluation) => evaluation.name === "avg_accuracy",
  )?.value;

  if (typeof accuracy !== "number" || accuracy < THRESHOLD) {
    throw new RegressionError({
      // Attach the result so the action can include scores in the PR comment and `result_json` output.
      result,
      metric: "avg_accuracy",
      value: typeof accuracy === "number" ? accuracy : 0,
      threshold: THRESHOLD,
    });
  }

  return result;
}

async function answerSupportQuestion(item: ExperimentTaskParams) {
  const { question } = item.input as { question: string };

  // Replace this with your application logic, for example calling your agent.
  return await supportAgent(question);
}

async function supportAgent(question: string) {
  return question;
}

async function exactMatch({
  output,
  expectedOutput,
}: {
  output: string;
  expectedOutput?: string;
}): Promise<Evaluation> {
  const passed = output.trim() === expectedOutput?.trim();
  return { name: "exact_match", value: passed ? 1 : 0 };
}

async function avgAccuracy({
  itemResults,
}: {
  itemResults: Array<{ evaluations: Evaluation[] }>;
}): Promise<Evaluation> {
  const scores = itemResults
    .flatMap((item) => item.evaluations)
    .filter((evaluation) => evaluation.name === "exact_match")
    .map((evaluation) => Number(evaluation.value))
    .filter((score) => Number.isFinite(score));

  return {
    name: "avg_accuracy",
    value: scores.length
      ? scores.reduce((sum, score) => sum + score, 0) / scores.length
      : 0,
  };
}

import json
from datetime import datetime
from pathlib import Path

from langfuse import RegressionError

EVALUATOR_VERSION = "refund-window-v1"


def parse_version(value: str | datetime) -> datetime:
    parsed = (
        datetime.fromisoformat(value.replace("Z", "+00:00"))
        if isinstance(value, str)
        else value
    )
    if parsed.utcoffset() is None:
        raise ValueError("Dataset version must include a timezone")
    return parsed


def check_approved_baseline(result, *, dataset_version: str | datetime):
    baseline = json.loads(Path("experiments/approved-baseline.json").read_text())
    if (
        parse_version(baseline["dataset_version"])
        != parse_version(dataset_version)
        or baseline["evaluator_version"] != EVALUATOR_VERSION
    ):
        raise ValueError(
            "Dataset or evaluator version differs from the approved baseline"
        )
    approved = baseline["cases"]
    if not approved or any(type(value) is not bool for value in approved.values()):
        raise ValueError("Baseline must contain reviewed boolean verdicts")

    current = {}
    for row in result.item_results:
        item = row.item
        metadata = item.get("metadata") if isinstance(item, dict) else item.metadata
        case_id = (metadata or {}).get("case_id")
        scores = [e.value for e in row.evaluations if e.name == "refund_window"]
        if (
            not isinstance(case_id, str)
            or case_id in current
            or len(scores) != 1
            or scores[0] not in (0, 1)
        ):
            raise ValueError("Incomplete, duplicate, or invalid case result")
        current[case_id] = scores[0] == 1
    if current.keys() != approved.keys():
        raise ValueError("Candidate and baseline contain different cases")

    regressions = [
        case_id for case_id in approved if approved[case_id] and not current[case_id]
    ]
    if regressions:
        print("Newly failing cases:", ", ".join(regressions))
        raise RegressionError(
            result=result,
            metric="newly_failing_cases",
            value=float(len(regressions)),
            threshold=0.0,
        )

import { readFileSync } from "node:fs";
import { RegressionError, type ExperimentResult } from "@langfuse/client";

const EVALUATOR_VERSION = "refund-window-v1";

export function checkApprovedBaseline(
  result: ExperimentResult,
  datasetVersion: string,
) {
  const baseline = JSON.parse(
    readFileSync("experiments/approved-baseline.json", "utf8"),
  );
  if (
    !Number.isFinite(Date.parse(datasetVersion)) ||
    Date.parse(baseline.dataset_version) !== Date.parse(datasetVersion) ||
    baseline.evaluator_version !== EVALUATOR_VERSION
  ) {
    throw new Error(
      "Dataset or evaluator version differs from the approved baseline",
    );
  }
  const approved = baseline.cases as Record<string, boolean>;
  if (
    !approved ||
    Object.keys(approved).length === 0 ||
    Object.values(approved).some((value) => typeof value !== "boolean")
  ) {
    throw new Error("Baseline must contain reviewed boolean verdicts");
  }

  const current = new Map<string, boolean>();
  for (const row of result.itemResults) {
    const metadata = row.item.metadata as { case_id?: unknown } | undefined;
    const caseId = metadata?.case_id;
    const scores = row.evaluations.filter((e) => e.name === "refund_window");
    if (
      typeof caseId !== "string" ||
      current.has(caseId) ||
      scores.length !== 1 ||
      (scores[0].value !== 0 && scores[0].value !== 1)
    ) {
      throw new Error("Incomplete, duplicate, or invalid case result");
    }
    current.set(caseId, scores[0].value === 1);
  }
  if (
    current.size !== Object.keys(approved).length ||
    Object.keys(approved).some((id) => !current.has(id))
  ) {
    throw new Error("Candidate and baseline contain different cases");
  }

  const regressions = Object.keys(approved).filter(
    (id) => approved[id] && !current.get(id),
  );
  if (regressions.length) {
    console.error("Newly failing cases:", regressions.join(", "));
    throw new RegressionError({
      result,
      metric: "newly_failing_cases",
      value: regressions.length,
      threshold: 0,
    });
  }
}

import hashlib
import json
import os
from pathlib import Path

from langfuse import Evaluation, get_client
from my_app import run_application  # Replace with your application's import.
from my_checks import grade  # Reuse your existing boolean grader.


def test_application_checks():
    case_file = Path("cases.json").read_bytes()
    cases = json.loads(case_file)
    ids = [case["id"] for case in cases]
    assert ids and all(ids) and len(set(ids)) == len(ids), "Invalid case IDs"

    def task(*, item, **kwargs):
        return run_application(item["input"])

    def evaluator(*, output, expected_output, **kwargs):
        passed = grade(output, expected_output)
        if type(passed) is not bool:
            raise ValueError("grade must return a boolean")
        return Evaluation(name="existing_checks", value=int(passed))

    langfuse = get_client()
    try:
        result = langfuse.run_experiment(
            name="Application checks",
            data=[{
                "input": case["input"],
                "expected_output": case["expected_output"],
                "metadata": {"case_id": case["id"]},
            } for case in cases],
            task=task,
            evaluators=[evaluator],
            metadata={
                "application_version": os.environ["APP_VERSION"],
                "cases_sha256": hashlib.sha256(case_file).hexdigest(),
                "evaluator_version": "existing-checks-v1",
            },
        )
        print(result.format())  # Includes the experiment link.
        returned_ids = [item.item["metadata"]["case_id"] for item in result.item_results]
        assert sorted(returned_ids) == sorted(ids), "Incomplete results"
        for item in result.item_results:
            assert item.output is not None, "Missing application output"
            scores = [e for e in item.evaluations if e.name == "existing_checks"]
            assert len(scores) == 1 and scores[0].value == 1, (
                f"Failed or missing check: {item.item['metadata']['case_id']}"
            )
    finally:
        langfuse.flush()

import { createHash } from "node:crypto";
import { readFileSync } from "node:fs";
import { it, expect } from "vitest";
import { LangfuseClient } from "@langfuse/client";
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { NodeSDK } from "@opentelemetry/sdk-node";
import { runApplication } from "../src/app"; // Replace with your application's import.
import { grade } from "../src/checks"; // Reuse your existing boolean grader.

it("passes every required application check", async () => {
  const caseFile = readFileSync("cases.json");
  const cases: { id: string; input: unknown; expected_output: unknown }[] =
    JSON.parse(caseFile.toString("utf8"));
  const ids = cases.map((item) => item.id);
  expect(ids.length).toBeGreaterThan(0);
  expect(ids.every(Boolean)).toBe(true);
  expect(new Set(ids).size).toBe(ids.length);
  const applicationVersion = process.env.APP_VERSION;
  if (!applicationVersion) throw new Error("Set APP_VERSION");

  const otel = new NodeSDK({ spanProcessors: [new LangfuseSpanProcessor()] });
  otel.start();
  try {
    const langfuse = new LangfuseClient();
    const result = await langfuse.experiment.run({
      name: "Application checks",
      data: cases.map((item) => ({
        input: item.input,
        expectedOutput: item.expected_output,
        metadata: { case_id: item.id },
      })),
      task: async (item) => runApplication(item.input),
      evaluators: [async ({ output, expectedOutput }) => {
        const passed = await grade(output, expectedOutput);
        if (typeof passed !== "boolean") throw new Error("grade must return a boolean");
        return { name: "existing_checks", value: Number(passed) };
      }],
      metadata: {
        application_version: applicationVersion,
        cases_sha256: createHash("sha256").update(caseFile).digest("hex"),
        evaluator_version: "existing-checks-v1",
      },
    });
    console.log(await result.format()); // Includes the experiment link.
    expect(result.itemResults.map((item) =>
      (item.item.metadata as { case_id: string }).case_id,
    ).sort())
      .toEqual([...ids].sort());
    for (const item of result.itemResults) {
      expect(item.output).not.toBeNull();
      expect(item.output).not.toBeUndefined();
      const scores = item.evaluations.filter((score) => score.name === "existing_checks");
      expect(scores).toHaveLength(1);
      expect(scores[0].value).toBe(1);
    }
  } finally {
    await otel.shutdown();
  }
}, 60_000); // Adjust for the runtime of your application and case set.