diff --git a/packages/client/src/launchdarkly_ai_server/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py index 9a07b05..9b2fce8 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -22,11 +22,15 @@ set_conversation_id_if_absent, ) from .evaluations import ( + Criterion, + DatasetRow, EvalRunResult, EvaluationsError, EvaluationsModule, GenerationConfig, + Judge, RunSummary, + Scorer, init_evaluations, ) from .graph import GraphInstance, graph, resolve_graph @@ -165,10 +169,14 @@ "VariationMeta", # evaluations "EvalRunResult", + "Criterion", + "DatasetRow", "EvaluationsError", "EvaluationsModule", "GenerationConfig", + "Judge", "RunSummary", + "Scorer", "init_evaluations", # utils "create_handler", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py index 6516f4a..b110a55 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -9,19 +9,24 @@ Transport, urllib_transport, ) +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", "GenerationConfig", "HttpResponse", + "Judge", "LDApiClient", "LDApiError", "RunSummary", + "Scorer", "Transport", "Usage", "init_evaluations", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/criteria.py b/packages/client/src/launchdarkly_ai_server/evaluations/criteria.py new file mode 100644 index 0000000..f77d641 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/criteria.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Any + +from .types import DatasetRow + +type ScorerFn = Callable[[DatasetRow, 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)``, 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 + 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 Criterion = Judge | Scorer + + +def _validate_thresholds( + *, + threshold: float | None, + pass_rate_threshold: float | None, +) -> None: + for name, value in ( + ("threshold", threshold), + ("pass_rate_threshold", pass_rate_threshold), + ): + if value is 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/events.py b/packages/client/src/launchdarkly_ai_server/evaluations/events.py new file mode 100644 index 0000000..97de093 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/events.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from typing import Any + + +class EvaluationStatus(StrEnum): + COMPLETE = "COMPLETE" + ERROR = "ERROR" + + +class EvaluationEventKind(StrEnum): + JUDGE = "judge" + SCORER = "scorer" + + +@dataclass(frozen=True) +class TokenUsage: + """Token usage reported by an LD Judge provider call.""" + + input_tokens: int + output_tokens: int + + def to_wire(self) -> dict[str, int]: + return { + "inputTokens": self.input_tokens, + "outputTokens": self.output_tokens, + } + + +@dataclass(frozen=True, kw_only=True) +class EvaluationEventPayload: + """Common fields emitted for every SDK-run evaluation criterion result.""" + + project_key: str + evaluation_id: str + evaluation_run_id: str + run_id: str + dataset_id: str + row_index: int + criterion_type: str + kind: EvaluationEventKind + event_id: str + emitted_at: str + evaluation_key: str + dataset_key: str + status: EvaluationStatus + started_at: str + evaluated_at: str + latency_ms: int + evaluation_version: int | None = None + score: float | None = None + reason: str | None = None + error: dict[str, Any] | None = None + error_message: str | None = None + + def to_track_payload(self) -> dict[str, Any]: + payload: dict[str, Any] = { + "projectKey": self.project_key, + "evaluationId": self.evaluation_id, + "evaluationRunId": self.evaluation_run_id, + "runId": self.run_id, + "datasetId": self.dataset_id, + "rowIndex": self.row_index, + "criterionType": self.criterion_type, + "kind": self.kind.value, + "eventId": self.event_id, + "emittedAt": self.emitted_at, + "evaluationKey": self.evaluation_key, + "evaluationVersion": self.evaluation_version, + "datasetKey": self.dataset_key, + "status": self.status.value, + "startedAt": self.started_at, + "evaluatedAt": self.evaluated_at, + "latencyMs": self.latency_ms, + "score": self.score, + "reason": self.reason, + "error": self.error, + "errorMessage": self.error_message, + } + return {key: value for key, value in payload.items() if value is not None} + + +@dataclass(frozen=True, kw_only=True) +class LDJudgeEvaluationEventPayload(EvaluationEventPayload): + """Payload for one LaunchDarkly AI Judge result on one dataset row.""" + + kind: EvaluationEventKind = EvaluationEventKind.JUDGE + judge_key: str + variation_key: str + version: int | None = None + usage: TokenUsage | None = None + + def to_track_payload(self) -> dict[str, Any]: + payload = super().to_track_payload() + payload["judgeKey"] = self.judge_key + payload["variationKey"] = self.variation_key + if self.version is not None: + payload["version"] = self.version + if self.usage is not None: + payload["usage"] = self.usage.to_wire() + return payload + + +@dataclass(frozen=True, kw_only=True) +class 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/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 4a3ffad..57a7752 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 .criteria import Criterion, Judge from .runner import EvalHandler, EvaluationsRunner, ToolImplementation, _segment from .types import EvalRunResult, GenerationConfig, RunSummary @@ -93,12 +94,19 @@ async def run( handler: EvalHandler, generation: GenerationConfig, tools: Mapping[str, ToolImplementation] | None = None, + criteria: list[Criterion] | None = None, concurrency: int = 10, poll_interval_seconds: float | None = None, poll_timeout_seconds: float | None = None, ) -> EvalRunResult: """ - Create and run a generation-only evaluation in the caller's process. + Create and run an evaluation in the caller's process. + + Each dataset row is generated with ``handler``; every entry in + ``criteria`` — LaunchDarkly :class:`Judge` references and local + deterministic :class:`Scorer` functions — is then run against each + generated row, and one evaluation event is emitted per + ``(row, criterion)`` result. The returned pass/fail result is derived from LaunchDarkly's run summary. A CI script can exit with ``0 if result.passed else 1`` after awaiting @@ -121,14 +129,20 @@ async def run( poll_timeout_seconds=poll_timeout_seconds, ) 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) + ] client = await self._resolve_client() # The management API client is synchronous; running it in a worker thread # keeps the caller's event loop free. - # Tool verification is deliberately first: a typo must not create records. + # Tool/judge verification is deliberately first: a typo must not create records. resolved_tools = await asyncio.to_thread( self._runner._resolve_tools, project_key, run_tools ) + resolved_judges = await self._runner._resolve_judges(project_key, ld_judges) dataset_ref = await asyncio.to_thread( self._runner._fetch_dataset, project_key, dataset ) @@ -141,12 +155,12 @@ async def run( key, generation, resolved_tools, + run_criteria, ) evaluation_run = await asyncio.to_thread( self._runner._create_evaluation_run, project_key, evaluation.id, - len(rows), dataset_ref.id, ) config = self._runner._build_handler_config(generation, resolved_tools) @@ -157,17 +171,38 @@ async def run( run_tools, concurrency, ) - self._runner._emit_generation_events( - client, - project_key=project_key, - evaluation=evaluation, - evaluation_run=evaluation_run, - dataset=dataset_ref, - results=results, - ) - flush_result = client.flush() - if inspect.isawaitable(flush_result): - await flush_result + try: + self._runner._emit_generation_events( + client, + project_key=project_key, + evaluation=evaluation, + evaluation_run=evaluation_run, + dataset=dataset_ref, + results=results, + ) + if run_criteria: + criterion_results = await self._runner._run_criteria_for_results( + results, + handler, + run_tools, + run_criteria, + resolved_judges, + concurrency, + ) + self._runner._emit_evaluation_events( + client, + project_key=project_key, + evaluation=evaluation, + evaluation_run=evaluation_run, + dataset=dataset_ref, + results=criterion_results, + ) + finally: + # Generation results already queued on the SDK event buffer must + # reach LaunchDarkly even when the criteria phase fails. + flush_result = client.flush() + if inspect.isawaitable(flush_result): + await flush_result summary = await self._poll_summary_until_terminal( project_key, evaluation.id, @@ -242,6 +277,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 951213b..5a2e4ee 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -2,28 +2,52 @@ import asyncio import hashlib +import inspect import json +import logging import time import urllib.parse from collections.abc import Awaitable, Callable, Mapping from datetime import UTC, datetime from typing import Any +from ..judge_scoring import ( + FORMATTING_INSTRUCTIONS, + numeric_score, + parse_judge_response, +) +from ..lifecycle import extract_variation from ..types import NativeTool -from ..utils import parse_template, parse_usage, to_ld_context +from ..utils import ( + parse_template, + parse_usage, + to_ld_context, +) from .api import EvaluationsError, LDApiClient, LDApiError +from .criteria import Criterion, Judge, Scorer +from .events import ( + DeterministicScorerEvaluationEventPayload, + EvaluationEventPayload, + EvaluationStatus, + LDJudgeEvaluationEventPayload, + TokenUsage, +) from .types import ( DatasetRef, DatasetRow, EvaluationRef, EvaluationRunRef, GenerationConfig, + ResolvedJudge, ResolvedTool, RunSummary, ) +logger = logging.getLogger(__name__) + DATASET_PAGE_SIZE = 200 GENERATION_EVENT_NAME = "$ld:ai:offline-evals:generation" +EVALUATION_EVENT_NAME = "$ld:ai:offline-evals:evaluation" EvalHandler = Callable[..., Awaitable[dict[str, Any]]] ToolImplementation = Callable[..., Any] | NativeTool @@ -124,6 +148,44 @@ def _resolve_tools( ) return resolved + 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] = {} + # variation() rejects a context without kind and key; use the same + # context shape the emitted evaluation events are attributed to. + context: dict[str, Any] = {"kind": "evaluation", "key": project_key} + for judge in judges: + try: + variation = await extract_variation(judge.key, context) + except Exception as error: + raise EvaluationsError( + f"Failed to resolve LaunchDarkly judge {judge.key!r}: {error} " + f"If the judge does not exist in project {project_key!r}, " + "create it in the LaunchDarkly UI and try again." + ) from error + config = variation.get("config") + meta_value = variation.get("meta") + meta: Mapping[str, Any] = ( + meta_value if isinstance(meta_value, Mapping) else {} + ) + if not isinstance(config, Mapping): + raise EvaluationsError( + f"LaunchDarkly judge {judge.key!r} returned an invalid AI config variation" + ) + 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 +293,7 @@ def _create_evaluation( key: str, generation: GenerationConfig, tools: Mapping[str, ResolvedTool], + criteria: list[Criterion] | None = None, ) -> EvaluationRef: body: dict[str, Any] = { "name": key, @@ -253,6 +316,8 @@ def _create_evaluation( body["tools"] = [ {"key": tool.key, "version": tool.version} for tool in tools.values() ] + if criteria: + body["criteria"] = [criterion.to_criteria_wire() for criterion in criteria] path = f"projects/{_segment(project_key)}/evaluations" raw = _mapping(self._api.post(path, body=body), description="evaluation") @@ -269,22 +334,18 @@ def _create_evaluation_run( self, project_key: str, evaluation_id: str, - row_count: int, dataset_id: str, ) -> EvaluationRunRef: path = ( f"projects/{_segment(project_key)}/evaluations/" f"{_segment(evaluation_id)}/runs" ) + body: dict[str, Any] = { + "source": "api", + "datasetId": dataset_id, + } raw = _mapping( - self._api.post( - path, - body={ - "source": "api", - "rowCount": row_count, - "datasetId": dataset_id, - }, - ), + self._api.post(path, body=body), description="evaluation run", ) return self._run_ref(raw) @@ -475,9 +536,348 @@ def _emit_generation_events( if "usage" in generated: payload["usage"] = generated["usage"] client.track(GENERATION_EVENT_NAME, context, payload, 1) - print( - f"{GENERATION_EVENT_NAME} emittedAt={emitted_at} eventId={event_id}", - flush=True, + logger.info( + "%s emittedAt=%s eventId=%s", + GENERATION_EVENT_NAME, + emitted_at, + event_id, + ) + + def _judge_variables( + self, + row_result: Mapping[str, Any], + judge: Judge, + ) -> dict[str, Any]: + """Variables available to the judge config's ``{{...}}`` placeholders. + + Absent values become empty strings: ``parse_template`` leaves a + placeholder with a ``None`` value as-is, and literal mustache text must + not reach the judge model. + """ + variables = dict(row_result.get("variables") or {}) + output = row_result.get("output") + expected = row_result.get("expected_output") + ground_truth = judge.ground_truth_context + if ground_truth is not None: + ground_truth = parse_template(ground_truth, variables) + elif expected is not None: + ground_truth = str(expected) + variables.update( + { + "input": row_result.get("input") or "", + "response_to_evaluate": output if output is not None else "", + "message_history": "\n\n".join( + str(value) + for value in (row_result.get("input"), output) + if value is not None + ), + "expected_output": expected if expected is not None else "", + "ground_truth_context": ( + ground_truth if ground_truth is not None else "" + ), + } + ) + return variables + + def _criterion_error_result( + self, + base: Mapping[str, Any], + started_clock: float, + code: str, + message: str, + ) -> dict[str, Any]: + completed = datetime.now(UTC) + return { + **base, + "status": "ERROR", + "error": {"code": code, "message": message}, + "evaluated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round((time.perf_counter() - started_clock) * 1000), + } + + async def _run_scorer_for_result( + self, + row: Mapping[str, Any], + scorer: Scorer, + ) -> dict[str, Any]: + started = datetime.now(UTC) + started_clock = time.perf_counter() + base: dict[str, Any] = { + "row_index": row["row_index"], + "criterion_type": scorer.criterion_type, + "kind": "scorer", + "started_at": started.isoformat().replace("+00:00", "Z"), + } + if row.get("status") != "COMPLETE": + return self._criterion_error_result( + base, + started_clock, + "generation_incomplete", + "generation did not complete", + ) + dataset_row = DatasetRow( + row_index=row["row_index"], + input=row.get("input"), + expected_output=row.get("expected_output"), + variables=dict(row.get("variables") or {}), + metadata=row.get("metadata"), + ) + try: + score_value = scorer.fn(dataset_row, row.get("output")) + if inspect.isawaitable(score_value): + score_value = await score_value + except Exception as error: + return self._criterion_error_result( + base, started_clock, "scorer_raised", f"scorer fn raised: {error}" + ) + if isinstance(score_value, bool): + score: float = 1.0 if score_value else 0.0 + else: + maybe_score = numeric_score(score_value) + if maybe_score is None: + return self._criterion_error_result( + base, + started_clock, + "invalid_score", + "scorer fn must return a bool or a finite number, " + f"got {score_value!r}", + ) + score = maybe_score + if score < 0 or score > 1: + return self._criterion_error_result( + base, + started_clock, + "invalid_score", + f"scorer fn score must be between 0 and 1, got {score_value!r}", + ) + completed = datetime.now(UTC) + return { + **base, + "status": "COMPLETE", + "score": score, + "reason": None, + "evaluated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round((time.perf_counter() - started_clock) * 1000), + } + + async def _run_ld_judge_for_result( + self, + row: Mapping[str, Any], + 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, + "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: + result = await handler( + dict(resolved.config), + row.get("output"), + tool_handlers, + { + **variables, + "formatting_instructions": FORMATTING_INSTRUCTIONS, + }, + ) + except Exception as error: + return self._criterion_error_result( + base, started_clock, "handler_raised", f"judge handler raised: {error}" + ) + if not isinstance(result, Mapping): + return self._criterion_error_result( + base, + started_clock, + "invalid_judge_output", + "judge handler result must be a mapping", + ) + try: + raw_score, reason = parse_judge_response( + result.get("output", result.get("response")) + ) + except ValueError as error: + return self._criterion_error_result( + base, started_clock, "invalid_judge_output", str(error) + ) + score = numeric_score(raw_score) + if score is None or score < 0 or score > 1: + return self._criterion_error_result( + base, + started_clock, + "invalid_score", + f"judge score must be a number between 0 and 1, got {raw_score!r}", + ) + completed = datetime.now(UTC) + event = { + **base, + "status": "COMPLETE", + "score": score, + "reason": reason, + "evaluated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round((time.perf_counter() - started_clock) * 1000), + } + usage = result.get("usage") + if isinstance(usage, Mapping): + event["usage"] = dict(usage) + return event + + async def _run_criteria_for_results( + self, + rows: list[dict[str, Any]], + handler: EvalHandler, + tool_handlers: dict[str, ToolImplementation], + criteria: list[Criterion], + resolved_judges: Mapping[str, ResolvedJudge], + concurrency: int, + ) -> list[dict[str, Any]]: + """Run every (row, criterion) pair, bounded by the run's concurrency.""" + controller = ConcurrencyController(concurrency) + + async def run_one( + row: Mapping[str, Any], criterion: Criterion + ) -> dict[str, Any]: + await controller.acquire() + try: + if isinstance(criterion, Scorer): + return await self._run_scorer_for_result(row, criterion) + return await self._run_ld_judge_for_result( + row, + 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, + 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: + # 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 + ) + except Exception: + logger.exception( + "Skipping evaluation event for row %s criterion %s", + result.get("row_index"), + result.get("criterion_type"), + ) + continue + logger.info( + "%s emittedAt=%s eventId=%s", + EVALUATION_EVENT_NAME, + emitted_at, + event_id, ) def _get_summary( diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/types.py b/packages/client/src/launchdarkly_ai_server/evaluations/types.py index 5f4f4d5..b5d0e64 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/src/launchdarkly_ai_server/judge_scoring.py b/packages/client/src/launchdarkly_ai_server/judge_scoring.py new file mode 100644 index 0000000..8375b41 --- /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 ecff6b1..934c271 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 5784559..38104dc 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -9,8 +9,11 @@ import pytest from launchdarkly_ai_server.evaluations import ( + DatasetRow, EvaluationsError, HttpResponse, + Judge, + Scorer, init_evaluations, ) @@ -117,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( @@ -278,7 +282,6 @@ async def test_complete_run_with_zero_failed_and_error_rows_passes( ) assert transport.requests[5]["body"] == { "source": "api", - "rowCount": 2, "datasetId": "33333333-3333-3333-3333-333333333333", } @@ -306,9 +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']}" ) @@ -1033,3 +1040,594 @@ async def fake_init_client(options: dict[str, Any]) -> MagicMock: assert "inputTokens" not in error_event assert "outputTokens" not in error_event assert {"input", "expected_output", "metadata", "variables"}.isdisjoint(error_event) + + +@pytest.mark.asyncio +async def test_run_with_ld_judge_emits_per_criterion_evaluation_event( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": 42, + "input": "Question {{id}}", + "expectedOutput": "Answer {{id}}", + "variables": {"id": "A"}, + } + ], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + assert key == "$ld:ai:judge:accuracy" + # An empty or kindless context is invalid to the real LD SDK and would + # make every judge resolution fail. + assert context == {"kind": "evaluation", "key": "proj"} + return { + "config": { + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + "instructions": "Judge {{response_to_evaluate}} against {{expected_output}}", + }, + "meta": {"variationKey": "default", "version": 12}, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + assert user_input == "generated" + assert variables["response_to_evaluate"] == "generated" + assert variables["expected_output"] == "Answer A" + # The SDK hands the judge config over unrendered; the handler owns + # the single template pass. + assert config["instructions"] == ( + "Judge {{response_to_evaluate}} against {{expected_output}}" + ) + assert variables["formatting_instructions"].startswith( + "Your response MUST be in valid JSON" + ) + return { + "output": '{"score": 0.86, "reasoning": "matches policy"}', + "usage": {"input_tokens": 640, "output_tokens": 48}, + } + return { + "output": "generated", + "usage": {"input_tokens": 10, "output_tokens": 4}, + } + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + 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=r"Failed to resolve LaunchDarkly judge 'security-judge': not found", + ): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="security-judge")], + ) + + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_run_with_deterministic_scorer_emits_scorer_evaluation_event( + stub_sdk_client: MagicMock, +) -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "support-golden-v3"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": 42, + "input": "Ticket {{id}}", + "expectedOutput": "refund row", + "variables": {"id": "A"}, + } + ], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return { + "output": "refund exists", + "usage": {"input_tokens": 10, "output_tokens": 4}, + } + + def check_refund(row: DatasetRow, output: Any) -> bool: + assert row.row_index == 42 + assert row.input == "Ticket A" + assert output == "refund exists" + return "refund" in str(output) + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="support-golden-v3", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Scorer(name="refund-exists", fn=check_refund)], + ) + + assert result.passed is True + 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 + + +def judge_run_transport(*, summary: dict[str, Any] | None = None) -> SequencedTransport: + """Transport for a one-row run that resolves a dataset, evaluation, and run.""" + return SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": 7, + "input": "Question {{id}}", + "expectedOutput": "Answer {{id}}", + "variables": {"id": "A"}, + } + ], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + summary + or { + "statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0} + }, + ), + ] + ) + + +def accuracy_judge_variation(monkeypatch: pytest.MonkeyPatch) -> None: + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + return { + "config": { + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + "instructions": "Judge {{response_to_evaluate}} against {{expected_output}}", + }, + "meta": {"variationKey": "default", "version": 12}, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + + +@pytest.mark.parametrize( + ("judge_output", "expected_code"), + [ + ('{"score": "high (0.9)", "reasoning": "confident"}', "invalid_score"), + ('{"score": 3, "reasoning": "confident"}', "invalid_score"), + ('{"score": NaN, "reasoning": "confident"}', "invalid_score"), + ("the answer looks right to me", "invalid_judge_output"), + ], +) +@pytest.mark.asyncio +async def test_bad_judge_output_emits_error_event_instead_of_crashing( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, + judge_output: str, + expected_code: str, +) -> None: + transport = judge_run_transport() + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + return {"output": judge_output} + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + judge_event = next(event for event in events if event.get("kind") == "judge") + assert judge_event["status"] == "ERROR" + assert judge_event["error"]["code"] == expected_code + assert judge_event["errorMessage"] == judge_event["error"]["message"] + assert "score" not in judge_event + stub_sdk_client.flush.assert_awaited() + + +@pytest.mark.asyncio +async def test_generated_placeholders_are_not_expanded_into_judge_prompt( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + from launchdarkly_ai_server import parse_template + + transport = judge_run_transport() + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + rendered = parse_template(config["instructions"], variables) + # The placeholder smuggled in via the generated output must stay + # literal text after the handler's single render pass. + assert rendered == "Judge {{expected_output}} leaked? against Answer A" + return {"output": '{"score": 1, "reasoning": "ok"}'} + return {"output": "{{expected_output}} leaked?"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + judge_event = next(event for event in events if event.get("kind") == "judge") + assert judge_event["status"] == "COMPLETE" + assert judge_event["score"] == 1.0 + + +@pytest.mark.asyncio +async def test_missing_expected_output_renders_empty_judge_variables( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page([{"rowIndex": 7, "input": "Question"}], total=1), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + assert variables["expected_output"] == "" + assert variables["ground_truth_context"] == "" + return {"output": '{"score": 1, "reasoning": "ok"}'} + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + assert result.passed is True + + +@pytest.mark.asyncio +async def test_duplicate_criteria_rejected_before_any_request() -> None: + transport = SequencedTransport([]) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + with pytest.raises(EvaluationsError, match="Duplicate evaluation criteria"): + await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[ + Judge(key="accuracy"), + Scorer(name="accuracy", fn=lambda row, output: True), + ], + ) + + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_errored_generation_row_emits_generation_incomplete_criterion_event( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + transport = judge_run_transport( + summary={"statusCounts": {"total": 1, "passed": 0, "error": 1, "pending": 0}} + ) + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + raise AssertionError("judges must not run for errored generations") + raise RuntimeError("provider unavailable") + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is False + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + judge_event = next(event for event in events if event.get("kind") == "judge") + assert judge_event["status"] == "ERROR" + assert judge_event["error"]["code"] == "generation_incomplete" + + +@pytest.mark.asyncio +async def test_failed_evaluation_event_tracking_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/packages/client/tests/test_judge_scoring.py b/packages/client/tests/test_judge_scoring.py new file mode 100644 index 0000000..4613b67 --- /dev/null +++ b/packages/client/tests/test_judge_scoring.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from launchdarkly_ai_server.judge_scoring import parse_judge_response + + +class TestParseJudgeResponse: + def test_parses_plain_json(self) -> None: + assert parse_judge_response('{"score": 0.9, "reasoning": "solid"}') == ( + 0.9, + "solid", + ) + + def test_parses_fenced_json(self) -> None: + raw = '```json\n{"score": 1, "reasoning": "ok"}\n```' + assert parse_judge_response(raw) == (1, "ok") + + def test_accepts_already_decoded_mapping(self) -> None: + assert parse_judge_response({"score": 0.5, "reasoning": "meh"}) == ( + 0.5, + "meh", + ) + + def test_falls_back_to_reason_key(self) -> None: + assert parse_judge_response({"score": 0.5, "reason": "alt key"}) == ( + 0.5, + "alt key", + ) + + def test_null_reasoning_becomes_empty_string_not_none_literal(self) -> None: + assert parse_judge_response({"score": 0.5, "reasoning": None}) == (0.5, "") + + def test_score_returned_untouched_for_caller_policy(self) -> None: + score, _ = parse_judge_response({"score": "high", "reasoning": "?"}) + assert score == "high" + + @pytest.mark.parametrize( + "raw", + [ + "the answer looks correct", + "{}", + {}, + None, + 42, + ["not", "a", "mapping"], + '["not", "a", "mapping"]', + ], + ) + def test_rejects_non_object_responses(self, raw: Any) -> None: + with pytest.raises(ValueError, match="Invalid JSON from judge"): + parse_judge_response(raw) diff --git a/packages/client/tests/test_judges.py b/packages/client/tests/test_judges.py index 582db51..01998bf 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