feat(evaluations): run generation-only evaluations from code

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.

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.
 

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:

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",

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",
+]

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)

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