From ccfdc0a2a3586143d24cb82f39a874f694af8c1b Mon Sep 17 00:00:00 2001 From: Dylan O'Neill Date: Wed, 2 Sep 2026 12:36:44 -0700 Subject: [PATCH 1/5] feat(evaluations): add LD judge event support --- packages/client/pyproject.toml | 2 +- .../src/launchdarkly_ai_server/__init__.py | 4 + .../evaluations/__init__.py | 3 + .../evaluations/events.py | 73 ++++ .../evaluations/judges.py | 114 ++++++ .../evaluations/module.py | 25 +- .../evaluations/runner.py | 342 +++++++++++++++++- .../evaluations/types.py | 10 + packages/client/tests/test_evaluations_run.py | 228 +++++++++++- uv.lock | 2 + 10 files changed, 788 insertions(+), 15 deletions(-) create mode 100644 packages/client/src/launchdarkly_ai_server/evaluations/events.py create mode 100644 packages/client/src/launchdarkly_ai_server/evaluations/judges.py diff --git a/packages/client/pyproject.toml b/packages/client/pyproject.toml index 9ea3ce74..ee8f995c 100644 --- a/packages/client/pyproject.toml +++ b/packages/client/pyproject.toml @@ -2,7 +2,7 @@ name = "launchdarkly-ai-server" version = "0.1.3" requires-python = ">=3.12" -dependencies = ["opentelemetry-api>=1.25"] +dependencies = ["opentelemetry-api>=1.25", "pydantic>=2"] description = "LaunchDarkly AI SDK core client for Python" readme = "README.md" license = "Apache-2.0" diff --git a/packages/client/src/launchdarkly_ai_server/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py index 9a07b056..94267723 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -26,7 +26,9 @@ EvaluationsError, EvaluationsModule, GenerationConfig, + Judge, RunSummary, + Scorer, init_evaluations, ) from .graph import GraphInstance, graph, resolve_graph @@ -168,7 +170,9 @@ "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..f6623983 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -9,6 +9,7 @@ Transport, urllib_transport, ) +from .judges import Judge, Scorer from .module import EvaluationsModule, init_evaluations from .types import EvalRunResult, GenerationConfig, RunSummary, Usage @@ -19,9 +20,11 @@ "EvaluationsModule", "GenerationConfig", "HttpResponse", + "Judge", "LDApiClient", "LDApiError", "RunSummary", + "Scorer", "Transport", "Usage", "init_evaluations", 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..ba2d3e04 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/events.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + + +class EvaluationStatus(StrEnum): + COMPLETE = "COMPLETE" + ERROR = "ERROR" + + +class EvaluationEventKind(StrEnum): + JUDGE = "judge" + SCORER = "scorer" + + +class TokenUsage(BaseModel): + """Token usage reported by an LD Judge provider call.""" + + model_config = ConfigDict(populate_by_name=True, extra="forbid") + + input_tokens: int = Field(alias="inputTokens") + output_tokens: int = Field(alias="outputTokens") + + +class EvaluationEventPayload(BaseModel): + """Common fields emitted for every SDK-run evaluation criterion result.""" + + model_config = ConfigDict( + populate_by_name=True, extra="forbid", use_enum_values=True + ) + + project_key: str = Field(alias="projectKey") + evaluation_id: str = Field(alias="evaluationId") + evaluation_run_id: str = Field(alias="evaluationRunId") + run_id: str = Field(alias="runId") + dataset_id: str = Field(alias="datasetId") + row_index: int = Field(alias="rowIndex") + criterion_type: str = Field(alias="criterionType") + kind: EvaluationEventKind + event_id: str = Field(alias="eventId") + emitted_at: str = Field(alias="emittedAt") + evaluation_key: str = Field(alias="evaluationKey") + evaluation_version: int | None = Field(default=None, alias="evaluationVersion") + dataset_key: str = Field(alias="datasetKey") + status: EvaluationStatus + started_at: str = Field(alias="startedAt") + evaluated_at: str = Field(alias="evaluatedAt") + latency_ms: int = Field(alias="latencyMs") + score: float | int | None = None + reason: str | None = None + error: dict[str, Any] | None = None + + def to_track_payload(self) -> dict[str, Any]: + return self.model_dump(by_alias=True, exclude_none=True) + + +class LDJudgeEvaluationEventPayload(EvaluationEventPayload): + """Payload for one LaunchDarkly AI Judge result on one dataset row.""" + + kind: EvaluationEventKind = EvaluationEventKind.JUDGE + judge_key: str = Field(alias="judgeKey") + variation_key: str = Field(alias="variationKey") + version: int | None = None + usage: TokenUsage | None = None + + +class DeterministicScorerEvaluationEventPayload(EvaluationEventPayload): + """Payload for one local deterministic scorer result on one dataset row.""" + + kind: EvaluationEventKind = EvaluationEventKind.SCORER diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/judges.py b/packages/client/src/launchdarkly_ai_server/evaluations/judges.py new file mode 100644 index 00000000..44230df3 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/judges.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass +from typing import Any + +type ScorerFn = Callable[ + [Mapping[str, Any], Any], float | bool | Awaitable[float | bool] +] + + +@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. + """ + + key: str + threshold: float | None = None + 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, "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)``. 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. + """ + + name: str + fn: ScorerFn + threshold: float | None = 1.0 + pass_rate_threshold: float | None = None + + 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, + "options": _criteria_options( + threshold=self.threshold, + pass_rate_threshold=self.pass_rate_threshold, + ), + } + + +type JudgeReference = 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 not None and (value < 0 or value > 1): + raise ValueError(f"{name} must be 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/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 4a3ffad3..7f8067da 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -17,6 +17,7 @@ Transport, urllib_transport, ) +from .judges import Judge, JudgeReference from .runner import EvalHandler, EvaluationsRunner, ToolImplementation, _segment from .types import EvalRunResult, GenerationConfig, RunSummary @@ -93,6 +94,7 @@ async def run( handler: EvalHandler, generation: GenerationConfig, tools: Mapping[str, ToolImplementation] | None = None, + judges: list[JudgeReference] | None = None, concurrency: int = 10, poll_interval_seconds: float | None = None, poll_timeout_seconds: float | None = None, @@ -121,14 +123,17 @@ async def run( poll_timeout_seconds=poll_timeout_seconds, ) run_tools = dict(tools or {}) + run_judges = list(judges or []) + ld_judges = [judge for judge in run_judges if isinstance(judge, 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(ld_judges) dataset_ref = await asyncio.to_thread( self._runner._fetch_dataset, project_key, dataset ) @@ -141,12 +146,12 @@ async def run( key, generation, resolved_tools, + run_judges, ) 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) @@ -165,6 +170,22 @@ async def run( dataset=dataset_ref, results=results, ) + if run_judges: + judge_results = await self._runner._run_judges_for_results( + results, + handler, + run_tools, + run_judges, + resolved_judges, + ) + self._runner._emit_evaluation_events( + client, + project_key=project_key, + evaluation=evaluation, + evaluation_run=evaluation_run, + dataset=dataset_ref, + results=judge_results, + ) flush_result = client.flush() if inspect.isawaitable(flush_result): await flush_result diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index 951213b5..132d8361 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -2,6 +2,7 @@ import asyncio import hashlib +import inspect import json import time import urllib.parse @@ -9,21 +10,44 @@ from datetime import UTC, datetime from typing import Any +from ..lifecycle import extract_variation from ..types import NativeTool -from ..utils import parse_template, parse_usage, to_ld_context +from ..utils import ( + parse_json_with_possible_fences, + parse_template, + parse_usage, + to_ld_context, +) from .api import EvaluationsError, LDApiClient, LDApiError +from .events import ( + DeterministicScorerEvaluationEventPayload, + EvaluationEventPayload, + LDJudgeEvaluationEventPayload, + TokenUsage, +) +from .judges import Judge, JudgeReference, Scorer from .types import ( DatasetRef, DatasetRow, EvaluationRef, EvaluationRunRef, GenerationConfig, + ResolvedJudge, ResolvedTool, RunSummary, ) DATASET_PAGE_SIZE = 200 GENERATION_EVENT_NAME = "$ld:ai:offline-evals:generation" +EVALUATION_EVENT_NAME = "$ld:ai:offline-evals:evaluation" +JUDGE_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.", + "Do not include ```json tags.", + ] +) EvalHandler = Callable[..., Awaitable[dict[str, Any]]] ToolImplementation = Callable[..., Any] | NativeTool @@ -124,6 +148,40 @@ def _resolve_tools( ) return resolved + async def _resolve_judges( + self, + judges: list[Judge], + ) -> dict[str, ResolvedJudge]: + """Resolve LD Judge configs before any evaluation records are created.""" + resolved: dict[str, ResolvedJudge] = {} + context: dict[str, Any] = {} + for judge in judges: + try: + variation = await extract_variation(judge.key, context) + except Exception as error: + raise EvaluationsError( + f"LaunchDarkly judge {judge.key!r} was not found or could not be resolved " + "for this project. Create the judge 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" + ) + resolved[judge.key] = 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, + ) + 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 +289,7 @@ def _create_evaluation( key: str, generation: GenerationConfig, tools: Mapping[str, ResolvedTool], + judges: list[JudgeReference] | None = None, ) -> EvaluationRef: body: dict[str, Any] = { "name": key, @@ -253,6 +312,8 @@ def _create_evaluation( body["tools"] = [ {"key": tool.key, "version": tool.version} for tool in tools.values() ] + if judges: + body["criteria"] = [judge.to_criteria_wire() for judge in judges] path = f"projects/{_segment(project_key)}/evaluations" raw = _mapping(self._api.post(path, body=body), description="evaluation") @@ -269,22 +330,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) @@ -480,6 +537,271 @@ def _emit_generation_events( flush=True, ) + def _judge_variables( + self, + row_result: Mapping[str, Any], + judge: Judge, + ) -> dict[str, Any]: + 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) + variables.update( + { + "input": row_result.get("input"), + "response_to_evaluate": output, + "message_history": "\n\n".join( + str(value) + for value in (row_result.get("input"), output) + if value is not None + ), + "expected_output": expected, + "ground_truth_context": ground_truth, + } + ) + return variables + + def _render_config_value(self, value: Any, variables: Mapping[str, Any]) -> Any: + if isinstance(value, str): + return parse_template(value, dict(variables)) + if isinstance(value, list): + return [self._render_config_value(item, variables) for item in value] + if isinstance(value, Mapping): + return { + str(key): self._render_config_value(item, variables) + for key, item in value.items() + } + return value + + 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, "scorer_error", "generation did not complete" + ) + try: + score_value = scorer.fn(row, row.get("output")) + if inspect.isawaitable(score_value): + score_value = await score_value + if isinstance(score_value, bool): + score: float = 1.0 if score_value else 0.0 + elif isinstance(score_value, int | float): + score = float(score_value) + else: + raise TypeError("scorer fn must return a bool or numeric score") + if score < 0 or score > 1: + raise ValueError("scorer fn score must be between 0 and 1") + 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), + } + except Exception as error: + return self._criterion_error_result( + base, started_clock, "scorer_error", str(error) + ) + + async def _run_ld_judge_for_result( + self, + row: Mapping[str, Any], + handler: EvalHandler, + tool_handlers: dict[str, ToolImplementation], + judge: Judge, + resolved: ResolvedJudge, + ) -> dict[str, Any]: + 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, "judge_error", "generation did not complete" + ) + try: + variables = self._judge_variables(row, judge) + rendered_config = self._render_config_value(resolved.config, variables) + result = await handler( + rendered_config, + row.get("output"), + tool_handlers, + { + **variables, + "formatting_instructions": JUDGE_FORMATTING_INSTRUCTIONS, + }, + ) + if not isinstance(result, Mapping): + raise TypeError("judge handler result must be a mapping") + raw = result.get("output", result.get("response")) + parsed = ( + parse_json_with_possible_fences(raw) + if isinstance(raw, str) + else raw + if isinstance(raw, Mapping) + else None + ) + if not isinstance(parsed, Mapping): + raise ValueError("Invalid JSON from judge") + completed = datetime.now(UTC) + event = { + **base, + "status": "COMPLETE", + "score": parsed.get("score"), + "reason": str(parsed.get("reasoning", parsed.get("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 + except Exception as error: + return self._criterion_error_result( + base, started_clock, "judge_error", str(error) + ) + + async def _run_judges_for_results( + self, + rows: list[dict[str, Any]], + handler: EvalHandler, + tool_handlers: dict[str, ToolImplementation], + judge_refs: list[JudgeReference], + resolved_judges: Mapping[str, ResolvedJudge], + ) -> list[dict[str, Any]]: + results: list[dict[str, Any]] = [] + for row in rows: + for judge in judge_refs: + if isinstance(judge, Scorer): + results.append(await self._run_scorer_for_result(row, judge)) + else: + results.append( + await self._run_ld_judge_for_result( + row, + handler, + tool_handlers, + judge, + resolved_judges[judge.key], + ) + ) + return results + + 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, + }, + ) + for result in results: + 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( + inputTokens=normalized_usage["input"], + outputTokens=normalized_usage["output"], + ) + common_payload = { + **identity, + "eventId": event_id, + "emittedAt": emitted_at, + "evaluationKey": evaluation.key, + "evaluationVersion": evaluation.version, + "datasetKey": dataset.key, + "status": result["status"], + "startedAt": result["started_at"], + "evaluatedAt": result["evaluated_at"], + "latencyMs": result["latency_ms"], + "score": result.get("score"), + "reason": result.get("reason"), + "error": result.get("error"), + } + payload_model: EvaluationEventPayload + if result["kind"] == "judge": + payload_model = LDJudgeEvaluationEventPayload( + **common_payload, + judgeKey=result["judge_key"], + variationKey=result["variation_key"], + version=result.get("version"), + usage=usage, + ) + else: + payload_model = DeterministicScorerEvaluationEventPayload( + **common_payload + ) + client.track( + EVALUATION_EVENT_NAME, context, payload_model.to_track_payload(), 1 + ) + print( + f"{EVALUATION_EVENT_NAME} emittedAt={emitted_at} eventId={event_id}", + flush=True, + ) + def _get_summary( self, project_key: str, evaluation_id: str, run_id: str ) -> RunSummary: diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/types.py b/packages/client/src/launchdarkly_ai_server/evaluations/types.py index 5f4f4d5c..b5d0e647 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/types.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/types.py @@ -67,6 +67,16 @@ class ResolvedTool: schema: dict[str, Any] = field(default_factory=dict) +@dataclass +class ResolvedJudge: + """A LaunchDarkly AI Judge config variation resolved for an evaluation run.""" + + key: str + config: dict[str, Any] + variation_key: str = "" + version: int | None = None + + @dataclass class EvaluationRef: """Identifiers returned after creating an evaluation.""" diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index 5784559e..62a3b614 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from collections.abc import Callable +from collections.abc import Callable, Mapping from datetime import datetime from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -11,6 +11,8 @@ from launchdarkly_ai_server.evaluations import ( EvaluationsError, HttpResponse, + Judge, + Scorer, init_evaluations, ) @@ -278,7 +280,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", } @@ -1033,3 +1034,226 @@ 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" + assert context == {} + 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" + assert "generated" in config["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"}, + judges=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + assert transport.requests[2]["body"]["criteria"] == [ + { + "criterionType": "$ld:ai:judge:accuracy", + "options": {}, + } + ] + 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 + + +@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="Create the judge in the LaunchDarkly UI" + ): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + judges=[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: Mapping[str, Any], 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"}, + judges=[Scorer(name="refund-exists", fn=check_refund)], + ) + + assert result.passed is True + assert transport.requests[2]["body"]["criteria"] == [ + {"criterionType": "refund-exists", "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 diff --git a/uv.lock b/uv.lock index 7d93a3cd..fa21dbca 100644 --- a/uv.lock +++ b/uv.lock @@ -922,6 +922,7 @@ version = "0.1.3" source = { editable = "packages/client" } dependencies = [ { name = "opentelemetry-api" }, + { name = "pydantic" }, ] [package.optional-dependencies] @@ -935,6 +936,7 @@ requires-dist = [ { name = "opentelemetry-api", specifier = ">=1.25" }, { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'otel'", specifier = ">=1.25" }, { name = "opentelemetry-sdk", marker = "extra == 'otel'", specifier = ">=1.25" }, + { name = "pydantic", specifier = ">=2" }, ] provides-extras = ["otel"] From f63cb3cafd2c77775e1d455f040b9f31f2f80073 Mon Sep 17 00:00:00 2001 From: Dylan O'Neill Date: Wed, 2 Sep 2026 16:45:45 -0700 Subject: [PATCH 2/5] refactor(evaluations): criteria naming + shared judge scoring contract Rename the public run() parameter judges= to criteria= (with JudgeReference -> Criterion and evaluations/judges.py -> criteria.py): the wire format already calls these criteria, and a Scorer is not a judge. Scorer callbacks now receive the public DatasetRow instead of the internal result dict, so internal key renames cannot break customer scorers; Criterion and DatasetRow are exported. Extract the judge response contract (formatting instructions, JSON score parsing, finite-number guard) into judge_scoring.py shared by the online judge path and the offline evaluations runner. The two copies had already drifted textually. Co-Authored-By: Claude Fable 5 --- .../src/launchdarkly_ai_server/__init__.py | 4 ++ .../evaluations/__init__.py | 6 +- .../evaluations/{judges.py => criteria.py} | 22 +++--- .../evaluations/module.py | 20 +++--- .../evaluations/runner.py | 61 ++++++++--------- .../launchdarkly_ai_server/judge_scoring.py | 61 +++++++++++++++++ .../src/launchdarkly_ai_server/judges.py | 67 ++++++------------- packages/client/tests/test_evaluations_run.py | 15 +++-- packages/client/tests/test_judges.py | 16 ++--- 9 files changed, 157 insertions(+), 115 deletions(-) rename packages/client/src/launchdarkly_ai_server/evaluations/{judges.py => criteria.py} (80%) create mode 100644 packages/client/src/launchdarkly_ai_server/judge_scoring.py diff --git a/packages/client/src/launchdarkly_ai_server/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py index 94267723..9b2fce80 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -22,6 +22,8 @@ set_conversation_id_if_absent, ) from .evaluations import ( + Criterion, + DatasetRow, EvalRunResult, EvaluationsError, EvaluationsModule, @@ -167,6 +169,8 @@ "VariationMeta", # evaluations "EvalRunResult", + "Criterion", + "DatasetRow", "EvaluationsError", "EvaluationsModule", "GenerationConfig", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py index f6623983..b110a550 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -9,12 +9,14 @@ Transport, urllib_transport, ) -from .judges import Judge, Scorer +from .criteria import Criterion, Judge, Scorer 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", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/judges.py b/packages/client/src/launchdarkly_ai_server/evaluations/criteria.py similarity index 80% rename from packages/client/src/launchdarkly_ai_server/evaluations/judges.py rename to packages/client/src/launchdarkly_ai_server/evaluations/criteria.py index 44230df3..f77d6415 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/judges.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/criteria.py @@ -1,12 +1,12 @@ from __future__ import annotations -from collections.abc import Awaitable, Callable, Mapping +from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Any -type ScorerFn = Callable[ - [Mapping[str, Any], Any], float | bool | Awaitable[float | bool] -] +from .types import DatasetRow + +type ScorerFn = Callable[[DatasetRow, Any], float | bool | Awaitable[float | bool]] @dataclass(frozen=True) @@ -48,9 +48,15 @@ def to_criteria_wire(self) -> dict[str, Any]: class Scorer: """Local deterministic scorer run for each generated evaluation row. - ``fn`` may be sync or async and receives ``(row, 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. + ``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. """ name: str @@ -82,7 +88,7 @@ def to_criteria_wire(self) -> dict[str, Any]: } -type JudgeReference = Judge | Scorer +type Criterion = Judge | Scorer def _validate_thresholds( diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 7f8067da..9bb3399c 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -17,7 +17,7 @@ Transport, urllib_transport, ) -from .judges import Judge, JudgeReference +from .criteria import Criterion, Judge from .runner import EvalHandler, EvaluationsRunner, ToolImplementation, _segment from .types import EvalRunResult, GenerationConfig, RunSummary @@ -94,7 +94,7 @@ async def run( handler: EvalHandler, generation: GenerationConfig, tools: Mapping[str, ToolImplementation] | None = None, - judges: list[JudgeReference] | None = None, + criteria: list[Criterion] | None = None, concurrency: int = 10, poll_interval_seconds: float | None = None, poll_timeout_seconds: float | None = None, @@ -123,8 +123,10 @@ async def run( poll_timeout_seconds=poll_timeout_seconds, ) run_tools = dict(tools or {}) - run_judges = list(judges or []) - ld_judges = [judge for judge in run_judges if isinstance(judge, Judge)] + run_criteria = list(criteria or []) + 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 @@ -146,7 +148,7 @@ async def run( key, generation, resolved_tools, - run_judges, + run_criteria, ) evaluation_run = await asyncio.to_thread( self._runner._create_evaluation_run, @@ -170,12 +172,12 @@ async def run( dataset=dataset_ref, results=results, ) - if run_judges: - judge_results = await self._runner._run_judges_for_results( + if run_criteria: + criterion_results = await self._runner._run_criteria_for_results( results, handler, run_tools, - run_judges, + run_criteria, resolved_judges, ) self._runner._emit_evaluation_events( @@ -184,7 +186,7 @@ async def run( evaluation=evaluation, evaluation_run=evaluation_run, dataset=dataset_ref, - results=judge_results, + results=criterion_results, ) flush_result = client.flush() if inspect.isawaitable(flush_result): diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index 132d8361..99c4e673 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -10,22 +10,25 @@ from datetime import UTC, datetime from typing import Any +from ..judge_scoring import ( + FORMATTING_INSTRUCTIONS, + parse_judge_response, +) from ..lifecycle import extract_variation from ..types import NativeTool from ..utils import ( - parse_json_with_possible_fences, parse_template, parse_usage, to_ld_context, ) from .api import EvaluationsError, LDApiClient, LDApiError +from .criteria import Criterion, Judge, Scorer from .events import ( DeterministicScorerEvaluationEventPayload, EvaluationEventPayload, LDJudgeEvaluationEventPayload, TokenUsage, ) -from .judges import Judge, JudgeReference, Scorer from .types import ( DatasetRef, DatasetRow, @@ -40,14 +43,6 @@ DATASET_PAGE_SIZE = 200 GENERATION_EVENT_NAME = "$ld:ai:offline-evals:generation" EVALUATION_EVENT_NAME = "$ld:ai:offline-evals:evaluation" -JUDGE_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.", - "Do not include ```json tags.", - ] -) EvalHandler = Callable[..., Awaitable[dict[str, Any]]] ToolImplementation = Callable[..., Any] | NativeTool @@ -289,7 +284,7 @@ def _create_evaluation( key: str, generation: GenerationConfig, tools: Mapping[str, ResolvedTool], - judges: list[JudgeReference] | None = None, + criteria: list[Criterion] | None = None, ) -> EvaluationRef: body: dict[str, Any] = { "name": key, @@ -312,8 +307,8 @@ def _create_evaluation( body["tools"] = [ {"key": tool.key, "version": tool.version} for tool in tools.values() ] - if judges: - body["criteria"] = [judge.to_criteria_wire() for judge in judges] + 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") @@ -611,7 +606,14 @@ async def _run_scorer_for_result( base, started_clock, "scorer_error", "generation did not complete" ) try: - score_value = scorer.fn(row, row.get("output")) + 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"), + ) + score_value = scorer.fn(dataset_row, row.get("output")) if inspect.isawaitable(score_value): score_value = await score_value if isinstance(score_value, bool): @@ -668,27 +670,20 @@ async def _run_ld_judge_for_result( tool_handlers, { **variables, - "formatting_instructions": JUDGE_FORMATTING_INSTRUCTIONS, + "formatting_instructions": FORMATTING_INSTRUCTIONS, }, ) if not isinstance(result, Mapping): raise TypeError("judge handler result must be a mapping") - raw = result.get("output", result.get("response")) - parsed = ( - parse_json_with_possible_fences(raw) - if isinstance(raw, str) - else raw - if isinstance(raw, Mapping) - else None + score, reason = parse_judge_response( + result.get("output", result.get("response")) ) - if not isinstance(parsed, Mapping): - raise ValueError("Invalid JSON from judge") completed = datetime.now(UTC) event = { **base, "status": "COMPLETE", - "score": parsed.get("score"), - "reason": str(parsed.get("reasoning", parsed.get("reason", ""))), + "score": score, + "reason": reason, "evaluated_at": completed.isoformat().replace("+00:00", "Z"), "latency_ms": round((time.perf_counter() - started_clock) * 1000), } @@ -701,27 +696,27 @@ async def _run_ld_judge_for_result( base, started_clock, "judge_error", str(error) ) - async def _run_judges_for_results( + async def _run_criteria_for_results( self, rows: list[dict[str, Any]], handler: EvalHandler, tool_handlers: dict[str, ToolImplementation], - judge_refs: list[JudgeReference], + criteria: list[Criterion], resolved_judges: Mapping[str, ResolvedJudge], ) -> list[dict[str, Any]]: results: list[dict[str, Any]] = [] for row in rows: - for judge in judge_refs: - if isinstance(judge, Scorer): - results.append(await self._run_scorer_for_result(row, judge)) + for criterion in criteria: + if isinstance(criterion, Scorer): + results.append(await self._run_scorer_for_result(row, criterion)) else: results.append( await self._run_ld_judge_for_result( row, handler, tool_handlers, - judge, - resolved_judges[judge.key], + criterion, + resolved_judges[criterion.key], ) ) return results 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..934c2716 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,17 @@ 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: + if score is None: + score = 0.0 + 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 62a3b614..dbbabab0 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from collections.abc import Callable, Mapping +from collections.abc import Callable from datetime import datetime from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -9,6 +9,7 @@ import pytest from launchdarkly_ai_server.evaluations import ( + DatasetRow, EvaluationsError, HttpResponse, Judge, @@ -1116,7 +1117,7 @@ async def handler( dataset="golden", handler=handler, generation={"provider": "OpenAI", "model": "gpt-4o"}, - judges=[Judge(key="$ld:ai:judge:accuracy")], + criteria=[Judge(key="$ld:ai:judge:accuracy")], ) assert result.passed is True @@ -1169,7 +1170,7 @@ async def handler(*args: object) -> dict[str, Any]: dataset="golden", handler=handler, generation={"provider": "OpenAI", "model": "gpt-4o"}, - judges=[Judge(key="security-judge")], + criteria=[Judge(key="security-judge")], ) assert transport.requests == [] @@ -1215,9 +1216,9 @@ async def handler(*args: object) -> dict[str, Any]: "usage": {"input_tokens": 10, "output_tokens": 4}, } - def check_refund(row: Mapping[str, Any], output: Any) -> bool: - assert row["row_index"] == 42 - assert row["input"] == "Ticket A" + 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) @@ -1227,7 +1228,7 @@ def check_refund(row: Mapping[str, Any], output: Any) -> bool: dataset="support-golden-v3", handler=handler, generation={"provider": "OpenAI", "model": "gpt-4o"}, - judges=[Scorer(name="refund-exists", fn=check_refund)], + criteria=[Scorer(name="refund-exists", fn=check_refund)], ) assert result.passed is True diff --git a/packages/client/tests/test_judges.py b/packages/client/tests/test_judges.py index 582db513..01998bf5 100644 --- a/packages/client/tests/test_judges.py +++ b/packages/client/tests/test_judges.py @@ -391,18 +391,18 @@ 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 From 1eb197615739c45dd9a7c261a4573ae92a9f6479 Mon Sep 17 00:00:00 2001 From: Dylan O'Neill Date: Wed, 2 Sep 2026 16:50:22 -0700 Subject: [PATCH 3/5] fix(evaluations): correct judge resolution, score validation, and rendering - Resolve LD judges with a valid {kind: evaluation, key: project} context. The previous empty context is invalid to the real LD SDK, so every resolution returned the None default and failed as 'not found'. The resolution error now also carries the underlying cause instead of always claiming the judge does not exist. - Validate judge scores when the judge responds: non-JSON output, non-numeric, non-finite, and out-of-range scores become per-criterion ERROR events with cause codes (invalid_judge_output, invalid_score, handler_raised, generation_incomplete, scorer_raised) instead of crashing the run at event-build time after all LLM spend. - Pass judge configs to the handler unrendered. The handler owns the single template pass, so {{...}} sequences inside generated output or dataset values can no longer be expanded into the judge prompt. Absent judge variables render as empty strings rather than leaving literal mustache in the prompt. - Reject duplicate criterion identities (judge keys / scorer names) before any records are created; they would share event identity. - Flush queued generation events in a finally so they reach LaunchDarkly even when the criteria phase fails. Co-Authored-By: Claude Fable 5 --- .../evaluations/module.py | 70 +++-- .../evaluations/runner.py | 176 +++++++----- packages/client/tests/test_evaluations_run.py | 269 +++++++++++++++++- 3 files changed, 423 insertions(+), 92 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 9bb3399c..4b4dbe78 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -124,6 +124,7 @@ async def run( ) run_tools = dict(tools or {}) run_criteria = list(criteria or []) + self._validate_criteria(run_criteria) ld_judges = [ criterion for criterion in run_criteria if isinstance(criterion, Judge) ] @@ -135,7 +136,7 @@ async def run( resolved_tools = await asyncio.to_thread( self._runner._resolve_tools, project_key, run_tools ) - resolved_judges = await self._runner._resolve_judges(ld_judges) + resolved_judges = await self._runner._resolve_judges(project_key, ld_judges) dataset_ref = await asyncio.to_thread( self._runner._fetch_dataset, project_key, dataset ) @@ -164,33 +165,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, - ) - if run_criteria: - criterion_results = await self._runner._run_criteria_for_results( - results, - handler, - run_tools, - run_criteria, - resolved_judges, - ) - self._runner._emit_evaluation_events( + try: + self._runner._emit_generation_events( client, project_key=project_key, evaluation=evaluation, evaluation_run=evaluation_run, dataset=dataset_ref, - results=criterion_results, + results=results, ) - flush_result = client.flush() - if inspect.isawaitable(flush_result): - await flush_result + if run_criteria: + criterion_results = await self._runner._run_criteria_for_results( + results, + handler, + run_tools, + run_criteria, + resolved_judges, + ) + 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, @@ -265,6 +270,27 @@ 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. + """ + seen: set[str] = set() + duplicates: list[str] = [] + for criterion in criteria: + criterion_type = criterion.criterion_type + if criterion_type in seen and criterion_type not in duplicates: + duplicates.append(criterion_type) + seen.add(criterion_type) + 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." + ) + @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 99c4e673..3fbd49cb 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -4,6 +4,7 @@ import hashlib import inspect import json +import logging import time import urllib.parse from collections.abc import Awaitable, Callable, Mapping @@ -12,6 +13,7 @@ from ..judge_scoring import ( FORMATTING_INSTRUCTIONS, + numeric_score, parse_judge_response, ) from ..lifecycle import extract_variation @@ -40,6 +42,8 @@ RunSummary, ) +logger = logging.getLogger(__name__) + DATASET_PAGE_SIZE = 200 GENERATION_EVENT_NAME = "$ld:ai:offline-evals:generation" EVALUATION_EVENT_NAME = "$ld:ai:offline-evals:evaluation" @@ -145,18 +149,22 @@ def _resolve_tools( async def _resolve_judges( self, + project_key: str, judges: list[Judge], ) -> dict[str, ResolvedJudge]: """Resolve LD Judge configs before any evaluation records are created.""" resolved: dict[str, ResolvedJudge] = {} - context: dict[str, Any] = {} + # 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"LaunchDarkly judge {judge.key!r} was not found or could not be resolved " - "for this project. Create the judge in the LaunchDarkly UI and try again." + 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") @@ -537,6 +545,12 @@ def _judge_variables( 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") @@ -547,31 +561,21 @@ def _judge_variables( ground_truth = str(expected) variables.update( { - "input": row_result.get("input"), - "response_to_evaluate": output, + "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) if value is not None ), - "expected_output": expected, - "ground_truth_context": ground_truth, + "expected_output": expected if expected is not None else "", + "ground_truth_context": ( + ground_truth if ground_truth is not None else "" + ), } ) return variables - def _render_config_value(self, value: Any, variables: Mapping[str, Any]) -> Any: - if isinstance(value, str): - return parse_template(value, dict(variables)) - if isinstance(value, list): - return [self._render_config_value(item, variables) for item in value] - if isinstance(value, Mapping): - return { - str(key): self._render_config_value(item, variables) - for key, item in value.items() - } - return value - def _criterion_error_result( self, base: Mapping[str, Any], @@ -603,40 +607,55 @@ async def _run_scorer_for_result( } if row.get("status") != "COMPLETE": return self._criterion_error_result( - base, started_clock, "scorer_error", "generation did not complete" + 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: - 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"), - ) score_value = scorer.fn(dataset_row, row.get("output")) if inspect.isawaitable(score_value): score_value = await score_value - if isinstance(score_value, bool): - score: float = 1.0 if score_value else 0.0 - elif isinstance(score_value, int | float): - score = float(score_value) - else: - raise TypeError("scorer fn must return a bool or numeric score") - if score < 0 or score > 1: - raise ValueError("scorer fn score must be between 0 and 1") - 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), - } except Exception as error: return self._criterion_error_result( - base, started_clock, "scorer_error", str(error) + 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, @@ -659,13 +678,18 @@ async def _run_ld_judge_for_result( } if row.get("status") != "COMPLETE": return self._criterion_error_result( - base, started_clock, "judge_error", "generation did not complete" + 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) try: - variables = self._judge_variables(row, judge) - rendered_config = self._render_config_value(resolved.config, variables) result = await handler( - rendered_config, + dict(resolved.config), row.get("output"), tool_handlers, { @@ -673,28 +697,46 @@ async def _run_ld_judge_for_result( "formatting_instructions": FORMATTING_INSTRUCTIONS, }, ) - if not isinstance(result, Mapping): - raise TypeError("judge handler result must be a mapping") - score, reason = parse_judge_response( + 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")) ) - completed = datetime.now(UTC) - 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 - except Exception as error: + except ValueError as error: return self._criterion_error_result( - base, started_clock, "judge_error", str(error) + 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) + 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, diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index dbbabab0..4dcbf743 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -1075,7 +1075,9 @@ async def fake_extract_variation( key: str, context: dict[str, Any] ) -> dict[str, Any]: assert key == "$ld:ai:judge:accuracy" - assert context == {} + # 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"}, @@ -1101,7 +1103,14 @@ async def handler( assert user_input == "generated" assert variables["response_to_evaluate"] == "generated" assert variables["expected_output"] == "Answer A" - assert "generated" in config["instructions"] + # 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" + ) return { "output": '{"score": 0.86, "reasoning": "matches policy"}', "usage": {"input_tokens": 640, "output_tokens": 48}, @@ -1162,7 +1171,8 @@ async def handler(*args: object) -> dict[str, Any]: return {"output": "generated"} with pytest.raises( - EvaluationsError, match="Create the judge in the LaunchDarkly UI" + EvaluationsError, + match=r"Failed to resolve LaunchDarkly judge 'security-judge': not found", ): await evals.run( project_key="proj", @@ -1258,3 +1268,256 @@ def check_refund(row: DatasetRow, output: Any) -> bool: 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 "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_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" From 46bfa001b24f1266105b2bbf6b2c1ff27afe98be Mon Sep 17 00:00:00 2001 From: Dylan O'Neill Date: Wed, 2 Sep 2026 16:52:44 -0700 Subject: [PATCH 4/5] refactor(evaluations): drop pydantic, run criteria concurrently Rewrite the evaluation event payloads as frozen dataclasses with an explicit to_track_payload(), matching the Usage/RunSummary wire pattern used everywhere else in the SDK, and remove the pydantic>=2 dependency. The models validated the SDK's own dicts, and a validation failure surfaced as a run-aborting crash at emission time; payload construction is now also wrapped per result so one bad criterion result is logged and skipped instead of dropping the whole batch. ERROR events carry a top-level errorMessage for parity with generation events. Run (row x criterion) pairs through the same ConcurrencyController and concurrency parameter the generation phase uses, instead of one criterion at a time: a 200-row dataset with 3 judges was 600 serial LLM calls. Co-Authored-By: Claude Fable 5 --- packages/client/pyproject.toml | 2 +- .../evaluations/events.py | 101 +++++++---- .../evaluations/module.py | 1 + .../evaluations/runner.py | 165 +++++++++++------- packages/client/tests/test_evaluations_run.py | 105 +++++++++++ uv.lock | 2 - 6 files changed, 277 insertions(+), 99 deletions(-) diff --git a/packages/client/pyproject.toml b/packages/client/pyproject.toml index ee8f995c..9ea3ce74 100644 --- a/packages/client/pyproject.toml +++ b/packages/client/pyproject.toml @@ -2,7 +2,7 @@ name = "launchdarkly-ai-server" version = "0.1.3" requires-python = ">=3.12" -dependencies = ["opentelemetry-api>=1.25", "pydantic>=2"] +dependencies = ["opentelemetry-api>=1.25"] description = "LaunchDarkly AI SDK core client for Python" readme = "README.md" license = "Apache-2.0" diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/events.py b/packages/client/src/launchdarkly_ai_server/evaluations/events.py index ba2d3e04..97de0938 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/events.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/events.py @@ -1,10 +1,9 @@ from __future__ import annotations +from dataclasses import dataclass from enum import StrEnum from typing import Any -from pydantic import BaseModel, ConfigDict, Field - class EvaluationStatus(StrEnum): COMPLETE = "COMPLETE" @@ -16,57 +15,95 @@ class EvaluationEventKind(StrEnum): SCORER = "scorer" -class TokenUsage(BaseModel): +@dataclass(frozen=True) +class TokenUsage: """Token usage reported by an LD Judge provider call.""" - model_config = ConfigDict(populate_by_name=True, extra="forbid") + input_tokens: int + output_tokens: int - input_tokens: int = Field(alias="inputTokens") - output_tokens: int = Field(alias="outputTokens") + def to_wire(self) -> dict[str, int]: + return { + "inputTokens": self.input_tokens, + "outputTokens": self.output_tokens, + } -class EvaluationEventPayload(BaseModel): +@dataclass(frozen=True, kw_only=True) +class EvaluationEventPayload: """Common fields emitted for every SDK-run evaluation criterion result.""" - model_config = ConfigDict( - populate_by_name=True, extra="forbid", use_enum_values=True - ) - - project_key: str = Field(alias="projectKey") - evaluation_id: str = Field(alias="evaluationId") - evaluation_run_id: str = Field(alias="evaluationRunId") - run_id: str = Field(alias="runId") - dataset_id: str = Field(alias="datasetId") - row_index: int = Field(alias="rowIndex") - criterion_type: str = Field(alias="criterionType") + project_key: str + evaluation_id: str + evaluation_run_id: str + run_id: str + dataset_id: str + row_index: int + criterion_type: str kind: EvaluationEventKind - event_id: str = Field(alias="eventId") - emitted_at: str = Field(alias="emittedAt") - evaluation_key: str = Field(alias="evaluationKey") - evaluation_version: int | None = Field(default=None, alias="evaluationVersion") - dataset_key: str = Field(alias="datasetKey") + event_id: str + emitted_at: str + evaluation_key: str + dataset_key: str status: EvaluationStatus - started_at: str = Field(alias="startedAt") - evaluated_at: str = Field(alias="evaluatedAt") - latency_ms: int = Field(alias="latencyMs") - score: float | int | None = None + 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]: - return self.model_dump(by_alias=True, exclude_none=True) - - + 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 LDJudgeEvaluationEventPayload(EvaluationEventPayload): """Payload for one LaunchDarkly AI Judge result on one dataset row.""" kind: EvaluationEventKind = EvaluationEventKind.JUDGE - judge_key: str = Field(alias="judgeKey") - variation_key: str = Field(alias="variationKey") + 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 DeterministicScorerEvaluationEventPayload(EvaluationEventPayload): """Payload for one local deterministic scorer result on one dataset row.""" diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 4b4dbe78..3ee59d09 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -181,6 +181,7 @@ async def run( run_tools, run_criteria, resolved_judges, + concurrency, ) self._runner._emit_evaluation_events( client, diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index 3fbd49cb..0aeaee18 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -28,6 +28,7 @@ from .events import ( DeterministicScorerEvaluationEventPayload, EvaluationEventPayload, + EvaluationStatus, LDJudgeEvaluationEventPayload, TokenUsage, ) @@ -745,23 +746,33 @@ async def _run_criteria_for_results( tool_handlers: dict[str, ToolImplementation], criteria: list[Criterion], resolved_judges: Mapping[str, ResolvedJudge], + concurrency: int, ) -> list[dict[str, Any]]: - results: list[dict[str, Any]] = [] - for row in rows: - for criterion in criteria: + """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): - results.append(await self._run_scorer_for_result(row, criterion)) - else: - results.append( - await self._run_ld_judge_for_result( - row, - handler, - tool_handlers, - criterion, - resolved_judges[criterion.key], - ) - ) - return results + return await self._run_scorer_for_result(row, criterion) + return await self._run_ld_judge_for_result( + row, + handler, + 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, @@ -783,57 +794,83 @@ def _emit_evaluation_events( }, ) for result in results: - 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( - inputTokens=normalized_usage["input"], - outputTokens=normalized_usage["output"], - ) - common_payload = { - **identity, - "eventId": event_id, - "emittedAt": emitted_at, - "evaluationKey": evaluation.key, - "evaluationVersion": evaluation.version, - "datasetKey": dataset.key, - "status": result["status"], - "startedAt": result["started_at"], - "evaluatedAt": result["evaluated_at"], - "latencyMs": result["latency_ms"], - "score": result.get("score"), - "reason": result.get("reason"), - "error": result.get("error"), - } - payload_model: EvaluationEventPayload - if result["kind"] == "judge": - payload_model = LDJudgeEvaluationEventPayload( - **common_payload, - judgeKey=result["judge_key"], - variationKey=result["variation_key"], - version=result.get("version"), - usage=usage, + # One bad criterion result must not abort the run or drop the + # events queued for the results that preceded it. + 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": EvaluationStatus(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: EvaluationEventPayload + if result["kind"] == "judge": + payload_model = LDJudgeEvaluationEventPayload( + **common_payload, + judge_key=result["judge_key"], + variation_key=result["variation_key"], + version=result.get("version"), + usage=usage, + ) + else: + payload_model = DeterministicScorerEvaluationEventPayload( + **common_payload + ) + client.track( + EVALUATION_EVENT_NAME, context, payload_model.to_track_payload(), 1 ) - else: - payload_model = DeterministicScorerEvaluationEventPayload( - **common_payload + except Exception: + logger.exception( + "Skipping evaluation event for row %s criterion %s", + result.get("row_index"), + result.get("criterion_type"), ) - client.track( - EVALUATION_EVENT_NAME, context, payload_model.to_track_payload(), 1 - ) + continue print( f"{EVALUATION_EVENT_NAME} emittedAt={emitted_at} eventId={event_id}", flush=True, diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index 4dcbf743..ac73ed59 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -1368,6 +1368,7 @@ async def handler( 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() @@ -1521,3 +1522,107 @@ async def handler( 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_skips_event_but_completes_run( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = judge_run_transport() + accuracy_judge_variation(monkeypatch) + + def track(event_name: str, *args: Any) -> None: + if event_name == "$ld:ai:offline-evals:evaluation": + 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"} + + 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 + stub_sdk_client.flush.assert_awaited() + + +@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/uv.lock b/uv.lock index fa21dbca..7d93a3cd 100644 --- a/uv.lock +++ b/uv.lock @@ -922,7 +922,6 @@ version = "0.1.3" source = { editable = "packages/client" } dependencies = [ { name = "opentelemetry-api" }, - { name = "pydantic" }, ] [package.optional-dependencies] @@ -936,7 +935,6 @@ requires-dist = [ { name = "opentelemetry-api", specifier = ">=1.25" }, { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'otel'", specifier = ">=1.25" }, { name = "opentelemetry-sdk", marker = "extra == 'otel'", specifier = ">=1.25" }, - { name = "pydantic", specifier = ">=2" }, ] provides-extras = ["otel"] From da69d336dca0e6ae936bb262aeec477feee3b14a Mon Sep 17 00:00:00 2001 From: Dylan O'Neill Date: Wed, 2 Sep 2026 16:53:54 -0700 Subject: [PATCH 5/5] polish(evaluations): event logging, docstrings, judge scoring tests Emit per-event telemetry lines through the module logger instead of printing to the host application's stdout, refresh the run() docstring (no longer generation-only), and cover the shared judge response parser with unit tests. Co-Authored-By: Claude Fable 5 --- .../evaluations/module.py | 8 ++- .../evaluations/runner.py | 16 +++--- packages/client/tests/test_evaluations_run.py | 13 +++-- packages/client/tests/test_judge_scoring.py | 54 +++++++++++++++++++ 4 files changed, 80 insertions(+), 11 deletions(-) create mode 100644 packages/client/tests/test_judge_scoring.py diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 3ee59d09..57a77523 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -100,7 +100,13 @@ async def run( 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. 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 diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index 0aeaee18..5a2e4ee1 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -536,9 +536,11 @@ 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( @@ -871,9 +873,11 @@ def _emit_evaluation_events( result.get("criterion_type"), ) continue - print( - f"{EVALUATION_EVENT_NAME} emittedAt={emitted_at} eventId={event_id}", - flush=True, + logger.info( + "%s emittedAt=%s eventId=%s", + EVALUATION_EVENT_NAME, + emitted_at, + event_id, ) def _get_summary( diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index ac73ed59..38104dc5 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -120,8 +120,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( @@ -308,9 +309,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']}" ) 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)