diff --git a/packages/ai/README.md b/packages/ai/README.md index 7bbf073f..b1b47a70 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -54,10 +54,10 @@ Never raises. Returns `{"enabled": bool, "config": dict | None, "meta": dict | N ## Evaluations from code -`init_evaluations` and the evaluations result types are also re-exported: +`init_evaluations`, the criterion types, and the evaluations result types are all re-exported: ```python -from launchdarkly_ai_python import init_evaluations +from launchdarkly_ai_python import Judge, Scorer, init_evaluations evals = init_evaluations() result = await evals.run( @@ -66,10 +66,14 @@ result = await evals.run( dataset="golden-dataset", handler=my_handler, generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[ + Judge(key="accuracy-judge"), + Scorer(name="mentions-policy", fn=lambda row, output: "policy" in (output or "")), + ], ) ``` -`LD_API_TOKEN` is required. Configure `LD_SDK_KEY` — or initialize your own client with `init_client(client=...)` — to emit one `$ld:ai:offline-evals:generation` event per generated row through the standard SDK event transport. Use `LD_API_BASE_URI` for staging or local management API traffic; it is separate from the SDK delivery setting `LD_BASE_URI`. Evaluation-run links use the explicit `ui_base_uri` option or `LD_UI_BASE_URI` (for example, `https://ld-stg.launchdarkly.com` in staging), defaulting to `https://app.launchdarkly.com`. See the [core evaluations guide](../client/README.md#run-an-evaluation-from-code). +`LD_API_TOKEN` is required. Configure `LD_SDK_KEY` — or initialize your own client with `init_client(client=...)` — to emit one `$ld:ai:offline-evals:generation` event per generated row, plus one `$ld:ai:offline-evals:criterion` event per `(row, criterion)` when `criteria` are supplied, through the standard SDK event transport. The SDK reports scores; LaunchDarkly rules on them at ingest. A judge served by a different provider than `generation` needs a handler for it in `judge_handlers`. Use `LD_API_BASE_URI` for staging or local management API traffic; it is separate from the SDK delivery setting `LD_BASE_URI`. Evaluation-run links use the explicit `ui_base_uri` option or `LD_UI_BASE_URI` (for example, `https://ld-stg.launchdarkly.com` in staging), defaulting to `https://app.launchdarkly.com`. See the [core evaluations guide](../client/README.md#run-an-evaluation-from-code). --- diff --git a/packages/client/README.md b/packages/client/README.md index 852f5eff..0c5c4b26 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -46,9 +46,9 @@ No code changes are required — `init_client()` detects the packages at runtime ### Run an evaluation from code -The generation-only evaluations harness reads an LD-hosted dataset, creates a new evaluation and API-source run, and invokes your handler once per row. Each success or error queues a `$ld:ai:offline-evals:generation` custom event containing the evaluation, run, dataset, and row identifiers plus output or error (`errorMessage` is included for `ERROR` rows), nested `usage.inputTokens`/`usage.outputTokens`, timing, and stable hashes. Dataset-owned input, expected output, metadata, and variables are not duplicated in the event. Each queued event prints a line to stdout with its RFC3339 UTC `emittedAt` timestamp and stable `eventId`, making it possible to compare SDK emission time with ClickHouse arrival time. The same `emittedAt` value is included in the event payload. Events are flushed before the summary is fetched and the call returns; handlers are never rerun to retry event delivery. Pass/fail is derived from LaunchDarkly's run summary. +The evaluations harness reads an LD-hosted dataset, creates a new evaluation and API-source run, and invokes your handler once per row. Rows can then be scored by LaunchDarkly judges and local scorer functions; see [Score rows with judges and scorers](#score-rows-with-judges-and-scorers). Each success or error queues a `$ld:ai:offline-evals:generation` custom event containing the evaluation, run, dataset, and row identifiers plus output or error (`errorMessage` is included for `ERROR` rows), nested `usage.inputTokens`/`usage.outputTokens`, timing, and stable hashes. Dataset-owned input, expected output, metadata, and variables are not duplicated in the event. Each queued event is logged at `INFO` on the `launchdarkly_ai_server.evaluations.runner` logger with its RFC3339 UTC `emittedAt` timestamp and stable `eventId`, making it possible to compare SDK emission time with ClickHouse arrival time once that logger is enabled. The same `emittedAt` value is included in the event payload. Events are flushed before the summary is fetched and the call returns; handlers are never rerun to retry event delivery. Pass/fail is derived from LaunchDarkly's run summary. -Result links use `ui_base_uri`, then `LD_UI_BASE_URI`, then `https://app.launchdarkly.com`; this is independent of `LD_API_BASE_URI`. After flushing generation events, the harness polls the run summary endpoint until passed + failed + error rows fully account for a nonzero total with no pending rows, polling every `poll_interval_seconds` (default 2s) up to `poll_timeout_seconds` (default 180s); pass either to `run()` to widen both for large datasets. The summary endpoint does not return run state, so `RunSummary` exposes row counts only. A generation result passes only when the completed summary has no error or pending rows. Evaluation keys must be unique because every call creates a new evaluation with `POST`. +Result links use `ui_base_uri`, then `LD_UI_BASE_URI`, then `https://app.launchdarkly.com`; this is independent of `LD_API_BASE_URI`. After flushing generation events, the harness polls the run summary endpoint until passed + failed + error rows fully account for a nonzero total with no pending rows, polling every `poll_interval_seconds` (default 2s) up to `poll_timeout_seconds` (default 180s); pass either to `run()` to widen both for large datasets. The summary endpoint does not return run state, so `RunSummary` exposes row counts only. A run passes only when the completed summary has no failed, error, or pending rows. `failed_rows` counts rows whose criteria were scored and did not meet their threshold, so a gate that ignored it would exit 0 on a run where every row failed its judge. Evaluation keys must be unique because every call creates a new evaluation with `POST`. ```python import asyncio @@ -80,7 +80,45 @@ sys.exit(asyncio.run(main())) `project_key` is supplied per run rather than during initialization. `generation.instructions` is shorthand for one system message; use `generation.messages` instead for a full message list, but do not supply both. The harness never retries a handler invocation because doing so could repeat tool side effects. Its retries apply only to LaunchDarkly management API requests. -Generation events are the only path by which row results reach LaunchDarkly, so `init_evaluations()` raises rather than creating a run that can never complete unless it can resolve an event transport: either an SDK key (`sdk_key` or `LD_SDK_KEY`) or a client already initialized through `init_client(client=...)`. Bringing your own client lets a process emit evaluation events without an SDK key in scope. Every generated row is emitted and flushed unconditionally; no feature flag gates event publishing. The harness then polls the summary endpoint until row accounting shows processing is complete. +Generation and criterion events are the only path by which row results reach LaunchDarkly, so `init_evaluations()` raises rather than creating a run that can never complete unless it can resolve an event transport: either an SDK key (`sdk_key` or `LD_SDK_KEY`) or a client already initialized through `init_client(client=...)`. Bringing your own client lets a process emit evaluation events without an SDK key in scope. Every generated row is emitted and flushed unconditionally; no feature flag gates event publishing. The harness then polls the summary endpoint until row accounting shows processing is complete. + +### Score rows with judges and scorers + +Pass `criteria` to `run()` to score every generated row. A `Judge` references an AI Judge config that already exists in LaunchDarkly — the SDK creates no judges and ships none of its own — and a `Scorer` wraps a local function, so a run can mix model-graded and deterministic checks. Each criterion runs once per generated row, bounded by the same `concurrency` as generation, and emits one `$ld:ai:offline-evals:criterion` event per `(row, criterion)` carrying the criterion identity, the judge's variation key and version, the validated score, its reason, usage, and timings. + +```python +from launchdarkly_ai_claude_messages import create_claude_messages_handler +from launchdarkly_ai_openai_messages import create_openai_messages_handler +from launchdarkly_ai_server import DatasetRow, Judge, Scorer, init_evaluations + + +def mentions_policy(row: DatasetRow, output: str | None) -> bool: + return "refund policy" in (output or "").lower() + + +result = await init_evaluations().run( + project_key="my-project", + key="support-qa-2026-08-20", + dataset="support-golden", + handler=create_openai_messages_handler(), + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[ + Judge(key="accuracy-judge", threshold=0.8), + Scorer(name="mentions-policy", fn=mentions_policy), + ], + # Needed only because this judge is served by a different provider than + # the generation config above. + judge_handlers=[create_claude_messages_handler()], +) +``` + +`Scorer.fn` receives the `DatasetRow` the output was generated from plus the generated output, may be sync or async, and must return a bool or a number from 0 to 1; booleans become 1.0 or 0.0. `Judge.threshold` defaults to 0.5 and `Scorer.threshold` to 1.0 — a perfect score, which is what a boolean scorer wants — and both accept an optional `pass_rate_threshold`. Judge keys and scorer names share one `criterionType` namespace and must be unique within a run, case-insensitively, because that name is part of each result's deterministic event identity. `Judge.ground_truth_context` overrides what the judge is graded against when the dataset row's expected output is not it. + +**The SDK reports scores and never rules on them.** LaunchDarkly derives each row's verdict at ingest by comparing the score against the criterion's stored threshold and success direction, so pass/fail policy is one server-side implementation that applies to every SDK version and to runs already recorded. A judge's direction lives on its AI Config and is injected server-side, keeping the one input a verdict turns on server-attested; a `Scorer` has no LaunchDarkly-side config to read, so it declares its own `success_direction` (default `"higher_is_better"` — set `"lower_is_better"` for a scorer that counts something unwanted, like a regex hit count). + +**Judges are independent AI Configs, so handlers are routed per judge.** A judge may resolve to a different provider or mode than `generation`, and a handler built for one provider cannot execute another's config. `handler` runs a judge when it provides for that judge's provider; pass handlers for any other providers in `judge_handlers`. Selection prefers a handler naming the judge's provider outright over a wildcard multi-provider adapter, and an agent-mode handler can serve a messages-mode judge with its messages collapsed into one instructions block. A plain callable that declares no `provides_for` routes itself, exactly as it already does for the generation config. + +Judges are resolved through flag delivery, and handlers are matched to them, **before** any evaluation records are created — a missing judge or one no handler covers fails the run up front rather than after the generation spend. After that point a criterion failure never aborts the run: an unparseable judge response, an out-of-range score, a raising handler or scorer, and a row whose generation errored each become a per-criterion `ERROR` event with a cause code (`invalid_judge_output`, `invalid_score`, `handler_raised`, `scorer_raised`, `generation_incomplete`) and a top-level `errorMessage`. Event *delivery* is different: the backend needs one result per `(row, criterion)` to finish row accounting, so if tracking a criterion event fails, every remaining result is still attempted and flushed and then `run()` raises — rather than polling to its timeout with the cause hidden. The client uses **lazy initialization**: importing the package does not connect to LaunchDarkly. The singleton is created automatically on the first API call that needs it (`config().invoke()`, `graph().invoke()`, `resolve_graph()`, etc.), as long as `LD_SDK_KEY` is set in the environment. diff --git a/packages/client/src/launchdarkly_ai_server/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py index 9a07b056..9b2fce80 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -22,11 +22,15 @@ set_conversation_id_if_absent, ) from .evaluations import ( + Criterion, + DatasetRow, EvalRunResult, EvaluationsError, EvaluationsModule, GenerationConfig, + Judge, RunSummary, + Scorer, init_evaluations, ) from .graph import GraphInstance, graph, resolve_graph @@ -165,10 +169,14 @@ "VariationMeta", # evaluations "EvalRunResult", + "Criterion", + "DatasetRow", "EvaluationsError", "EvaluationsModule", "GenerationConfig", + "Judge", "RunSummary", + "Scorer", "init_evaluations", # utils "create_handler", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py index 6516f4a0..bd3eb0a4 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -9,19 +9,25 @@ Transport, urllib_transport, ) +from .criteria import Criterion, Judge, Scorer, SuccessDirection from .module import EvaluationsModule, init_evaluations -from .types import EvalRunResult, GenerationConfig, RunSummary, Usage +from .types import DatasetRow, EvalRunResult, GenerationConfig, RunSummary, Usage __all__ = [ "DEFAULT_BASE_URI", + "Criterion", + "DatasetRow", "EvalRunResult", "EvaluationsError", "EvaluationsModule", "GenerationConfig", "HttpResponse", + "Judge", "LDApiClient", "LDApiError", "RunSummary", + "Scorer", + "SuccessDirection", "Transport", "Usage", "init_evaluations", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/criteria.py b/packages/client/src/launchdarkly_ai_server/evaluations/criteria.py new file mode 100644 index 00000000..ee0962fd --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/criteria.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from math import isnan +from typing import Any, Literal + +from .types import DatasetRow + +type ScorerFn = Callable[[DatasetRow, Any], float | bool | Awaitable[float | bool]] + +type SuccessDirection = Literal["higher_is_better", "lower_is_better"] + +#: The threshold a judge is held to when its reference sets none. A criterion +#: with no threshold gives LaunchDarkly nothing to compare a score against, so +#: it would be recorded and never ruled on; defaulting is what keeps a judge +#: usable as a gate without every caller restating the obvious. +DEFAULT_JUDGE_THRESHOLD = 0.5 + + +@dataclass(frozen=True) +class Judge: + """Reference to a LaunchDarkly AI Judge config to run for each eval row. + + The SDK does not create or provide built-in judges. Pass the key of a judge + that exists in LaunchDarkly. Resolution uses LaunchDarkly flag delivery for + the currently served variation. + + A judge's success direction is not set here. It lives on the judge's AI + Config as ``isInverted`` and is injected onto the criterion by LaunchDarkly + when the evaluation is created, so the one input a verdict is ruled on stays + server-attested even though the score beside it is client-reported. + """ + + key: str + threshold: float | None = DEFAULT_JUDGE_THRESHOLD + pass_rate_threshold: float | None = None + ground_truth_context: str | None = None + + def __post_init__(self) -> None: + if not isinstance(self.key, str) or not self.key.strip(): + raise ValueError("judge key must not be blank") + _validate_thresholds( + threshold=self.threshold, + pass_rate_threshold=self.pass_rate_threshold, + ) + + @property + def criterion_type(self) -> str: + return self.key + + def to_criteria_wire(self) -> dict[str, Any]: + options = _criteria_options( + threshold=self.threshold, + pass_rate_threshold=self.pass_rate_threshold, + ground_truth_context=self.ground_truth_context, + ) + return { + "criterionType": self.criterion_type, + "kind": "judge", + "judgeKey": self.key, + "options": options, + } + + +@dataclass(frozen=True) +class Scorer: + """Local deterministic scorer run for each generated evaluation row. + + ``fn`` may be sync or async and receives ``(row, output)``, where ``row`` + is the :class:`~launchdarkly_ai_server.evaluations.types.DatasetRow` the + output was generated from and ``output`` is the generated output. It must + return a boolean or a numeric score from 0 to 1. Boolean results are + converted to 1.0 or 0.0 before being emitted as evaluation events. + + ``threshold`` defaults to 1.0: a row passes only on a perfect score, which + matches the common case of boolean scorers. Pass a lower threshold for + graded numeric scorers. + + ``success_direction`` says which way the score points, and defaults to + higher-is-better. Unlike a judge, a scorer has no LaunchDarkly-side config + to read a direction from, so the declaration here is the only source -- + LaunchDarkly derives each row's verdict by comparing score to threshold in + this direction. Set ``"lower_is_better"`` for a scorer that counts + something unwanted, e.g. a regex hit count or an edit distance. + """ + + name: str + fn: ScorerFn + threshold: float | None = 1.0 + pass_rate_threshold: float | None = None + success_direction: SuccessDirection = "higher_is_better" + + def __post_init__(self) -> None: + if not isinstance(self.name, str) or not self.name.strip(): + raise ValueError("scorer name must not be blank") + if not callable(self.fn): + raise ValueError("scorer fn must be callable") + _validate_thresholds( + threshold=self.threshold, + pass_rate_threshold=self.pass_rate_threshold, + ) + + @property + def criterion_type(self) -> str: + return self.name + + def to_criteria_wire(self) -> dict[str, Any]: + return { + "criterionType": self.criterion_type, + "kind": "scorer", + "successDirection": self.success_direction, + "options": _criteria_options( + threshold=self.threshold, + pass_rate_threshold=self.pass_rate_threshold, + ), + } + + +type Criterion = Judge | Scorer + + +def _validate_thresholds( + *, + threshold: float | None, + pass_rate_threshold: float | None, +) -> None: + for name, value in ( + ("threshold", threshold), + ("pass_rate_threshold", pass_rate_threshold), + ): + if value is None: + continue + # NaN passes both range comparisons, so without an explicit check it + # reaches to_criteria_wire and is serialized as a bare ``NaN`` literal + # the management API rejects -- an evaluation that fails to be created + # rather than a threshold that fails to validate. + if isnan(value) or value < 0 or value > 1: + raise ValueError(f"{name} must be a number between 0 and 1") + + +def _criteria_options( + *, + threshold: float | None, + pass_rate_threshold: float | None, + ground_truth_context: str | None = None, +) -> dict[str, Any]: + options: dict[str, Any] = {} + if threshold is not None: + options["threshold"] = threshold + if pass_rate_threshold is not None: + options["passRateThreshold"] = pass_rate_threshold + if ground_truth_context is not None: + options["groundTruthContext"] = ground_truth_context + return options diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/events.py b/packages/client/src/launchdarkly_ai_server/evaluations/events.py new file mode 100644 index 00000000..4098e61b --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/events.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from typing import Any + + +class CriterionStatus(StrEnum): + COMPLETE = "COMPLETE" + ERROR = "ERROR" + + +class CriterionEventKind(StrEnum): + JUDGE = "judge" + SCORER = "scorer" + + +@dataclass(frozen=True) +class TokenUsage: + """Token usage reported by an LD Judge provider call.""" + + input_tokens: int + output_tokens: int + + def to_wire(self) -> dict[str, int]: + return { + "inputTokens": self.input_tokens, + "outputTokens": self.output_tokens, + } + + +@dataclass(frozen=True, kw_only=True) +class CriterionEventPayload: + """Common fields emitted for every SDK-run evaluation criterion result.""" + + project_key: str + evaluation_id: str + evaluation_run_id: str + run_id: str + dataset_id: str + row_index: int + criterion_type: str + kind: CriterionEventKind + event_id: str + emitted_at: str + evaluation_key: str + dataset_key: str + status: CriterionStatus + started_at: str + evaluated_at: str + latency_ms: int + evaluation_version: int | None = None + score: float | None = None + reason: str | None = None + error: dict[str, Any] | None = None + error_message: str | None = None + + def to_track_payload(self) -> dict[str, Any]: + payload: dict[str, Any] = { + "projectKey": self.project_key, + "evaluationId": self.evaluation_id, + "evaluationRunId": self.evaluation_run_id, + "runId": self.run_id, + "datasetId": self.dataset_id, + "rowIndex": self.row_index, + "criterionType": self.criterion_type, + "kind": self.kind.value, + "eventId": self.event_id, + "emittedAt": self.emitted_at, + "evaluationKey": self.evaluation_key, + "evaluationVersion": self.evaluation_version, + "datasetKey": self.dataset_key, + "status": self.status.value, + "startedAt": self.started_at, + "evaluatedAt": self.evaluated_at, + "latencyMs": self.latency_ms, + "score": self.score, + "reason": self.reason, + "error": self.error, + "errorMessage": self.error_message, + } + return {key: value for key, value in payload.items() if value is not None} + + +@dataclass(frozen=True, kw_only=True) +class LDJudgeCriterionEventPayload(CriterionEventPayload): + """Payload for one LaunchDarkly AI Judge result on one dataset row.""" + + kind: CriterionEventKind = CriterionEventKind.JUDGE + judge_key: str + variation_key: str + version: int | None = None + usage: TokenUsage | None = None + + def to_track_payload(self) -> dict[str, Any]: + payload = super().to_track_payload() + payload["judgeKey"] = self.judge_key + payload["variationKey"] = self.variation_key + if self.version is not None: + payload["version"] = self.version + if self.usage is not None: + payload["usage"] = self.usage.to_wire() + return payload + + +@dataclass(frozen=True, kw_only=True) +class DeterministicScorerCriterionEventPayload(CriterionEventPayload): + """Payload for one local deterministic scorer result on one dataset row.""" + + kind: CriterionEventKind = CriterionEventKind.SCORER diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 4a3ffad3..80d22ece 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -17,7 +17,14 @@ Transport, urllib_transport, ) -from .runner import EvalHandler, EvaluationsRunner, ToolImplementation, _segment +from .criteria import Criterion, Judge +from .runner import ( + EvalHandler, + EvaluationsRunner, + ToolImplementation, + _provides_for, + _segment, +) from .types import EvalRunResult, GenerationConfig, RunSummary logger = logging.getLogger(__name__) @@ -93,12 +100,26 @@ async def run( handler: EvalHandler, generation: GenerationConfig, tools: Mapping[str, ToolImplementation] | None = None, + criteria: list[Criterion] | None = None, + judge_handlers: list[EvalHandler] | None = None, concurrency: int = 10, poll_interval_seconds: float | None = None, poll_timeout_seconds: float | None = None, ) -> EvalRunResult: """ - Create and run a generation-only evaluation in the caller's process. + Create and run an evaluation in the caller's process. + + Each dataset row is generated with ``handler``; every entry in + ``criteria`` — LaunchDarkly :class:`Judge` references and local + deterministic :class:`Scorer` functions — is then run against each + generated row, and one evaluation event is emitted per + ``(row, criterion)`` result. + + A :class:`Judge` is an independent AI Config and may be served by a + different provider or mode than ``generation``. ``handler`` runs a judge + only when it provides for that judge's provider; pass handlers for any + other providers your judges use in ``judge_handlers``. A judge no + handler covers fails the run before any records are created. The returned pass/fail result is derived from LaunchDarkly's run summary. A CI script can exit with ``0 if result.passed else 1`` after awaiting @@ -121,14 +142,24 @@ async def run( poll_timeout_seconds=poll_timeout_seconds, ) run_tools = dict(tools or {}) + run_criteria = list(criteria or []) + run_judge_handlers = list(judge_handlers or []) + self._validate_criteria(run_criteria) + self._validate_judge_handlers(run_judge_handlers) + ld_judges = [ + criterion for criterion in run_criteria if isinstance(criterion, Judge) + ] client = await self._resolve_client() # The management API client is synchronous; running it in a worker thread # keeps the caller's event loop free. - # Tool verification is deliberately first: a typo must not create records. + # Tool/judge verification is deliberately first: a typo must not create records. resolved_tools = await asyncio.to_thread( self._runner._resolve_tools, project_key, run_tools ) + resolved_judges = await self._runner._resolve_judges( + project_key, ld_judges, handler, run_judge_handlers + ) dataset_ref = await asyncio.to_thread( self._runner._fetch_dataset, project_key, dataset ) @@ -141,12 +172,12 @@ async def run( key, generation, resolved_tools, + run_criteria, ) evaluation_run = await asyncio.to_thread( self._runner._create_evaluation_run, project_key, evaluation.id, - len(rows), dataset_ref.id, ) config = self._runner._build_handler_config(generation, resolved_tools) @@ -157,17 +188,37 @@ async def run( run_tools, concurrency, ) - self._runner._emit_generation_events( - client, - project_key=project_key, - evaluation=evaluation, - evaluation_run=evaluation_run, - dataset=dataset_ref, - results=results, - ) - flush_result = client.flush() - if inspect.isawaitable(flush_result): - await flush_result + try: + self._runner._emit_generation_events( + client, + project_key=project_key, + evaluation=evaluation, + evaluation_run=evaluation_run, + dataset=dataset_ref, + results=results, + ) + if run_criteria: + criterion_results = await self._runner._run_criteria_for_results( + results, + run_tools, + run_criteria, + resolved_judges, + concurrency, + ) + self._runner._emit_evaluation_events( + client, + project_key=project_key, + evaluation=evaluation, + evaluation_run=evaluation_run, + dataset=dataset_ref, + results=criterion_results, + ) + finally: + # Generation results already queued on the SDK event buffer must + # reach LaunchDarkly even when the criteria phase fails. + flush_result = client.flush() + if inspect.isawaitable(flush_result): + await flush_result summary = await self._poll_summary_until_terminal( project_key, evaluation.id, @@ -180,7 +231,16 @@ async def run( f"{_segment(evaluation.id)}/runs/{_segment(evaluation_run.id)}" ) return EvalRunResult( - passed=(summary.error_rows == 0 and summary.pending_rows == 0), + # failed_rows counts rows whose criteria were scored and did not + # meet their threshold, so a gate that ignores it exits 0 on a run + # where every row failed its judge. It was omissible while runs were + # generation-only -- a row either generated or errored, and nothing + # produced a fail -- and stops being so the moment criteria exist. + passed=( + summary.error_rows == 0 + and summary.failed_rows == 0 + and summary.pending_rows == 0 + ), url=url, run_id=evaluation_run.id, summary=summary, @@ -242,6 +302,52 @@ async def _resolve_client(self) -> Any: ) return await init_client({"sdkKey": self._sdk_key}) + @staticmethod + def _validate_criteria(criteria: list[Criterion]) -> None: + """Reject duplicate criterion identities before any records are created. + + A judge key and a scorer name that collide would share a criterionType, + and with it the deterministic event identity of their results. Case- + insensitive, matching the API's own dedup: the worker's retry gate + lowercases criterion types, so two criteria differing only by case + would still collide there even though they look distinct here. + """ + seen: set[str] = set() + duplicates: list[str] = [] + for criterion in criteria: + criterion_type = criterion.criterion_type + normalized = criterion_type.lower() + if normalized in seen and criterion_type not in duplicates: + duplicates.append(criterion_type) + seen.add(normalized) + if duplicates: + raise EvaluationsError( + "Duplicate evaluation criteria: " + + ", ".join(repr(name) for name in duplicates) + + ". Judge keys and scorer names must be unique within a run " + "(case-insensitive)." + ) + + @staticmethod + def _validate_judge_handlers(judge_handlers: list[EvalHandler]) -> None: + """Reject judge handlers that cannot be routed by provider and mode. + + A judge handler is only ever chosen by matching its ``provides_for`` + against the judge's resolved provider and mode. One without that + metadata could never be selected, so it would silently fall through to + the generation handler instead of running the judge it was passed for. + """ + for index, candidate in enumerate(judge_handlers): + if not callable(candidate): + raise EvaluationsError(f"judge_handlers[{index}] must be callable") + if _provides_for(candidate) is None: + raise EvaluationsError( + f"judge_handlers[{index}] does not declare provides_for. " + "Build judge handlers with create_handler() (or a provider " + "package's create_*_handler()) so they can be matched to a " + "judge's provider and mode." + ) + @staticmethod def _validate_run_args( *, diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index 951213b5..164ea57d 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -2,33 +2,161 @@ import asyncio import hashlib +import inspect import json +import logging import time import urllib.parse from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass from datetime import UTC, datetime -from typing import Any +from typing import Any, Literal +from ..judge_scoring import ( + FORMATTING_INSTRUCTIONS, + numeric_score, + parse_judge_response, +) +from ..lifecycle import extract_variation from ..types import NativeTool -from ..utils import parse_template, parse_usage, to_ld_context +from ..utils import ( + collapse_messages_to_instructions, + normalize_mode, + parse_template, + parse_usage, + to_ld_context, +) from .api import EvaluationsError, LDApiClient, LDApiError +from .criteria import Criterion, Judge, Scorer +from .events import ( + CriterionEventPayload, + CriterionStatus, + DeterministicScorerCriterionEventPayload, + LDJudgeCriterionEventPayload, + TokenUsage, +) from .types import ( DatasetRef, DatasetRow, EvaluationRef, EvaluationRunRef, GenerationConfig, + ResolvedJudge, ResolvedTool, RunSummary, ) +logger = logging.getLogger(__name__) + DATASET_PAGE_SIZE = 200 GENERATION_EVENT_NAME = "$ld:ai:offline-evals:generation" +CRITERION_EVENT_NAME = "$ld:ai:offline-evals:criterion" EvalHandler = Callable[..., Awaitable[dict[str, Any]]] ToolImplementation = Callable[..., Any] | NativeTool +@dataclass(frozen=True) +class JudgeExecution: + """A resolved judge paired with the handler selected to run its config.""" + + resolved: ResolvedJudge + handler: EvalHandler + collapse_messages: bool = False + + +def _provides_for( + handler: EvalHandler, +) -> tuple[str, Literal["agent", "messages"]] | None: + provides_for = getattr(handler, "provides_for", None) + if ( + isinstance(provides_for, tuple | list) + and len(provides_for) == 2 + and isinstance(provides_for[0], str) + ): + return (provides_for[0], normalize_mode(provides_for[1])) + return None + + +def _covers_provider( + provides_for: tuple[str, Literal["agent", "messages"]], + provider: str | None, +) -> bool: + return provides_for[0] == provider or provides_for[0] == "*" + + +def _find_judge_handler( + judge_handlers: list[EvalHandler], + provider: str | None, + mode: Literal["agent", "messages"], +) -> EvalHandler | None: + """Find a handler for ``provider`` in ``mode``, exact match before wildcard. + + A wildcard handler is a fallback for multi-provider adapters, so it is only + chosen when no handler names the provider outright -- the priority + ``config()`` already applies to a generation config. Searching in one pass + would instead let the order the caller happened to list its handlers in + decide, sending an OpenAI judge through a LangChain adapter that was merely + listed first. + """ + for exact in (True, False): + for candidate in judge_handlers: + provides_for = _provides_for(candidate) + if provides_for is None or provides_for[1] != mode: + continue + if exact: + if provides_for[0] == provider: + return candidate + elif provides_for[0] == "*": + return candidate + return None + + +def _select_judge_handler( + resolved: ResolvedJudge, + handler: EvalHandler, + judge_handlers: list[EvalHandler], +) -> JudgeExecution | None: + """Pick the handler that can run this judge's config, or ``None``. + + A judge is an independent AI Config: it may resolve to a different provider + and mode than the evaluation's generation config, and a handler built for + one provider cannot execute another's config. The priority mirrors the + online path (``judges.run_judges``): + + 1. a judge handler in the judge's mode, naming its provider outright + before any wildcard adapter; + 2. an agent-mode judge handler for a messages-mode judge, whose messages + are collapsed into a single instructions block; + 3. the generation handler, when it covers the judge's provider. + + A handler that declares no ``provides_for`` is a plain callable doing its + own routing -- the same contract it already honours for the generation + config -- so it is treated as covering every judge. + """ + match = _find_judge_handler(judge_handlers, resolved.provider, resolved.mode) + if match is not None: + return JudgeExecution(resolved=resolved, handler=match) + if resolved.mode == "messages": + agent_fallback = _find_judge_handler(judge_handlers, resolved.provider, "agent") + if agent_fallback is not None: + return JudgeExecution( + resolved=resolved, handler=agent_fallback, collapse_messages=True + ) + generation_provides_for = _provides_for(handler) + if generation_provides_for is None: + return JudgeExecution(resolved=resolved, handler=handler) + if _covers_provider(generation_provides_for, resolved.provider): + return JudgeExecution( + resolved=resolved, + handler=handler, + collapse_messages=( + generation_provides_for[1] == "agent" and resolved.mode == "messages" + ), + ) + return None + + def _segment(value: str) -> str: return urllib.parse.quote(value, safe="") @@ -124,6 +252,74 @@ def _resolve_tools( ) return resolved + async def _resolve_judges( + self, + project_key: str, + judges: list[Judge], + handler: EvalHandler, + judge_handlers: list[EvalHandler] | None = None, + ) -> dict[str, JudgeExecution]: + """Resolve LD Judge configs before any evaluation records are created. + + Each judge is paired with the handler that can execute its config here, + rather than at scoring time, so a judge no handler covers fails the run + before any records exist or any generation spend happens. + """ + available_judge_handlers = list(judge_handlers or []) + resolved: dict[str, JudgeExecution] = {} + # variation() rejects a context without kind and key; use the same + # context shape the emitted evaluation events are attributed to. + context: dict[str, Any] = {"kind": "evaluation", "key": project_key} + for judge in judges: + try: + variation = await extract_variation(judge.key, context) + except Exception as error: + raise EvaluationsError( + f"Failed to resolve LaunchDarkly judge {judge.key!r}: {error} " + f"If the judge does not exist in project {project_key!r}, " + "create it in the LaunchDarkly UI and try again." + ) from error + config = variation.get("config") + meta_value = variation.get("meta") + meta: Mapping[str, Any] = ( + meta_value if isinstance(meta_value, Mapping) else {} + ) + if not isinstance(config, Mapping): + raise EvaluationsError( + f"LaunchDarkly judge {judge.key!r} returned an invalid AI config variation" + ) + provider_value = config.get("provider") + provider = ( + provider_value.get("name") + if isinstance(provider_value, Mapping) + else None + ) + resolved_judge = ResolvedJudge( + key=judge.key, + config=dict(config), + variation_key=str(meta.get("variationKey") or ""), + version=int(meta["version"]) + if isinstance(meta.get("version"), int) + else None, + provider=str(provider) if isinstance(provider, str) else None, + mode=normalize_mode( + meta.get("mode") if isinstance(meta.get("mode"), str) else None + ), + ) + execution = _select_judge_handler( + resolved_judge, handler, available_judge_handlers + ) + if execution is None: + raise EvaluationsError( + f"No handler can run LaunchDarkly judge {judge.key!r}: its " + f"config is served by provider {resolved_judge.provider!r} in " + f"{resolved_judge.mode!r} mode, which neither the generation " + "handler nor any judge_handlers entry provides for. Pass a " + "handler for that provider to run(judge_handlers=[...])." + ) + resolved[judge.key] = execution + return resolved + def _fetch_dataset(self, project_key: str, dataset_key: str) -> DatasetRef: path = f"projects/{_segment(project_key)}/datasets/{_segment(dataset_key)}" try: @@ -231,6 +427,7 @@ def _create_evaluation( key: str, generation: GenerationConfig, tools: Mapping[str, ResolvedTool], + criteria: list[Criterion] | None = None, ) -> EvaluationRef: body: dict[str, Any] = { "name": key, @@ -253,6 +450,8 @@ def _create_evaluation( body["tools"] = [ {"key": tool.key, "version": tool.version} for tool in tools.values() ] + if criteria: + body["criteria"] = [criterion.to_criteria_wire() for criterion in criteria] path = f"projects/{_segment(project_key)}/evaluations" raw = _mapping(self._api.post(path, body=body), description="evaluation") @@ -269,22 +468,18 @@ def _create_evaluation_run( self, project_key: str, evaluation_id: str, - row_count: int, dataset_id: str, ) -> EvaluationRunRef: path = ( f"projects/{_segment(project_key)}/evaluations/" f"{_segment(evaluation_id)}/runs" ) + body: dict[str, Any] = { + "source": "api", + "datasetId": dataset_id, + } raw = _mapping( - self._api.post( - path, - body={ - "source": "api", - "rowCount": row_count, - "datasetId": dataset_id, - }, - ), + self._api.post(path, body=body), description="evaluation run", ) return self._run_ref(raw) @@ -475,9 +670,389 @@ def _emit_generation_events( if "usage" in generated: payload["usage"] = generated["usage"] client.track(GENERATION_EVENT_NAME, context, payload, 1) - print( - f"{GENERATION_EVENT_NAME} emittedAt={emitted_at} eventId={event_id}", - flush=True, + logger.info( + "%s emittedAt=%s eventId=%s", + GENERATION_EVENT_NAME, + emitted_at, + event_id, + ) + + def _judge_variables( + self, + row_result: Mapping[str, Any], + judge: Judge, + ) -> dict[str, Any]: + """Variables available to the judge config's ``{{...}}`` placeholders. + + Absent values become empty strings: ``parse_template`` leaves a + placeholder with a ``None`` value as-is, and literal mustache text must + not reach the judge model. + """ + variables = dict(row_result.get("variables") or {}) + output = row_result.get("output") + expected = row_result.get("expected_output") + ground_truth = judge.ground_truth_context + if ground_truth is not None: + ground_truth = parse_template(ground_truth, variables) + elif expected is not None: + ground_truth = str(expected) + # message_history carries FORMATTING_INSTRUCTIONS the same way the + # online path builds it (judges.run_judges), because that -- not the + # standalone formatting_instructions variable below -- is what every + # judge built from the AI Library's default templates (accuracy, + # relevance, toxicity, and any judge cloned from them) actually + # references. A judge authored before this variable existed must keep + # getting scored without edits. + variables.update( + { + "input": row_result.get("input") or "", + "response_to_evaluate": output if output is not None else "", + "message_history": "\n\n".join( + str(value) + for value in ( + row_result.get("input"), + output, + FORMATTING_INSTRUCTIONS, + ) + if value + ), + "expected_output": expected if expected is not None else "", + "ground_truth_context": ( + ground_truth if ground_truth is not None else "" + ), + } + ) + return variables + + def _criterion_error_result( + self, + base: Mapping[str, Any], + started_clock: float, + code: str, + message: str, + ) -> dict[str, Any]: + completed = datetime.now(UTC) + return { + **base, + "status": "ERROR", + "error": {"code": code, "message": message}, + "evaluated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round((time.perf_counter() - started_clock) * 1000), + } + + async def _run_scorer_for_result( + self, + row: Mapping[str, Any], + scorer: Scorer, + ) -> dict[str, Any]: + started = datetime.now(UTC) + started_clock = time.perf_counter() + base: dict[str, Any] = { + "row_index": row["row_index"], + "criterion_type": scorer.criterion_type, + "kind": "scorer", + "started_at": started.isoformat().replace("+00:00", "Z"), + } + if row.get("status") != "COMPLETE": + return self._criterion_error_result( + base, + started_clock, + "generation_incomplete", + "generation did not complete", + ) + dataset_row = DatasetRow( + row_index=row["row_index"], + input=row.get("input"), + expected_output=row.get("expected_output"), + variables=dict(row.get("variables") or {}), + metadata=row.get("metadata"), + ) + try: + score_value = scorer.fn(dataset_row, row.get("output")) + if inspect.isawaitable(score_value): + score_value = await score_value + except Exception as error: + return self._criterion_error_result( + base, started_clock, "scorer_raised", f"scorer fn raised: {error}" + ) + if isinstance(score_value, bool): + score: float = 1.0 if score_value else 0.0 + else: + maybe_score = numeric_score(score_value) + if maybe_score is None: + return self._criterion_error_result( + base, + started_clock, + "invalid_score", + "scorer fn must return a bool or a finite number, " + f"got {score_value!r}", + ) + score = maybe_score + if score < 0 or score > 1: + return self._criterion_error_result( + base, + started_clock, + "invalid_score", + f"scorer fn score must be between 0 and 1, got {score_value!r}", + ) + completed = datetime.now(UTC) + return { + **base, + "status": "COMPLETE", + "score": score, + "reason": None, + "evaluated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round((time.perf_counter() - started_clock) * 1000), + } + + async def _run_ld_judge_for_result( + self, + row: Mapping[str, Any], + tool_handlers: dict[str, ToolImplementation], + judge: Judge, + execution: JudgeExecution, + ) -> dict[str, Any]: + resolved = execution.resolved + started = datetime.now(UTC) + started_clock = time.perf_counter() + base: dict[str, Any] = { + "row_index": row["row_index"], + "criterion_type": judge.criterion_type, + "kind": "judge", + "judge_key": judge.key, + "started_at": started.isoformat().replace("+00:00", "Z"), + "variation_key": resolved.variation_key, + "version": resolved.version, + } + if row.get("status") != "COMPLETE": + return self._criterion_error_result( + base, + started_clock, + "generation_incomplete", + "generation did not complete", + ) + # The config is passed unrendered: the handler owns the single + # parse_template pass, so ``{{...}}`` sequences inside generated output + # or dataset values are never re-expanded into the judge prompt. + variables = self._judge_variables(row, judge) + # An agent-mode handler standing in for a messages-mode judge needs the + # messages folded into one instructions block, exactly as the online + # path does before handing a judge config to an agent handler. + judge_config = ( + collapse_messages_to_instructions(resolved.config) + if execution.collapse_messages + else resolved.config + ) + try: + result = await execution.handler( + dict(judge_config), + row.get("output"), + tool_handlers, + { + **variables, + "formatting_instructions": FORMATTING_INSTRUCTIONS, + }, + ) + except Exception as error: + return self._criterion_error_result( + base, started_clock, "handler_raised", f"judge handler raised: {error}" + ) + if not isinstance(result, Mapping): + return self._criterion_error_result( + base, + started_clock, + "invalid_judge_output", + "judge handler result must be a mapping", + ) + try: + raw_score, reason = parse_judge_response( + result.get("output", result.get("response")) + ) + except ValueError as error: + return self._criterion_error_result( + base, started_clock, "invalid_judge_output", str(error) + ) + score = numeric_score(raw_score) + if score is None or score < 0 or score > 1: + return self._criterion_error_result( + base, + started_clock, + "invalid_score", + f"judge score must be a number between 0 and 1, got {raw_score!r}", + ) + completed = datetime.now(UTC) + # No verdict: the SDK reports the score and LaunchDarkly rules on it. The + # criterion carries the threshold and the judge's success direction, and + # ai-evaluator compares them at ingest -- so pass/fail policy is one + # server-side implementation that applies to every SDK version and to runs + # already recorded, rather than one frozen into each release of each + # language's SDK. + event = { + **base, + "status": "COMPLETE", + "score": score, + "reason": reason, + "evaluated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round((time.perf_counter() - started_clock) * 1000), + } + usage = result.get("usage") + if isinstance(usage, Mapping): + event["usage"] = dict(usage) + return event + + async def _run_criteria_for_results( + self, + rows: list[dict[str, Any]], + tool_handlers: dict[str, ToolImplementation], + criteria: list[Criterion], + resolved_judges: Mapping[str, JudgeExecution], + concurrency: int, + ) -> list[dict[str, Any]]: + """Run every (row, criterion) pair, bounded by the run's concurrency.""" + controller = ConcurrencyController(concurrency) + + async def run_one( + row: Mapping[str, Any], criterion: Criterion + ) -> dict[str, Any]: + await controller.acquire() + try: + if isinstance(criterion, Scorer): + return await self._run_scorer_for_result(row, criterion) + return await self._run_ld_judge_for_result( + row, + tool_handlers, + criterion, + resolved_judges[criterion.key], + ) + finally: + controller.release() + + return list( + await asyncio.gather( + *(run_one(row, criterion) for row in rows for criterion in criteria) + ) + ) + + def _emit_evaluation_events( + self, + client: Any, + *, + project_key: str, + evaluation: EvaluationRef, + evaluation_run: EvaluationRunRef, + dataset: DatasetRef, + results: list[dict[str, Any]], + ) -> None: + context = to_ld_context( + client, + { + "kind": "evaluation", + "key": evaluation_run.id, + "projectKey": project_key, + "evaluationId": evaluation.id, + }, + ) + failures: list[str] = [] + for result in results: + # One bad criterion result must not stop the remaining results from + # being emitted, or drop the events already queued for the ones + # before it -- so every result is attempted and the failures are + # raised together once the loop is done. + try: + identity = { + "projectKey": project_key, + "evaluationId": evaluation.id, + "evaluationRunId": evaluation_run.id, + "runId": evaluation_run.id, + "datasetId": dataset.id, + "rowIndex": result["row_index"], + "criterionType": result["criterion_type"], + } + event_id = hashlib.sha256( + json.dumps(identity, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + emitted_at = datetime.now(UTC).isoformat().replace("+00:00", "Z") + usage: TokenUsage | None = None + if result["kind"] == "judge" and isinstance( + result.get("usage"), Mapping + ): + normalized_usage = parse_usage(dict(result["usage"])) + usage = TokenUsage( + input_tokens=normalized_usage["input"], + output_tokens=normalized_usage["output"], + ) + error = result.get("error") + error_message: str | None = None + if result["status"] == "ERROR": + if isinstance(error, Mapping) and error.get("message"): + error_message = str(error["message"]) + else: + error_message = str(error) if error else "Unknown error" + common_payload: dict[str, Any] = { + "project_key": project_key, + "evaluation_id": evaluation.id, + "evaluation_run_id": evaluation_run.id, + "run_id": evaluation_run.id, + "dataset_id": dataset.id, + "row_index": result["row_index"], + "criterion_type": result["criterion_type"], + "event_id": event_id, + "emitted_at": emitted_at, + "evaluation_key": evaluation.key, + "evaluation_version": evaluation.version, + "dataset_key": dataset.key, + "status": CriterionStatus(result["status"]), + "started_at": result["started_at"], + "evaluated_at": result["evaluated_at"], + "latency_ms": result["latency_ms"], + "score": result.get("score"), + "reason": result.get("reason"), + "error": error, + "error_message": error_message, + } + payload_model: CriterionEventPayload + if result["kind"] == "judge": + payload_model = LDJudgeCriterionEventPayload( + **common_payload, + judge_key=result["judge_key"], + variation_key=result["variation_key"], + version=result.get("version"), + usage=usage, + ) + else: + payload_model = DeterministicScorerCriterionEventPayload( + **common_payload + ) + client.track( + CRITERION_EVENT_NAME, context, payload_model.to_track_payload(), 1 + ) + except Exception as error: + logger.exception( + "Failed to emit evaluation event for row %s criterion %s", + result.get("row_index"), + result.get("criterion_type"), + ) + failures.append( + f"row {result.get('row_index')} criterion " + f"{result.get('criterion_type')!r}: {error}" + ) + continue + logger.info( + "%s emittedAt=%s eventId=%s", + CRITERION_EVENT_NAME, + emitted_at, + event_id, + ) + if failures: + # The evaluation was created with a fixed criterion list, so the + # backend needs one result per (row, criterion) before it can finish + # row accounting. A dropped event is never converted into an error + # result by anything downstream -- swallowing it here would leave + # run() polling to its timeout instead of reporting what failed. + raise EvaluationsError( + f"Failed to emit {len(failures)} of {len(results)} evaluation " + "criterion events, so the run cannot be fully accounted for: " + + "; ".join(failures) ) def _get_summary( diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/types.py b/packages/client/src/launchdarkly_ai_server/evaluations/types.py index 5f4f4d5c..f397859b 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/types.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/types.py @@ -2,7 +2,7 @@ from collections.abc import Mapping from dataclasses import dataclass, field -from typing import Any, TypedDict +from typing import Any, Literal, TypedDict @dataclass @@ -67,6 +67,24 @@ class ResolvedTool: schema: dict[str, Any] = field(default_factory=dict) +@dataclass +class ResolvedJudge: + """A LaunchDarkly AI Judge config variation resolved for an evaluation run. + + ``provider`` and ``mode`` come from the judge's own variation, not the + evaluation's generation config: a judge is an independent AI Config and may + be served by a different provider in a different mode. They are kept here + because they are what selects the handler that can actually run this config. + """ + + key: str + config: dict[str, Any] + variation_key: str = "" + version: int | None = None + provider: str | None = None + mode: Literal["agent", "messages"] = "messages" + + @dataclass class EvaluationRef: """Identifiers returned after creating an evaluation.""" diff --git a/packages/client/src/launchdarkly_ai_server/judge_scoring.py b/packages/client/src/launchdarkly_ai_server/judge_scoring.py new file mode 100644 index 00000000..8375b413 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/judge_scoring.py @@ -0,0 +1,61 @@ +"""Shared scoring contract for LaunchDarkly AI Judge invocations. + +Both judge execution paths — the online path (``judges.run_judges``, sampled +per invocation) and the offline evaluations path (``evaluations.runner``) — +prompt a judge model for the same ``{"score": <0-1>, "reasoning": }`` +JSON shape and must parse it the same way. This module owns that contract so +the two paths cannot drift. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from math import isfinite +from typing import Any + +from .utils import parse_json_with_possible_fences + +FORMATTING_INSTRUCTIONS = "\n".join( + [ + "Your response MUST be in valid JSON format with the following structure:", + '{ "score": , "reasoning": }', + "The output must be valid, parseable JSON. Do not include additional tags, comments, " + "formatting, or newlines.", + "It should be returned in a format that is immediately parseable by a JSON parsing " + "function. Do not include ```json tags.", + ] +) + + +def numeric_score(score: Any) -> float | None: + """Return ``score`` as a float only when it already is a finite number. + + Never raises. A judge that returns ``"0.9 (high)"`` or ``None`` must not take down the + evaluation metric track that follows, and must not put a string where semconv defines a double. + """ + if isinstance(score, bool) or not isinstance(score, (int, float)): + return None + value = float(score) + return value if isfinite(value) else None + + +def parse_judge_response(raw: Any) -> tuple[Any, str]: + """Parse a judge model response into ``(score, reasoning)``. + + Accepts a JSON string (possibly wrapped in markdown fences) or an + already-decoded mapping. The score is returned untouched — callers apply + their own policy to non-numeric values via :func:`numeric_score`. + + Raises ``ValueError`` when the response is not a non-empty JSON object. + """ + parsed: Any + if isinstance(raw, Mapping): + parsed = raw + elif isinstance(raw, str): + parsed = parse_json_with_possible_fences(raw) + else: + parsed = None + if not isinstance(parsed, Mapping) or not parsed: + raise ValueError("Invalid JSON from judge") + reasoning = parsed.get("reasoning") or parsed.get("reason") or "" + return parsed.get("score"), str(reasoning) diff --git a/packages/client/src/launchdarkly_ai_server/judges.py b/packages/client/src/launchdarkly_ai_server/judges.py index ecff6b13..5878d6b1 100644 --- a/packages/client/src/launchdarkly_ai_server/judges.py +++ b/packages/client/src/launchdarkly_ai_server/judges.py @@ -3,10 +3,14 @@ import logging import random from collections.abc import Callable -from math import isfinite from typing import Any from .conversation import with_judge_evaluation +from .judge_scoring import ( + FORMATTING_INSTRUCTIONS, + numeric_score, + parse_judge_response, +) from .types import ( AiConfigRep, JudgeResult, @@ -22,7 +26,6 @@ ) from .utils import ( normalize_mode, - parse_json_with_possible_fences, to_ld_context, to_usage_dict, ) @@ -38,29 +41,6 @@ def _provider_matches(handler: ProviderHandler, provider: str | None) -> bool: logger = logging.getLogger(__name__) -_FORMATTING_INSTRUCTIONS = "\n".join( - [ - "Your response MUST be in valid JSON format with the following structure:", - '{ "score": , "reasoning": }', - "The output must be valid, parseable JSON. Do not include additional tags, comments, " - "formatting, or newlines.", - "It should be returned in a format that is immediately parseable by a JSON parsing " - "function. Do not include ```json tags.", - ] -) - - -def _numeric_score(score: Any) -> float | None: - """Return ``score`` as a float only when it already is a finite number. - - Never raises. A judge that returns ``"0.9 (high)"`` or ``None`` must not take down the - evaluation metric track that follows, and must not put a string where semconv defines a double. - """ - if isinstance(score, bool) or not isinstance(score, (int, float)): - return None - value = float(score) - return value if isfinite(value) else None - async def run_judges( *, @@ -166,7 +146,7 @@ async def run_judges( ) message_history = "\n\n".join( - filter(None, [user_input, llm_response, _FORMATTING_INSTRUCTIONS]) + filter(None, [user_input, llm_response, FORMATTING_INSTRUCTIONS]) ) async with with_judge_evaluation(judge_key) as record_evaluation: @@ -185,24 +165,16 @@ async def run_judges( }, ) - raw = result["response"] - judge_response = raw if isinstance(raw, str) else str(raw) - - parsed = parse_json_with_possible_fences(judge_response) - if not parsed: - raise ValueError("Invalid JSON from judge") - - score = parsed.get("score") - reasoning = parsed.get("reasoning", "") + score, reasoning = parse_judge_response(result["response"]) judge_results[judge_key] = JudgeResult( usage=to_usage_dict(result["usage"]), response=reasoning, score=score, ) - numeric_score = _numeric_score(score) - if numeric_score is not None: + metric_score = numeric_score(score) + if metric_score is not None: record_evaluation( - numeric_score, + metric_score, reasoning if judge_handler.capture_content else None, ) @@ -403,7 +375,7 @@ def _matches(h: ProviderHandler) -> bool: ) message_history = "\n\n".join( - filter(None, [task.actual_output, _FORMATTING_INSTRUCTIONS]) + filter(None, [task.actual_output, FORMATTING_INSTRUCTIONS]) ) async with with_judge_evaluation(task.config_key) as record_evaluation: @@ -422,18 +394,19 @@ def _matches(h: ProviderHandler) -> bool: }, ) - raw = result["response"] - judge_response = raw if isinstance(raw, str) else str(raw) - parsed = parse_json_with_possible_fences(judge_response) - if not parsed: + try: + score, reasoning = parse_judge_response(result["response"]) + except ValueError: return None - score = parsed.get("score", 0.0) - reasoning = parsed.get("reasoning", "") - numeric_score = _numeric_score(score) - if numeric_score is not None: + # The score is reported as the judge gave it. A missing or null score + # is not a zero: coercing it would record a gen_ai.evaluation of 0 -- + # indistinguishable from a judge that scored the output a hard fail -- + # where every other non-numeric judge output skips the metric instead. + metric_score = numeric_score(score) + if metric_score is not None: record_evaluation( - numeric_score, + metric_score, reasoning if judge_handler.capture_content else None, ) raw_usage = result["usage"] diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index 5784559e..c7ca278b 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -8,9 +8,13 @@ import pytest +from launchdarkly_ai_server import create_handler from launchdarkly_ai_server.evaluations import ( + DatasetRow, EvaluationsError, HttpResponse, + Judge, + Scorer, init_evaluations, ) @@ -117,8 +121,9 @@ def lookup_order(order_id: str) -> str: @pytest.mark.asyncio async def test_complete_run_with_zero_failed_and_error_rows_passes( monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], + caplog: pytest.LogCaptureFixture, ) -> None: + caplog.set_level("INFO", logger="launchdarkly_ai_server.evaluations.runner") monkeypatch.delenv("LD_SDK_KEY", raising=False) init_client = AsyncMock() monkeypatch.setattr( @@ -278,7 +283,6 @@ async def test_complete_run_with_zero_failed_and_error_rows_passes( ) assert transport.requests[5]["body"] == { "source": "api", - "rowCount": 2, "datasetId": "33333333-3333-3333-3333-333333333333", } @@ -306,9 +310,13 @@ async def test_complete_run_with_zero_failed_and_error_rows_passes( assert event["emittedAt"].endswith("Z") assert datetime.fromisoformat(event["emittedAt"]).tzinfo is not None assert {"input", "expected_output", "metadata", "variables"}.isdisjoint(event) - output_lines = capsys.readouterr().out.splitlines() - assert len(output_lines) == 2 - assert output_lines[0] == ( + emit_logs = [ + record.getMessage() + for record in caplog.records + if record.name == "launchdarkly_ai_server.evaluations.runner" + ] + assert len(emit_logs) == 2 + assert emit_logs[0] == ( "$ld:ai:offline-evals:generation " f"emittedAt={event['emittedAt']} eventId={event['eventId']}" ) @@ -805,7 +813,15 @@ async def handler(*args: object) -> dict[str, Any]: @pytest.mark.asyncio -async def test_generation_failed_rows_do_not_fail_the_result() -> None: +async def test_failed_rows_fail_the_result() -> None: + """A row the server scored and marked failed must fail the gate. + + This reverses the previous assertion, which was written when runs were + generation-only -- a row then either generated or errored, and nothing + produced a "failed", so excluding failed_rows was unobservable. With + criteria it is the normal way a run fails, and a gate that ignores it exits + 0 on a run where every row failed its judge. + """ transport = SequencedTransport( [ response(200, {"id": "dataset-id", "name": "golden"}), @@ -849,7 +865,7 @@ async def handler(*args: object) -> dict[str, Any]: ) assert result.summary.failed_rows == 1 - assert result.passed is True + assert result.passed is False @pytest.mark.asyncio @@ -1033,3 +1049,1203 @@ async def fake_init_client(options: dict[str, Any]) -> MagicMock: assert "inputTokens" not in error_event assert "outputTokens" not in error_event assert {"input", "expected_output", "metadata", "variables"}.isdisjoint(error_event) + + +@pytest.mark.asyncio +async def test_run_with_ld_judge_emits_per_criterion_evaluation_event( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": 42, + "input": "Question {{id}}", + "expectedOutput": "Answer {{id}}", + "variables": {"id": "A"}, + } + ], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + assert key == "$ld:ai:judge:accuracy" + # An empty or kindless context is invalid to the real LD SDK and would + # make every judge resolution fail. + assert context == {"kind": "evaluation", "key": "proj"} + return { + "config": { + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + "instructions": "Judge {{response_to_evaluate}} against {{expected_output}}", + }, + "meta": {"variationKey": "default", "version": 12}, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + assert user_input == "generated" + assert variables["response_to_evaluate"] == "generated" + assert variables["expected_output"] == "Answer A" + # The SDK hands the judge config over unrendered; the handler owns + # the single template pass. + assert config["instructions"] == ( + "Judge {{response_to_evaluate}} against {{expected_output}}" + ) + assert variables["formatting_instructions"].startswith( + "Your response MUST be in valid JSON" + ) + # message_history must carry the formatting instructions the same + # way judges.run_judges (the online path) builds it: every judge + # built from the AI Library's default templates references + # {{message_history}}, not the standalone formatting_instructions + # variable above, to ask for the {score, reasoning} JSON shape. + assert ( + "Your response MUST be in valid JSON format" + in (variables["message_history"]) + ) + return { + "output": '{"score": 0.86, "reasoning": "matches policy"}', + "usage": {"input_tokens": 640, "output_tokens": 48}, + } + return { + "output": "generated", + "usage": {"input_tokens": 10, "output_tokens": 4}, + } + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + # kind and judgeKey are what let ai-evaluator store this as a judge rather + # than default it to a deepeval metric and reject the key. successDirection + # is deliberately absent: LaunchDarkly injects it from the judge's AI Config + # on the way through, so the SDK must not assert a direction of its own. + assert transport.requests[2]["body"]["criteria"] == [ + { + "criterionType": "$ld:ai:judge:accuracy", + "kind": "judge", + "judgeKey": "$ld:ai:judge:accuracy", + "options": {"threshold": 0.5}, + } + ] + assert transport.requests[3]["body"] == {"source": "api", "datasetId": "dataset-id"} + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + judge_event = next(event for event in events if event.get("kind") == "judge") + assert judge_event["criterionType"] == "$ld:ai:judge:accuracy" + assert judge_event["judgeKey"] == "$ld:ai:judge:accuracy" + assert judge_event["status"] == "COMPLETE" + assert judge_event["score"] == 0.86 + assert judge_event["reason"] == "matches policy" + assert judge_event["usage"] == {"inputTokens": 640, "outputTokens": 48} + assert judge_event["variationKey"] == "default" + assert judge_event["version"] == 12 + assert len(judge_event["eventId"]) == 64 + # The SDK reports the score and never rules on it: ai-evaluator derives the + # verdict at ingest from the criterion's stored threshold and direction. + assert "verdict" not in judge_event + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("is_inverted", "threshold"), + [ + # The score is fixed at 0.86 below. Under every direction and on both + # sides of the threshold, the SDK reports the same thing: a score. + (False, 0.8), + (False, 0.9), + (True, 0.9), + (True, 0.5), + (None, 0.8), + ], +) +async def test_run_with_ld_judge_never_sends_a_verdict( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, + is_inverted: bool | None, + threshold: float, +) -> None: + """Pass/fail is ai-evaluator's ruling, not the SDK's. + + Parametrized over isInverted -- including the served-payload value -- to + pin that the SDK does not compare even when it could: verdict policy has to + be able to change server-side and apply to runs already recorded, which it + cannot if each SDK release freezes its own comparison. + """ + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": 42, + "input": "Question {{id}}", + "expectedOutput": "Answer {{id}}", + "variables": {"id": "A"}, + } + ], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + config: dict[str, Any] = { + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + "instructions": "Judge {{response_to_evaluate}} against {{expected_output}}", + } + if is_inverted is not None: + config["isInverted"] = is_inverted + return { + "config": config, + "meta": {"variationKey": "default", "version": 12}, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + return { + "output": '{"score": 0.86, "reasoning": "matches policy"}', + "usage": {"input_tokens": 640, "output_tokens": 48}, + } + return { + "output": "generated", + "usage": {"input_tokens": 10, "output_tokens": 4}, + } + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy", threshold=threshold)], + ) + + assert result.passed is True + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + judge_event = next(event for event in events if event.get("kind") == "judge") + assert judge_event["score"] == 0.86 + assert "verdict" not in judge_event + assert "successDirection" not in judge_event + + +@pytest.mark.asyncio +async def test_judges_resolve_once_per_run_not_once_per_row( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + """One resolution for the whole run, however many rows it has. + + extract_variation reads flag delivery, an in-memory store that updates + within seconds of a UI edit, so resolving per row would let an edit + mid-run change the rubric text, judge model, and provider between one row + and the next -- rows in a single run scored against different judges. The + online path does resolve per invocation (judges.build_judge_tasks), so + routing the offline runner through it for convenience is a live way to + reintroduce this. + """ + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + {"rowIndex": 0, "input": "one", "variables": {}}, + {"rowIndex": 1, "input": "two", "variables": {}}, + {"rowIndex": 2, "input": "three", "variables": {}}, + ], + total=3, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa"}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 3, "passed": 3, "error": 0, "pending": 0}}, + ), + ] + ) + + resolutions: list[str] = [] + + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + resolutions.append(key) + return { + "config": { + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + "instructions": "Judge {{response_to_evaluate}}", + }, + "meta": {"variationKey": "default", "version": 12}, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + return {"output": '{"score": 0.9, "reasoning": "fine"}'} + return {"output": "generated"} + + await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert resolutions == ["$ld:ai:judge:accuracy"] + + +def test_scorer_lower_is_better_reaches_the_criteria_wire() -> None: + """A scorer counting something unwanted -- regex hits, edit distance -- + inverts, and only the SDK knows: there is no AI Config for the proxy to + read a scorer's direction off, so what the caller declares is the sole + source ai-evaluator derives its verdict from.""" + + def count_violations(row: DatasetRow, output: Any) -> float: + return 0.0 + + scorer = Scorer( + name="policy-violations", + fn=count_violations, + threshold=0.0, + success_direction="lower_is_better", + ) + + assert scorer.to_criteria_wire() == { + "criterionType": "policy-violations", + "kind": "scorer", + "successDirection": "lower_is_better", + "options": {"threshold": 0.0}, + } + + +def test_judge_threshold_defaults_so_a_criterion_is_always_rulable() -> None: + """A judge with no threshold gives LaunchDarkly nothing to compare against, + so the criterion would be stored and never ruled on.""" + assert Judge(key="$ld:ai:judge:accuracy").to_criteria_wire() == { + "criterionType": "$ld:ai:judge:accuracy", + "kind": "judge", + "judgeKey": "$ld:ai:judge:accuracy", + "options": {"threshold": 0.5}, + } + + +@pytest.mark.asyncio +async def test_missing_ld_judge_aborts_before_mutating_request( + monkeypatch: pytest.MonkeyPatch, +) -> None: + transport = SequencedTransport([]) + + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + raise RuntimeError("not found") + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + with pytest.raises( + EvaluationsError, + match=r"Failed to resolve LaunchDarkly judge 'security-judge': not found", + ): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="security-judge")], + ) + + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_run_with_deterministic_scorer_emits_scorer_evaluation_event( + stub_sdk_client: MagicMock, +) -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "support-golden-v3"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": 42, + "input": "Ticket {{id}}", + "expectedOutput": "refund row", + "variables": {"id": "A"}, + } + ], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return { + "output": "refund exists", + "usage": {"input_tokens": 10, "output_tokens": 4}, + } + + def check_refund(row: DatasetRow, output: Any) -> bool: + assert row.row_index == 42 + assert row.input == "Ticket A" + assert output == "refund exists" + return "refund" in str(output) + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="support-golden-v3", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Scorer(name="refund-exists", fn=check_refund)], + ) + + assert result.passed is True + # A scorer has no LaunchDarkly-side config, so unlike a judge it declares + # its own direction and the proxy leaves it alone. + assert transport.requests[2]["body"]["criteria"] == [ + { + "criterionType": "refund-exists", + "kind": "scorer", + "successDirection": "higher_is_better", + "options": {"threshold": 1.0}, + } + ] + assert transport.requests[3]["body"] == {"source": "api", "datasetId": "dataset-id"} + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + scorer_event = next(event for event in events if event.get("kind") == "scorer") + assert scorer_event["projectKey"] == "proj" + assert scorer_event["evaluationId"] == "evaluation-id" + assert scorer_event["evaluationRunId"] == "run-id" + assert scorer_event["runId"] == "run-id" + assert scorer_event["datasetId"] == "dataset-id" + assert scorer_event["rowIndex"] == 42 + assert scorer_event["criterionType"] == "refund-exists" + assert scorer_event["evaluationKey"] == "support-qa" + assert scorer_event["evaluationVersion"] == 3 + assert scorer_event["datasetKey"] == "support-golden-v3" + assert scorer_event["status"] == "COMPLETE" + assert scorer_event["score"] == 1 + assert "reason" not in scorer_event + assert "usage" not in scorer_event + assert scorer_event["latencyMs"] >= 0 + assert scorer_event["startedAt"].endswith("Z") + assert scorer_event["evaluatedAt"].endswith("Z") + assert "judgeKey" not in scorer_event + assert "variationKey" not in scorer_event + assert "version" not in scorer_event + + +def judge_run_transport(*, summary: dict[str, Any] | None = None) -> SequencedTransport: + """Transport for a one-row run that resolves a dataset, evaluation, and run.""" + return SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": 7, + "input": "Question {{id}}", + "expectedOutput": "Answer {{id}}", + "variables": {"id": "A"}, + } + ], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + summary + or { + "statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0} + }, + ), + ] + ) + + +def accuracy_judge_variation(monkeypatch: pytest.MonkeyPatch) -> None: + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + return { + "config": { + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + "instructions": "Judge {{response_to_evaluate}} against {{expected_output}}", + }, + "meta": {"variationKey": "default", "version": 12}, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + + +@pytest.mark.parametrize( + ("judge_output", "expected_code"), + [ + ('{"score": "high (0.9)", "reasoning": "confident"}', "invalid_score"), + ('{"score": 3, "reasoning": "confident"}', "invalid_score"), + ('{"score": NaN, "reasoning": "confident"}', "invalid_score"), + ("the answer looks right to me", "invalid_judge_output"), + ], +) +@pytest.mark.asyncio +async def test_bad_judge_output_emits_error_event_instead_of_crashing( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, + judge_output: str, + expected_code: str, +) -> None: + transport = judge_run_transport() + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + return {"output": judge_output} + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + judge_event = next(event for event in events if event.get("kind") == "judge") + assert judge_event["status"] == "ERROR" + assert judge_event["error"]["code"] == expected_code + assert judge_event["errorMessage"] == judge_event["error"]["message"] + assert "score" not in judge_event + stub_sdk_client.flush.assert_awaited() + + +@pytest.mark.asyncio +async def test_generated_placeholders_are_not_expanded_into_judge_prompt( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + from launchdarkly_ai_server import parse_template + + transport = judge_run_transport() + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + rendered = parse_template(config["instructions"], variables) + # The placeholder smuggled in via the generated output must stay + # literal text after the handler's single render pass. + assert rendered == "Judge {{expected_output}} leaked? against Answer A" + return {"output": '{"score": 1, "reasoning": "ok"}'} + return {"output": "{{expected_output}} leaked?"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + judge_event = next(event for event in events if event.get("kind") == "judge") + assert judge_event["status"] == "COMPLETE" + assert judge_event["score"] == 1.0 + + +@pytest.mark.asyncio +async def test_missing_expected_output_renders_empty_judge_variables( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page([{"rowIndex": 7, "input": "Question"}], total=1), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + assert variables["expected_output"] == "" + assert variables["ground_truth_context"] == "" + return {"output": '{"score": 1, "reasoning": "ok"}'} + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + assert result.passed is True + + +@pytest.mark.asyncio +async def test_duplicate_criteria_rejected_before_any_request() -> None: + transport = SequencedTransport([]) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + with pytest.raises(EvaluationsError, match="Duplicate evaluation criteria"): + await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[ + Judge(key="accuracy"), + Scorer(name="accuracy", fn=lambda row, output: True), + ], + ) + + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_duplicate_criteria_rejected_case_insensitively() -> None: + """Matches the API's own dedup, which lowercases criterionType before + comparing: the worker's retry gate does the same, so criteria differing + only by case would still collide there even though they'd look distinct + to a case-sensitive check.""" + transport = SequencedTransport([]) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + with pytest.raises(EvaluationsError, match="Duplicate evaluation criteria"): + await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[ + Judge(key="Accuracy"), + Scorer(name="accuracy", fn=lambda row, output: True), + ], + ) + + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_errored_generation_row_emits_generation_incomplete_criterion_event( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = judge_run_transport( + summary={"statusCounts": {"total": 1, "passed": 0, "error": 1, "pending": 0}} + ) + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + raise AssertionError("judges must not run for errored generations") + raise RuntimeError("provider unavailable") + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is False + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + judge_event = next(event for event in events if event.get("kind") == "judge") + assert judge_event["status"] == "ERROR" + assert judge_event["error"]["code"] == "generation_incomplete" + + +@pytest.mark.asyncio +async def test_failed_evaluation_event_tracking_raises_after_attempting_every_result( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + """A dropped criterion event is a delivery failure, not a silent skip. + + The evaluation is created with a fixed criterion list, so the backend needs + one result per (row, criterion) before row accounting can finish. Swallowing + the failure leaves run() polling to its timeout and hides the cause, so the + run attempts every result, flushes what it queued, and then raises. + """ + transport = judge_run_transport() + accuracy_judge_variation(monkeypatch) + + attempted: list[str] = [] + + def track(event_name: str, *args: Any) -> None: + if event_name == "$ld:ai:offline-evals:criterion": + attempted.append(args[1]["criterionType"]) + raise RuntimeError("event pipeline unavailable") + + stub_sdk_client.track = MagicMock(side_effect=track) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + return {"output": '{"score": 1, "reasoning": "ok"}'} + return {"output": "generated"} + + with pytest.raises(EvaluationsError) as error: + await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[ + Judge(key="$ld:ai:judge:accuracy"), + Scorer(name="nonempty", fn=lambda row, output: bool(output)), + ], + ) + + # Every criterion is attempted before the failures are reported together, + # so one bad result never drops the ones behind it. + assert attempted == ["$ld:ai:judge:accuracy", "nonempty"] + assert "Failed to emit 2 of 2 evaluation criterion events" in str(error.value) + assert "event pipeline unavailable" in str(error.value) + stub_sdk_client.flush.assert_awaited() + + +@pytest.mark.parametrize("value", [float("nan"), -0.1, 1.1]) +@pytest.mark.parametrize("field", ["threshold", "pass_rate_threshold"]) +def test_criteria_reject_thresholds_outside_zero_to_one( + field: str, value: float +) -> None: + """NaN passes both range comparisons, so it needs its own rejection. + + Left in, it is serialized into the criteria wire payload as a bare ``NaN`` + literal and the management API rejects the whole evaluation. + """ + with pytest.raises(ValueError, match=f"{field} must be a number between 0 and 1"): + Judge(key="$ld:ai:judge:accuracy", **{field: value}) + with pytest.raises(ValueError, match=f"{field} must be a number between 0 and 1"): + Scorer(name="nonempty", fn=lambda row, output: True, **{field: value}) + + +def judge_variation( + monkeypatch: pytest.MonkeyPatch, + *, + provider: str, + mode: str | None = None, + config: dict[str, Any] | None = None, +) -> None: + """Serve one judge variation for the given provider and mode.""" + + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + meta: dict[str, Any] = {"variationKey": "default", "version": 12} + if mode is not None: + meta["mode"] = mode + return { + "config": { + "provider": {"name": provider}, + "model": {"name": "judge-model"}, + **(config or {"instructions": "Judge {{response_to_evaluate}}"}), + }, + "meta": meta, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + + +async def _generation_only( + config: dict[str, Any], + user_input: str | None = None, + tool_handlers: dict[str, Callable[..., Any]] | None = None, + variables: dict[str, Any] | None = None, + history: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + return {"output": "generated"} + + +@pytest.mark.asyncio +async def test_judge_on_another_provider_fails_before_any_records_are_created( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A provider handler cannot execute another provider's judge config. + + Passing it anyway spent the generation budget and then recorded every row + as handler_raised, so the mismatch is caught while it is still only a + configuration error: before the dataset is read or any record is created. + """ + transport = judge_run_transport() + judge_variation(monkeypatch, provider="Anthropic") + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + with pytest.raises(EvaluationsError) as error: + await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=create_handler(("OpenAI", "messages"), _generation_only), + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert "No handler can run LaunchDarkly judge" in str(error.value) + assert "'Anthropic'" in str(error.value) + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_judge_handlers_route_a_judge_to_its_own_provider( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = judge_run_transport() + judge_variation(monkeypatch, provider="Anthropic") + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + judged: list[dict[str, Any]] = [] + + async def anthropic_judge( + config: dict[str, Any], + user_input: str | None = None, + tool_handlers: dict[str, Callable[..., Any]] | None = None, + variables: dict[str, Any] | None = None, + history: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + judged.append(config) + return {"output": '{"score": 0.75, "reasoning": "ok"}'} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=create_handler(("OpenAI", "messages"), _generation_only), + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + judge_handlers=[create_handler(("Anthropic", "messages"), anthropic_judge)], + ) + + assert result.passed is True + assert [config["provider"]["name"] for config in judged] == ["Anthropic"] + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + judge_event = next(event for event in events if event.get("kind") == "judge") + assert judge_event["status"] == "COMPLETE" + assert judge_event["score"] == 0.75 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("wildcard_first", [True, False]) +async def test_exact_provider_judge_handler_beats_a_wildcard_adapter( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, + wildcard_first: bool, +) -> None: + """A wildcard is a fallback, so the order handlers are listed in cannot decide. + + Taking the first provider-or-wildcard match would send an Anthropic judge + through a multi-provider adapter that merely happened to be listed first. + """ + transport = judge_run_transport() + judge_variation(monkeypatch, provider="Anthropic") + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + chosen: list[str] = [] + + def judge_handler(name: str) -> Any: + async def run( + config: dict[str, Any], + user_input: str | None = None, + tool_handlers: dict[str, Callable[..., Any]] | None = None, + variables: dict[str, Any] | None = None, + history: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + chosen.append(name) + return {"output": '{"score": 1, "reasoning": "ok"}'} + + return run + + wildcard = create_handler(("*", "messages"), judge_handler("wildcard")) + exact = create_handler(("Anthropic", "messages"), judge_handler("exact")) + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=create_handler(("OpenAI", "messages"), _generation_only), + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + judge_handlers=[wildcard, exact] if wildcard_first else [exact, wildcard], + ) + + assert result.passed is True + assert chosen == ["exact"] + + +@pytest.mark.asyncio +async def test_wildcard_judge_handler_runs_a_judge_no_handler_names( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = judge_run_transport() + judge_variation(monkeypatch, provider="Anthropic") + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + judged: list[dict[str, Any]] = [] + + async def wildcard_judge( + config: dict[str, Any], + user_input: str | None = None, + tool_handlers: dict[str, Callable[..., Any]] | None = None, + variables: dict[str, Any] | None = None, + history: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + judged.append(config) + return {"output": '{"score": 1, "reasoning": "ok"}'} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=create_handler(("OpenAI", "messages"), _generation_only), + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + judge_handlers=[create_handler(("*", "messages"), wildcard_judge)], + ) + + assert result.passed is True + assert [config["provider"]["name"] for config in judged] == ["Anthropic"] + + +@pytest.mark.asyncio +async def test_agent_handler_runs_a_messages_mode_judge_with_collapsed_messages( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + """Mirrors the online path's agent-mode fallback for a messages-mode judge.""" + transport = judge_run_transport() + judge_variation( + monkeypatch, + provider="Anthropic", + mode="messages", + config={ + "messages": [ + {"role": "system", "content": "Grade strictly."}, + {"role": "user", "content": "Judge {{response_to_evaluate}}"}, + ] + }, + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + judged: list[dict[str, Any]] = [] + + async def anthropic_agent_judge( + config: dict[str, Any], + user_input: str | None = None, + tool_handlers: dict[str, Callable[..., Any]] | None = None, + variables: dict[str, Any] | None = None, + history: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + judged.append(config) + return {"output": '{"score": 1, "reasoning": "ok"}'} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=create_handler(("OpenAI", "messages"), _generation_only), + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + judge_handlers=[create_handler(("Anthropic", "agent"), anthropic_agent_judge)], + ) + + assert result.passed is True + assert judged[0]["instructions"] == ( + "Grade strictly.\n\nJudge {{response_to_evaluate}}" + ) + assert judged[0]["messages"] == [] + + +@pytest.mark.asyncio +async def test_generation_handler_runs_a_judge_on_the_same_provider( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = judge_run_transport() + judge_variation(monkeypatch, provider="OpenAI") + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + calls: list[str | None] = [] + + async def openai_handler( + config: dict[str, Any], + user_input: str | None = None, + tool_handlers: dict[str, Callable[..., Any]] | None = None, + variables: dict[str, Any] | None = None, + history: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + calls.append(config.get("instructions")) + if "Judge" in (config.get("instructions") or ""): + return {"output": '{"score": 0.9, "reasoning": "ok"}'} + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=create_handler(("OpenAI", "messages"), openai_handler), + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + assert any("Judge" in (instructions or "") for instructions in calls) + + +@pytest.mark.asyncio +async def test_judge_handlers_must_declare_the_provider_they_serve( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An unrouted judge handler would silently never be selected.""" + transport = judge_run_transport() + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + with pytest.raises(EvaluationsError, match="does not declare provides_for"): + await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=_generation_only, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + judge_handlers=[_generation_only], + ) + + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_criteria_run_concurrently_within_the_concurrency_bound( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + import asyncio + + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + {"rowIndex": index, "input": f"Question {index}"} + for index in range(3) + ], + total=3, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 3, "passed": 3, "error": 0, "pending": 0}}, + ), + ] + ) + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + in_flight = 0 + max_in_flight = 0 + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + nonlocal in_flight, max_in_flight + if "Judge" in config.get("instructions", ""): + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + await asyncio.sleep(0.01) + in_flight -= 1 + return {"output": '{"score": 1, "reasoning": "ok"}'} + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + concurrency=2, + ) + + assert result.passed is True + assert max_in_flight == 2 diff --git a/packages/client/tests/test_judge_scoring.py b/packages/client/tests/test_judge_scoring.py new file mode 100644 index 00000000..4613b679 --- /dev/null +++ b/packages/client/tests/test_judge_scoring.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from launchdarkly_ai_server.judge_scoring import parse_judge_response + + +class TestParseJudgeResponse: + def test_parses_plain_json(self) -> None: + assert parse_judge_response('{"score": 0.9, "reasoning": "solid"}') == ( + 0.9, + "solid", + ) + + def test_parses_fenced_json(self) -> None: + raw = '```json\n{"score": 1, "reasoning": "ok"}\n```' + assert parse_judge_response(raw) == (1, "ok") + + def test_accepts_already_decoded_mapping(self) -> None: + assert parse_judge_response({"score": 0.5, "reasoning": "meh"}) == ( + 0.5, + "meh", + ) + + def test_falls_back_to_reason_key(self) -> None: + assert parse_judge_response({"score": 0.5, "reason": "alt key"}) == ( + 0.5, + "alt key", + ) + + def test_null_reasoning_becomes_empty_string_not_none_literal(self) -> None: + assert parse_judge_response({"score": 0.5, "reasoning": None}) == (0.5, "") + + def test_score_returned_untouched_for_caller_policy(self) -> None: + score, _ = parse_judge_response({"score": "high", "reasoning": "?"}) + assert score == "high" + + @pytest.mark.parametrize( + "raw", + [ + "the answer looks correct", + "{}", + {}, + None, + 42, + ["not", "a", "mapping"], + '["not", "a", "mapping"]', + ], + ) + def test_rejects_non_object_responses(self, raw: Any) -> None: + with pytest.raises(ValueError, match="Invalid JSON from judge"): + parse_judge_response(raw) diff --git a/packages/client/tests/test_judges.py b/packages/client/tests/test_judges.py index 582db513..38b289c8 100644 --- a/packages/client/tests/test_judges.py +++ b/packages/client/tests/test_judges.py @@ -391,18 +391,90 @@ class TestScoreGuard: """`float(score)` used to sit ahead of the evaluation-metric track, so a junk score killed it.""" def test_rejects_non_numeric_scores_without_raising(self) -> None: - from launchdarkly_ai_server.judges import _numeric_score + from launchdarkly_ai_server.judge_scoring import numeric_score for junk in ("0.9 (high)", "85%", None, {"v": 1}, [], True, False): - assert _numeric_score(junk) is None + assert numeric_score(junk) is None def test_accepts_finite_numbers(self) -> None: from math import inf, nan - from launchdarkly_ai_server.judges import _numeric_score + from launchdarkly_ai_server.judge_scoring import numeric_score - assert _numeric_score(0.9) == 0.9 - assert _numeric_score(1) == 1.0 - assert _numeric_score(0) == 0.0 - assert _numeric_score(inf) is None - assert _numeric_score(nan) is None + assert numeric_score(0.9) == 0.9 + assert numeric_score(1) == 1.0 + assert numeric_score(0) == 0.0 + assert numeric_score(inf) is None + assert numeric_score(nan) is None + + +class TestRunJudgeScoreReporting: + """A judge that returns no score has not scored the output a zero.""" + + @pytest.mark.asyncio + async def test_null_score_skips_the_metric_instead_of_recording_zero( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + from contextlib import asynccontextmanager + + import launchdarkly_ai_server.judges as judges_module + import launchdarkly_ai_server.tracking as tracking_module + from launchdarkly_ai_server import JudgeTask, run_judge + + recorded: list[tuple[float, str | None]] = [] + + @asynccontextmanager + async def fake_with_judge_evaluation(name: str) -> Any: + def record(score: float, explanation: str | None = None) -> None: + recorded.append((score, explanation)) + + yield record + + monkeypatch.setattr( + judges_module, "with_judge_evaluation", fake_with_judge_evaluation + ) + + async def fake_execute_and_track(**kwargs: Any) -> dict[str, Any]: + return { + "response": '{"score": null, "reasoning": "cannot tell"}', + "usage": {"input_tokens": 1, "output_tokens": 1}, + "track_data": {"runId": "run-1"}, + } + + monkeypatch.setattr( + tracking_module, "execute_and_track", fake_execute_and_track + ) + + async def judge_fn( + config, user_input, tool_handlers, variables, history=None + ) -> dict: # type: ignore[override] + raise AssertionError("execute_and_track is stubbed") + + handler = ProviderHandler( + fn=judge_fn, provides_for=("TestProvider", "messages") + ) # type: ignore[arg-type] + + task = JudgeTask( + config_key="judge-key", + judge_config={ + "model": {"name": "gpt-4"}, + "provider": {"name": "TestProvider"}, + "instructions": "judge", + }, + judge_meta={"enabled": True, "variationKey": "j1", "version": 1}, + actual_output="response", + user_context=CONTEXT, + judge_provider="TestProvider", + judge_mode="messages", + collapse_messages=False, + parent_track_data={"runId": "run-1"}, + ) + + result = await run_judge(task, [handler]) + + assert result is not None + # A gen_ai.evaluation of 0 is indistinguishable from a judge that + # scored the output a hard fail, so no metric is recorded at all. + assert recorded == [] + assert result.score is None + assert result.response == "cannot tell"