diff --git a/packages/ai/README.md b/packages/ai/README.md index e35135b..e1539fd 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -52,6 +52,25 @@ 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 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"}, +) +``` + +`LD_API_TOKEN` is required. 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 8691bab..08a0917 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -39,6 +39,44 @@ 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 generation-only evaluations harness reads an LD-hosted dataset, creates a new evaluation and client-source run, invokes your handler once per row, 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 init_evaluations + + +async def main() -> int: + evals = init_evaluations() # LD_API_TOKEN required; LD_SDK_KEY optional + 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.", + }, + ) + print(result.url, result.summary) + 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. + +When `LD_SDK_KEY` is configured, generation-result publishing is controlled by the `enable-batch-ingest-in-evals-from-code` flag evaluated for the project. Results are uploaded only when the variation is exactly `true`; false, malformed, or failed evaluations skip publishing. Without an SDK key, publishing retains its existing behavior. 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 61d99cf..a28753d 100644 --- a/packages/client/agents.md +++ b/packages/client/agents.md @@ -31,6 +31,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, and generation-only `EvaluationsModule.run()` orchestration | | `src/launchdarkly_ai_server/__init__.py` | Public barrel — the only surface handler packages import from | --- @@ -68,7 +69,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`). @@ -125,6 +126,14 @@ 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 runs; when set it enables the normal handler observability path and gates generation-result ingest on the `enable-batch-ingest-in-evals-from-code` flag (only a strictly `true` variation publishes; false, default, malformed, or evaluation-error results skip publish safely). Without an SDK key the gate cannot be evaluated and ingest runs unconditionally. + +`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, batches generation ingest when the gate permits, and trusts only the server's stored verdict. + +--- + ## Conversation grouping LaunchDarkly's conversation view groups spans on `gen_ai.conversation.id`. Bind a caller-supplied id around any `invoke()` / `stream()` / `graph().invoke()` call: diff --git a/packages/client/src/launchdarkly_ai_server/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py index 80b959b..9a07b05 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -21,6 +21,14 @@ conversation_id, set_conversation_id_if_absent, ) +from .evaluations import ( + EvalRunResult, + EvaluationsError, + EvaluationsModule, + GenerationConfig, + RunSummary, + init_evaluations, +) from .graph import GraphInstance, graph, resolve_graph from .judges import build_judge_tasks, run_judge, run_judges from .lifecycle import ( @@ -155,6 +163,13 @@ "text_message", "to_semconv_finish_reason", "VariationMeta", + # evaluations + "EvalRunResult", + "EvaluationsError", + "EvaluationsModule", + "GenerationConfig", + "RunSummary", + "init_evaluations", # 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 0000000..6516f4a --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -0,0 +1,29 @@ +"""Run LaunchDarkly evaluations from your own environment.""" + +from .api import ( + DEFAULT_BASE_URI, + EvaluationsError, + HttpResponse, + LDApiClient, + LDApiError, + Transport, + urllib_transport, +) +from .module import EvaluationsModule, init_evaluations +from .types import EvalRunResult, GenerationConfig, RunSummary, Usage + +__all__ = [ + "DEFAULT_BASE_URI", + "EvalRunResult", + "EvaluationsError", + "EvaluationsModule", + "GenerationConfig", + "HttpResponse", + "LDApiClient", + "LDApiError", + "RunSummary", + "Transport", + "Usage", + "init_evaluations", + "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 0000000..957a092 --- /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 0000000..ee1a138 --- /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/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py new file mode 100644 index 0000000..58f48f1 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +import logging +import os +from collections.abc import Mapping + +from ..lifecycle import init_client +from .api import ( + DEFAULT_BASE_URI, + EvaluationsError, + LDApiClient, + Transport, + urllib_transport, +) +from .flags import is_generation_result_batch_ingest_enabled +from .runner import EvalHandler, EvaluationsRunner, ToolImplementation, _segment +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, + concurrency: int = 10, + timeout: float = 300.0, + ) -> EvalRunResult: + """ + Create and run a generation-only evaluation in the caller's process. + + 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 {}) + 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 + ) + + # 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 = await self._runner._run_rows( + rows, + handler, + config, + run_tools, + concurrency, + ) + 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, + ) + + @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 0000000..0da6814 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -0,0 +1,469 @@ +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 .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 + + +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.""" + + 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, + ) -> list[dict[str, Any]]: + controller = ConcurrencyController(concurrency) + + async def invoke(row: DatasetRow) -> dict[str, Any]: + 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) + controller.record_success(config["provider"]["name"]) + return payload + 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() + + return list(await asyncio.gather(*(invoke(row) for row in rows))) + + 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/types.py b/packages/client/src/launchdarkly_ai_server/evaluations/types.py new file mode 100644 index 0000000..dda010a --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/types.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any, TypedDict + + +@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 verdict of an evaluation run, as computed and stored by LaunchDarkly.""" + + passed: bool + url: str + run_id: str + summary: RunSummary diff --git a/packages/client/tests/test_evaluation_flags.py b/packages/client/tests/test_evaluation_flags.py new file mode 100644 index 0000000..8533cbf --- /dev/null +++ b/packages/client/tests/test_evaluation_flags.py @@ -0,0 +1,58 @@ +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 +@pytest.mark.parametrize("malformed_value", [None, 1, "true", {}]) +async def test_malformed_flag_disables_generation_result_batch_ingest( + malformed_value: object, +) -> None: + client = MagicMock() + client.variation = AsyncMock(return_value=malformed_value) + + 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_evaluations.py b/packages/client/tests/test_evaluations.py new file mode 100644 index 0000000..c950429 --- /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 0000000..828563e --- /dev/null +++ b/packages/client/tests/test_evaluations_run.py @@ -0,0 +1,524 @@ +from __future__ import annotations + +import json +from collections.abc import Callable +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from launchdarkly_ai_server.evaluations import ( + EvaluationsError, + HttpResponse, + init_evaluations, +) + + +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) + init_client = AsyncMock() + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.init_client", init_client + ) + 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" + init_client.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("flag_value", "expected_ingest"), + [ + pytest.param(True, True, id="enabled"), + pytest.param(False, False, id="disabled-default"), + pytest.param("true", False, id="malformed"), + pytest.param( + RuntimeError("delivery unavailable"), False, id="evaluation-error" + ), + ], +) +async def test_batch_ingest_flag_controls_generation_result_publishing( + monkeypatch: pytest.MonkeyPatch, + flag_value: object, + expected_ingest: bool, +) -> None: + responses = [ + 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", + }, + ), + ] + if expected_ingest: + responses.append(response(202, {})) + responses.extend( + [ + response( + 200, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "COMPLETE", + "verdict": "passed", + }, + ), + response(200, {"statusCounts": {"total": 1, "passed": 1}}), + ] + ) + transport = SequencedTransport(responses) + client = MagicMock() + if isinstance(flag_value, Exception): + client.variation = AsyncMock(side_effect=flag_value) + else: + client.variation = AsyncMock(return_value=flag_value) + + 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 ( + any( + request["url"].endswith("/generation-results") + for request in transport.requests + ) + is expected_ingest + ) + + +@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 de575a7..7d93a3c 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,7 +918,7 @@ provides-extras = ["otel"] [[package]] name = "launchdarkly-ai-server" -version = "0.1.1" +version = "0.1.3" source = { editable = "packages/client" } dependencies = [ { name = "opentelemetry-api" },