diff --git a/AGENTS.md b/AGENTS.md index 69f1975c..caac1990 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -703,3 +703,10 @@ response = await graph( }, ).invoke(user_input, context) ``` + +## Maintaining this file + +Keep this file for knowledge useful to almost every future agent session in this project. +Do not repeat what the codebase already shows; point to the authoritative file or command instead. +Prefer rewriting or pruning existing entries over appending new ones. +When updating this file, preserve this bar for all agents and keep entries concise. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 00000000..a9d4d269 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,2 @@ + +@AGENTS.md diff --git a/packages/ai/README.md b/packages/ai/README.md index e35135b9..140a1a8e 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -52,6 +52,30 @@ if result["enabled"]: Never raises. Returns `{"enabled": bool, "config": dict | None, "meta": dict | None}`. +## Evaluations from code + +`init_evaluations` and the evaluations result types are also re-exported: + +```python +from launchdarkly_ai_python import Accuracy, Scorer, init_evaluations + +evals = init_evaluations() +result = await evals.run( + project_key="my-project", + key="unique-evaluation-key", + dataset="golden-dataset", + handler=my_handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + judges=[ + Accuracy(), + Scorer(name="exact-match", fn=lambda row, output: output == row.expected_output), + ], +) +print(result.evaluation_results) +``` + +`LD_API_TOKEN` is required, and LaunchDarkly judges also require `LD_SDK_KEY`; deterministic `Scorer` values do not. Use `LD_API_BASE_URI` for staging or local management API traffic; it is separate from the SDK delivery setting `LD_BASE_URI`. See the [core evaluations guide](../client/README.md#run-an-evaluation-from-code). + --- All exports, types, and behaviors are identical to `launchdarkly-ai-server`. See the [core client README](../client/README.md) for the full API reference. diff --git a/packages/client/README.md b/packages/client/README.md index 8691babd..d2cfe505 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -39,6 +39,53 @@ No code changes are required — `init_client()` detects the packages at runtime | `LD_SERVICE_NAME` | No | OTel `service.name` resource attribute (default: `python-sdk`) | | `LD_ENVIRONMENT` | No | `deployment.environment` resource attribute attached to telemetry | | `OTEL_EXPORTER_OTLP_ENDPOINT` | No | OTLP endpoint override (default: LaunchDarkly Observability backend) | +| `LD_API_TOKEN` | For evaluations | API access token used by the evaluations management API | +| `LD_API_BASE_URI` | No | Evaluations management API host override; intentionally separate from `LD_BASE_URI` | + +### Run an evaluation from code + +The evaluations harness reads an LD-hosted dataset, creates a new evaluation and client-source run, invokes your handler once per row, optionally runs typed LaunchDarkly judges and deterministic scorers in the same worker, uploads the generations, and returns LaunchDarkly's stored verdict. Evaluation keys must be unique because every call creates a new evaluation with `POST`. + +```python +import asyncio +import sys + +from launchdarkly_ai_openai_messages import create_openai_messages_handler +from launchdarkly_ai_server import Accuracy, Judge, Scorer, init_evaluations + + +async def main() -> int: + evals = init_evaluations() # LD_API_TOKEN + LD_SDK_KEY for LD judges + result = await evals.run( + project_key="my-project", + key="support-qa-2026-08-20", + dataset="support-golden", + handler=create_openai_messages_handler(), + generation={ + "provider": "OpenAI", + "model": "gpt-4o", + "instructions": "You are a support agent.", + }, + judges=[ + Accuracy(), + Judge(key="security-judge", threshold=0.7), + Scorer( + name="exact-match", + fn=lambda row, output: output == row.expected_output, + ), + ], + ) + print(result.url, result.summary) + print(result.evaluation_results) + return 0 if result.passed else 1 + + +sys.exit(asyncio.run(main())) +``` + +`project_key` is supplied per run rather than during initialization. `generation.instructions` is shorthand for one system message; use `generation.messages` instead for a full message list, but do not supply both. The harness never retries a handler invocation because doing so could repeat tool side effects. Its retries apply only to LaunchDarkly management API requests. + +`judges` accepts only typed `JudgeReference` values (`Accuracy`, `AnswerRelevancy`, `Likeness`, `Bias`, `Toxicity`, `Misinformation`, or `Judge`) and `Scorer` values. LaunchDarkly judges require `LD_SDK_KEY` and a generation handler built with `create_handler()` so the resolved judge model can be routed safely. Scorers may be synchronous or asynchronous and receive `(row, generation_output)`; `row` includes the rendered row index, input, expected output, variables, and metadata. Results are returned in `EvalRunResult.evaluation_results`. This judging foundation does not yet submit those local scores to LaunchDarkly's evaluation-results endpoint, so `result.passed` remains the stored generation-run verdict. The client uses **lazy initialization**: importing the package does not connect to LaunchDarkly. The singleton is created automatically on the first API call that needs it (`config().invoke()`, `graph().invoke()`, `resolve_graph()`, etc.), as long as `LD_SDK_KEY` is set in the environment. diff --git a/packages/client/agents.md b/packages/client/agents.md index 97802964..6a6b03ec 100644 --- a/packages/client/agents.md +++ b/packages/client/agents.md @@ -30,6 +30,7 @@ No other `launchdarkly-ai-*` package may define or duplicate these. They import | `src/launchdarkly_ai_server/utils.py` | `parse_template`, `parse_json_with_possible_fences`, `create_handler`, `parse_usage`, `make_track_data`, `to_ld_context` | | `src/launchdarkly_ai_server/registry.py` | `Registry`, `global_registry`, `compose`, `resolve_handlers`, `resolve_tools` | | `src/launchdarkly_ai_server/judges.py` | `run_judges`, `build_judge_tasks`, `run_judge` | +| `src/launchdarkly_ai_server/evaluations/` | `init_evaluations`, the private management API operations, offline judge/scorer resolution, and `EvaluationsModule.run()` orchestration | | `src/launchdarkly_ai_server/__init__.py` | Public barrel — the only surface handler packages import from | --- @@ -66,7 +67,7 @@ from launchdarkly_ai_server import Registry, global_registry, compose, resolve_h from launchdarkly_ai_server import execute_and_track, execute_and_stream, wrap_tool_handlers # Entry points -from launchdarkly_ai_server import config, graph, resolve_graph +from launchdarkly_ai_server import config, graph, resolve_graph, init_evaluations ``` When adding a new export, add it to `__init__.py`'s imports and `__all__`. Handler packages must never import from sub-paths (e.g. `launchdarkly_ai_server.client`). @@ -123,6 +124,12 @@ Handlers may return any of these — the client normalizes them before emitting --- +## SDK-run evaluations + +`init_evaluations()` creates an evaluations harness using `LD_API_TOKEN` and the management API host `LD_API_BASE_URI`. Do not reuse `LD_BASE_URI`: that variable configures SDK flag delivery and may point at a relay proxy. `LD_SDK_KEY` is optional for generation-only or deterministic-scorer runs, and required when `judges` contains a LaunchDarkly judge reference. + +`await EvaluationsModule.run(...)` takes `project_key` per call. Dataset lookup/row pagination, evaluation creation, and run creation are private helpers; only `run()` is public. Each call creates a new evaluation with `POST`, so its key must be unique. The harness directly invokes the supplied handler once per row, never retries it, then runs typed `JudgeReference` / `Scorer` values with the generated output and full rendered row context. Generation results are batch-ingested; local evaluation outcomes are returned in `EvalRunResult.evaluation_results`, while `passed` remains the server's stored generation-run verdict until evaluation-results ingest lands. + ## OTel Setup The core client owns all OTel initialization. `init_client()` configures a `TracerProvider` with a `BatchSpanProcessor` and an OTLP HTTP exporter when the optional OTel packages are installed. 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 73bfc3dc..233357ff 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -16,6 +16,34 @@ text_message, to_semconv_finish_reason, ) +from .evaluations import ( + Accuracy, + AnswerRelevancy, + Bias, + EvalRunResult, + EvaluationMethod, + EvaluationsError, + EvaluationsModule, + GenerationConfig, + Judge, + JudgeEvaluationError, + JudgeEvaluationResult, + JudgeIdentity, + JudgeReference, + JudgeUsage, + LaunchDarklyJudgeEvaluation, + Likeness, + Misinformation, + RunSummary, + Scorer, + ScorerError, + ScorerResult, + ScorerRow, + ScoreValue, + Toxicity, + init_evaluations, + resolve_launchdarkly_judges, +) from .graph import GraphInstance, graph, resolve_graph from .judges import build_judge_tasks, run_judge, run_judges from .lifecycle import ( @@ -150,6 +178,33 @@ "text_message", "to_semconv_finish_reason", "VariationMeta", + # evaluations + "Accuracy", + "AnswerRelevancy", + "Bias", + "EvalRunResult", + "EvaluationMethod", + "EvaluationsError", + "EvaluationsModule", + "GenerationConfig", + "Judge", + "JudgeEvaluationError", + "JudgeEvaluationResult", + "JudgeIdentity", + "JudgeReference", + "JudgeUsage", + "LaunchDarklyJudgeEvaluation", + "Likeness", + "Misinformation", + "RunSummary", + "ScoreValue", + "Scorer", + "ScorerError", + "ScorerResult", + "ScorerRow", + "Toxicity", + "init_evaluations", + "resolve_launchdarkly_judges", # utils "create_handler", "make_track_data", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py new file mode 100644 index 00000000..9f696926 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -0,0 +1,67 @@ +"""Run LaunchDarkly evaluations from your own environment.""" + +from .api import ( + DEFAULT_BASE_URI, + EvaluationsError, + HttpResponse, + LDApiClient, + LDApiError, + Transport, + urllib_transport, +) +from .judges import ( + Accuracy, + AnswerRelevancy, + Bias, + EvaluationMethod, + Judge, + JudgeEvaluationError, + JudgeEvaluationResult, + JudgeIdentity, + JudgeReference, + JudgeUsage, + LaunchDarklyJudgeEvaluation, + Likeness, + Misinformation, + Toxicity, + resolve_launchdarkly_judges, +) +from .module import EvaluationsModule, init_evaluations +from .scorers import Scorer, ScorerError, ScorerResult, ScorerRow, ScoreValue +from .types import EvalRunResult, GenerationConfig, RunSummary, Usage + +__all__ = [ + "DEFAULT_BASE_URI", + "Accuracy", + "AnswerRelevancy", + "Bias", + "EvalRunResult", + "EvaluationMethod", + "EvaluationsError", + "EvaluationsModule", + "GenerationConfig", + "HttpResponse", + "Judge", + "JudgeEvaluationError", + "JudgeEvaluationResult", + "JudgeIdentity", + "JudgeReference", + "JudgeUsage", + "LDApiClient", + "LDApiError", + "LaunchDarklyJudgeEvaluation", + "Likeness", + "Misinformation", + "RunSummary", + "ScoreValue", + "Scorer", + "ScorerError", + "ScorerResult", + "ScorerRow", + "Toxicity", + "Transport", + "Usage", + "init_evaluations", + "resolve_launchdarkly_judges", + "urllib_transport", +] diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/api.py b/packages/client/src/launchdarkly_ai_server/evaluations/api.py new file mode 100644 index 00000000..957a0922 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/api.py @@ -0,0 +1,182 @@ +from __future__ import annotations + +import json +import random +import time +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import UTC, datetime +from email.utils import parsedate_to_datetime +from typing import Any, Protocol + +DEFAULT_BASE_URI = "https://app.launchdarkly.com" + + +class EvaluationsError(Exception): + """Base error for the evaluations harness.""" + + +class LDApiError(EvaluationsError): + """A non-2xx response from the LaunchDarkly API.""" + + def __init__(self, status: int, method: str, path: str, body: str) -> None: + super().__init__( + f"LaunchDarkly API {method} {path} failed with {status}: {body}" + ) + self.status = status + self.method = method + self.path = path + self.body = body + + +@dataclass +class HttpResponse: + status: int + body: str + headers: dict[str, str] = field(default_factory=dict) + + +class Transport(Protocol): + """Seam the API client sends requests through; replaced in tests.""" + + def __call__( + self, + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: ... + + +def urllib_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, +) -> HttpResponse: + request = urllib.request.Request(url, data=body, headers=headers, method=method) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + return HttpResponse( + status=response.status, + body=response.read().decode("utf-8"), + headers={k.lower(): v for k, v in response.headers.items()}, + ) + except urllib.error.HTTPError as error: + return HttpResponse( + status=error.code, + body=error.read().decode("utf-8"), + headers={k.lower(): v for k, v in error.headers.items()}, + ) + + +class LDApiClient: + """Minimal retrying client for the LaunchDarkly public management API.""" + + def __init__( + self, + api_token: str, + base_uri: str = DEFAULT_BASE_URI, + transport: Transport = urllib_transport, + timeout: float = 30.0, + max_retries: int = 3, + sleep: Callable[[float], None] = time.sleep, + random_value: Callable[[], float] = random.random, + ) -> None: + self.api_token = api_token + self.base_uri = base_uri.rstrip("/") + self._transport = transport + self._timeout = timeout + self._max_retries = max(0, max_retries) + self._sleep = sleep + self._random_value = random_value + + def url_for(self, path: str, params: dict[str, Any] | None = None) -> str: + url = f"{self.base_uri}/api/v2/{path.lstrip('/')}" + if params: + query = {k: str(v) for k, v in params.items() if v is not None} + if query: + url = f"{url}?{urllib.parse.urlencode(query)}" + return url + + def _retry_delay(self, attempt: int, response: HttpResponse | None = None) -> float: + if response is not None: + retry_after = response.headers.get("retry-after") or response.headers.get( + "Retry-After" + ) + if retry_after: + try: + return max(0.0, float(retry_after)) + except ValueError: + try: + when: datetime = parsedate_to_datetime(retry_after) + now = datetime.now(UTC) + return max(0.0, (when - now).total_seconds()) + except (TypeError, ValueError, OverflowError): + pass + exponential = float(min(30.0, 0.5 * (2**attempt))) + jitter = float(self._random_value()) * min(1.0, exponential) + return exponential + jitter + + def request( + self, + method: str, + path: str, + body: Any = None, + params: dict[str, Any] | None = None, + ) -> Any: + headers = { + "Authorization": self.api_token, + "Accept": "application/json", + "User-Agent": "launchdarkly-ai-evaluations-python", + } + payload: bytes | None = None + if body is not None: + headers["Content-Type"] = "application/json" + payload = json.dumps(body).encode("utf-8") + + response: HttpResponse | None = None + for attempt in range(self._max_retries + 1): + try: + response = self._transport( + method, self.url_for(path, params), headers, payload, self._timeout + ) + except (TimeoutError, urllib.error.URLError) as error: + if attempt >= self._max_retries: + raise EvaluationsError( + f"LaunchDarkly API {method} {path} failed after retries: {error}" + ) from error + self._sleep(self._retry_delay(attempt)) + continue + + retryable = response.status == 429 or response.status >= 500 + if retryable and attempt < self._max_retries: + self._sleep(self._retry_delay(attempt, response)) + continue + break + + if response is None: + raise EvaluationsError( + f"LaunchDarkly API {method} {path} returned no response" + ) + if response.status < 200 or response.status >= 300: + raise LDApiError(response.status, method, path, response.body) + if not response.body: + return None + try: + return json.loads(response.body) + except json.JSONDecodeError as error: + raise EvaluationsError( + f"LaunchDarkly API {method} {path} returned invalid JSON" + ) from error + + def get(self, path: str, params: dict[str, Any] | None = None) -> Any: + return self.request("GET", path, params=params) + + def post(self, path: str, body: Any = None) -> Any: + return self.request("POST", path, body=body) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/flags.py b/packages/client/src/launchdarkly_ai_server/evaluations/flags.py new file mode 100644 index 00000000..ee1a1380 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/flags.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import inspect +import logging +from typing import Any, Final + +from ..utils import to_ld_context + +logger = logging.getLogger(__name__) + +ENABLE_BATCH_INGEST_IN_EVALS_FROM_CODE_FLAG_KEY: Final[str] = ( + "enable-batch-ingest-in-evals-from-code" +) +"""Canonical rollout flag for generation-result batch ingestion.""" + + +async def is_generation_result_batch_ingest_enabled( + client: Any, + project_key: str, +) -> bool: + """Return whether the rollout flag enables generation-result batch ingest. + + Flag evaluation is fail-safe: false, malformed, or failed evaluations disable + the gated batch-ingest path. + """ + try: + context = to_ld_context( + client, + {"kind": "project", "key": project_key}, + ) + result = client.variation( + ENABLE_BATCH_INGEST_IN_EVALS_FROM_CODE_FLAG_KEY, + context, + False, + ) + value = await result if inspect.isawaitable(result) else result + return value is True + except Exception: + logger.warning( + "Unable to evaluate %s; generation results will not be batch ingested", + ENABLE_BATCH_INGEST_IN_EVALS_FROM_CODE_FLAG_KEY, + exc_info=True, + ) + return False 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..203e6e76 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/judges.py @@ -0,0 +1,412 @@ +from __future__ import annotations + +import math +from collections.abc import Awaitable, Callable, Mapping, Sequence +from typing import Any, Literal, Protocol + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from ..judges import _FORMATTING_INSTRUCTIONS +from ..lifecycle import extract_variation, init_client +from ..types import AiConfigRep, LDContext, ProviderHandler, VariationMeta +from ..utils import ( + collapse_messages_to_instructions, + normalize_mode, + parse_json_with_possible_fences, + parse_template, + parse_usage, +) +from .api import EvaluationsError + + +class JudgeReference(BaseModel): + """A reference to a LaunchDarkly judge config and its evaluation thresholds.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + key: str + threshold: float = Field(default=0.5, ge=0.0, le=1.0) + pass_rate_threshold: float = Field(default=1.0, ge=0.0, le=1.0) + ground_truth_context: str | None = None + + @field_validator("key") + @classmethod + def _key_must_not_be_blank(cls, value: str) -> str: + if not value.strip(): + raise ValueError("judge key must not be blank") + return value + + def to_criterion(self) -> dict[str, Any]: + """Build the existing evaluation criteria wire representation.""" + options: dict[str, Any] = { + "threshold": self.threshold, + "passRateThreshold": self.pass_rate_threshold, + } + if self.ground_truth_context is not None: + options["groundTruthContext"] = self.ground_truth_context + return {"criterionType": self.key, "options": options} + + +class Judge(JudgeReference): + """A reference to any customer or LaunchDarkly judge config.""" + + +class Accuracy(JudgeReference): + key: Literal["$ld:ai:judge:accuracy"] = "$ld:ai:judge:accuracy" + + +class AnswerRelevancy(JudgeReference): + key: Literal["$ld:ai:judge:relevance"] = "$ld:ai:judge:relevance" + + +class Likeness(JudgeReference): + key: Literal["$ld:ai:judge:likeness"] = "$ld:ai:judge:likeness" + ground_truth_context: str | None = "{{expected_output}}" + + +class Bias(JudgeReference): + key: Literal["$ld:ai:judge:bias"] = "$ld:ai:judge:bias" + threshold: float = Field(default=0.3, ge=0.0, le=1.0) + + +class Toxicity(JudgeReference): + key: Literal["$ld:ai:judge:toxicity"] = "$ld:ai:judge:toxicity" + + +class Misinformation(JudgeReference): + key: Literal["$ld:ai:judge:misinformation"] = "$ld:ai:judge:misinformation" + threshold: float = Field(default=0.3, ge=0.0, le=1.0) + ground_truth_context: str | None = "{{expected_output}}" + + +class JudgeIdentity(BaseModel): + """Pinned judge identity retained for later evaluation-results ingest.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + key: str + variation_key: str + version: int + provider: str + model: str + mode: Literal["agent", "messages"] + is_inverted: bool = False + + +class JudgeUsage(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + input: int = 0 + output: int = 0 + total: int = 0 + + +JudgeErrorCode = Literal[ + "rate_limit_exhausted", "judge_timeout", "judge_parse_error", "judge_error" +] + + +class JudgeEvaluationError(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + code: JudgeErrorCode + message: str + + +class JudgeEvaluationResult(BaseModel): + """One offline score, including its row and pinned judge identity.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + row_index: int + judge: JudgeIdentity + status: Literal["complete", "error"] + score: float | None = Field(default=None, ge=0.0, le=1.0) + reasoning: str | None = None + usage: JudgeUsage = Field(default_factory=JudgeUsage) + error: JudgeEvaluationError | None = None + + +class EvaluationMethod(Protocol): + """Seam for evaluating one generation with its stable dataset-row context.""" + + async def evaluate( + self, + generation_output: str, + *, + row_index: int, + rendered_input: str | None, + expected_output: str | None, + variables: Mapping[str, Any], + metadata: Mapping[str, Any] | None, + ) -> JudgeEvaluationResult: ... + + +JudgeVariationResolver = Callable[[str, LDContext], Awaitable[dict[str, Any]]] +JudgeClientInitializer = Callable[[dict[str, Any]], Awaitable[Any]] + + +def _required_mapping(value: Any, description: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise EvaluationsError(f"Resolved judge has invalid {description}") + return value + + +def _required_non_blank_string(value: Any, description: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise EvaluationsError(f"Resolved judge has no {description}") + return value + + +def _select_handler( + handlers: Sequence[ProviderHandler], provider: str, mode: str +) -> tuple[ProviderHandler, bool] | None: + exact = next( + (handler for handler in handlers if handler.provides_for == (provider, mode)), + None, + ) + wildcard = next( + (handler for handler in handlers if handler.provides_for == ("*", mode)), + None, + ) + selected = exact if exact is not None else wildcard + if selected is not None: + return selected, False + + if mode == "messages": + exact_agent = next( + ( + handler + for handler in handlers + if handler.provides_for == (provider, "agent") + ), + None, + ) + wildcard_agent = next( + (handler for handler in handlers if handler.provides_for == ("*", "agent")), + None, + ) + selected_agent = exact_agent if exact_agent is not None else wildcard_agent + if selected_agent is not None: + return selected_agent, True + return None + + +def _usage(raw: Any) -> JudgeUsage: + normalized = parse_usage(dict(raw) if isinstance(raw, Mapping) else {}) + return JudgeUsage( + input=normalized["input"], + output=normalized["output"], + total=normalized["total"], + ) + + +def _error_code(error: Exception) -> JudgeErrorCode: + if isinstance(error, TimeoutError): + return "judge_timeout" + if ( + getattr(error, "status", None) == 429 + or getattr(error, "status_code", None) == 429 + ): + return "rate_limit_exhausted" + return "judge_error" + + +class LaunchDarklyJudgeEvaluation: + """Resolved, metric-free evaluation method backed by one LD judge config.""" + + def __init__( + self, + *, + reference: JudgeReference, + config: AiConfigRep, + identity: JudgeIdentity, + handler: ProviderHandler, + collapse_messages: bool, + ) -> None: + self.reference = reference + self.identity = identity + self._config = ( + collapse_messages_to_instructions(config) if collapse_messages else config + ) + self._handler = handler + + async def evaluate( + self, + generation_output: str, + *, + row_index: int, + rendered_input: str | None, + expected_output: str | None, + variables: Mapping[str, Any], + metadata: Mapping[str, Any] | None, + ) -> JudgeEvaluationResult: + """Evaluate a generation without emitting online evaluation metrics.""" + stable_variables: dict[str, Any] = { + **variables, + "row_index": row_index, + "input": rendered_input, + "expected_output": expected_output, + "metadata": dict(metadata) if metadata is not None else None, + "response_to_evaluate": generation_output, + } + history_parts = [rendered_input, generation_output, _FORMATTING_INSTRUCTIONS] + stable_variables["message_history"] = "\n\n".join( + part for part in history_parts if part + ) + if self.reference.ground_truth_context is not None: + stable_variables["ground_truth_context"] = parse_template( + self.reference.ground_truth_context, stable_variables + ) + + try: + response = await self._handler( + self._config, + generation_output, + None, + stable_variables, + None, + ) + if not isinstance(response, Mapping): + raise TypeError("judge handler result must be a mapping") + usage = _usage(response.get("usage")) + output = response.get("output") + parsed = parse_json_with_possible_fences( + output if isinstance(output, str) else str(output or "") + ) + if not isinstance(parsed, Mapping): + return self._parse_error( + row_index, usage, "Judge returned invalid JSON" + ) + score = parsed.get("score") + reasoning = parsed.get("reasoning") + if ( + isinstance(score, bool) + or not isinstance(score, int | float) + or not math.isfinite(float(score)) + or not 0.0 <= float(score) <= 1.0 + or not isinstance(reasoning, str) + ): + return self._parse_error( + row_index, + usage, + "Judge response must contain a score from 0 to 1 and string reasoning", + ) + return JudgeEvaluationResult( + row_index=row_index, + judge=self.identity, + status="complete", + score=float(score), + reasoning=reasoning, + usage=usage, + ) + except Exception as error: + return JudgeEvaluationResult( + row_index=row_index, + judge=self.identity, + status="error", + error=JudgeEvaluationError( + code=_error_code(error), message=f"Judge invocation failed: {error}" + ), + ) + + def _parse_error( + self, row_index: int, usage: JudgeUsage, message: str + ) -> JudgeEvaluationResult: + return JudgeEvaluationResult( + row_index=row_index, + judge=self.identity, + status="error", + usage=usage, + error=JudgeEvaluationError(code="judge_parse_error", message=message), + ) + + +async def resolve_launchdarkly_judges( + references: Sequence[JudgeReference], + handlers: Sequence[ProviderHandler], + *, + sdk_key: str | None, + context: LDContext | None = None, + resolver: JudgeVariationResolver = extract_variation, + initialize_client: JudgeClientInitializer = init_client, +) -> list[LaunchDarklyJudgeEvaluation]: + """Resolve all judges before evaluation/run records are created. + + The integration layer should call this during preflight. Missing credentials, + unknown/disabled judge keys, invalid variation metadata, and incompatible + handlers are hard failures, so no partially configured offline run is started. + """ + if any(not isinstance(reference, JudgeReference) for reference in references): + raise EvaluationsError( + "judges must contain typed JudgeReference objects, not strings or mappings" + ) + if not references: + return [] + if not sdk_key or not sdk_key.strip(): + raise EvaluationsError( + "LaunchDarkly judging requires an SDK key. Set LD_SDK_KEY or pass " + "sdk_key to init_evaluations()." + ) + + await initialize_client({"sdkKey": sdk_key}) + resolution_context = context or { + "kind": "user", + "key": "offline-evaluation-judge-resolution", + } + evaluations: list[LaunchDarklyJudgeEvaluation] = [] + for reference in references: + try: + variation = await resolver(reference.key, resolution_context) + except Exception as error: + raise EvaluationsError( + f"LaunchDarkly judge {reference.key!r} was not found or is unavailable; " + "create or enable it in the LaunchDarkly UI before starting the run" + ) from error + + config = dict(_required_mapping(variation.get("config"), "config")) + meta: VariationMeta = dict( + _required_mapping(variation.get("meta"), "variation metadata") + ) + provider = _required_non_blank_string( + _required_mapping(config.get("provider"), "provider").get("name"), + "provider name", + ) + model = _required_non_blank_string( + _required_mapping(config.get("model"), "model").get("name"), + "model name", + ) + variation_key = _required_non_blank_string( + meta.get("variationKey"), "variation key" + ) + version = meta.get("version") + if isinstance(version, bool) or not isinstance(version, int): + raise EvaluationsError( + f"Resolved judge {reference.key!r} has no integer version" + ) + mode = normalize_mode(meta.get("mode")) + selected = _select_handler(handlers, provider, mode) + if selected is None: + raise EvaluationsError( + f"No handler can execute LaunchDarkly judge {reference.key!r} " + f"for provider {provider!r} in {mode!r} mode" + ) + handler, collapse_messages = selected + evaluations.append( + LaunchDarklyJudgeEvaluation( + reference=reference, + config=config, + identity=JudgeIdentity( + key=reference.key, + variation_key=variation_key, + version=version, + provider=provider, + model=model, + mode=mode, + is_inverted=bool(config.get("isInverted", False)), + ), + handler=handler, + collapse_messages=collapse_messages, + ) + ) + return evaluations diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py new file mode 100644 index 00000000..f8a1c236 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import logging +import os +from collections.abc import Mapping, Sequence +from typing import cast + +from ..lifecycle import init_client +from ..types import ProviderHandler +from .api import ( + DEFAULT_BASE_URI, + EvaluationsError, + LDApiClient, + Transport, + urllib_transport, +) +from .flags import is_generation_result_batch_ingest_enabled +from .judges import ( + JudgeReference, + LaunchDarklyJudgeEvaluation, + resolve_launchdarkly_judges, +) +from .runner import ( + EvalHandler, + EvaluationsRunner, + OfflineEvaluation, + ToolImplementation, + _segment, +) +from .scorers import Scorer +from .types import EvalRunResult, GenerationConfig + +logger = logging.getLogger(__name__) + + +def _env(name: str) -> str | None: + """Read an env var, treating blank/whitespace-only values as unset.""" + value = os.environ.get(name, "").strip() + return value if value else None + + +class EvaluationsModule: + """Entry point for running LaunchDarkly evaluations from customer code.""" + + def __init__(self, api_client: LDApiClient, sdk_key: str | None = None) -> None: + self._api = api_client + self._sdk_key = sdk_key + self._runner = EvaluationsRunner(api_client) + + @property + def api(self) -> LDApiClient: + return self._api + + @property + def sdk_key(self) -> str | None: + """SDK key used for observability traces; ``None`` disables tracing.""" + return self._sdk_key + + async def run( + self, + *, + project_key: str, + key: str, + dataset: str, + handler: EvalHandler, + generation: GenerationConfig, + tools: Mapping[str, ToolImplementation] | None = None, + judges: Sequence[JudgeReference | Scorer] | None = None, + concurrency: int = 10, + timeout: float = 300.0, + ) -> EvalRunResult: + """ + Create and run an evaluation in the caller's process. + + Typed judges and scorers run after each successful generation. The + returned verdict is computed by LaunchDarkly. A CI script can exit + with ``0 if result.passed else 1`` after awaiting this method. + """ + self._validate_run_args( + project_key=project_key, + key=key, + dataset=dataset, + handler=handler, + generation=generation, + concurrency=concurrency, + timeout=timeout, + ) + run_tools = dict(tools or {}) + requested_evaluations = list(judges or []) + self._validate_evaluations(requested_evaluations) + batch_ingest_enabled = True + if self._sdk_key: + client = await init_client({"sdkKey": self._sdk_key}) + batch_ingest_enabled = await is_generation_result_batch_ingest_enabled( + client, project_key + ) + evaluation_methods = await self._resolve_evaluations( + requested_evaluations, handler + ) + + # Tool verification is deliberately first: a typo must not create records. + resolved_tools = self._runner._resolve_tools(project_key, run_tools) + dataset_ref = self._runner._fetch_dataset(project_key, dataset) + rows = self._runner._get_dataset_rows(project_key, dataset) + evaluation = self._runner._create_evaluation( + project_key, key, generation, resolved_tools + ) + evaluation_run = self._runner._create_evaluation_run( + project_key, evaluation.id, len(rows), dataset_ref.id + ) + config = self._runner._build_handler_config(generation, resolved_tools) + results, evaluation_results = await self._runner._run_rows( + rows, + handler, + config, + run_tools, + concurrency, + evaluation_methods, + ) + self._runner._ingest_results( + project_key, + evaluation.id, + evaluation_run.id, + results, + batch_ingest_enabled=batch_ingest_enabled, + ) + completed = await self._runner._poll_run( + project_key, evaluation.id, evaluation_run.id, timeout + ) + summary = self._runner._get_summary( + project_key, evaluation.id, evaluation_run.id + ) + url = ( + f"{self._api.base_uri}/projects/{_segment(project_key)}/ai/evaluations/" + f"{_segment(evaluation.id)}/runs/{_segment(evaluation_run.id)}" + ) + return EvalRunResult( + passed=completed.verdict == "passed", + url=url, + run_id=evaluation_run.id, + summary=summary, + evaluation_results=evaluation_results, + ) + + def _validate_evaluations( + self, evaluations: Sequence[JudgeReference | Scorer] + ) -> None: + if any( + not isinstance(evaluation, (JudgeReference, Scorer)) + for evaluation in evaluations + ): + raise EvaluationsError( + "judges must contain typed JudgeReference or Scorer objects" + ) + + async def _resolve_evaluations( + self, + evaluations: Sequence[JudgeReference | Scorer], + generation_handler: EvalHandler, + ) -> list[OfflineEvaluation]: + judge_references = [ + evaluation + for evaluation in evaluations + if isinstance(evaluation, JudgeReference) + ] + resolved_judges: list[LaunchDarklyJudgeEvaluation] = [] + if judge_references: + if not hasattr(generation_handler, "provides_for"): + raise EvaluationsError( + "LaunchDarkly judges require a ProviderHandler created with " + "create_handler()" + ) + resolved_judges = await resolve_launchdarkly_judges( + judge_references, + [cast(ProviderHandler, generation_handler)], + sdk_key=self._sdk_key, + ) + + judge_iterator = iter(resolved_judges) + return [ + next(judge_iterator) + if isinstance(evaluation, JudgeReference) + else evaluation + for evaluation in evaluations + ] + + @staticmethod + def _validate_run_args( + *, + project_key: str, + key: str, + dataset: str, + handler: EvalHandler, + generation: GenerationConfig, + concurrency: int, + timeout: float, + ) -> None: + for name, value in ( + ("project_key", project_key), + ("key", key), + ("dataset", dataset), + ): + if not value.strip(): + raise EvaluationsError(f"{name} must not be blank") + if not callable(handler): + raise EvaluationsError("handler must be callable") + provider = generation.get("provider") + model = generation.get("model") + if not isinstance(provider, str) or not provider.strip(): + raise EvaluationsError("generation.provider is required") + if not isinstance(model, str) or not model.strip(): + raise EvaluationsError("generation.model is required") + if "instructions" in generation and "messages" in generation: + raise EvaluationsError( + "generation.instructions and generation.messages are mutually exclusive" + ) + if concurrency < 1: + raise EvaluationsError("concurrency must be at least 1") + if timeout <= 0: + raise EvaluationsError("timeout must be greater than zero") + + +def init_evaluations( + api_token: str | None = None, + sdk_key: str | None = None, + base_uri: str | None = None, + transport: Transport = urllib_transport, +) -> EvaluationsModule: + """Resolve credentials and construct the evaluations module.""" + token = api_token or _env("LD_API_TOKEN") + if not token: + raise EvaluationsError( + "No LaunchDarkly API access token provided. Set the LD_API_TOKEN " + "environment variable or pass api_token to init_evaluations()." + ) + + resolved_sdk_key = sdk_key or _env("LD_SDK_KEY") + if not resolved_sdk_key: + logger.info( + "No LaunchDarkly SDK key provided; evaluation runs will not emit traces." + ) + + api_client = LDApiClient( + api_token=token, + base_uri=base_uri or _env("LD_API_BASE_URI") or DEFAULT_BASE_URI, + transport=transport, + ) + return EvaluationsModule(api_client=api_client, sdk_key=resolved_sdk_key) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py new file mode 100644 index 00000000..4df791e4 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -0,0 +1,524 @@ +from __future__ import annotations + +import asyncio +import json +import time +import urllib.parse +from collections.abc import Awaitable, Callable, Mapping +from datetime import UTC, datetime +from typing import Any + +from ..types import NativeTool +from ..utils import parse_template +from .api import EvaluationsError, LDApiClient, LDApiError +from .judges import JudgeEvaluationResult, LaunchDarklyJudgeEvaluation +from .scorers import Scorer, ScorerResult, ScorerRow +from .types import ( + DatasetRef, + DatasetRow, + EvaluationRef, + EvaluationRunRef, + GenerationConfig, + ResolvedTool, + RunSummary, +) + +DATASET_PAGE_SIZE = 200 +INGEST_BATCH_SIZE = 50 +MAX_INGEST_ROW_BYTES = 256 * 1024 + +EvalHandler = Callable[..., Awaitable[dict[str, Any]]] +ToolImplementation = Callable[..., Any] | NativeTool +OfflineEvaluation = LaunchDarklyJudgeEvaluation | Scorer +OfflineEvaluationResult = JudgeEvaluationResult | ScorerResult + + +def _segment(value: str) -> str: + return urllib.parse.quote(value, safe="") + + +def _mapping(value: Any, *, description: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise EvaluationsError( + f"LaunchDarkly returned an invalid {description} response" + ) + return value + + +def _required_string(data: Mapping[str, Any], key: str, description: str) -> str: + value = data.get(key) + if not isinstance(value, str) or not value: + raise EvaluationsError( + f"LaunchDarkly {description} response is missing string field {key!r}" + ) + return value + + +class ConcurrencyController: + """Owns row-worker permits; adaptive signals arrive in Phase 3.""" + + def __init__(self, limit: int = 10) -> None: + if limit < 1: + raise EvaluationsError("concurrency must be at least 1") + self._semaphore = asyncio.Semaphore(limit) + + async def acquire(self, provider: str | None = None) -> None: + del provider + await self._semaphore.acquire() + + def release(self) -> None: + self._semaphore.release() + + def record_success( + self, + provider: str | None = None, + headers: Mapping[str, str] | None = None, + ) -> None: + del provider, headers + + def record_rate_limit( + self, + provider: str | None = None, + retry_after: float | None = None, + ) -> None: + del provider, retry_after + + +class EvaluationsRunner: + """Private API operations and orchestration used by EvaluationsModule.run().""" + + def __init__(self, api: LDApiClient) -> None: + self._api = api + + def _resolve_tools( + self, + project_key: str, + tools: Mapping[str, ToolImplementation], + ) -> dict[str, ResolvedTool]: + resolved: dict[str, ResolvedTool] = {} + for key, implementation in tools.items(): + if not callable(implementation) and not isinstance( + implementation, NativeTool + ): + raise EvaluationsError( + f"Tool {key!r} must be callable or a NativeTool instance" + ) + path = f"projects/{_segment(project_key)}/ai-tools/{_segment(key)}" + try: + raw = _mapping(self._api.get(path), description=f"tool {key!r}") + except LDApiError as error: + if error.status == 404: + raise EvaluationsError( + f"LaunchDarkly AI tool {key!r} was not found in project {project_key!r}" + ) from error + raise + version = raw.get("version") + if not isinstance(version, int): + raise EvaluationsError( + f"LaunchDarkly AI tool {key!r} has no integer version" + ) + schema = raw.get("schema") + if not isinstance(schema, Mapping): + schema = {} + resolved[key] = ResolvedTool( + key=key, + version=version, + description=str(raw.get("description") or ""), + schema=dict(schema), + ) + return resolved + + def _fetch_dataset(self, project_key: str, dataset_key: str) -> DatasetRef: + path = f"projects/{_segment(project_key)}/datasets/{_segment(dataset_key)}" + try: + raw = _mapping(self._api.get(path), description=f"dataset {dataset_key!r}") + except LDApiError as error: + if error.status == 404: + raise EvaluationsError( + f"LaunchDarkly dataset {dataset_key!r} was not found in project {project_key!r}" + ) from error + raise + dataset_id = _required_string(raw, "id", "dataset") + response_key = raw.get("key", raw.get("name", dataset_key)) + return DatasetRef(id=dataset_id, key=str(response_key)) + + def _fetch_dataset_rows_page( + self, + project_key: str, + dataset_key: str, + *, + offset: int, + ) -> Mapping[str, Any]: + path = f"projects/{_segment(project_key)}/datasets/{_segment(dataset_key)}/rows" + return _mapping( + self._api.get( + path, + params={ + "mode": "all", + "limit": DATASET_PAGE_SIZE, + "offset": offset, + }, + ), + description=f"rows for dataset {dataset_key!r}", + ) + + def _get_dataset_rows(self, project_key: str, dataset_key: str) -> list[DatasetRow]: + rows: list[DatasetRow] = [] + offset = 0 + total: int | None = None + while total is None or len(rows) < total: + page = self._fetch_dataset_rows_page( + project_key, dataset_key, offset=offset + ) + items = page.get("items") + page_total = page.get("totalCount") + if not isinstance(items, list) or not isinstance(page_total, int): + raise EvaluationsError( + f"LaunchDarkly returned invalid rows for dataset {dataset_key!r}" + ) + total = page_total + if not items: + break + for item_value in items: + item = _mapping(item_value, description="dataset row") + row_index = item.get("rowIndex") + if not isinstance(row_index, int): + raise EvaluationsError( + "A dataset row is missing its integer rowIndex" + ) + variables_value = item.get("variables") + variables = ( + dict(variables_value) + if isinstance(variables_value, Mapping) + else {} + ) + input_value = item.get("input") + expected_value = item.get("expectedOutput") + rendered_input = ( + parse_template(input_value, variables) + if isinstance(input_value, str) + else None + ) + rendered_expected = ( + parse_template(expected_value, variables) + if isinstance(expected_value, str) + else None + ) + variables["input"] = rendered_input + variables["expected_output"] = rendered_expected + metadata_value = item.get("metadata") + rows.append( + DatasetRow( + row_index=row_index, + input=rendered_input, + expected_output=rendered_expected, + variables=variables, + metadata=( + dict(metadata_value) + if isinstance(metadata_value, Mapping) + else None + ), + ) + ) + offset += len(items) + if not rows: + raise EvaluationsError(f"Dataset {dataset_key!r} is empty") + if total is not None and len(rows) != total: + raise EvaluationsError( + f"Dataset {dataset_key!r} returned {len(rows)} of {total} rows" + ) + return rows + + def _create_evaluation( + self, + project_key: str, + key: str, + generation: GenerationConfig, + tools: Mapping[str, ResolvedTool], + ) -> EvaluationRef: + body: dict[str, Any] = { + "name": key, + "generationProvider": generation["provider"], + "generationModel": generation["model"], + } + if "parameters" in generation: + body["parameters"] = generation["parameters"] + if "instructions" in generation: + body["messages"] = [ + {"role": "system", "content": generation["instructions"]} + ] + elif "messages" in generation: + body["messages"] = generation["messages"] + else: + body["messages"] = [] + if "prompt_snippets" in generation: + body["promptSnippets"] = generation["prompt_snippets"] + if tools: + body["tools"] = [ + {"key": tool.key, "version": tool.version} for tool in tools.values() + ] + + path = f"projects/{_segment(project_key)}/evaluations" + raw = _mapping(self._api.post(path, body=body), description="evaluation") + evaluation_id = _required_string(raw, "id", "evaluation") + response_key = raw.get("name", raw.get("label", key)) + version = raw.get("version") + return EvaluationRef( + id=evaluation_id, + key=str(response_key), + version=version if isinstance(version, int) else None, + ) + + 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" + ) + raw = _mapping( + self._api.post( + path, + body={ + "source": "client", + "rowCount": row_count, + "datasetId": dataset_id, + }, + ), + description="evaluation run", + ) + return self._run_ref(raw) + + def _run_ref(self, raw: Mapping[str, Any]) -> EvaluationRunRef: + return EvaluationRunRef( + id=_required_string(raw, "id", "evaluation run"), + evaluation_id=_required_string(raw, "evaluationId", "evaluation run"), + state=_required_string(raw, "state", "evaluation run"), + verdict=(str(raw["verdict"]) if raw.get("verdict") is not None else None), + status_reason=( + str(raw["statusReason"]) + if raw.get("statusReason") is not None + else None + ), + ) + + def _build_handler_config( + self, + generation: GenerationConfig, + tools: Mapping[str, ResolvedTool], + ) -> dict[str, Any]: + parameters = generation.get("parameters") + config: dict[str, Any] = { + "provider": {"name": generation["provider"]}, + "model": {"name": generation["model"], "parameters": parameters}, + "tools": { + key: { + "description": tool.description, + "parameters": tool.schema, + } + for key, tool in tools.items() + }, + } + snippet_variables = {"snippet": generation.get("prompt_snippets", {})} + if "instructions" in generation: + config["instructions"] = parse_template( + generation["instructions"], snippet_variables + ) + elif "messages" in generation: + config["messages"] = [ + { + **message, + "content": parse_template(message["content"], snippet_variables) + if isinstance(message.get("content"), str) + else message.get("content"), + } + for message in generation["messages"] + ] + if "output_format" in generation: + config["outputFormat"] = generation["output_format"] + return config + + async def _run_rows( + self, + rows: list[DatasetRow], + handler: EvalHandler, + config: dict[str, Any], + tool_handlers: dict[str, ToolImplementation], + concurrency: int, + evaluations: list[OfflineEvaluation] | None = None, + ) -> tuple[list[dict[str, Any]], list[OfflineEvaluationResult]]: + controller = ConcurrencyController(concurrency) + evaluation_methods = evaluations or [] + + async def invoke( + row: DatasetRow, + ) -> tuple[dict[str, Any], list[OfflineEvaluationResult]]: + await controller.acquire(config["provider"]["name"]) + started = datetime.now(UTC) + started_clock = time.perf_counter() + try: + result = await handler( + config, row.input, tool_handlers, dict(row.variables) + ) + if not isinstance(result, Mapping): + raise TypeError("handler result must be a mapping") + completed = datetime.now(UTC) + payload: dict[str, Any] = { + "row_index": row.row_index, + "input": row.input, + "expected_output": row.expected_output, + "variables": row.variables, + "metadata": row.metadata, + "output": {"generation": result.get("output")}, + "started_at": started.isoformat().replace("+00:00", "Z"), + "generated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round((time.perf_counter() - started_clock) * 1000), + "status": "COMPLETE", + } + usage = result.get("usage") + if isinstance(usage, Mapping): + payload["output"]["usage"] = dict(usage) + generation_output = result.get("output") + evaluation_results: list[OfflineEvaluationResult] = [] + for evaluation in evaluation_methods: + if isinstance(evaluation, Scorer): + evaluation_results.append( + await evaluation.execute( + ScorerRow( + row_index=row.row_index, + input=row.input, + expected_output=row.expected_output, + variables=row.variables, + metadata=row.metadata, + ), + generation_output + if isinstance(generation_output, str) + else None, + ) + ) + else: + evaluation_results.append( + await evaluation.evaluate( + generation_output + if isinstance(generation_output, str) + else "", + row_index=row.row_index, + rendered_input=row.input, + expected_output=row.expected_output, + variables=row.variables, + metadata=row.metadata, + ) + ) + controller.record_success(config["provider"]["name"]) + return payload, evaluation_results + except Exception as error: + completed = datetime.now(UTC) + return ( + { + "row_index": row.row_index, + "input": row.input, + "expected_output": row.expected_output, + "variables": row.variables, + "metadata": row.metadata, + "started_at": started.isoformat().replace("+00:00", "Z"), + "generated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round( + (time.perf_counter() - started_clock) * 1000 + ), + "status": "ERROR", + "error": { + "code": 5001, + "message": f"handler raised: {error}", + }, + }, + [], + ) + finally: + controller.release() + + row_results = await asyncio.gather(*(invoke(row) for row in rows)) + return ( + [generation for generation, _ in row_results], + [ + evaluation + for _, evaluations_for_row in row_results + for evaluation in evaluations_for_row + ], + ) + + def _ingest_results( + self, + project_key: str, + evaluation_id: str, + run_id: str, + results: list[dict[str, Any]], + *, + batch_ingest_enabled: bool = True, + ) -> None: + if not batch_ingest_enabled: + return + path = ( + f"projects/{_segment(project_key)}/evaluations/{_segment(evaluation_id)}" + f"/runs/{_segment(run_id)}/generation-results" + ) + for result in results: + size = len(json.dumps(result).encode("utf-8")) + if size > MAX_INGEST_ROW_BYTES: + raise EvaluationsError( + f"Generation result row {result['row_index']} exceeds the " + f"{MAX_INGEST_ROW_BYTES}-byte limit" + ) + for start in range(0, len(results), INGEST_BATCH_SIZE): + self._api.post( + path, body={"results": results[start : start + INGEST_BATCH_SIZE]} + ) + + async def _poll_run( + self, + project_key: str, + evaluation_id: str, + run_id: str, + timeout: float, + ) -> EvaluationRunRef: + path = ( + f"projects/{_segment(project_key)}/evaluations/{_segment(evaluation_id)}" + f"/runs/{_segment(run_id)}" + ) + deadline = time.monotonic() + timeout + delay = 0.25 + while True: + run = self._run_ref( + _mapping(self._api.get(path), description="evaluation run") + ) + if run.state == "COMPLETE": + if run.verdict not in {"passed", "failed"}: + raise EvaluationsError( + f"Evaluation run {run_id!r} completed without a verdict" + ) + return run + if run.state in {"CANCELLED", "TEMPORARY_ERROR", "PERMANENT_ERROR"}: + reason = f": {run.status_reason}" if run.status_reason else "" + raise EvaluationsError( + f"Evaluation run {run_id!r} failed in state {run.state}{reason}" + ) + if time.monotonic() >= deadline: + raise EvaluationsError( + f"Evaluation run {run_id!r} is still in progress after {timeout} seconds" + ) + await asyncio.sleep(delay) + delay = min(5.0, delay * 2) + + def _get_summary( + self, project_key: str, evaluation_id: str, run_id: str + ) -> RunSummary: + path = ( + f"projects/{_segment(project_key)}/evaluations/{_segment(evaluation_id)}" + f"/runs/{_segment(run_id)}/summary" + ) + return RunSummary.from_wire( + _mapping(self._api.get(path), description="evaluation run summary") + ) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/scorers.py b/packages/client/src/launchdarkly_ai_server/evaluations/scorers.py new file mode 100644 index 00000000..d6f88da5 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/scorers.py @@ -0,0 +1,211 @@ +"""Deterministic function scorers for client-side evaluations.""" + +from __future__ import annotations + +import inspect +import math +import time +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass +from datetime import UTC, datetime +from types import MappingProxyType +from typing import Any, Literal + +ScoreValue = bool | int | float +ScorerFunction = Callable[["ScorerRow", str | None], ScoreValue | Awaitable[ScoreValue]] +ScorerStatus = Literal["COMPLETE", "ERROR"] +ScorerErrorCode = Literal["invalid_score", "scorer_error"] + + +@dataclass(frozen=True, slots=True) +class ScorerRow: + """Complete rendered dataset-row context passed to a scorer function.""" + + row_index: int + input: str | None + expected_output: str | None + variables: Mapping[str, Any] + metadata: Mapping[str, Any] | None + + def __post_init__(self) -> None: + if type(self.row_index) is not int or self.row_index < 0: + raise ValueError("row_index must be a non-negative integer") + if self.input is not None and not isinstance(self.input, str): + raise TypeError("input must be a string or None") + if self.expected_output is not None and not isinstance( + self.expected_output, str + ): + raise TypeError("expected_output must be a string or None") + object.__setattr__( + self, "variables", _validated_mapping(self.variables, name="variables") + ) + if self.metadata is not None: + object.__setattr__( + self, "metadata", _validated_mapping(self.metadata, name="metadata") + ) + + +@dataclass(frozen=True, slots=True) +class ScorerError: + """Structured scorer failure details for later evaluation-results ingest.""" + + code: ScorerErrorCode + message: str + exception_type: str + + +@dataclass(frozen=True, slots=True) +class ScorerResult: + """The normalized outcome and execution metadata for one row and scorer.""" + + scorer_name: str + row_index: int + score: float | None + started_at: datetime + evaluated_at: datetime + latency_ms: float + status: ScorerStatus + error: ScorerError | None = None + + def __post_init__(self) -> None: + if self.status not in {"COMPLETE", "ERROR"}: + raise ValueError(f"unknown scorer result status: {self.status!r}") + if self.status == "COMPLETE": + if self.score is None or self.error is not None: + raise ValueError( + "a COMPLETE scorer result requires a score and no error" + ) + elif self.score is not None or self.error is None: + raise ValueError("an ERROR scorer result requires an error and no score") + + +@dataclass(frozen=True, slots=True) +class Scorer: + """A named deterministic scorer with an async execution method. + + The scorer function follows the Phase 3 protocol ``fn(row, output)`` and may + be synchronous or asynchronous. ``execute`` converts function failures and + invalid return values into typed error results so evaluation orchestration + can continue processing the remaining rows. + """ + + name: str + fn: ScorerFunction + threshold: float = 1.0 + + 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 TypeError("scorer fn must be callable") + if isinstance(self.threshold, bool) or not isinstance( + self.threshold, (int, float) + ): + raise TypeError("scorer threshold must be numeric") + normalized_threshold = float(self.threshold) + if ( + not math.isfinite(normalized_threshold) + or not 0 <= normalized_threshold <= 1 + ): + raise ValueError("scorer threshold must be between 0 and 1") + object.__setattr__(self, "threshold", normalized_threshold) + + async def execute(self, row: ScorerRow, output: str | None) -> ScorerResult: + """Run this scorer for one generation and return a normalized result.""" + if not isinstance(row, ScorerRow): + raise TypeError("row must be a ScorerRow") + if output is not None and not isinstance(output, str): + raise TypeError("output must be a string or None") + + started_at = datetime.now(UTC) + started_clock = time.perf_counter() + try: + value = self.fn(row, output) + if inspect.isawaitable(value): + value = await value + score = _normalize_score(value) + except _InvalidScore as error: + return self._error_result( + row=row, + started_at=started_at, + started_clock=started_clock, + code="invalid_score", + error=error, + ) + except Exception as error: + return self._error_result( + row=row, + started_at=started_at, + started_clock=started_clock, + code="scorer_error", + error=error, + ) + + evaluated_at = datetime.now(UTC) + return ScorerResult( + scorer_name=self.name, + row_index=row.row_index, + score=score, + started_at=started_at, + evaluated_at=evaluated_at, + latency_ms=_elapsed_ms(started_clock), + status="COMPLETE", + ) + + def _error_result( + self, + *, + row: ScorerRow, + started_at: datetime, + started_clock: float, + code: ScorerErrorCode, + error: Exception, + ) -> ScorerResult: + evaluated_at = datetime.now(UTC) + message = str(error) or type(error).__name__ + return ScorerResult( + scorer_name=self.name, + row_index=row.row_index, + score=None, + started_at=started_at, + evaluated_at=evaluated_at, + latency_ms=_elapsed_ms(started_clock), + status="ERROR", + error=ScorerError( + code=code, + message=message, + exception_type=type(error).__name__, + ), + ) + + +class _InvalidScore(ValueError): + pass + + +def _validated_mapping(value: object, *, name: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise TypeError(f"{name} must be a mapping") + if any(not isinstance(key, str) for key in value): + raise TypeError(f"{name} keys must be strings") + return MappingProxyType(dict(value)) + + +def _normalize_score(value: object) -> float: + if isinstance(value, bool): + return float(value) + if not isinstance(value, (int, float)): + raise _InvalidScore( + "scorer must return bool or a numeric score between 0 and 1; " + f"got {type(value).__name__}" + ) + score = float(value) + if not math.isfinite(score) or not 0 <= score <= 1: + raise _InvalidScore( + f"scorer must return a finite numeric score between 0 and 1; got {value!r}" + ) + return score + + +def _elapsed_ms(started_clock: float) -> float: + return round((time.perf_counter() - started_clock) * 1000, 3) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/types.py b/packages/client/src/launchdarkly_ai_server/evaluations/types.py new file mode 100644 index 00000000..281126e6 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/types.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, TypedDict + +if TYPE_CHECKING: + from .judges import JudgeEvaluationResult + from .scorers import ScorerResult + + +@dataclass +class Usage: + """Token counts for one generation, using the ingest wire field names.""" + + input_tokens: int + output_tokens: int + + def to_wire(self) -> dict[str, int]: + return { + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + } + + @classmethod + def from_wire(cls, data: Mapping[str, Any]) -> Usage: + return cls( + input_tokens=int(data.get("input_tokens") or 0), + output_tokens=int(data.get("output_tokens") or 0), + ) + + +class GenerationConfig(TypedDict, total=False): + """Generation settings stored on the evaluation and passed to its handler.""" + + provider: str + model: str + parameters: dict[str, Any] + instructions: str + messages: list[dict[str, Any]] + prompt_snippets: dict[str, str] + output_format: dict[str, Any] + + +@dataclass +class DatasetRef: + """Identifiers returned when resolving a dataset by key.""" + + id: str + key: str + + +@dataclass +class DatasetRow: + """A rendered dataset row ready for handler invocation and ingest.""" + + row_index: int + input: str | None = None + expected_output: str | None = None + variables: dict[str, Any] = field(default_factory=dict) + metadata: dict[str, Any] | None = None + + +@dataclass +class ResolvedTool: + """The schema and pinned version returned by the LaunchDarkly tool API.""" + + key: str + version: int + description: str = "" + schema: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class EvaluationRef: + """Identifiers returned after creating an evaluation.""" + + id: str + key: str + version: int | None = None + + +@dataclass +class EvaluationRunRef: + """Identifiers and state returned by the evaluation-run API.""" + + id: str + evaluation_id: str + state: str + verdict: str | None = None + status_reason: str | None = None + + +@dataclass +class RunSummary: + """Row counts for a finished evaluation run.""" + + total_rows: int = 0 + passed_rows: int = 0 + failed_rows: int = 0 + error_rows: int = 0 + + @classmethod + def from_wire(cls, data: Mapping[str, Any] | None) -> RunSummary: + data = data or {} + counts_value = data.get("statusCounts") + counts = counts_value if isinstance(counts_value, Mapping) else data + return cls( + total_rows=int(counts.get("total", counts.get("total_rows", 0)) or 0), + passed_rows=int(counts.get("passed", counts.get("passed_rows", 0)) or 0), + failed_rows=int(counts.get("failed", counts.get("failed_rows", 0)) or 0), + error_rows=int(counts.get("error", counts.get("error_rows", 0)) or 0), + ) + + +@dataclass +class EvalRunResult: + """The server verdict and local judge/scorer results for an evaluation run.""" + + passed: bool + url: str + run_id: str + summary: RunSummary + evaluation_results: list[JudgeEvaluationResult | ScorerResult] = field( + default_factory=list + ) diff --git a/packages/client/tests/test_evaluation_flags.py b/packages/client/tests/test_evaluation_flags.py new file mode 100644 index 00000000..9113b623 --- /dev/null +++ b/packages/client/tests/test_evaluation_flags.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from launchdarkly_ai_server.evaluations.flags import ( + ENABLE_BATCH_INGEST_IN_EVALS_FROM_CODE_FLAG_KEY, + is_generation_result_batch_ingest_enabled, +) + + +@pytest.mark.asyncio +async def test_enabled_flag_enables_generation_result_batch_ingest() -> None: + client = MagicMock() + client.variation = AsyncMock(return_value=True) + + assert ( + await is_generation_result_batch_ingest_enabled(client, "project-key") is True + ) + client.variation.assert_awaited_once_with( + ENABLE_BATCH_INGEST_IN_EVALS_FROM_CODE_FLAG_KEY, + {"kind": "project", "key": "project-key"}, + False, + ) + + +@pytest.mark.asyncio +async def test_disabled_flag_disables_generation_result_batch_ingest() -> None: + client = MagicMock() + client.variation = AsyncMock(return_value=False) + + assert ( + await is_generation_result_batch_ingest_enabled(client, "project-key") is False + ) + + +@pytest.mark.asyncio +async def test_flag_evaluation_error_disables_generation_result_batch_ingest() -> None: + client = MagicMock() + client.variation = AsyncMock(side_effect=RuntimeError("delivery unavailable")) + + assert ( + await is_generation_result_batch_ingest_enabled(client, "project-key") is False + ) diff --git a/packages/client/tests/test_evaluation_judges.py b/packages/client/tests/test_evaluation_judges.py new file mode 100644 index 00000000..7092ca29 --- /dev/null +++ b/packages/client/tests/test_evaluation_judges.py @@ -0,0 +1,386 @@ +from __future__ import annotations + +import re +from collections.abc import Mapping, Sequence +from typing import Any, Literal, cast +from unittest.mock import AsyncMock + +import pytest +from pydantic import ValidationError + +from launchdarkly_ai_server.evaluations.api import EvaluationsError +from launchdarkly_ai_server.evaluations.judges import ( + Accuracy, + AnswerRelevancy, + Bias, + Judge, + JudgeReference, + Likeness, + Misinformation, + Toxicity, + resolve_launchdarkly_judges, +) +from launchdarkly_ai_server.types import ProviderHandler +from launchdarkly_ai_server.utils import create_handler + + +def judge_variation( + *, + key: str = "served-variation", + version: int = 12, + provider: str = "OpenAI", + mode: str = "messages", + inverted: bool = False, +) -> dict[str, Any]: + return { + "config": { + "provider": {"name": provider}, + "model": {"name": "judge-model"}, + "instructions": "Evaluate {{response_to_evaluate}}", + "isInverted": inverted, + "evaluationMetricKey": "must-not-be-emitted-offline", + }, + "meta": { + "enabled": True, + "variationKey": key, + "version": version, + "mode": mode, + }, + } + + +def handler( + fn: Any, + *, + provider: str = "OpenAI", + mode: Literal["agent", "messages"] = "messages", +) -> ProviderHandler: + return create_handler((provider, mode), fn) + + +def test_judge_reference_defaults_and_criteria_wire_shape() -> None: + assert Accuracy().to_criterion() == { + "criterionType": "$ld:ai:judge:accuracy", + "options": {"threshold": 0.5, "passRateThreshold": 1.0}, + } + assert AnswerRelevancy().key == "$ld:ai:judge:relevance" + assert Toxicity().threshold == 0.5 + assert Bias().threshold == 0.3 + assert Likeness().ground_truth_context == "{{expected_output}}" + assert Misinformation().to_criterion()["options"] == { + "threshold": 0.3, + "passRateThreshold": 1.0, + "groundTruthContext": "{{expected_output}}", + } + + +def test_judge_references_forbid_typos_and_invalid_thresholds() -> None: + with pytest.raises(ValidationError, match="extra_forbidden"): + Accuracy(threshhold=0.7) # type: ignore[call-arg] + with pytest.raises(ValidationError, match="less_than_equal"): + Judge(key="security", threshold=1.1) + with pytest.raises(ValidationError, match="judge key must not be blank"): + Judge(key=" ") + with pytest.raises(ValidationError, match="literal_error"): + Accuracy(key="different") # type: ignore[arg-type] + + +@pytest.mark.asyncio +async def test_missing_sdk_key_fails_before_initialization_or_resolution() -> None: + initialize = AsyncMock() + resolver = AsyncMock(return_value=judge_variation()) + + with pytest.raises(EvaluationsError, match="LD_SDK_KEY"): + await resolve_launchdarkly_judges( + [Accuracy()], + [], + sdk_key=" ", + resolver=resolver, + initialize_client=initialize, + ) + + initialize.assert_not_awaited() + resolver.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_only_typed_judge_references_are_accepted() -> None: + raw_references = cast(Sequence[JudgeReference], ["security-judge"]) + + with pytest.raises(EvaluationsError, match="typed JudgeReference"): + await resolve_launchdarkly_judges( + raw_references, + [], + sdk_key="sdk-key", + initialize_client=AsyncMock(), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("reference", "expected_key"), + [ + (Judge(key="security-judge"), "security-judge"), + (Accuracy(), "$ld:ai:judge:accuracy"), + ], +) +async def test_unknown_judge_fails_clearly_during_preflight( + reference: JudgeReference, expected_key: str +) -> None: + async def missing(key: str, context: dict[str, Any]) -> dict[str, Any]: + del key, context + raise RuntimeError("variation returned None") + + with pytest.raises( + EvaluationsError, match=rf"{re.escape(expected_key)}.*LaunchDarkly UI" + ): + await resolve_launchdarkly_judges( + [reference], + [], + sdk_key="sdk-key", + resolver=missing, + initialize_client=AsyncMock(), + ) + + +@pytest.mark.asyncio +async def test_resolved_evaluation_preserves_context_score_and_judge_identity() -> None: + received: dict[str, Any] = {} + + async def judge_handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Any] | None, + variables: dict[str, Any] | None, + history: list[dict[str, Any]] | None, + ) -> dict[str, Any]: + received.update( + config=config, + user_input=user_input, + tool_handlers=tool_handlers, + variables=variables, + history=history, + ) + return { + "output": '```json\n{"score": 0.82, "reasoning": "matches"}\n```', + "usage": {"input_tokens": 31, "output_tokens": 7}, + } + + resolver = AsyncMock( + return_value=judge_variation(key="variation-abc", version=27, inverted=True) + ) + methods = await resolve_launchdarkly_judges( + [ + Judge( + key="security-judge", + threshold=0.7, + ground_truth_context="Known: {{expected_output}} / {{account}}", + ) + ], + [handler(judge_handler)], + sdk_key="sdk-key", + resolver=resolver, + initialize_client=AsyncMock(), + ) + + result = await methods[0].evaluate( + "generated answer", + row_index=41, + rendered_input="Where is order A19?", + expected_output="Order A19 shipped", + variables={"account": "enterprise", "input": "unrendered"}, + metadata={"suite": "orders", "case_id": "stable-41"}, + ) + + resolver.assert_awaited_once_with( + "security-judge", + {"kind": "user", "key": "offline-evaluation-judge-resolution"}, + ) + assert result.status == "complete" + assert result.row_index == 41 + assert result.score == 0.82 + assert result.reasoning == "matches" + assert result.usage.model_dump() == {"input": 31, "output": 7, "total": 38} + assert result.judge.model_dump() == { + "key": "security-judge", + "variation_key": "variation-abc", + "version": 27, + "provider": "OpenAI", + "model": "judge-model", + "mode": "messages", + "is_inverted": True, + } + + assert received["user_input"] == "generated answer" + assert received["tool_handlers"] is None + assert received["history"] is None + judge_variables = cast(Mapping[str, Any], received["variables"]) + assert judge_variables["row_index"] == 41 + assert judge_variables["input"] == "Where is order A19?" + assert judge_variables["expected_output"] == "Order A19 shipped" + assert judge_variables["account"] == "enterprise" + assert judge_variables["metadata"] == { + "suite": "orders", + "case_id": "stable-41", + } + assert judge_variables["response_to_evaluate"] == "generated answer" + assert judge_variables["ground_truth_context"] == ( + "Known: Order A19 shipped / enterprise" + ) + assert "Where is order A19?" in judge_variables["message_history"] + assert "generated answer" in judge_variables["message_history"] + + +@pytest.mark.asyncio +async def test_offline_evaluation_never_emits_online_metric_events() -> None: + tracked: list[tuple[Any, ...]] = [] + + class FakeClient: + def track(self, *args: Any) -> None: + tracked.append(args) + + client = FakeClient() + + async def initialize(options: dict[str, Any]) -> FakeClient: + assert options == {"sdkKey": "sdk-key"} + return client + + async def judge_handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Any] | None, + variables: dict[str, Any] | None, + history: list[dict[str, Any]] | None, + ) -> dict[str, Any]: + del config, user_input, tool_handlers, variables, history + return {"output": '{"score": 1, "reasoning": "safe"}', "usage": {}} + + methods = await resolve_launchdarkly_judges( + [Judge(key="security-judge")], + [handler(judge_handler)], + sdk_key="sdk-key", + resolver=AsyncMock(return_value=judge_variation()), + initialize_client=initialize, + ) + result = await methods[0].evaluate( + "answer", + row_index=0, + rendered_input="question", + expected_output=None, + variables={}, + metadata=None, + ) + + assert result.status == "complete" + assert tracked == [] + + +@pytest.mark.asyncio +async def test_unparseable_judge_response_becomes_diagnosable_error_result() -> None: + async def judge_handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Any] | None, + variables: dict[str, Any] | None, + history: list[dict[str, Any]] | None, + ) -> dict[str, Any]: + del config, user_input, tool_handlers, variables, history + return { + "output": "not json", + "usage": {"inputTokens": 4, "outputTokens": 2}, + } + + methods = await resolve_launchdarkly_judges( + [Accuracy()], + [handler(judge_handler)], + sdk_key="sdk-key", + resolver=AsyncMock(return_value=judge_variation()), + initialize_client=AsyncMock(), + ) + result = await methods[0].evaluate( + "answer", + row_index=3, + rendered_input="question", + expected_output="expected", + variables={}, + metadata=None, + ) + + assert result.status == "error" + assert result.score is None + assert result.reasoning is None + assert result.usage.total == 6 + assert result.error is not None + assert result.error.code == "judge_parse_error" + assert "invalid JSON" in result.error.message + + +@pytest.mark.asyncio +async def test_messages_judge_uses_agent_fallback_with_collapsed_prompt() -> None: + received_configs: list[dict[str, Any]] = [] + + async def agent_handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Any] | None, + variables: dict[str, Any] | None, + history: list[dict[str, Any]] | None, + ) -> dict[str, Any]: + del user_input, tool_handlers, variables, history + received_configs.append(config) + return {"output": '{"score": 0.5, "reasoning": "ok"}', "usage": {}} + + variation = judge_variation() + variation["config"].pop("instructions") + variation["config"]["messages"] = [ + {"role": "system", "content": "Apply the rubric."}, + {"role": "user", "content": "Score the response."}, + ] + methods = await resolve_launchdarkly_judges( + [Accuracy()], + [handler(agent_handler, mode="agent")], + sdk_key="sdk-key", + resolver=AsyncMock(return_value=variation), + initialize_client=AsyncMock(), + ) + + await methods[0].evaluate( + "answer", + row_index=1, + rendered_input="question", + expected_output=None, + variables={}, + metadata=None, + ) + + assert received_configs == [ + { + **variation["config"], + "instructions": "Apply the rubric.\n\nScore the response.", + "messages": [], + } + ] + + +@pytest.mark.asyncio +async def test_missing_compatible_handler_fails_during_resolution() -> None: + async def unused_handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Any] | None, + variables: dict[str, Any] | None, + history: list[dict[str, Any]] | None, + ) -> dict[str, Any]: + del config, user_input, tool_handlers, variables, history + return {} + + with pytest.raises(EvaluationsError, match=r"No handler.*Anthropic"): + await resolve_launchdarkly_judges( + [Accuracy()], + [handler(unused_handler, provider="OpenAI")], + sdk_key="sdk-key", + resolver=AsyncMock( + return_value=judge_variation(provider="Anthropic", mode="messages") + ), + initialize_client=AsyncMock(), + ) diff --git a/packages/client/tests/test_evaluation_scorers.py b/packages/client/tests/test_evaluation_scorers.py new file mode 100644 index 00000000..2974e525 --- /dev/null +++ b/packages/client/tests/test_evaluation_scorers.py @@ -0,0 +1,160 @@ +from __future__ import annotations + +from typing import Any, cast + +import pytest + +from launchdarkly_ai_server.evaluations.scorers import ( + Scorer, + ScorerRow, + ScoreValue, +) + + +def scorer_row() -> ScorerRow: + return ScorerRow( + row_index=7, + input="Rendered order A19", + expected_output="Refund A19", + variables={"order_id": "A19", "input": "Rendered order A19"}, + metadata={"suite": "refunds", "priority": 1}, + ) + + +@pytest.mark.asyncio +async def test_sync_scorer_receives_generation_output_and_row_context() -> None: + received: dict[str, Any] = {} + + def score(row: ScorerRow, output: str | None) -> float: + received.update( + { + "row_index": row.row_index, + "input": row.input, + "expected_output": row.expected_output, + "variables": dict(row.variables), + "metadata": dict(row.metadata or {}), + "output": output, + } + ) + return 0.75 + + result = await Scorer(name="refund-exists", fn=score).execute( + scorer_row(), "Refund created" + ) + + assert received == { + "row_index": 7, + "input": "Rendered order A19", + "expected_output": "Refund A19", + "variables": {"order_id": "A19", "input": "Rendered order A19"}, + "metadata": {"suite": "refunds", "priority": 1}, + "output": "Refund created", + } + assert result.scorer_name == "refund-exists" + assert result.row_index == 7 + assert result.score == 0.75 + assert result.status == "COMPLETE" + assert result.error is None + assert result.started_at.tzinfo is not None + assert result.evaluated_at >= result.started_at + assert result.latency_ms >= 0 + + +@pytest.mark.asyncio +async def test_async_scorer_is_awaited() -> None: + called = False + + async def score(row: ScorerRow, output: str | None) -> float: + nonlocal called + called = True + assert row.metadata == {"suite": "refunds", "priority": 1} + assert output == "done" + return 0.4 + + result = await Scorer(name="async-check", fn=score).execute(scorer_row(), "done") + + assert called is True + assert result.status == "COMPLETE" + assert result.score == 0.4 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("raw_score", "normalized"), + [(True, 1.0), (False, 0.0), (0, 0.0), (1, 1.0), (0.625, 0.625)], +) +async def test_bool_and_numeric_scores_are_normalized( + raw_score: ScoreValue, normalized: float +) -> None: + def score(row: ScorerRow, output: str | None) -> ScoreValue: + del row, output + return raw_score + + result = await Scorer(name="normalized", fn=score).execute(scorer_row(), "ok") + + assert result.status == "COMPLETE" + assert result.score == normalized + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "raw_score", [None, "1", -0.01, 1.01, float("nan"), float("inf")] +) +async def test_invalid_scorer_results_are_clear_error_results( + raw_score: object, +) -> None: + def score(row: ScorerRow, output: str | None) -> ScoreValue: + del row, output + return cast(ScoreValue, raw_score) + + result = await Scorer(name="bad-result", fn=score).execute(scorer_row(), "ok") + + assert result.scorer_name == "bad-result" + assert result.row_index == 7 + assert result.status == "ERROR" + assert result.score is None + assert result.error is not None + assert result.error.code == "invalid_score" + assert "between 0 and 1" in result.error.message + assert result.started_at.tzinfo is not None + assert result.evaluated_at >= result.started_at + assert result.latency_ms >= 0 + + +@pytest.mark.asyncio +async def test_scorer_exception_is_preserved_as_error_result() -> None: + def score(row: ScorerRow, output: str | None) -> float: + del row, output + raise RuntimeError("database unavailable") + + result = await Scorer(name="db-check", fn=score).execute(scorer_row(), "ok") + + assert result.status == "ERROR" + assert result.score is None + assert result.error is not None + assert result.error.code == "scorer_error" + assert result.error.exception_type == "RuntimeError" + assert result.error.message == "database unavailable" + + +def test_scorer_and_row_dtos_validate_strictly() -> None: + with pytest.raises(ValueError, match="name"): + Scorer(name=" ", fn=lambda row, output: True) + with pytest.raises(ValueError, match="between 0 and 1"): + Scorer(name="check", fn=lambda row, output: True, threshold=1.1) + with pytest.raises(ValueError, match="row_index"): + ScorerRow( + row_index=-1, + input=None, + expected_output=None, + variables={}, + metadata=None, + ) + with pytest.raises(TypeError, match="metadata keys"): + ScorerRow( + row_index=0, + input=None, + expected_output=None, + variables={}, + metadata=cast(dict[str, Any], {1: "invalid"}), + ) diff --git a/packages/client/tests/test_evaluations.py b/packages/client/tests/test_evaluations.py new file mode 100644 index 00000000..c9504298 --- /dev/null +++ b/packages/client/tests/test_evaluations.py @@ -0,0 +1,240 @@ +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from launchdarkly_ai_server.evaluations import ( + DEFAULT_BASE_URI, + EvalRunResult, + EvaluationsError, + HttpResponse, + LDApiClient, + LDApiError, + RunSummary, + Usage, + init_evaluations, +) + + +class RecordingTransport: + """Mocked LD API — records requests and replays canned responses.""" + + def __init__(self, responses: list[HttpResponse] | None = None) -> None: + self.requests: list[dict[str, Any]] = [] + self.responses = responses or [HttpResponse(status=200, body="{}")] + + def __call__( + self, + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + self.requests.append( + { + "method": method, + "url": url, + "headers": headers, + "body": json.loads(body) if body else None, + "timeout": timeout, + } + ) + index = min(len(self.requests) - 1, len(self.responses) - 1) + return self.responses[index] + + +def failing_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, +) -> HttpResponse: + raise AssertionError("no network I/O expected") + + +def test_init_resolves_credentials_from_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token-from-env") + monkeypatch.setenv("LD_SDK_KEY", "sdk-key-from-env") + + evals = init_evaluations(transport=RecordingTransport()) + + assert evals.api.api_token == "api-token-from-env" + assert evals.sdk_key == "sdk-key-from-env" + assert evals.api.base_uri == DEFAULT_BASE_URI + + +def test_init_prefers_explicit_credentials(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token-from-env") + monkeypatch.setenv("LD_SDK_KEY", "sdk-key-from-env") + + evals = init_evaluations( + api_token="explicit-token", + sdk_key="explicit-sdk-key", + transport=RecordingTransport(), + ) + + assert evals.api.api_token == "explicit-token" + assert evals.sdk_key == "explicit-sdk-key" + + +def test_missing_api_token_raises_before_network_io( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("LD_API_TOKEN", raising=False) + monkeypatch.setenv("LD_SDK_KEY", "sdk-key") + + with pytest.raises(EvaluationsError, match="LD_API_TOKEN"): + init_evaluations(transport=failing_transport) + + +def test_blank_api_token_env_is_treated_as_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LD_API_TOKEN", " ") + + with pytest.raises(EvaluationsError): + init_evaluations(transport=failing_transport) + + +def test_missing_sdk_key_is_allowed(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token") + monkeypatch.delenv("LD_SDK_KEY", raising=False) + + evals = init_evaluations(transport=RecordingTransport()) + + assert evals.sdk_key is None + + +def test_base_uri_override_isolated_from_sdk_delivery_uri( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token") + monkeypatch.setenv("LD_API_BASE_URI", "https://api.staging.example.com/") + monkeypatch.setenv("LD_BASE_URI", "https://relay.example.com/") + + from_env = init_evaluations(transport=RecordingTransport()) + explicit = init_evaluations( + base_uri="https://other.example.com", transport=RecordingTransport() + ) + + assert from_env.api.base_uri == "https://api.staging.example.com" + assert explicit.api.base_uri == "https://other.example.com" + + +def test_requests_carry_token_auth_and_json_body() -> None: + transport = RecordingTransport([HttpResponse(status=201, body='{"key": "run-1"}')]) + client = LDApiClient(api_token="api-token", transport=transport) + + result = client.post("projects/proj/evaluations", body={"key": "support-qa"}) + + assert result == {"key": "run-1"} + request = transport.requests[0] + assert request["method"] == "POST" + assert request["url"] == f"{DEFAULT_BASE_URI}/api/v2/projects/proj/evaluations" + assert request["headers"]["Authorization"] == "api-token" + assert request["headers"]["Content-Type"] == "application/json" + assert request["body"] == {"key": "support-qa"} + + +def test_get_encodes_query_params_and_omits_none() -> None: + transport = RecordingTransport([HttpResponse(status=200, body='{"items": []}')]) + client = LDApiClient( + api_token="api-token", base_uri="https://ld.example.com", transport=transport + ) + + client.get("projects/proj/datasets/golden", params={"limit": 50, "offset": None}) + + request = transport.requests[0] + assert ( + request["url"] + == "https://ld.example.com/api/v2/projects/proj/datasets/golden?limit=50" + ) + assert "Content-Type" not in request["headers"] + + +def test_rate_limit_retries_and_honors_retry_after() -> None: + transport = RecordingTransport( + [ + HttpResponse( + status=429, + body='{"message": "slow down"}', + headers={"retry-after": "2"}, + ), + HttpResponse(status=200, body='{"items": []}'), + ] + ) + sleeps: list[float] = [] + client = LDApiClient( + api_token="api-token", + transport=transport, + max_retries=1, + sleep=sleeps.append, + random_value=lambda: 0.0, + ) + + assert client.get("projects/proj/datasets") == {"items": []} + assert len(transport.requests) == 2 + assert sleeps == [2.0] + + +def test_forbidden_response_is_not_retried() -> None: + transport = RecordingTransport( + [HttpResponse(status=403, body='{"message": "forbidden"}')] + ) + client = LDApiClient(api_token="api-token", transport=transport, max_retries=3) + + with pytest.raises(LDApiError) as excinfo: + client.get("projects/proj/evaluations") + + assert excinfo.value.status == 403 + assert len(transport.requests) == 1 + + +def test_error_response_raises_ld_api_error() -> None: + transport = RecordingTransport( + [HttpResponse(status=404, body='{"message": "nope"}')] + ) + client = LDApiClient(api_token="api-token", transport=transport) + + with pytest.raises(LDApiError) as excinfo: + client.get("projects/proj/ai-tools/missing") + + assert excinfo.value.status == 404 + assert excinfo.value.path == "projects/proj/ai-tools/missing" + + +def test_empty_response_body_is_none() -> None: + transport = RecordingTransport([HttpResponse(status=204, body="")]) + client = LDApiClient(api_token="api-token", transport=transport) + + assert client.post("projects/proj/evaluations/support-qa/runs") is None + + +def test_usage_matches_ingest_wire_shape() -> None: + usage = Usage(input_tokens=812, output_tokens=96) + + assert usage.to_wire() == {"input_tokens": 812, "output_tokens": 96} + assert Usage.from_wire({"input_tokens": 1, "output_tokens": 2}) == Usage(1, 2) + assert Usage.from_wire({}) == Usage(0, 0) + + +def test_run_summary_and_result() -> None: + summary = RunSummary.from_wire( + {"total_rows": 500, "passed_rows": 498, "failed_rows": 1, "error_rows": 1} + ) + result = EvalRunResult( + passed=False, + url="https://app.launchdarkly.com/run", + run_id="run-1", + summary=summary, + ) + + assert summary.total_rows == 500 + assert summary.error_rows == 1 + assert RunSummary.from_wire(None) == RunSummary() + assert result.passed is False + assert result.run_id == "run-1" diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py new file mode 100644 index 00000000..f065d0f6 --- /dev/null +++ b/packages/client/tests/test_evaluations_run.py @@ -0,0 +1,740 @@ +from __future__ import annotations + +import json +from collections.abc import Callable, Mapping, Sequence +from typing import Any, cast +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from launchdarkly_ai_server.evaluations import ( + EvaluationsError, + HttpResponse, + Judge, + JudgeEvaluationResult, + JudgeIdentity, + JudgeReference, + LaunchDarklyJudgeEvaluation, + Scorer, + ScorerResult, + ScorerRow, + init_evaluations, +) +from launchdarkly_ai_server.types import ProviderHandler +from launchdarkly_ai_server.utils import create_handler + + +class SequencedTransport: + """Records requests and returns one response for each expected request.""" + + def __init__(self, responses: list[HttpResponse]) -> None: + self.responses = responses + self.requests: list[dict[str, Any]] = [] + + def __call__( + self, + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + index = len(self.requests) + self.requests.append( + { + "method": method, + "url": url, + "headers": headers, + "body": json.loads(body) if body else None, + "timeout": timeout, + } + ) + if index >= len(self.responses): + raise AssertionError(f"unexpected request: {method} {url}") + return self.responses[index] + + +def response(status: int, body: dict[str, Any] | None = None) -> HttpResponse: + return HttpResponse( + status=status, body=json.dumps(body) if body is not None else "" + ) + + +def dataset_page( + items: list[dict[str, Any]], total: int, next_href: str | None = None +) -> dict[str, Any]: + links: dict[str, Any] = {"self": {"href": "https://api.test/current"}} + if next_href: + links["next"] = {"href": next_href} + return {"items": items, "totalCount": total, "_links": links} + + +async def successful_handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], +) -> dict[str, Any]: + assert config["provider"] == {"name": "OpenAI"} + assert config["model"] == { + "name": "gpt-4o", + "parameters": {"temperature": 0.2}, + } + assert config["tools"]["lookup_order"] == { + "description": "Look up an order", + "parameters": {"type": "object"}, + } + assert "lookup_order" in tool_handlers + assert variables["input"] == user_input + return { + "output": f"generated: {user_input}", + "usage": {"input_tokens": 10, "output_tokens": 4}, + } + + +def lookup_order(order_id: str) -> str: + return order_id + + +@pytest.mark.asyncio +async def test_run_calls_private_operations_in_order_and_returns_server_verdict( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("LD_SDK_KEY", raising=False) + transport = SequencedTransport( + [ + response( + 200, + { + "key": "lookup_order", + "version": 7, + "description": "Look up an order", + "schema": {"type": "object"}, + }, + ), + response( + 200, + { + "id": "33333333-3333-3333-3333-333333333333", + "name": "golden", + }, + ), + response( + 200, + dataset_page( + [ + { + "rowIndex": 4, + "input": "Order {{order_id}}", + "expectedOutput": "Found {{order_id}}", + "variables": {"order_id": "A19"}, + "metadata": {"suite": "orders"}, + } + ], + total=2, + next_href="https://api.test/api/v2/projects/proj/datasets/key/golden/preview?limit=1&offset=1", + ), + ), + response( + 200, + dataset_page( + [ + { + "rowIndex": 9, + "input": "Order {{order_id}}", + "expectedOutput": None, + "variables": {"order_id": "B20"}, + "metadata": None, + } + ], + total=2, + ), + ), + response( + 201, + { + "id": "11111111-1111-1111-1111-111111111111", + "name": "support-qa-unique", + "version": 1, + }, + ), + response( + 201, + { + "id": "22222222-2222-2222-2222-222222222222", + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "source": "client", + "state": "PENDING", + "createdAt": 1, + }, + ), + response(202, {}), + response( + 200, + { + "id": "22222222-2222-2222-2222-222222222222", + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "source": "client", + "state": "COMPLETE", + "verdict": "passed", + "createdAt": 1, + }, + ), + response( + 200, + { + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "evaluationRunId": "22222222-2222-2222-2222-222222222222", + "statusCounts": { + "total": 2, + "passed": 2, + "failed": 0, + "error": 0, + "pending": 0, + }, + "createdAt": 1, + }, + ), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + assert evals.sdk_key is None + + result = await evals.run( + project_key="proj", + key="support-qa-unique", + dataset="golden", + handler=successful_handler, + tools={"lookup_order": lookup_order}, + generation={ + "provider": "OpenAI", + "model": "gpt-4o", + "parameters": {"temperature": 0.2}, + "instructions": "Help the user.", + }, + concurrency=2, + ) + + assert result.passed is True + assert result.run_id == "22222222-2222-2222-2222-222222222222" + assert result.summary.total_rows == 2 + + assert [request["method"] for request in transport.requests] == [ + "GET", + "GET", + "GET", + "GET", + "POST", + "POST", + "POST", + "GET", + "GET", + ] + assert transport.requests[0]["url"].endswith( + "/api/v2/projects/proj/ai-tools/lookup_order" + ) + assert transport.requests[1]["url"].endswith( + "/api/v2/projects/proj/datasets/golden" + ) + assert "/projects/proj/datasets/golden/rows" in transport.requests[2]["url"] + assert "mode=all" in transport.requests[2]["url"] + assert transport.requests[4]["body"] == { + "name": "support-qa-unique", + "generationProvider": "OpenAI", + "generationModel": "gpt-4o", + "parameters": {"temperature": 0.2}, + "messages": [{"role": "system", "content": "Help the user."}], + "tools": [{"key": "lookup_order", "version": 7}], + } + assert transport.requests[5]["url"].endswith( + "/api/v2/projects/proj/evaluations/11111111-1111-1111-1111-111111111111/runs" + ) + assert transport.requests[5]["body"] == { + "source": "client", + "rowCount": 2, + "datasetId": "33333333-3333-3333-3333-333333333333", + } + + ingested = transport.requests[6]["body"]["results"] + assert [row["row_index"] for row in ingested] == [4, 9] + assert ingested[0]["input"] == "Order A19" + assert ingested[0]["expected_output"] == "Found A19" + assert ingested[0]["variables"]["input"] == "Order A19" + assert ingested[0]["variables"]["expected_output"] == "Found A19" + + +@pytest.mark.asyncio +async def test_disabled_batch_ingest_flag_skips_generation_result_ingestion( + monkeypatch: pytest.MonkeyPatch, +) -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [{"rowIndex": 3, "input": "hello", "variables": {}}], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "PENDING", + }, + ), + response( + 200, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "COMPLETE", + "verdict": "passed", + }, + ), + response(200, {"statusCounts": {"total": 1, "passed": 1}}), + ] + ) + client = MagicMock() + client.variation = AsyncMock(return_value=False) + + async def fake_init_client(options: dict[str, Any]) -> MagicMock: + assert options == {"sdkKey": "sdk-key"} + return client + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.init_client", fake_init_client + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert result.passed is True + assert not any( + request["url"].endswith("/generation-results") for request in transport.requests + ) + + +@pytest.mark.asyncio +async def test_run_executes_scorer_with_generation_and_complete_row_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("LD_SDK_KEY", raising=False) + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": 8, + "input": "Order {{order_id}}", + "expectedOutput": "Found {{order_id}}", + "variables": {"order_id": "A19"}, + "metadata": {"suite": "orders"}, + } + ], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "PENDING", + }, + ), + response(202, {}), + response( + 200, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "COMPLETE", + "verdict": "passed", + }, + ), + response(200, {"statusCounts": {"total": 1, "passed": 1}}), + ] + ) + received: dict[str, Any] = {} + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "Found A19"} + + def score(row: ScorerRow, output: str | None) -> bool: + received.update( + row_index=row.row_index, + input=row.input, + expected_output=row.expected_output, + variables=dict(row.variables), + metadata=dict(row.metadata or {}), + output=output, + ) + return True + + result = await init_evaluations(api_token="token", transport=transport).run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + judges=[Scorer(name="exact-match", fn=score)], + ) + + assert received == { + "row_index": 8, + "input": "Order A19", + "expected_output": "Found A19", + "variables": { + "order_id": "A19", + "input": "Order A19", + "expected_output": "Found A19", + }, + "metadata": {"suite": "orders"}, + "output": "Found A19", + } + assert len(result.evaluation_results) == 1 + scorer_result = result.evaluation_results[0] + assert isinstance(scorer_result, ScorerResult) + assert scorer_result.score == 1.0 + + +@pytest.mark.asyncio +async def test_run_resolves_and_executes_launchdarkly_judge( + monkeypatch: pytest.MonkeyPatch, +) -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": 5, + "input": "Question", + "expectedOutput": "Expected", + "variables": {"account": "enterprise"}, + "metadata": {"suite": "judge"}, + } + ], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "PENDING", + }, + ), + response(202, {}), + response( + 200, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "COMPLETE", + "verdict": "passed", + }, + ), + response(200, {"statusCounts": {"total": 1, "passed": 1}}), + ] + ) + received: dict[str, Any] = {} + + async def generation_handler(*args: object) -> dict[str, Any]: + return {"output": "Generated answer"} + + async def judge_handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]] | None, + variables: dict[str, Any] | None, + history: list[dict[str, Any]] | None, + ) -> dict[str, Any]: + del config, tool_handlers, history + received.update(user_input=user_input, variables=variables) + return {"output": '{"score": 0.9, "reasoning": "good"}'} + + reference = Judge(key="security-judge") + resolved = LaunchDarklyJudgeEvaluation( + reference=reference, + config={ + "provider": {"name": "OpenAI"}, + "model": {"name": "judge-model"}, + "instructions": "Judge the response", + }, + identity=JudgeIdentity( + key="security-judge", + variation_key="variation-key", + version=3, + provider="OpenAI", + model="judge-model", + mode="messages", + ), + handler=create_handler(("OpenAI", "messages"), judge_handler), + collapse_messages=False, + ) + generation = create_handler(("OpenAI", "messages"), generation_handler) + flag_client = MagicMock() + flag_client.variation = AsyncMock(return_value=True) + + async def fake_init_client(options: dict[str, Any]) -> MagicMock: + assert options == {"sdkKey": "sdk-key"} + return flag_client + + async def fake_resolve( + references: Sequence[JudgeReference], + handlers: Sequence[ProviderHandler], + *, + sdk_key: str | None, + ) -> list[LaunchDarklyJudgeEvaluation]: + assert references == [reference] + assert handlers == [generation] + assert sdk_key == "sdk-key" + return [resolved] + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.init_client", fake_init_client + ) + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.resolve_launchdarkly_judges", + fake_resolve, + ) + + result = await init_evaluations( + api_token="token", sdk_key="sdk-key", transport=transport + ).run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=generation, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + judges=[reference], + ) + + judge_result = result.evaluation_results[0] + assert isinstance(judge_result, JudgeEvaluationResult) + assert judge_result.score == 0.9 + assert received["user_input"] == "Generated answer" + judge_variables = cast(Mapping[str, Any], received["variables"]) + assert judge_variables["row_index"] == 5 + assert judge_variables["input"] == "Question" + assert judge_variables["expected_output"] == "Expected" + assert judge_variables["account"] == "enterprise" + assert judge_variables["metadata"] == {"suite": "judge"} + assert judge_variables["response_to_evaluate"] == "Generated answer" + + +@pytest.mark.asyncio +async def test_run_rejects_untyped_judges_before_network_io() -> None: + transport = SequencedTransport([]) + evals = init_evaluations(api_token="token", transport=transport) + + with pytest.raises(EvaluationsError, match="typed JudgeReference or Scorer"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=successful_handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + judges=["accuracy"], # type: ignore[list-item] + ) + + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_run_rejects_instructions_and_messages_before_network_io() -> None: + transport = SequencedTransport([]) + evals = init_evaluations(api_token="token", transport=transport) + + with pytest.raises(EvaluationsError, match=r"instructions.*messages"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=successful_handler, + generation={ + "provider": "OpenAI", + "model": "gpt-4o", + "instructions": "System prompt", + "messages": [{"role": "user", "content": "{{input}}"}], + }, + ) + + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_missing_tool_aborts_before_any_mutating_request() -> None: + transport = SequencedTransport( + [response(404, {"code": "not_found", "message": "not found"})] + ) + evals = init_evaluations(api_token="token", transport=transport) + + with pytest.raises(EvaluationsError, match="missing_tool"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=successful_handler, + tools={"missing_tool": lookup_order}, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert [request["method"] for request in transport.requests] == ["GET"] + + +@pytest.mark.asyncio +async def test_empty_dataset_fails_before_evaluation_or_run_creation() -> None: + transport = SequencedTransport( + [ + response( + 200, + { + "id": "33333333-3333-3333-3333-333333333333", + "name": "golden", + }, + ), + response(200, dataset_page([], total=0)), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + + with pytest.raises(EvaluationsError, match="empty"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=successful_handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert [request["method"] for request in transport.requests] == ["GET", "GET"] + + +@pytest.mark.asyncio +async def test_handler_error_is_ingested_and_other_rows_continue() -> None: + calls: list[str | None] = [] + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + calls.append(user_input) + if user_input == "bad": + raise RuntimeError("provider failed") + return {"output": "ok"} + + transport = SequencedTransport( + [ + response( + 200, + { + "id": "33333333-3333-3333-3333-333333333333", + "name": "golden", + }, + ), + response( + 200, + dataset_page( + [ + {"rowIndex": 0, "input": "bad", "variables": {}}, + {"rowIndex": 1, "input": "good", "variables": {}}, + ], + total=2, + ), + ), + response( + 201, + { + "id": "11111111-1111-1111-1111-111111111111", + "name": "eval-key", + "version": 1, + }, + ), + response( + 201, + { + "id": "22222222-2222-2222-2222-222222222222", + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "source": "client", + "state": "PENDING", + "createdAt": 1, + }, + ), + response(202, {}), + response( + 200, + { + "id": "22222222-2222-2222-2222-222222222222", + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "source": "client", + "state": "COMPLETE", + "verdict": "failed", + "createdAt": 1, + }, + ), + response( + 200, + { + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "evaluationRunId": "22222222-2222-2222-2222-222222222222", + "statusCounts": { + "total": 2, + "passed": 1, + "failed": 0, + "error": 1, + "pending": 0, + }, + "createdAt": 1, + }, + ), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert set(calls) == {"bad", "good"} + assert len(calls) == 2 + assert result.passed is False + rows = transport.requests[4]["body"]["results"] + assert {row["status"] for row in rows} == {"COMPLETE", "ERROR"} + error_row = next(row for row in rows if row["status"] == "ERROR") + assert error_row["row_index"] == 0 + assert "provider failed" in error_row["error"]["message"] diff --git a/uv.lock b/uv.lock index de575a71..fa21dbca 100644 --- a/uv.lock +++ b/uv.lock @@ -790,7 +790,7 @@ wheels = [ [[package]] name = "launchdarkly-ai-claude-agents" -version = "0.1.1" +version = "0.1.4" source = { editable = "packages/claude-agents" } dependencies = [ { name = "anthropic" }, @@ -809,7 +809,7 @@ requires-dist = [ [[package]] name = "launchdarkly-ai-claude-messages" -version = "0.1.1" +version = "0.1.4" source = { editable = "packages/claude-messages" } dependencies = [ { name = "anthropic" }, @@ -826,7 +826,7 @@ requires-dist = [ [[package]] name = "launchdarkly-ai-langchain-agents" -version = "0.1.1" +version = "0.1.4" source = { editable = "packages/langchain-agents" } dependencies = [ { name = "langchain-core" }, @@ -845,7 +845,7 @@ requires-dist = [ [[package]] name = "launchdarkly-ai-langchain-messages" -version = "0.1.1" +version = "0.1.4" source = { editable = "packages/langchain-messages" } dependencies = [ { name = "langchain-core" }, @@ -862,7 +862,7 @@ requires-dist = [ [[package]] name = "launchdarkly-ai-openai-agents" -version = "0.1.1" +version = "0.1.4" source = { editable = "packages/openai-agents" } dependencies = [ { name = "launchdarkly-ai-server" }, @@ -881,7 +881,7 @@ requires-dist = [ [[package]] name = "launchdarkly-ai-openai-messages" -version = "0.1.1" +version = "0.1.4" source = { editable = "packages/openai-messages" } dependencies = [ { name = "launchdarkly-ai-server" }, @@ -898,7 +898,7 @@ requires-dist = [ [[package]] name = "launchdarkly-ai-python" -version = "0.1.1" +version = "0.1.3" source = { editable = "packages/ai" } dependencies = [ { name = "launchdarkly-ai-server" }, @@ -918,10 +918,11 @@ provides-extras = ["otel"] [[package]] name = "launchdarkly-ai-server" -version = "0.1.1" +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"]