From 864867533d6b65537638429f0a724ff93ab9dba8 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 14 Oct 2026 21:33:48 +0000 Subject: [PATCH 1/8] feat(evaluations): export offline evaluation spans through the SDK exporter Stamp the shared UUID5 conversation id and evaluation-run attributes on handler spans, and record a verdict-free gen_ai.evaluation.result on an evaluate span so a plain function still exports through the SDK-key provider init_client already registered. Co-authored-by: christopher.schmitz --- .../launchdarkly_ai_server/conversation.py | 86 +++- .../evaluations/module.py | 30 +- .../evaluations/runner.py | 99 +++- .../src/launchdarkly_ai_server/lifecycle.py | 21 + packages/client/tests/test_conversation.py | 55 +++ packages/client/tests/test_evaluations_run.py | 1 + .../client/tests/test_evaluations_spans.py | 439 ++++++++++++++++++ 7 files changed, 706 insertions(+), 25 deletions(-) create mode 100644 packages/client/tests/test_evaluations_spans.py diff --git a/packages/client/src/launchdarkly_ai_server/conversation.py b/packages/client/src/launchdarkly_ai_server/conversation.py index d3af0e9e..8fc51106 100644 --- a/packages/client/src/launchdarkly_ai_server/conversation.py +++ b/packages/client/src/launchdarkly_ai_server/conversation.py @@ -6,6 +6,7 @@ from __future__ import annotations +import uuid from collections.abc import AsyncGenerator, AsyncIterator, Callable, Iterator from contextlib import aclosing, asynccontextmanager, contextmanager from dataclasses import dataclass @@ -25,9 +26,16 @@ _SpanProcessorBase = object GEN_AI_CONVERSATION_ID = "gen_ai.conversation.id" +LD_EVALUATION_RUN_ID = "launchdarkly.evaluation_run.id" +LD_EVALUATION_RUN_ROW_INDEX = "launchdarkly.evaluation_run.row_index" + +# Same namespace ai-evaluator uses for conversationId on a result row. +# Changing it re-keys every offline evaluation conversation. +_EVALUATION_CONVERSATION_NAMESPACE = uuid.UUID("6f0c2a4e-3b6d-4d0e-9c1a-2e6b7f8a9d01") _CONV_KEY = otel_context.create_key("launchdarkly.gen_ai.conversation.id") _EVAL_KEY = otel_context.create_key("launchdarkly.judge.evaluation") +_EVAL_ROW_KEY = otel_context.create_key("launchdarkly.evaluation_run.row") @dataclass @@ -81,6 +89,72 @@ def _conversation_id_from(ctx: otel_context.Context | None) -> str | None: return value if isinstance(value, str) and value else None +def conversation_id_for_evaluation_row(run_id: str, row_index: int | None) -> str: + """UUID5 of ``{run_id}:{row_index}`` in the ai-evaluator namespace. + + Row index ``0`` is the suffix ``:0``. A missing index is row 0, matching + the workers that persist a missing index that way. + """ + index = 0 if row_index is None else int(row_index) + return str(uuid.uuid5(_EVALUATION_CONVERSATION_NAMESPACE, f"{run_id}:{index}")) + + +def _evaluation_row_attributes( + run_id: str, row_index: int | None +) -> dict[str, str | int]: + index = 0 if row_index is None else int(row_index) + return { + LD_EVALUATION_RUN_ID: str(run_id), + LD_EVALUATION_RUN_ROW_INDEX: index, + } + + +def _evaluation_row_from(ctx: otel_context.Context | None) -> dict[str, Any] | None: + value = otel_context.get_value(_EVAL_ROW_KEY, ctx) if ctx is not None else None + if value is None: + value = otel_context.get_value(_EVAL_ROW_KEY) + if not isinstance(value, dict) or not value: + return None + return value + + +@contextmanager +def bind_evaluation_row(run_id: str, row_index: int | None) -> Iterator[None]: + """Bind one offline-evaluation row onto SDK spans started in this block. + + Sets ``gen_ai.conversation.id`` to :func:`conversation_id_for_evaluation_row` + and ``launchdarkly.evaluation_run.id`` / ``.row_index``. Does not set + ``launchdarkly.run.id``; that key is the AI Config tracking id. + + An outer :func:`conversation_id` is shadowed until the block exits. The + evaluation row id is the join key for this row. + """ + index = 0 if row_index is None else int(row_index) + ctx = otel_context.set_value( + _CONV_KEY, conversation_id_for_evaluation_row(run_id, index) + ) + ctx = otel_context.set_value( + _EVAL_ROW_KEY, _evaluation_row_attributes(run_id, index), ctx + ) + token = otel_context.attach(ctx) + try: + yield + finally: + otel_context.detach(token) + + +def record_evaluation_result( + span: Any, name: str, score: float, explanation: str | None = None +) -> None: + """Write ``gen_ai.evaluation.result`` with no verdict and no score label. + + ``explanation`` is the criterion reason. Offline evaluation always records + it when the reason is non-empty: that text already travels on the criterion + event. This does not add a prompt, a completion, or a tool payload. + """ + _record_evaluation(span, name, score, explanation) + + def _record_evaluation( span: Any, name: str, score: float, explanation: str | None = None ) -> None: @@ -256,7 +330,7 @@ def record(score: float, explanation: str | None = None) -> None: class ConversationIdSpanProcessor(_SpanProcessorBase): - """Stamps ``gen_ai.conversation.id`` write-if-absent; delays judge ``invoke_agent`` end. + """Stamps conversation and evaluation-row attributes; delays judge ``invoke_agent`` end. Structurally a ``SpanProcessor``; the base is only real to a type checker so that an api-only install (no ``[otel]`` extra) still imports this module. @@ -275,6 +349,16 @@ def on_start( if conv and _is_launchdarkly_span(span): set_conversation_id_if_absent(span, conv) + row = _evaluation_row_from(ctx) + if row and _is_launchdarkly_span(span): + for key, value in row.items(): + if not isinstance(key, str) or _read_attribute(span, key) is not None: + continue + if isinstance(value, str): + span.set_attribute(key, value) + elif isinstance(value, int) and not isinstance(value, bool): + span.set_attribute(key, value) + capture = otel_context.get_value(_EVAL_KEY, ctx) if capture is None: capture = otel_context.get_value(_EVAL_KEY) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 072df9b0..5be8831f 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -10,7 +10,7 @@ from collections.abc import Mapping, Sequence from typing import Any, cast -from ..lifecycle import get_client, init_client +from ..lifecycle import _flush_tracer_provider, get_client, init_client from .api import ( DEFAULT_BASE_URI, EvaluationsError, @@ -426,14 +426,15 @@ async def run( ) raise config = self._runner._build_handler_config(generation, run_tools) - results = await self._runner._run_rows( - dataset_rows, - handler, - config, - run_tool_handlers, - concurrency, - ) try: + results = await self._runner._run_rows( + dataset_rows, + handler, + config, + run_tool_handlers, + concurrency, + evaluation_run_id=evaluation_run.id, + ) self._runner._emit_generation_events( client, project_key=self._project_key, @@ -449,6 +450,7 @@ async def run( run_criteria, resolved_judges, concurrency, + evaluation_run_id=evaluation_run.id, ) self._runner._emit_evaluation_events( client, @@ -460,10 +462,14 @@ async def run( ) finally: # Generation results already queued on the SDK event buffer must - # reach LaunchDarkly even when the criteria phase fails. - flush_result = client.flush() - if inspect.isawaitable(flush_result): - await flush_result + # reach LaunchDarkly even when the criteria phase fails. Spans + # follow on the provider init_client already registered. + try: + flush_result = client.flush() + if inspect.isawaitable(flush_result): + await flush_result + finally: + _flush_tracer_provider() summary = await self._poll_summary_until_terminal( evaluation.id, evaluation_run.id, diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index b12bab91..5ee06c37 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -7,11 +7,23 @@ import logging import time import urllib -from collections.abc import Awaitable, Callable, Mapping, Sequence +from collections.abc import Awaitable, Callable, Iterator, Mapping, Sequence +from contextlib import contextmanager from dataclasses import dataclass from datetime import UTC, datetime from typing import Any, Literal +from opentelemetry import trace +from opentelemetry.trace import Status, StatusCode + +from ..conversation import ( + GEN_AI_CONVERSATION_ID, + LD_EVALUATION_RUN_ID, + LD_EVALUATION_RUN_ROW_INDEX, + bind_evaluation_row, + conversation_id_for_evaluation_row, + record_evaluation_result, +) from ..judge_scoring import ( FORMATTING_INSTRUCTIONS, build_message_history, @@ -242,6 +254,55 @@ def record_rate_limit( del provider, retry_after +@contextmanager +def _criterion_span(run_id: str, row_index: int, criterion_type: str) -> Iterator[Any]: + """Open ``evaluate {criterion}`` on the SDK tracer. + + Correlation attributes are set here as well as by the span processor, so a + plain function still exports them when the processor is the one + ``init_client`` registered. This does not build an exporter. + """ + tracer = trace.get_tracer("@launchdarkly/ai-server") + with tracer.start_as_current_span(f"evaluate {criterion_type}") as span: + span.set_attribute("gen_ai.operation.name", "evaluate") + span.set_attribute( + GEN_AI_CONVERSATION_ID, + conversation_id_for_evaluation_row(run_id, row_index), + ) + span.set_attribute(LD_EVALUATION_RUN_ID, run_id) + span.set_attribute(LD_EVALUATION_RUN_ROW_INDEX, row_index) + yield span + + +def _finish_criterion_span(span: Any, result: Mapping[str, Any]) -> None: + """Record ``gen_ai.evaluation.result`` with no verdict. + + An ERROR criterion sets span status ERROR and ``error.type``, and does not + add a result event. Nothing here writes ``launchdarkly.run.id``. + """ + if result.get("status") == "COMPLETE": + score = result.get("score") + if isinstance(score, bool) or not isinstance(score, int | float): + span.set_status(Status(StatusCode.ERROR)) + span.set_attribute("error.type", "invalid_score") + return + reason = result.get("reason") + explanation = reason if isinstance(reason, str) and reason else None + record_evaluation_result( + span, + str(result.get("criterion_type") or ""), + float(score), + explanation, + ) + return + span.set_status(Status(StatusCode.ERROR)) + error = result.get("error") + if isinstance(error, Mapping): + code = error.get("code") + if isinstance(code, str) and code: + span.set_attribute("error.type", code) + + class EvaluationsRunner: """Private API operations and orchestration used by EvaluationsModule.run().""" @@ -635,6 +696,8 @@ async def _run_rows( config: dict[str, Any], tool_handlers: dict[str, ToolImplementation], concurrency: int, + *, + evaluation_run_id: str, ) -> list[dict[str, Any]]: controller = ConcurrencyController(concurrency) @@ -647,9 +710,10 @@ async def invoke(row: DatasetRow) -> dict[str, Any]: started = datetime.now(UTC) started_clock = time.perf_counter() try: - result = await handler( - config, row.input, row_tool_handlers, dict(row.variables) - ) + with bind_evaluation_row(evaluation_run_id, row.row_index): + result = await handler( + config, row.input, row_tool_handlers, dict(row.variables) + ) if not isinstance(result, Mapping): raise TypeError("handler result must be a mapping") completed = datetime.now(UTC) @@ -1011,6 +1075,8 @@ async def _run_criteria_for_results( criteria: list[Criterion], resolved_judges: Mapping[str, JudgeExecution], concurrency: int, + *, + evaluation_run_id: str, ) -> list[dict[str, Any]]: """Run every (row, criterion) pair, bounded by the run's concurrency.""" controller = ConcurrencyController(concurrency) @@ -1020,14 +1086,23 @@ async def run_one( ) -> dict[str, Any]: await controller.acquire() try: - if isinstance(criterion, Scorer): - return await self._run_scorer_for_result(row, criterion) - return await self._run_ld_judge_for_result( - row, - tool_handlers, - criterion, - resolved_judges[criterion.key], - ) + row_index = row.get("row_index") + index = row_index if isinstance(row_index, int) else 0 + with bind_evaluation_row(evaluation_run_id, index): + with _criterion_span( + evaluation_run_id, index, criterion.criterion_type + ) as span: + if isinstance(criterion, Scorer): + event = await self._run_scorer_for_result(row, criterion) + else: + event = await self._run_ld_judge_for_result( + row, + tool_handlers, + criterion, + resolved_judges[criterion.key], + ) + _finish_criterion_span(span, event) + return event finally: controller.release() diff --git a/packages/client/src/launchdarkly_ai_server/lifecycle.py b/packages/client/src/launchdarkly_ai_server/lifecycle.py index 5d4bbadc..d74f60e4 100644 --- a/packages/client/src/launchdarkly_ai_server/lifecycle.py +++ b/packages/client/src/launchdarkly_ai_server/lifecycle.py @@ -354,6 +354,27 @@ def _release_otel_globals() -> None: logger.debug("Could not release the global OTel tracer provider", exc_info=True) +def _flush_tracer_provider() -> None: + """Export spans sitting on the provider ``init_client`` already registered. + + Does not build an exporter, replace the endpoint, or invent a project id. + The resource attribute ``highlight.project_id`` stays the SDK key that + ``_setup_telemetry`` stamped. A provider without ``force_flush`` — the API + proxy, when the SDK extra is absent — is a no-op. + """ + try: + from opentelemetry import trace + except ImportError: + return + flush = getattr(trace.get_tracer_provider(), "force_flush", None) + if flush is None: + return + try: + flush() + except Exception: + logger.debug("Could not flush OpenTelemetry spans", exc_info=True) + + def _clear_experimental_state() -> None: """Clears experimental feature state. A failure is logged, never raised.""" try: diff --git a/packages/client/tests/test_conversation.py b/packages/client/tests/test_conversation.py index 83b5f8bf..9d599068 100644 --- a/packages/client/tests/test_conversation.py +++ b/packages/client/tests/test_conversation.py @@ -12,8 +12,12 @@ from launchdarkly_ai_server.conversation import ( GEN_AI_CONVERSATION_ID, + LD_EVALUATION_RUN_ID, + LD_EVALUATION_RUN_ROW_INDEX, ConversationIdSpanProcessor, + bind_evaluation_row, conversation_id, + conversation_id_for_evaluation_row, set_conversation_id_if_absent, with_judge_evaluation, ) @@ -229,3 +233,54 @@ async def test_omits_explanation_when_not_supplied(self) -> None: assert not any("explanation" in k for k in span.attributes) event = next(e for e in span.events if e.name == "gen_ai.evaluation.result") assert not any("explanation" in k for k in (event.attributes or {})) + + +class TestEvaluationRowBinding: + """Offline rows share ai-evaluator's UUID5 and must not reuse launchdarkly.run.id.""" + + def test_uuid5_matches_the_published_namespace(self) -> None: + assert ( + conversation_id_for_evaluation_row("run-id", 3) + == "faa03308-8fe7-5805-8c07-f82a3dbdb3f1" + ) + assert ( + conversation_id_for_evaluation_row("run-id", 0) + == "8b90044e-ac6d-5af1-88a3-b75fbb9c8268" + ) + assert conversation_id_for_evaluation_row("run-id", None) == ( + conversation_id_for_evaluation_row("run-id", 0) + ) + + def test_stamps_sdk_spans_and_not_run_id(self) -> None: + with bind_evaluation_row("run-id", 3): + span = _tracer.start_span("invoke_agent") + span.end() + recorded = finished()[0] + assert recorded.attributes is not None + assert ( + recorded.attributes[GEN_AI_CONVERSATION_ID] + == "faa03308-8fe7-5805-8c07-f82a3dbdb3f1" + ) + assert recorded.attributes[LD_EVALUATION_RUN_ID] == "run-id" + assert recorded.attributes[LD_EVALUATION_RUN_ROW_INDEX] == 3 + assert "launchdarkly.run.id" not in recorded.attributes + + def test_does_not_stamp_third_party_spans(self) -> None: + foreign = _provider.get_tracer("opentelemetry.instrumentation.httpx") + with bind_evaluation_row("run-id", 3): + span = foreign.start_span("POST") + span.end() + attributes = finished()[0].attributes or {} + assert GEN_AI_CONVERSATION_ID not in attributes + assert LD_EVALUATION_RUN_ID not in attributes + assert LD_EVALUATION_RUN_ROW_INDEX not in attributes + + def test_row_id_shadows_an_outer_conversation_id(self) -> None: + with conversation_id("thread-123"): + with bind_evaluation_row("run-id", 0): + span = _tracer.start_span("chat gpt-4o") + span.end() + assert ( + finished()[0].attributes[GEN_AI_CONVERSATION_ID] + == "8b90044e-ac6d-5af1-88a3-b75fbb9c8268" + ) diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index 5f273e6c..332e257d 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -3054,6 +3054,7 @@ async def handler( {"provider": {"name": "OpenAI"}, "model": {"name": "gpt-4o"}}, {"lookup_order": lambda args: "shipped"}, 1, + evaluation_run_id="run-id", ) assert results[0]["status"] == "ERROR" diff --git a/packages/client/tests/test_evaluations_spans.py b/packages/client/tests/test_evaluations_spans.py new file mode 100644 index 00000000..6c0e4115 --- /dev/null +++ b/packages/client/tests/test_evaluations_spans.py @@ -0,0 +1,439 @@ +"""Offline evaluation spans join the customer project on the existing exporter. + +Reference: TESTING.md §8.10. +""" + +from __future__ import annotations + +import json +from collections.abc import Callable, Iterator +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +from opentelemetry import trace as otel_trace +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import StatusCode +from opentelemetry.util._once import Once + +from launchdarkly_ai_server.conversation import ( + GEN_AI_CONVERSATION_ID, + LD_EVALUATION_RUN_ID, + LD_EVALUATION_RUN_ROW_INDEX, + ConversationIdSpanProcessor, +) +from launchdarkly_ai_server.evaluations import HttpResponse +from launchdarkly_ai_server.experimental.evaluations import ( + EvalTool, + Judge, + init_evaluations, +) + +_RUN_ID = "run-id" +_ROW_INDEX = 3 +_CONVERSATION_ID = "faa03308-8fe7-5805-8c07-f82a3dbdb3f1" +_TOOL_ARGUMENTS = '{"order_id": "A1"}' +_TOOL_RESULT = "shipped" + + +@pytest.fixture +def stub_sdk_client(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + monkeypatch.setenv("LD_SDK_KEY", "sdk-key") + client = MagicMock() + client.flush = AsyncMock() + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.get_client", + MagicMock(side_effect=RuntimeError("client not initialized")), + ) + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.init_client", + AsyncMock(return_value=client), + ) + return client + + +@pytest.fixture +def span_export() -> Iterator[tuple[InMemorySpanExporter, list[str]]]: + """Install the SDK processor on a provider whose project id is the SDK key.""" + saved = otel_trace._TRACER_PROVIDER + otel_trace._TRACER_PROVIDER = None + otel_trace._TRACER_PROVIDER_SET_ONCE = Once() + exporter = InMemorySpanExporter() + provider = TracerProvider( + resource=Resource.create( + {"service.name": "python-sdk", "highlight.project_id": "sdk-key"} + ) + ) + provider.add_span_processor(ConversationIdSpanProcessor()) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + otel_trace.set_tracer_provider(provider) + flushed: list[str] = [] + original_flush = provider.force_flush + + def spy(timeout_millis: int = 30000) -> bool: + flushed.append("traces") + return original_flush(timeout_millis) + + provider.force_flush = spy # type: ignore[method-assign] + try: + yield exporter, flushed + finally: + provider.shutdown() + once = Once() + otel_trace._TRACER_PROVIDER = saved + if saved is not None: + once.do_once(lambda: None) + otel_trace._TRACER_PROVIDER_SET_ONCE = once + + +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 "" + ) + + +class SequencedTransport: + 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 one_row_transport() -> SequencedTransport: + return SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + { + "items": [ + { + "rowIndex": _ROW_INDEX, + "input": "Where is order A1?", + "expectedOutput": "shipped", + "variables": {}, + } + ], + "totalCount": 1, + "_links": {"self": {"href": "https://api.test/current"}}, + }, + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + { + "id": _RUN_ID, + "evaluationId": "evaluation-id", + "state": "PENDING", + }, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + + +def patch_judge(monkeypatch: pytest.MonkeyPatch) -> None: + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + assert key == "$ld:ai:judge:accuracy" + return { + "config": { + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + "instructions": "Judge {{response_to_evaluate}}", + }, + "meta": {"variationKey": "default", "version": 12}, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + + +def forbid_new_exporter(monkeypatch: pytest.MonkeyPatch) -> None: + from opentelemetry.exporter.otlp.proto.http import trace_exporter + + def boom(*args: Any, **kwargs: Any) -> Any: + raise AssertionError("run() must not construct an OTLP exporter") + + monkeypatch.setattr(trace_exporter, "OTLPSpanExporter", boom) + + +def assert_no_verdict(attributes: Any) -> None: + keys = set(attributes or {}) + assert "gen_ai.evaluation.score.label" not in keys + assert "verdict" not in keys + assert "passed" not in keys + assert "successDirection" not in keys + assert "launchdarkly.run.id" not in keys + + +def assert_row_attributes(attributes: Any) -> None: + assert attributes[GEN_AI_CONVERSATION_ID] == _CONVERSATION_ID + assert attributes[LD_EVALUATION_RUN_ID] == _RUN_ID + assert attributes[LD_EVALUATION_RUN_ROW_INDEX] == _ROW_INDEX + assert_no_verdict(attributes) + + +@pytest.mark.asyncio +async def test_plain_function_exports_evaluate_span_on_the_sdk_key_provider( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, + span_export: tuple[InMemorySpanExporter, list[str]], +) -> None: + exporter, flushed = span_export + forbid_new_exporter(monkeypatch) + patch_judge(monkeypatch) + + async def client_flush() -> None: + flushed.append("client") + + stub_sdk_client.flush = client_flush + evals = init_evaluations( + project_key="proj", + api_key="token", + sdk_key="sdk-key", + transport=one_row_transport(), + ) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + assert "__ld" not in variables + if "Judge" in config.get("instructions", ""): + return { + "output": '{"score": 0.9, "reasoning": "Correct and concise."}', + } + return {"output": "shipped", "usage": {"input_tokens": 10, "output_tokens": 4}} + + result = await evals.run( + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + spans = exporter.get_finished_spans() + assert [span.name for span in spans] == ["evaluate $ld:ai:judge:accuracy"] + evaluate = spans[0] + assert evaluate.resource is not None + assert evaluate.resource.attributes["highlight.project_id"] == "sdk-key" + assert_row_attributes(evaluate.attributes) + assert evaluate.attributes["gen_ai.operation.name"] == "evaluate" + event = next( + item for item in evaluate.events if item.name == "gen_ai.evaluation.result" + ) + assert event.attributes is not None + assert event.attributes["gen_ai.evaluation.name"] == "$ld:ai:judge:accuracy" + assert event.attributes["gen_ai.evaluation.score.value"] == 0.9 + assert event.attributes["gen_ai.evaluation.explanation"] == "Correct and concise." + assert_no_verdict(event.attributes) + assert "gen_ai.input.messages" not in (evaluate.attributes or {}) + assert "gen_ai.output.messages" not in (evaluate.attributes or {}) + + payloads = [call.args[2] for call in stub_sdk_client.track.call_args_list] + names = [call.args[0] for call in stub_sdk_client.track.call_args_list] + assert names == [ + "$ld:ai:offline-evals:generation", + "$ld:ai:offline-evals:criterion", + ] + for payload in payloads: + assert "tool_calls" not in payload + assert "input" not in payload + assert "variables" not in payload + assert "metadata" not in payload + assert "verdict" not in payload + assert "passed" not in payload + generation = payloads[0] + assert generation["output"] == "shipped" + assert "message" not in generation + criterion = payloads[1] + assert criterion["score"] == 0.9 + assert criterion["reason"] == "Correct and concise." + assert flushed == ["client", "traces"] + + +@pytest.mark.asyncio +async def test_handler_spans_carry_the_row_and_the_tool_call( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, + span_export: tuple[InMemorySpanExporter, list[str]], +) -> None: + exporter, _flushed = span_export + forbid_new_exporter(monkeypatch) + patch_judge(monkeypatch) + evals = init_evaluations( + project_key="proj", + api_key="token", + sdk_key="sdk-key", + transport=one_row_transport(), + ) + tracer = otel_trace.get_tracer("@launchdarkly/ai-server") + foreign = otel_trace.get_tracer("opentelemetry.instrumentation.httpx") + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + assert "__ld" not in variables + if "Judge" in config.get("instructions", ""): + with tracer.start_as_current_span("invoke_agent"): + with tracer.start_as_current_span("chat gpt-4o"): + pass + return {"output": '{"score": 0.9, "reasoning": "Correct and concise."}'} + with tracer.start_as_current_span("invoke_agent"): + with tracer.start_as_current_span("chat gpt-4o"): + pass + with tracer.start_as_current_span("execute_tool lookup_order") as tool: + tool.set_attribute("gen_ai.operation.name", "execute_tool") + tool.set_attribute("gen_ai.tool.name", "lookup_order") + tool.set_attribute("gen_ai.tool.call.id", "call-1") + tool.set_attribute("gen_ai.tool.call.arguments", _TOOL_ARGUMENTS) + tool.set_attribute("gen_ai.tool.call.result", _TOOL_RESULT) + tool_handlers["lookup_order"]({"order_id": "A1"}) + http = foreign.start_span("POST /v1/chat") + http.end() + return {"output": "done"} + + await evals.run( + key="support-qa", + dataset="golden", + handler=handler, + tools=[ + EvalTool( + key="lookup_order", + implementation=lambda args: _TOOL_RESULT, + schema={"type": "object"}, + description="Look up an order", + ) + ], + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + spans = list(exporter.get_finished_spans()) + by_name: dict[str, list[Any]] = {} + for span in spans: + by_name.setdefault(span.name, []).append(span) + assert set(by_name) >= { + "invoke_agent", + "chat gpt-4o", + "execute_tool lookup_order", + "evaluate $ld:ai:judge:accuracy", + "POST /v1/chat", + } + for name in ( + "invoke_agent", + "chat gpt-4o", + "execute_tool lookup_order", + "evaluate $ld:ai:judge:accuracy", + ): + assert by_name[name] + for span in by_name[name]: + assert_row_attributes(span.attributes) + tool = by_name["execute_tool lookup_order"][0] + assert tool.attributes is not None + assert tool.attributes["gen_ai.tool.name"] == "lookup_order" + assert tool.attributes["gen_ai.tool.call.arguments"] == _TOOL_ARGUMENTS + assert tool.attributes["gen_ai.tool.call.result"] == _TOOL_RESULT + http_attributes = by_name["POST /v1/chat"][0].attributes or {} + assert GEN_AI_CONVERSATION_ID not in http_attributes + assert LD_EVALUATION_RUN_ID not in http_attributes + evaluate = by_name["evaluate $ld:ai:judge:accuracy"][0] + assert any(event.name == "gen_ai.evaluation.result" for event in evaluate.events) + generation = next( + call.args[2] + for call in stub_sdk_client.track.call_args_list + if call.args[0] == "$ld:ai:offline-evals:generation" + ) + assert "tool_calls" not in generation + assert _TOOL_ARGUMENTS not in json.dumps(generation) + assert _TOOL_RESULT not in json.dumps(generation) + + +@pytest.mark.asyncio +async def test_errored_judge_span_has_no_result_event( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, + span_export: tuple[InMemorySpanExporter, list[str]], +) -> None: + exporter, _flushed = span_export + patch_judge(monkeypatch) + evals = init_evaluations( + project_key="proj", + api_key="token", + sdk_key="sdk-key", + transport=one_row_transport(), + ) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + if "Judge" in config.get("instructions", ""): + return {"output": "not json"} + return {"output": "shipped"} + + await evals.run( + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + evaluate = exporter.get_finished_spans()[0] + assert evaluate.name == "evaluate $ld:ai:judge:accuracy" + assert evaluate.status.status_code == StatusCode.ERROR + assert evaluate.attributes is not None + assert evaluate.attributes["error.type"] == "invalid_judge_output" + assert_row_attributes(evaluate.attributes) + assert not any( + event.name == "gen_ai.evaluation.result" for event in evaluate.events + ) + criterion = next( + call.args[2] + for call in stub_sdk_client.track.call_args_list + if call.args[0] == "$ld:ai:offline-evals:criterion" + ) + assert criterion["status"] == "ERROR" + assert "verdict" not in criterion + assert "score" not in criterion From 4ae80ffd26b39c43b21895c7a47a8efc22d9c09a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 14 Oct 2026 22:24:15 +0000 Subject: [PATCH 2/8] feat(evaluations): time recorded tool calls and count handler spans ToolInvocation gains start and end times, excluded from equality, so the offline runner can place execute_tool spans. observe_handler_spans counts the SDK spans a handler opens, so the runner can tell a provider handler that traced itself from a plain function that did not. Co-authored-by: christopher.schmitz --- .../launchdarkly_ai_server/conversation.py | 28 +++++++++++++++++++ .../src/launchdarkly_ai_server/trajectory.py | 12 ++++++-- 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/conversation.py b/packages/client/src/launchdarkly_ai_server/conversation.py index 8fc51106..2cd86701 100644 --- a/packages/client/src/launchdarkly_ai_server/conversation.py +++ b/packages/client/src/launchdarkly_ai_server/conversation.py @@ -36,6 +36,7 @@ _CONV_KEY = otel_context.create_key("launchdarkly.gen_ai.conversation.id") _EVAL_KEY = otel_context.create_key("launchdarkly.judge.evaluation") _EVAL_ROW_KEY = otel_context.create_key("launchdarkly.evaluation_run.row") +_OBSERVER_KEY = otel_context.create_key("launchdarkly.evaluation_run.observer") @dataclass @@ -143,6 +144,29 @@ def bind_evaluation_row(run_id: str, row_index: int | None) -> Iterator[None]: otel_context.detach(token) +@dataclass +class HandlerSpanObserver: + """Counts SDK spans started while :func:`observe_handler_spans` is bound.""" + + count: int = 0 + + +@contextmanager +def observe_handler_spans() -> Iterator[HandlerSpanObserver]: + """Count the SDK spans a handler opens inside this block. + + The offline runner describes a plain-function handler's call itself, and + must not when a provider handler already traced it. Counting relies on + :class:`ConversationIdSpanProcessor`, which ``init_client`` registers. + """ + observer = HandlerSpanObserver() + token = otel_context.attach(otel_context.set_value(_OBSERVER_KEY, observer)) + try: + yield observer + finally: + otel_context.detach(token) + + def record_evaluation_result( span: Any, name: str, score: float, explanation: str | None = None ) -> None: @@ -349,6 +373,10 @@ def on_start( if conv and _is_launchdarkly_span(span): set_conversation_id_if_absent(span, conv) + observer = otel_context.get_value(_OBSERVER_KEY, ctx) + if isinstance(observer, HandlerSpanObserver) and _is_launchdarkly_span(span): + observer.count += 1 + row = _evaluation_row_from(ctx) if row and _is_launchdarkly_span(span): for key, value in row.items(): diff --git a/packages/client/src/launchdarkly_ai_server/trajectory.py b/packages/client/src/launchdarkly_ai_server/trajectory.py index 071925db..1e728112 100644 --- a/packages/client/src/launchdarkly_ai_server/trajectory.py +++ b/packages/client/src/launchdarkly_ai_server/trajectory.py @@ -21,7 +21,8 @@ import inspect import json from collections.abc import Callable, Collection, Mapping -from dataclasses import dataclass, replace +from dataclasses import dataclass, field, replace +from time import time_ns from typing import Any from .types import NativeTool @@ -57,6 +58,10 @@ class ToolInvocation: arguments: Any = None result: Any = None error: str | None = None + #: Epoch nanoseconds, for the offline runner's ``execute_tool`` spans. + #: Excluded from equality: a judge sees the same step whenever it ran. + started_ns: int | None = field(default=None, compare=False) + ended_ns: int | None = field(default=None, compare=False) class TrajectoryRecorder: @@ -177,7 +182,9 @@ def _reserve(self, name: str, arguments: Any) -> int | None: # later sees, and an evaluation holding many rows' calls in memory # until scoring must hold their bounded text, not each raw value. self._invocations.append( - ToolInvocation(name=name, arguments=_render_value(arguments)) + ToolInvocation( + name=name, arguments=_render_value(arguments), started_ns=time_ns() + ) ) return len(self._invocations) - 1 @@ -192,6 +199,7 @@ def _complete( self._invocations[slot], result=None if error is not None else _render_value(result), error=None if error is None else _truncate(error), + ended_ns=time_ns(), ) From 41eba3fba60253fd45fd758aba655a359b53c046 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 14 Oct 2026 22:24:16 +0000 Subject: [PATCH 3/8] feat(evaluations): trace each offline row as one conversation Each dataset row gets a launchdarkly.dataset_row root, and its evaluate spans parent under it, so generation and judging share one trace. When the handler is a plain function that opened no SDK span, the runner writes what it handed the function and what came back: a chat span with the rendered system instructions, input, output (tool_call parts first), model, and usage; an execute_tool span per recorded call; and a judge chat span under evaluate with the rendered judge prompt and completion. evaluate carries the judge model and OK or ERROR status. The offline-evals custom events are unchanged and still carry no transcript or tool payload. Co-authored-by: christopher.schmitz --- .../evaluations/module.py | 5 + .../evaluations/runner.py | 127 ++++++-- .../evaluations/spans.py | 277 ++++++++++++++++++ .../client/tests/test_evaluations_spans.py | 191 +++++++++++- 4 files changed, 569 insertions(+), 31 deletions(-) create mode 100644 packages/client/src/launchdarkly_ai_server/evaluations/spans.py diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 5be8831f..8940da95 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -426,6 +426,7 @@ async def run( ) raise config = self._runner._build_handler_config(generation, run_tools) + row_spans: dict[int, Any] = {} try: results = await self._runner._run_rows( dataset_rows, @@ -434,6 +435,7 @@ async def run( run_tool_handlers, concurrency, evaluation_run_id=evaluation_run.id, + row_spans=row_spans, ) self._runner._emit_generation_events( client, @@ -451,6 +453,7 @@ async def run( resolved_judges, concurrency, evaluation_run_id=evaluation_run.id, + row_spans=row_spans, ) self._runner._emit_evaluation_events( client, @@ -464,6 +467,8 @@ async def run( # Generation results already queued on the SDK event buffer must # reach LaunchDarkly even when the criteria phase fails. Spans # follow on the provider init_client already registered. + for row_span in row_spans.values(): + row_span.end() try: flush_result = client.flush() if inspect.isawaitable(flush_result): diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index 5ee06c37..b54cef5b 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -22,6 +22,7 @@ LD_EVALUATION_RUN_ROW_INDEX, bind_evaluation_row, conversation_id_for_evaluation_row, + observe_handler_spans, record_evaluation_result, ) from ..judge_scoring import ( @@ -59,6 +60,12 @@ LDJudgeCriterionEventPayload, TokenUsage, ) +from .spans import ( + record_chat, + record_tool_calls, + set_config_model_identity, + start_row_span, +) from .tools import ( EvalTool, ToolImplementation, @@ -255,15 +262,23 @@ def record_rate_limit( @contextmanager -def _criterion_span(run_id: str, row_index: int, criterion_type: str) -> Iterator[Any]: - """Open ``evaluate {criterion}`` on the SDK tracer. +def _criterion_span( + run_id: str, + row_index: int, + criterion_type: str, + row_span: Any = None, +) -> Iterator[Any]: + """Open ``evaluate {criterion}`` on the SDK tracer, under the row's root. Correlation attributes are set here as well as by the span processor, so a plain function still exports them when the processor is the one ``init_client`` registered. This does not build an exporter. """ tracer = trace.get_tracer("@launchdarkly/ai-server") - with tracer.start_as_current_span(f"evaluate {criterion_type}") as span: + parent = trace.set_span_in_context(row_span) if row_span is not None else None + with tracer.start_as_current_span( + f"evaluate {criterion_type}", context=parent + ) as span: span.set_attribute("gen_ai.operation.name", "evaluate") span.set_attribute( GEN_AI_CONVERSATION_ID, @@ -294,9 +309,13 @@ def _finish_criterion_span(span: Any, result: Mapping[str, Any]) -> None: float(score), explanation, ) + span.set_status(Status(StatusCode.OK)) return - span.set_status(Status(StatusCode.ERROR)) error = result.get("error") + message = error.get("message") if isinstance(error, Mapping) else None + span.set_status( + Status(StatusCode.ERROR, message if isinstance(message, str) else None) + ) if isinstance(error, Mapping): code = error.get("code") if isinstance(code, str) and code: @@ -698,7 +717,13 @@ async def _run_rows( concurrency: int, *, evaluation_run_id: str, + row_spans: dict[int, Any] | None = None, ) -> list[dict[str, Any]]: + """Generate every row. + + ``row_spans`` receives each row's open root span, keyed by row index, + for the criteria phase to parent under; the caller ends them. + """ controller = ConcurrencyController(concurrency) async def invoke(row: DatasetRow) -> dict[str, Any]: @@ -707,13 +732,42 @@ async def invoke(row: DatasetRow) -> dict[str, Any]: # same tool map, so a shared recorder would splice their calls. recorder = TrajectoryRecorder() row_tool_handlers = recorder.wrap(tool_handlers) + row_span: Any = trace.get_current_span() + if row_spans is not None: + row_span = start_row_span(evaluation_run_id, row.row_index) + row_spans[row.row_index] = row_span started = datetime.now(UTC) started_clock = time.perf_counter() + started_ns = time.time_ns() + result: Any = None + raised: BaseException | None = None try: - with bind_evaluation_row(evaluation_run_id, row.row_index): - result = await handler( - config, row.input, row_tool_handlers, dict(row.variables) - ) + with ( + bind_evaluation_row(evaluation_run_id, row.row_index), + trace.use_span(row_span, end_on_exit=False), + observe_handler_spans() as observer, + ): + try: + result = await handler( + config, row.input, row_tool_handlers, dict(row.variables) + ) + except Exception as error: + raised = error + raise + finally: + if observer.count == 0: + record_chat( + row_span, + config=config, + user_input=row.input, + variables=row.variables, + started_ns=started_ns, + ended_ns=time.time_ns(), + result=result if isinstance(result, Mapping) else None, + error=raised, + tool_calls=recorder.invocations, + ) + record_tool_calls(row_span, recorder.invocations) if not isinstance(result, Mapping): raise TypeError("handler result must be a mapping") completed = datetime.now(UTC) @@ -734,8 +788,13 @@ async def invoke(row: DatasetRow) -> dict[str, Any]: if isinstance(usage, Mapping): payload["usage"] = dict(usage) controller.record_success(config["provider"]["name"]) + if row_spans is not None: + row_span.set_status(Status(StatusCode.OK)) return payload except Exception as error: + if row_spans is not None: + row_span.set_attribute("error.type", type(error).__name__) + row_span.set_status(Status(StatusCode.ERROR, f"{error}")) completed = datetime.now(UTC) return { "row_index": row.row_index, @@ -1011,16 +1070,39 @@ async def _run_ld_judge_for_result( if execution.collapse_messages else resolved.config ) + judge_variables = { + **variables, + "formatting_instructions": FORMATTING_INSTRUCTIONS, + } + evaluate_span = trace.get_current_span() + set_config_model_identity(evaluate_span, judge_config) + started_ns = time.time_ns() + result: Any = None + raised: BaseException | None = None try: - result = await execution.handler( - dict(judge_config), - row.get("output"), - tool_handlers, - { - **variables, - "formatting_instructions": FORMATTING_INSTRUCTIONS, - }, - ) + with observe_handler_spans() as observer: + try: + result = await execution.handler( + dict(judge_config), + row.get("output"), + tool_handlers, + judge_variables, + ) + except Exception as error: + raised = error + raise + finally: + if observer.count == 0: + record_chat( + evaluate_span, + config=judge_config, + user_input=row.get("output"), + variables=judge_variables, + started_ns=started_ns, + ended_ns=time.time_ns(), + result=result if isinstance(result, Mapping) else None, + error=raised, + ) except Exception as error: return self._criterion_error_result( base, started_clock, "handler_raised", f"judge handler raised: {error}" @@ -1077,8 +1159,12 @@ async def _run_criteria_for_results( concurrency: int, *, evaluation_run_id: str, + row_spans: Mapping[int, Any] | None = None, ) -> list[dict[str, Any]]: - """Run every (row, criterion) pair, bounded by the run's concurrency.""" + """Run every (row, criterion) pair, bounded by the run's concurrency. + + Each ``evaluate`` span parents under its row's span in ``row_spans``. + """ controller = ConcurrencyController(concurrency) async def run_one( @@ -1090,7 +1176,10 @@ async def run_one( index = row_index if isinstance(row_index, int) else 0 with bind_evaluation_row(evaluation_run_id, index): with _criterion_span( - evaluation_run_id, index, criterion.criterion_type + evaluation_run_id, + index, + criterion.criterion_type, + (row_spans or {}).get(index), ) as span: if isinstance(criterion, Scorer): event = await self._run_scorer_for_result(row, criterion) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/spans.py b/packages/client/src/launchdarkly_ai_server/evaluations/spans.py new file mode 100644 index 00000000..25526b38 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/spans.py @@ -0,0 +1,277 @@ +"""Spans the offline runner writes for one dataset row. + +Every row gets a ``launchdarkly.dataset_row`` root, so its generation, tool +calls, and criteria share one trace. The root names no ``gen_ai`` operation and +no evaluation, so the conversation view treats it as an envelope and keeps the +generation in the transcript. + +A provider handler traces its own call. A plain function does not, so the +runner describes what it handed the function and what came back: one ``chat`` +span, an ``execute_tool`` span per recorded tool call, and, under each judge's +``evaluate`` span, a ``chat`` span for the judge call. These are written only +when the handler opened no SDK span of its own, so a call is never traced twice. + +These spans export to the customer's project on the SDK key, the copy that +always carries content. The ``$ld:ai:offline-evals:*`` events never do. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from typing import Any + +from opentelemetry import trace +from opentelemetry.trace import Span, Status, StatusCode + +from ..content import ( + SpanMessage, + SpanMessagePart, + ToolDefinitionInput, + set_input_content_attributes, + set_output_content_attributes, + set_tool_call_content_attributes, + text_message, +) +from ..conversation import ( + GEN_AI_CONVERSATION_ID, + LD_EVALUATION_RUN_ID, + LD_EVALUATION_RUN_ROW_INDEX, + conversation_id_for_evaluation_row, +) +from ..trajectory import ToolInvocation +from ..utils import ( + SpanUsage, + parse_template, + parse_usage, + set_model_identity_attributes, + set_usage_span_attributes, +) + +TRACER_NAME = "@launchdarkly/ai-server" +ROW_SPAN_NAME = "launchdarkly.dataset_row" + +#: The customer-project copy always carries content; see the module docstring. +_CAPTURE = True + + +def _tracer() -> trace.Tracer: + return trace.get_tracer(TRACER_NAME) + + +def set_row_attributes(span: Span, run_id: str, row_index: int) -> None: + span.set_attribute( + GEN_AI_CONVERSATION_ID, conversation_id_for_evaluation_row(run_id, row_index) + ) + span.set_attribute(LD_EVALUATION_RUN_ID, run_id) + span.set_attribute(LD_EVALUATION_RUN_ROW_INDEX, row_index) + + +def start_row_span(run_id: str, row_index: int) -> Span: + """Open the row's root span. The caller ends it after the row's criteria.""" + span = _tracer().start_span( + ROW_SPAN_NAME, context=trace.set_span_in_context(trace.INVALID_SPAN) + ) + set_row_attributes(span, run_id, row_index) + return span + + +def _provider_name(config: Mapping[str, Any]) -> str: + provider = config.get("provider") + name = provider.get("name") if isinstance(provider, Mapping) else None + return name.lower() if isinstance(name, str) else "" + + +def _model_name(config: Mapping[str, Any]) -> str: + model = config.get("model") + name = model.get("name") if isinstance(model, Mapping) else None + return name if isinstance(name, str) else "" + + +def _call_id(index: int) -> str: + """Row-local id joining a ``tool_call`` part to its ``execute_tool`` span. + + A plain function reports no provider call id, so the runner numbers the + calls it recorded, in start order. + """ + return f"call_{index + 1}" + + +def set_config_model_identity(span: Span, config: Mapping[str, Any]) -> None: + """Write the provider and model a config names onto ``span``.""" + set_model_identity_attributes(span, _provider_name(config), _model_name(config)) + + +def _text(value: Any) -> str: + if isinstance(value, str): + return value + return json.dumps(value, default=str) + + +def _prompt( + config: Mapping[str, Any], user_input: Any, variables: Mapping[str, Any] +) -> tuple[str | None, list[SpanMessage]]: + """The system instructions and messages the handler was given, rendered. + + Renders the config with the row's variables, the same single + ``parse_template`` pass a provider handler applies. ``user_input`` is the + final user turn, as it is for every handler. + """ + values = dict(variables) + system_parts: list[str] = [] + messages: list[SpanMessage] = [] + instructions = config.get("instructions") + if isinstance(instructions, str) and instructions: + system_parts.append(parse_template(instructions, values)) + for message in config.get("messages") or []: + if not isinstance(message, Mapping): + continue + content = message.get("content") + if not isinstance(content, str): + continue + rendered = parse_template(content, values) + role = message.get("role") + if role == "system": + system_parts.append(rendered) + elif isinstance(role, str): + messages.append(text_message(role, rendered)) + if user_input is not None: + messages.append(text_message("user", _text(user_input))) + return ("\n\n".join(system_parts) or None), messages + + +def _tool_definitions(config: Mapping[str, Any]) -> list[ToolDefinitionInput]: + tools = config.get("tools") + if not isinstance(tools, Mapping): + return [] + definitions: list[ToolDefinitionInput] = [] + for key, tool in tools.items(): + if not isinstance(tool, Mapping): + continue + name = tool.get("name", key) + if not isinstance(name, str) or not name: + continue + definitions.append( + ToolDefinitionInput( + name=name, + description=tool.get("description") or None, + parameters=tool.get("parameters"), + ) + ) + return definitions + + +def _span_usage(usage: Any) -> SpanUsage | None: + if not isinstance(usage, Mapping): + return None + parsed = parse_usage(dict(usage)) + details = parsed.get("input_details") or {} + return SpanUsage( + input=parsed["input"], + output=parsed["output"], + cache_read=details.get("cache_read", 0), + cache_creation=details.get("cache_creation", 0), + ) + + +def record_chat( + parent: Span, + *, + config: Mapping[str, Any], + user_input: Any, + variables: Mapping[str, Any], + started_ns: int, + ended_ns: int, + result: Mapping[str, Any] | None, + error: BaseException | None = None, + tool_calls: Sequence[ToolInvocation] = (), +) -> None: + """Write one ``chat {model}`` span for a call the handler did not trace. + + ``result`` is the handler's return value. ``error`` is what it raised; the + span then has ERROR status and no output. ``tool_calls`` are the calls the + handler made, written as ``tool_call`` parts ahead of the answer and tied + to their ``execute_tool`` spans by call id, so the transcript keeps their + order even when every step lands in the same millisecond. + """ + model = _model_name(config) + span = _tracer().start_span( + f"chat {model}".strip(), + context=trace.set_span_in_context(parent), + start_time=started_ns, + ) + try: + span.set_attribute("gen_ai.operation.name", "chat") + set_config_model_identity(span, config) + system_instructions, messages = _prompt(config, user_input, variables) + set_input_content_attributes( + span, + _CAPTURE, + system_instructions=system_instructions, + messages=messages, + tool_definitions=_tool_definitions(config), + ) + if error is not None: + span.record_exception(error) + span.set_attribute("error.type", type(error).__name__) + span.set_status(Status(StatusCode.ERROR, str(error))) + return + result = result or {} + usage = _span_usage(result.get("usage")) + if usage is not None: + set_usage_span_attributes(span, usage) + parts = [ + SpanMessagePart( + type="tool_call", + id=_call_id(index), + name=call.name, + arguments=call.arguments, + ) + for index, call in enumerate(tool_calls) + ] + output = result.get("output", result.get("response")) + if output is not None: + parts.append(SpanMessagePart(type="text", content=_text(output))) + if parts: + set_output_content_attributes( + span, _CAPTURE, [SpanMessage(role="assistant", parts=parts)] + ) + span.set_status(Status(StatusCode.OK)) + finally: + span.end(end_time=ended_ns) + + +def record_tool_calls(parent: Span, invocations: Sequence[ToolInvocation]) -> None: + """Write an ``execute_tool {name}`` span for each recorded call. + + Arguments and results are the recorder's rendered text, the same bounded + values a judge reads. A call that never completed has no end time and is + skipped rather than given one. + """ + context = trace.set_span_in_context(parent) + for index, invocation in enumerate(invocations): + if invocation.started_ns is None or invocation.ended_ns is None: + continue + span = _tracer().start_span( + f"execute_tool {invocation.name}", + context=context, + start_time=invocation.started_ns, + ) + try: + span.set_attribute("gen_ai.operation.name", "execute_tool") + span.set_attribute("gen_ai.tool.name", invocation.name) + span.set_attribute("gen_ai.tool.call.id", _call_id(index)) + span.set_attribute("gen_ai.tool.type", "function") + set_tool_call_content_attributes( + span, + _CAPTURE, + arguments=invocation.arguments, + result=invocation.result, + ) + if invocation.error is not None: + span.set_attribute("error.type", "tool_error") + span.set_status(Status(StatusCode.ERROR, invocation.error)) + else: + span.set_status(Status(StatusCode.OK)) + finally: + span.end(end_time=invocation.ended_ns) diff --git a/packages/client/tests/test_evaluations_spans.py b/packages/client/tests/test_evaluations_spans.py index 6c0e4115..2dcd6a09 100644 --- a/packages/client/tests/test_evaluations_spans.py +++ b/packages/client/tests/test_evaluations_spans.py @@ -37,6 +37,7 @@ _CONVERSATION_ID = "faa03308-8fe7-5805-8c07-f82a3dbdb3f1" _TOOL_ARGUMENTS = '{"order_id": "A1"}' _TOOL_RESULT = "shipped" +_TOOL_ARGUMENTS_RENDERED = '{"order_id":"A1"}' @pytest.fixture @@ -204,8 +205,28 @@ def assert_row_attributes(attributes: Any) -> None: assert_no_verdict(attributes) +def spans_by_name(exporter: InMemorySpanExporter) -> dict[str, list[Any]]: + by_name: dict[str, list[Any]] = {} + for span in exporter.get_finished_spans(): + by_name.setdefault(span.name, []).append(span) + return by_name + + +def parent_of(span: Any) -> int | None: + return span.parent.span_id if span.parent is not None else None + + +def lookup_order_tool() -> EvalTool: + return EvalTool( + key="lookup_order", + implementation=lambda args: _TOOL_RESULT, + schema={"type": "object"}, + description="Look up an order", + ) + + @pytest.mark.asyncio -async def test_plain_function_exports_evaluate_span_on_the_sdk_key_provider( +async def test_plain_function_row_is_one_trace_with_the_exchange_tools_and_judge( monkeypatch: pytest.MonkeyPatch, stub_sdk_client: MagicMock, span_export: tuple[InMemorySpanExporter, list[str]], @@ -235,25 +256,93 @@ async def handler( if "Judge" in config.get("instructions", ""): return { "output": '{"score": 0.9, "reasoning": "Correct and concise."}', + "usage": {"input_tokens": 30, "output_tokens": 9}, } + assert tool_handlers["lookup_order"]({"order_id": "A1"}) == _TOOL_RESULT return {"output": "shipped", "usage": {"input_tokens": 10, "output_tokens": 4}} result = await evals.run( key="support-qa", dataset="golden", handler=handler, - generation={"provider": "OpenAI", "model": "gpt-4o"}, + tools=[lookup_order_tool()], + generation={ + "provider": "OpenAI", + "model": "gpt-4o", + "instructions": "Answer about {{input}}", + }, criteria=[Judge(key="$ld:ai:judge:accuracy")], ) assert result.passed is True spans = exporter.get_finished_spans() - assert [span.name for span in spans] == ["evaluate $ld:ai:judge:accuracy"] - evaluate = spans[0] - assert evaluate.resource is not None - assert evaluate.resource.attributes["highlight.project_id"] == "sdk-key" - assert_row_attributes(evaluate.attributes) - assert evaluate.attributes["gen_ai.operation.name"] == "evaluate" + assert {span.context.trace_id for span in spans} == {spans[0].context.trace_id} + by_name = spans_by_name(exporter) + assert sorted(by_name) == [ + "chat gpt-4o", + "evaluate $ld:ai:judge:accuracy", + "execute_tool lookup_order", + "launchdarkly.dataset_row", + ] + for span in spans: + assert span.resource is not None + assert span.resource.attributes["highlight.project_id"] == "sdk-key" + assert_row_attributes(span.attributes) + assert span.status.status_code == StatusCode.OK + + (row,) = by_name["launchdarkly.dataset_row"] + assert row.parent is None + assert "gen_ai.operation.name" not in (row.attributes or {}) + (evaluate,) = by_name["evaluate $ld:ai:judge:accuracy"] + (tool,) = by_name["execute_tool lookup_order"] + generation_chat, judge_chat = sorted( + by_name["chat gpt-4o"], key=lambda span: span.start_time + ) + for child in (generation_chat, tool, evaluate): + assert parent_of(child) == row.context.span_id + assert parent_of(judge_chat) == evaluate.context.span_id + + attributes = generation_chat.attributes or {} + assert attributes["gen_ai.operation.name"] == "chat" + assert attributes["gen_ai.provider.name"] == "openai" + assert attributes["gen_ai.request.model"] == "gpt-4o" + assert attributes["gen_ai.usage.input_tokens"] == 10 + assert attributes["gen_ai.usage.output_tokens"] == 4 + assert json.loads(str(attributes["gen_ai.system_instructions"])) == [ + {"type": "text", "content": "Answer about Where is order A1?"} + ] + assert json.loads(str(attributes["gen_ai.input.messages"])) == [ + {"role": "user", "parts": [{"type": "text", "content": "Where is order A1?"}]} + ] + assert json.loads(str(attributes["gen_ai.output.messages"])) == [ + { + "role": "assistant", + "parts": [ + { + "type": "tool_call", + "id": "call_1", + "name": "lookup_order", + "arguments": _TOOL_ARGUMENTS_RENDERED, + }, + {"type": "text", "content": "shipped"}, + ], + } + ] + assert json.loads(str(attributes["gen_ai.tool.definitions"]))[0]["name"] == ( + "lookup_order" + ) + + tool_attributes = tool.attributes or {} + assert tool_attributes["gen_ai.operation.name"] == "execute_tool" + assert tool_attributes["gen_ai.tool.name"] == "lookup_order" + assert tool_attributes["gen_ai.tool.call.id"] == "call_1" + assert tool_attributes["gen_ai.tool.call.arguments"] == _TOOL_ARGUMENTS_RENDERED + assert tool_attributes["gen_ai.tool.call.result"] == _TOOL_RESULT + + evaluate_attributes = evaluate.attributes or {} + assert evaluate_attributes["gen_ai.operation.name"] == "evaluate" + assert evaluate_attributes["gen_ai.request.model"] == "gpt-4o" + assert "gen_ai.input.messages" not in evaluate_attributes event = next( item for item in evaluate.events if item.name == "gen_ai.evaluation.result" ) @@ -262,8 +351,27 @@ async def handler( assert event.attributes["gen_ai.evaluation.score.value"] == 0.9 assert event.attributes["gen_ai.evaluation.explanation"] == "Correct and concise." assert_no_verdict(event.attributes) - assert "gen_ai.input.messages" not in (evaluate.attributes or {}) - assert "gen_ai.output.messages" not in (evaluate.attributes or {}) + + judge_attributes = judge_chat.attributes or {} + assert judge_attributes["gen_ai.operation.name"] == "chat" + assert json.loads(str(judge_attributes["gen_ai.system_instructions"])) == [ + {"type": "text", "content": "Judge shipped"} + ] + assert json.loads(str(judge_attributes["gen_ai.input.messages"])) == [ + {"role": "user", "parts": [{"type": "text", "content": "shipped"}]} + ] + assert json.loads(str(judge_attributes["gen_ai.output.messages"])) == [ + { + "role": "assistant", + "parts": [ + { + "type": "text", + "content": '{"score": 0.9, "reasoning": "Correct and concise."}', + } + ], + } + ] + assert judge_attributes["gen_ai.usage.input_tokens"] == 30 payloads = [call.args[2] for call in stub_sdk_client.track.call_args_list] names = [call.args[0] for call in stub_sdk_client.track.call_args_list] @@ -278,6 +386,8 @@ async def handler( assert "metadata" not in payload assert "verdict" not in payload assert "passed" not in payload + assert _TOOL_ARGUMENTS_RENDERED not in json.dumps(payload) + assert "Where is order A1?" not in json.dumps(payload) generation = payloads[0] assert generation["output"] == "shipped" assert "message" not in generation @@ -287,6 +397,57 @@ async def handler( assert flushed == ["client", "traces"] +@pytest.mark.asyncio +async def test_plain_function_that_raises_records_the_failed_generation( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, + span_export: tuple[InMemorySpanExporter, list[str]], +) -> None: + exporter, _flushed = span_export + patch_judge(monkeypatch) + evals = init_evaluations( + project_key="proj", + api_key="token", + sdk_key="sdk-key", + transport=one_row_transport(), + ) + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + tool_handlers["lookup_order"]({"order_id": "A1"}) + raise RuntimeError("model refused") + + await evals.run( + key="support-qa", + dataset="golden", + handler=handler, + tools=[lookup_order_tool()], + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + by_name = spans_by_name(exporter) + (row,) = by_name["launchdarkly.dataset_row"] + assert row.status.status_code == StatusCode.ERROR + (chat,) = by_name["chat gpt-4o"] + assert chat.status.status_code == StatusCode.ERROR + assert chat.status.description == "model refused" + attributes = chat.attributes or {} + assert attributes["error.type"] == "RuntimeError" + assert "gen_ai.output.messages" not in attributes + assert "gen_ai.input.messages" in attributes + (tool,) = by_name["execute_tool lookup_order"] + assert (tool.attributes or {})["gen_ai.tool.call.result"] == _TOOL_RESULT + (evaluate,) = by_name["evaluate $ld:ai:judge:accuracy"] + assert evaluate.status.status_code == StatusCode.ERROR + assert (evaluate.attributes or {})["error.type"] == "generation_incomplete" + assert parent_of(evaluate) == row.context.span_id + + @pytest.mark.asyncio async def test_handler_spans_carry_the_row_and_the_tool_call( monkeypatch: pytest.MonkeyPatch, @@ -367,6 +528,13 @@ async def handler( assert by_name[name] for span in by_name[name]: assert_row_attributes(span.attributes) + # The handler traced itself, so the runner adds no chat or tool span. + assert len(by_name["chat gpt-4o"]) == 2 + assert len(by_name["execute_tool lookup_order"]) == 1 + (row,) = by_name["launchdarkly.dataset_row"] + assert { + span.context.trace_id for span in spans if span.name != "POST /v1/chat" + } == {row.context.trace_id} tool = by_name["execute_tool lookup_order"][0] assert tool.attributes is not None assert tool.attributes["gen_ai.tool.name"] == "lookup_order" @@ -420,8 +588,7 @@ async def handler( criteria=[Judge(key="$ld:ai:judge:accuracy")], ) - evaluate = exporter.get_finished_spans()[0] - assert evaluate.name == "evaluate $ld:ai:judge:accuracy" + (evaluate,) = spans_by_name(exporter)["evaluate $ld:ai:judge:accuracy"] assert evaluate.status.status_code == StatusCode.ERROR assert evaluate.attributes is not None assert evaluate.attributes["error.type"] == "invalid_judge_output" From 96be49fd939ff60f8fd8a15ecef7c5641898ef5d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 16 Oct 2026 20:57:16 +0000 Subject: [PATCH 4/8] fix(evaluations): open no evaluate span for a row that did not generate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TESTING.md §8.10 now matches §8.8.6: a row whose handler raised is never scored, so it has no evaluate span. The span test also pins the judge chat transcript with a messages-mode judge whose user turn expands {{message_history}}. Co-authored-by: christopher.schmitz --- .../evaluations/runner.py | 24 +++++---- .../client/tests/test_evaluations_spans.py | 52 +++++++++++++++---- 2 files changed, 56 insertions(+), 20 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index b54cef5b..ca23d5d7 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -1167,11 +1167,25 @@ async def _run_criteria_for_results( """ controller = ConcurrencyController(concurrency) + async def score(row: Mapping[str, Any], criterion: Criterion) -> dict[str, Any]: + if isinstance(criterion, Scorer): + return await self._run_scorer_for_result(row, criterion) + return await self._run_ld_judge_for_result( + row, + tool_handlers, + criterion, + resolved_judges[criterion.key], + ) + async def run_one( row: Mapping[str, Any], criterion: Criterion ) -> dict[str, Any]: await controller.acquire() try: + if row.get("status") != "COMPLETE": + # A row that did not generate is never scored, so there is + # no judge or scorer call for an ``evaluate`` span to wrap. + return await score(row, criterion) row_index = row.get("row_index") index = row_index if isinstance(row_index, int) else 0 with bind_evaluation_row(evaluation_run_id, index): @@ -1181,15 +1195,7 @@ async def run_one( criterion.criterion_type, (row_spans or {}).get(index), ) as span: - if isinstance(criterion, Scorer): - event = await self._run_scorer_for_result(row, criterion) - else: - event = await self._run_ld_judge_for_result( - row, - tool_handlers, - criterion, - resolved_judges[criterion.key], - ) + event = await score(row, criterion) _finish_criterion_span(span, event) return event finally: diff --git a/packages/client/tests/test_evaluations_spans.py b/packages/client/tests/test_evaluations_spans.py index 2dcd6a09..48fd28b4 100644 --- a/packages/client/tests/test_evaluations_spans.py +++ b/packages/client/tests/test_evaluations_spans.py @@ -160,7 +160,9 @@ def one_row_transport() -> SequencedTransport: ) -def patch_judge(monkeypatch: pytest.MonkeyPatch) -> None: +def patch_judge( + monkeypatch: pytest.MonkeyPatch, prompt: dict[str, Any] | None = None +) -> None: async def fake_extract_variation( key: str, context: dict[str, Any] ) -> dict[str, Any]: @@ -169,7 +171,7 @@ async def fake_extract_variation( "config": { "provider": {"name": "OpenAI"}, "model": {"name": "gpt-4o"}, - "instructions": "Judge {{response_to_evaluate}}", + **(prompt or {"instructions": "Judge {{response_to_evaluate}}"}), }, "meta": {"variationKey": "default", "version": 12}, } @@ -233,7 +235,15 @@ async def test_plain_function_row_is_one_trace_with_the_exchange_tools_and_judge ) -> None: exporter, flushed = span_export forbid_new_exporter(monkeypatch) - patch_judge(monkeypatch) + patch_judge( + monkeypatch, + { + "messages": [ + {"role": "system", "content": "Judge {{response_to_evaluate}}"}, + {"role": "user", "content": "Transcript:\n{{message_history}}"}, + ] + }, + ) async def client_flush() -> None: flushed.append("client") @@ -253,7 +263,7 @@ async def handler( variables: dict[str, Any], ) -> dict[str, Any]: assert "__ld" not in variables - if "Judge" in config.get("instructions", ""): + if config.get("messages"): return { "output": '{"score": 0.9, "reasoning": "Correct and concise."}', "usage": {"input_tokens": 30, "output_tokens": 9}, @@ -357,9 +367,20 @@ async def handler( assert json.loads(str(judge_attributes["gen_ai.system_instructions"])) == [ {"type": "text", "content": "Judge shipped"} ] - assert json.loads(str(judge_attributes["gen_ai.input.messages"])) == [ - {"role": "user", "parts": [{"type": "text", "content": "shipped"}]} - ] + history_turn, judged_turn = json.loads( + str(judge_attributes["gen_ai.input.messages"]) + ) + assert history_turn["role"] == "user" + (history_part,) = history_turn["parts"] + history = history_part["content"] + assert history.startswith("Transcript:\n") + assert "{{message_history}}" not in history + for fragment in ("Where is order A1?", "lookup_order", _TOOL_RESULT, "shipped"): + assert fragment in history + assert judged_turn == { + "role": "user", + "parts": [{"type": "text", "content": "shipped"}], + } assert json.loads(str(judge_attributes["gen_ai.output.messages"])) == [ { "role": "assistant", @@ -418,9 +439,13 @@ async def handler( tool_handlers: dict[str, Callable[..., Any]], variables: dict[str, Any], ) -> dict[str, Any]: + nonlocal calls + calls += 1 tool_handlers["lookup_order"]({"order_id": "A1"}) raise RuntimeError("model refused") + calls = 0 + await evals.run( key="support-qa", dataset="golden", @@ -442,10 +467,15 @@ async def handler( assert "gen_ai.input.messages" in attributes (tool,) = by_name["execute_tool lookup_order"] assert (tool.attributes or {})["gen_ai.tool.call.result"] == _TOOL_RESULT - (evaluate,) = by_name["evaluate $ld:ai:judge:accuracy"] - assert evaluate.status.status_code == StatusCode.ERROR - assert (evaluate.attributes or {})["error.type"] == "generation_incomplete" - assert parent_of(evaluate) == row.context.span_id + assert sorted(by_name) == [ + "chat gpt-4o", + "execute_tool lookup_order", + "launchdarkly.dataset_row", + ] + for child in (chat, tool): + assert parent_of(child) == row.context.span_id + assert child.context.trace_id == row.context.trace_id + assert calls == 1 @pytest.mark.asyncio From 0049c768715da8f94eda556a92edeeb18087308b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 16 Oct 2026 21:25:28 +0000 Subject: [PATCH 5/8] test(evaluations): assert each span wraps its call and the row is one trace Spans are stamped from timestamps captured around the call, so a child's window sits inside its parent's and the tool call's inside the generation chat's. Containment is the invariant; a duration floor would assert the machine's speed instead. Co-authored-by: christopher.schmitz --- .../client/tests/test_evaluations_spans.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/packages/client/tests/test_evaluations_spans.py b/packages/client/tests/test_evaluations_spans.py index 48fd28b4..bbffdcec 100644 --- a/packages/client/tests/test_evaluations_spans.py +++ b/packages/client/tests/test_evaluations_spans.py @@ -417,6 +417,24 @@ async def handler( assert criterion["reason"] == "Correct and concise." assert flushed == ["client", "traces"] + # Each span is timed to the call it describes, so its window sits inside + # its parent's. A canned handler makes these windows microseconds wide; + # a duration floor would assert the machine's speed, containment asserts + # that the span wrapped the call. + for span in spans: + assert span.end_time > span.start_time + for child, parent in ( + (generation_chat, row), + (tool, row), + (evaluate, row), + (judge_chat, evaluate), + # The tool ran inside the handler, so its window is inside the + # generation chat's even though the span is a sibling of it. + (tool, generation_chat), + ): + assert parent.start_time <= child.start_time + assert child.end_time <= parent.end_time + @pytest.mark.asyncio async def test_plain_function_that_raises_records_the_failed_generation( From 35717be4ff37b5580c85c30021cb385954ba94a0 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 19 Oct 2026 17:32:04 +0000 Subject: [PATCH 6/8] refactor(evaluations): keep evaluation names out of core conversation.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Core now exposes two feature-neutral span primitives, stamp_span_attributes and count_sdk_spans, and the processor's span-start path has no evaluation branch. The evaluation run keys, the ai-evaluator UUID5 namespace, the row-id derivation, bind_evaluation_row and record_evaluation_result move to evaluations/spans.py (TESTING.md §0.3, §8.10). Co-authored-by: christopher.schmitz --- .../launchdarkly_ai_server/conversation.py | 119 ++++++------------ .../evaluations/runner.py | 24 ++-- .../evaluations/spans.py | 60 ++++++++- .../src/launchdarkly_ai_server/trajectory.py | 2 +- packages/client/tests/test_conversation.py | 102 ++++++++------- .../tests/test_evaluations_row_binding.py | 90 +++++++++++++ .../client/tests/test_evaluations_spans.py | 6 +- 7 files changed, 255 insertions(+), 148 deletions(-) create mode 100644 packages/client/tests/test_evaluations_row_binding.py diff --git a/packages/client/src/launchdarkly_ai_server/conversation.py b/packages/client/src/launchdarkly_ai_server/conversation.py index 2cd86701..165d2eea 100644 --- a/packages/client/src/launchdarkly_ai_server/conversation.py +++ b/packages/client/src/launchdarkly_ai_server/conversation.py @@ -6,8 +6,13 @@ from __future__ import annotations -import uuid -from collections.abc import AsyncGenerator, AsyncIterator, Callable, Iterator +from collections.abc import ( + AsyncGenerator, + AsyncIterator, + Callable, + Iterator, + Mapping, +) from contextlib import aclosing, asynccontextmanager, contextmanager from dataclasses import dataclass from time import time_ns @@ -26,17 +31,11 @@ _SpanProcessorBase = object GEN_AI_CONVERSATION_ID = "gen_ai.conversation.id" -LD_EVALUATION_RUN_ID = "launchdarkly.evaluation_run.id" -LD_EVALUATION_RUN_ROW_INDEX = "launchdarkly.evaluation_run.row_index" - -# Same namespace ai-evaluator uses for conversationId on a result row. -# Changing it re-keys every offline evaluation conversation. -_EVALUATION_CONVERSATION_NAMESPACE = uuid.UUID("6f0c2a4e-3b6d-4d0e-9c1a-2e6b7f8a9d01") _CONV_KEY = otel_context.create_key("launchdarkly.gen_ai.conversation.id") _EVAL_KEY = otel_context.create_key("launchdarkly.judge.evaluation") -_EVAL_ROW_KEY = otel_context.create_key("launchdarkly.evaluation_run.row") -_OBSERVER_KEY = otel_context.create_key("launchdarkly.evaluation_run.observer") +_STAMP_KEY = otel_context.create_key("launchdarkly.span.stamped_attributes") +_COUNTER_KEY = otel_context.create_key("launchdarkly.span.counter") @dataclass @@ -90,54 +89,25 @@ def _conversation_id_from(ctx: otel_context.Context | None) -> str | None: return value if isinstance(value, str) and value else None -def conversation_id_for_evaluation_row(run_id: str, row_index: int | None) -> str: - """UUID5 of ``{run_id}:{row_index}`` in the ai-evaluator namespace. - - Row index ``0`` is the suffix ``:0``. A missing index is row 0, matching - the workers that persist a missing index that way. - """ - index = 0 if row_index is None else int(row_index) - return str(uuid.uuid5(_EVALUATION_CONVERSATION_NAMESPACE, f"{run_id}:{index}")) - - -def _evaluation_row_attributes( - run_id: str, row_index: int | None -) -> dict[str, str | int]: - index = 0 if row_index is None else int(row_index) - return { - LD_EVALUATION_RUN_ID: str(run_id), - LD_EVALUATION_RUN_ROW_INDEX: index, - } - - -def _evaluation_row_from(ctx: otel_context.Context | None) -> dict[str, Any] | None: - value = otel_context.get_value(_EVAL_ROW_KEY, ctx) if ctx is not None else None +def _stamped_attributes_from(ctx: otel_context.Context | None) -> Mapping[str, Any]: + value = otel_context.get_value(_STAMP_KEY, ctx) if ctx is not None else None if value is None: - value = otel_context.get_value(_EVAL_ROW_KEY) - if not isinstance(value, dict) or not value: - return None - return value + value = otel_context.get_value(_STAMP_KEY) + return value if isinstance(value, Mapping) else {} @contextmanager -def bind_evaluation_row(run_id: str, row_index: int | None) -> Iterator[None]: - """Bind one offline-evaluation row onto SDK spans started in this block. - - Sets ``gen_ai.conversation.id`` to :func:`conversation_id_for_evaluation_row` - and ``launchdarkly.evaluation_run.id`` / ``.row_index``. Does not set - ``launchdarkly.run.id``; that key is the AI Config tracking id. - - An outer :func:`conversation_id` is shadowed until the block exits. The - evaluation row id is the join key for this row. +def stamp_span_attributes(attributes: Mapping[str, str | int]) -> Iterator[None]: + """Write ``attributes`` onto every SDK span started in this block. + + Each key is written only when the span does not already carry it, and only + onto spans from this SDK's tracers. Values must be strings or integers; + anything else is skipped. A nested block adds to the outer set, and its own + values win on a shared key. Relies on :class:`ConversationIdSpanProcessor`, + which ``init_client`` registers. """ - index = 0 if row_index is None else int(row_index) - ctx = otel_context.set_value( - _CONV_KEY, conversation_id_for_evaluation_row(run_id, index) - ) - ctx = otel_context.set_value( - _EVAL_ROW_KEY, _evaluation_row_attributes(run_id, index), ctx - ) - token = otel_context.attach(ctx) + merged = {**_stamped_attributes_from(otel_context.get_current()), **attributes} + token = otel_context.attach(otel_context.set_value(_STAMP_KEY, merged)) try: yield finally: @@ -145,40 +115,27 @@ def bind_evaluation_row(run_id: str, row_index: int | None) -> Iterator[None]: @dataclass -class HandlerSpanObserver: - """Counts SDK spans started while :func:`observe_handler_spans` is bound.""" +class SpanCounter: + """Number of SDK spans started while :func:`count_sdk_spans` is bound.""" count: int = 0 @contextmanager -def observe_handler_spans() -> Iterator[HandlerSpanObserver]: - """Count the SDK spans a handler opens inside this block. +def count_sdk_spans() -> Iterator[SpanCounter]: + """Count the spans this SDK's tracers start inside this block. - The offline runner describes a plain-function handler's call itself, and - must not when a provider handler already traced it. Counting relies on + Only the innermost block counts a span. Relies on :class:`ConversationIdSpanProcessor`, which ``init_client`` registers. """ - observer = HandlerSpanObserver() - token = otel_context.attach(otel_context.set_value(_OBSERVER_KEY, observer)) + counter = SpanCounter() + token = otel_context.attach(otel_context.set_value(_COUNTER_KEY, counter)) try: - yield observer + yield counter finally: otel_context.detach(token) -def record_evaluation_result( - span: Any, name: str, score: float, explanation: str | None = None -) -> None: - """Write ``gen_ai.evaluation.result`` with no verdict and no score label. - - ``explanation`` is the criterion reason. Offline evaluation always records - it when the reason is non-empty: that text already travels on the criterion - event. This does not add a prompt, a completion, or a tool payload. - """ - _record_evaluation(span, name, score, explanation) - - def _record_evaluation( span: Any, name: str, score: float, explanation: str | None = None ) -> None: @@ -354,7 +311,7 @@ def record(score: float, explanation: str | None = None) -> None: class ConversationIdSpanProcessor(_SpanProcessorBase): - """Stamps conversation and evaluation-row attributes; delays judge ``invoke_agent`` end. + """Stamps the conversation id and bound attributes, counts SDK spans, delays judge ``invoke_agent`` end. Structurally a ``SpanProcessor``; the base is only real to a type checker so that an api-only install (no ``[otel]`` extra) still imports this module. @@ -373,13 +330,13 @@ def on_start( if conv and _is_launchdarkly_span(span): set_conversation_id_if_absent(span, conv) - observer = otel_context.get_value(_OBSERVER_KEY, ctx) - if isinstance(observer, HandlerSpanObserver) and _is_launchdarkly_span(span): - observer.count += 1 + counter = otel_context.get_value(_COUNTER_KEY, ctx) + if isinstance(counter, SpanCounter) and _is_launchdarkly_span(span): + counter.count += 1 - row = _evaluation_row_from(ctx) - if row and _is_launchdarkly_span(span): - for key, value in row.items(): + stamped = _stamped_attributes_from(ctx) + if stamped and _is_launchdarkly_span(span): + for key, value in stamped.items(): if not isinstance(key, str) or _read_attribute(span, key) is not None: continue if isinstance(value, str): diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index ca23d5d7..aba7091d 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -16,15 +16,7 @@ from opentelemetry import trace from opentelemetry.trace import Status, StatusCode -from ..conversation import ( - GEN_AI_CONVERSATION_ID, - LD_EVALUATION_RUN_ID, - LD_EVALUATION_RUN_ROW_INDEX, - bind_evaluation_row, - conversation_id_for_evaluation_row, - observe_handler_spans, - record_evaluation_result, -) +from ..conversation import count_sdk_spans from ..judge_scoring import ( FORMATTING_INSTRUCTIONS, build_message_history, @@ -61,9 +53,12 @@ TokenUsage, ) from .spans import ( + bind_evaluation_row, record_chat, + record_evaluation_result, record_tool_calls, set_config_model_identity, + set_row_attributes, start_row_span, ) from .tools import ( @@ -280,12 +275,7 @@ def _criterion_span( f"evaluate {criterion_type}", context=parent ) as span: span.set_attribute("gen_ai.operation.name", "evaluate") - span.set_attribute( - GEN_AI_CONVERSATION_ID, - conversation_id_for_evaluation_row(run_id, row_index), - ) - span.set_attribute(LD_EVALUATION_RUN_ID, run_id) - span.set_attribute(LD_EVALUATION_RUN_ROW_INDEX, row_index) + set_row_attributes(span, run_id, row_index) yield span @@ -745,7 +735,7 @@ async def invoke(row: DatasetRow) -> dict[str, Any]: with ( bind_evaluation_row(evaluation_run_id, row.row_index), trace.use_span(row_span, end_on_exit=False), - observe_handler_spans() as observer, + count_sdk_spans() as observer, ): try: result = await handler( @@ -1080,7 +1070,7 @@ async def _run_ld_judge_for_result( result: Any = None raised: BaseException | None = None try: - with observe_handler_spans() as observer: + with count_sdk_spans() as observer: try: result = await execution.handler( dict(judge_config), diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/spans.py b/packages/client/src/launchdarkly_ai_server/evaluations/spans.py index 25526b38..87647403 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/spans.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/spans.py @@ -18,7 +18,9 @@ from __future__ import annotations import json -from collections.abc import Mapping, Sequence +import uuid +from collections.abc import Iterator, Mapping, Sequence +from contextlib import contextmanager from typing import Any from opentelemetry import trace @@ -35,9 +37,9 @@ ) from ..conversation import ( GEN_AI_CONVERSATION_ID, - LD_EVALUATION_RUN_ID, - LD_EVALUATION_RUN_ROW_INDEX, - conversation_id_for_evaluation_row, + _record_evaluation, + conversation_id, + stamp_span_attributes, ) from ..trajectory import ToolInvocation from ..utils import ( @@ -54,6 +56,56 @@ #: The customer-project copy always carries content; see the module docstring. _CAPTURE = True +LD_EVALUATION_RUN_ID = "launchdarkly.evaluation_run.id" +LD_EVALUATION_RUN_ROW_INDEX = "launchdarkly.evaluation_run.row_index" + +# Same namespace ai-evaluator uses for conversationId on a result row. +# Changing it re-keys every offline evaluation conversation. +_EVALUATION_CONVERSATION_NAMESPACE = uuid.UUID("6f0c2a4e-3b6d-4d0e-9c1a-2e6b7f8a9d01") + + +def conversation_id_for_evaluation_row(run_id: str, row_index: int | None) -> str: + """UUID5 of ``{run_id}:{row_index}`` in the ai-evaluator namespace. + + Row index ``0`` is the suffix ``:0``. A missing index is row 0, matching + the workers that persist a missing index that way. + """ + index = 0 if row_index is None else int(row_index) + return str(uuid.uuid5(_EVALUATION_CONVERSATION_NAMESPACE, f"{run_id}:{index}")) + + +@contextmanager +def bind_evaluation_row(run_id: str, row_index: int | None) -> Iterator[None]: + """Bind one dataset row onto SDK spans started in this block. + + Sets ``gen_ai.conversation.id`` to :func:`conversation_id_for_evaluation_row` + and ``launchdarkly.evaluation_run.id`` / ``.row_index``. Does not set + ``launchdarkly.run.id``; that key is the AI Config tracking id. + + An outer conversation id is shadowed until the block exits. The evaluation + row id is the join key for this row. + """ + index = 0 if row_index is None else int(row_index) + with ( + conversation_id(conversation_id_for_evaluation_row(run_id, index)), + stamp_span_attributes( + {LD_EVALUATION_RUN_ID: str(run_id), LD_EVALUATION_RUN_ROW_INDEX: index} + ), + ): + yield + + +def record_evaluation_result( + span: Any, name: str, score: float, explanation: str | None = None +) -> None: + """Write ``gen_ai.evaluation.result`` with no verdict and no score label. + + ``explanation`` is the criterion reason. Offline evaluation always records + it when the reason is non-empty: that text already travels on the criterion + event. This does not add a prompt, a completion, or a tool payload. + """ + _record_evaluation(span, name, score, explanation) + def _tracer() -> trace.Tracer: return trace.get_tracer(TRACER_NAME) diff --git a/packages/client/src/launchdarkly_ai_server/trajectory.py b/packages/client/src/launchdarkly_ai_server/trajectory.py index 1e728112..002200e4 100644 --- a/packages/client/src/launchdarkly_ai_server/trajectory.py +++ b/packages/client/src/launchdarkly_ai_server/trajectory.py @@ -58,7 +58,7 @@ class ToolInvocation: arguments: Any = None result: Any = None error: str | None = None - #: Epoch nanoseconds, for the offline runner's ``execute_tool`` spans. + #: Epoch nanoseconds around the call, so a caller can time a span to it. #: Excluded from equality: a judge sees the same step whenever it ran. started_ns: int | None = field(default=None, compare=False) ended_ns: int | None = field(default=None, compare=False) diff --git a/packages/client/tests/test_conversation.py b/packages/client/tests/test_conversation.py index 9d599068..056b3235 100644 --- a/packages/client/tests/test_conversation.py +++ b/packages/client/tests/test_conversation.py @@ -12,13 +12,11 @@ from launchdarkly_ai_server.conversation import ( GEN_AI_CONVERSATION_ID, - LD_EVALUATION_RUN_ID, - LD_EVALUATION_RUN_ROW_INDEX, ConversationIdSpanProcessor, - bind_evaluation_row, conversation_id, - conversation_id_for_evaluation_row, + count_sdk_spans, set_conversation_id_if_absent, + stamp_span_attributes, with_judge_evaluation, ) @@ -235,52 +233,70 @@ async def test_omits_explanation_when_not_supplied(self) -> None: assert not any("explanation" in k for k in (event.attributes or {})) -class TestEvaluationRowBinding: - """Offline rows share ai-evaluator's UUID5 and must not reuse launchdarkly.run.id.""" +class TestStampSpanAttributes: + """The core primitive names no attribute of its own; callers supply the set.""" - def test_uuid5_matches_the_published_namespace(self) -> None: - assert ( - conversation_id_for_evaluation_row("run-id", 3) - == "faa03308-8fe7-5805-8c07-f82a3dbdb3f1" - ) - assert ( - conversation_id_for_evaluation_row("run-id", 0) - == "8b90044e-ac6d-5af1-88a3-b75fbb9c8268" - ) - assert conversation_id_for_evaluation_row("run-id", None) == ( - conversation_id_for_evaluation_row("run-id", 0) - ) - - def test_stamps_sdk_spans_and_not_run_id(self) -> None: - with bind_evaluation_row("run-id", 3): + def test_stamps_supplied_attributes_onto_sdk_spans(self) -> None: + with stamp_span_attributes({"example.key": "value", "example.count": 7}): span = _tracer.start_span("invoke_agent") span.end() - recorded = finished()[0] - assert recorded.attributes is not None - assert ( - recorded.attributes[GEN_AI_CONVERSATION_ID] - == "faa03308-8fe7-5805-8c07-f82a3dbdb3f1" - ) - assert recorded.attributes[LD_EVALUATION_RUN_ID] == "run-id" - assert recorded.attributes[LD_EVALUATION_RUN_ROW_INDEX] == 3 - assert "launchdarkly.run.id" not in recorded.attributes + attributes = finished()[0].attributes or {} + assert attributes["example.key"] == "value" + assert attributes["example.count"] == 7 + + def test_write_if_absent(self) -> None: + with stamp_span_attributes({"example.key": "bound"}): + span = _tracer.start_span("chat", attributes={"example.key": "own"}) + span.end() + assert (finished()[0].attributes or {})["example.key"] == "own" def test_does_not_stamp_third_party_spans(self) -> None: foreign = _provider.get_tracer("opentelemetry.instrumentation.httpx") - with bind_evaluation_row("run-id", 3): + with stamp_span_attributes({"example.key": "value"}): span = foreign.start_span("POST") span.end() - attributes = finished()[0].attributes or {} - assert GEN_AI_CONVERSATION_ID not in attributes - assert LD_EVALUATION_RUN_ID not in attributes - assert LD_EVALUATION_RUN_ROW_INDEX not in attributes + assert "example.key" not in (finished()[0].attributes or {}) - def test_row_id_shadows_an_outer_conversation_id(self) -> None: - with conversation_id("thread-123"): - with bind_evaluation_row("run-id", 0): - span = _tracer.start_span("chat gpt-4o") + def test_nested_block_adds_to_and_overrides_the_outer_set(self) -> None: + with stamp_span_attributes({"example.a": "outer", "example.b": "outer"}): + with stamp_span_attributes({"example.b": "inner"}): + span = _tracer.start_span("chat") span.end() - assert ( - finished()[0].attributes[GEN_AI_CONVERSATION_ID] - == "8b90044e-ac6d-5af1-88a3-b75fbb9c8268" - ) + attributes = finished()[0].attributes or {} + assert attributes["example.a"] == "outer" + assert attributes["example.b"] == "inner" + + def test_skips_values_that_are_not_str_or_int(self) -> None: + bad: dict[str, object] = {"example.flag": True, "example.ratio": 0.5} + with stamp_span_attributes(bad): # type: ignore[arg-type] + span = _tracer.start_span("chat") + span.end() + attributes = finished()[0].attributes or {} + assert "example.flag" not in attributes + assert "example.ratio" not in attributes + + def test_nothing_is_stamped_outside_the_block(self) -> None: + with stamp_span_attributes({"example.key": "value"}): + pass + span = _tracer.start_span("chat") + span.end() + assert "example.key" not in (finished()[0].attributes or {}) + + +class TestCountSdkSpans: + def test_counts_only_sdk_spans_started_in_the_block(self) -> None: + foreign = _provider.get_tracer("opentelemetry.instrumentation.httpx") + with count_sdk_spans() as counter: + _tracer.start_span("invoke_agent").end() + _tracer.start_span("chat").end() + foreign.start_span("POST").end() + _tracer.start_span("after").end() + assert counter.count == 2 + + def test_inner_block_counts_its_own_spans(self) -> None: + with count_sdk_spans() as outer: + with count_sdk_spans() as inner: + _tracer.start_span("chat").end() + _tracer.start_span("chat").end() + assert inner.count == 1 + assert outer.count == 1 diff --git a/packages/client/tests/test_evaluations_row_binding.py b/packages/client/tests/test_evaluations_row_binding.py new file mode 100644 index 00000000..60ec0008 --- /dev/null +++ b/packages/client/tests/test_evaluations_row_binding.py @@ -0,0 +1,90 @@ +"""The evaluations module's row binding, built on the core stamping primitives.""" + +from __future__ import annotations + +from collections.abc import Iterator + +import pytest +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from launchdarkly_ai_server.conversation import ( + GEN_AI_CONVERSATION_ID, + ConversationIdSpanProcessor, + conversation_id, +) +from launchdarkly_ai_server.evaluations.spans import ( + LD_EVALUATION_RUN_ID, + LD_EVALUATION_RUN_ROW_INDEX, + bind_evaluation_row, + conversation_id_for_evaluation_row, +) + +_exporter = InMemorySpanExporter() +_provider = TracerProvider() +_provider.add_span_processor(ConversationIdSpanProcessor()) +_provider.add_span_processor(SimpleSpanProcessor(_exporter)) +_tracer = _provider.get_tracer("@launchdarkly/ai-server") + + +@pytest.fixture(autouse=True) +def _reset_exporter() -> Iterator[None]: + _exporter.clear() + yield + _exporter.clear() + + +def finished() -> list[ReadableSpan]: + return list(_exporter.get_finished_spans()) + + +class TestEvaluationRowBinding: + """Offline rows share ai-evaluator's UUID5 and must not reuse launchdarkly.run.id.""" + + def test_uuid5_matches_the_published_namespace(self) -> None: + assert ( + conversation_id_for_evaluation_row("run-id", 3) + == "faa03308-8fe7-5805-8c07-f82a3dbdb3f1" + ) + assert ( + conversation_id_for_evaluation_row("run-id", 0) + == "8b90044e-ac6d-5af1-88a3-b75fbb9c8268" + ) + assert conversation_id_for_evaluation_row("run-id", None) == ( + conversation_id_for_evaluation_row("run-id", 0) + ) + + def test_stamps_sdk_spans_and_not_run_id(self) -> None: + with bind_evaluation_row("run-id", 3): + span = _tracer.start_span("invoke_agent") + span.end() + recorded = finished()[0] + assert recorded.attributes is not None + assert ( + recorded.attributes[GEN_AI_CONVERSATION_ID] + == "faa03308-8fe7-5805-8c07-f82a3dbdb3f1" + ) + assert recorded.attributes[LD_EVALUATION_RUN_ID] == "run-id" + assert recorded.attributes[LD_EVALUATION_RUN_ROW_INDEX] == 3 + assert "launchdarkly.run.id" not in recorded.attributes + + def test_does_not_stamp_third_party_spans(self) -> None: + foreign = _provider.get_tracer("opentelemetry.instrumentation.httpx") + with bind_evaluation_row("run-id", 3): + span = foreign.start_span("POST") + span.end() + attributes = finished()[0].attributes or {} + assert GEN_AI_CONVERSATION_ID not in attributes + assert LD_EVALUATION_RUN_ID not in attributes + assert LD_EVALUATION_RUN_ROW_INDEX not in attributes + + def test_row_id_shadows_an_outer_conversation_id(self) -> None: + with conversation_id("thread-123"): + with bind_evaluation_row("run-id", 0): + span = _tracer.start_span("chat gpt-4o") + span.end() + assert ( + finished()[0].attributes[GEN_AI_CONVERSATION_ID] + == "8b90044e-ac6d-5af1-88a3-b75fbb9c8268" + ) diff --git a/packages/client/tests/test_evaluations_spans.py b/packages/client/tests/test_evaluations_spans.py index bbffdcec..aea3b4b0 100644 --- a/packages/client/tests/test_evaluations_spans.py +++ b/packages/client/tests/test_evaluations_spans.py @@ -21,11 +21,13 @@ from launchdarkly_ai_server.conversation import ( GEN_AI_CONVERSATION_ID, - LD_EVALUATION_RUN_ID, - LD_EVALUATION_RUN_ROW_INDEX, ConversationIdSpanProcessor, ) from launchdarkly_ai_server.evaluations import HttpResponse +from launchdarkly_ai_server.evaluations.spans import ( + LD_EVALUATION_RUN_ID, + LD_EVALUATION_RUN_ROW_INDEX, +) from launchdarkly_ai_server.experimental.evaluations import ( EvalTool, Judge, From 836521f1484a7afcc1fc8f7ca6c3e13df941914b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 19 Oct 2026 21:02:11 +0000 Subject: [PATCH 7/8] docs(evaluations): warn that offline spans export content by default The process capture_content default stays false. Synthetic evaluation spans keep writing the transcript, and a test shows a normal handler span written with that resolved default still omits it. Co-authored-by: christopher.schmitz --- packages/client/README.md | 4 +- .../evaluations/spans.py | 10 +- .../client/tests/test_evaluations_spans.py | 151 ++++++++++++++++++ 3 files changed, 161 insertions(+), 4 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index bbe0170f..7edf2d1e 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -50,6 +50,8 @@ No code changes are required — `init_client()` detects the packages at runtime > none of these names is exported from the package root. They may change in a minor release, > so review the changelog when you upgrade. +**Privacy.** `capture_content` defaults to false, and ordinary handler spans omit prompts, completions, tool arguments, tool results, and judge explanations unless you set it true. An offline evaluation does not follow that default. The conversation view is not useful without the transcript, so `run()` exports it on the evaluation spans even when `capture_content` is false, including when a handler was built with `capture_content=False`. There is no per-run switch. What leaves the process on those spans is the row input and messages, the rendered system and judge instructions, the model output, tool arguments and results, and the judge completion and explanation. The `$ld:ai:offline-evals:generation` and `$ld:ai:offline-evals:criterion` events do not carry any of it. Do not put text in a dataset row that you are not willing to export as trace content. + The evaluations harness reads an LD-hosted dataset, creates a new evaluation and API-source run, and invokes your handler once per row. Rows can then be scored by LaunchDarkly judges and local scorer functions; see [Score rows with judges and scorers](#score-rows-with-judges-and-scorers). Each success or error queues a `$ld:ai:offline-evals:generation` custom event containing the evaluation, run, dataset, and row identifiers plus output or error (`errorMessage` is included for `ERROR` rows), nested `usage.inputTokens`/`usage.outputTokens`, timing, and stable hashes. Dataset-owned input, expected output, metadata, and variables are not duplicated in the event. Each queued event is logged at `INFO` on the `launchdarkly_ai_server.evaluations.runner` logger with its RFC3339 UTC `emittedAt` timestamp and stable `eventId`, making it possible to compare SDK emission time with ClickHouse arrival time once that logger is enabled. The same `emittedAt` value is included in the event payload. Events are flushed before the summary is fetched and the call returns; handlers are never rerun to retry event delivery. Pass/fail is derived from LaunchDarkly's run summary. Result links use `ui_base_uri`, then `LD_UI_BASE_URI`, then `https://app.launchdarkly.com`; this is independent of `LD_API_BASE_URI`. After flushing generation events, the harness polls the run summary endpoint until passed + failed + error rows fully account for a nonzero total with no pending rows, polling every `poll_interval_seconds` (default 2s) up to `poll_timeout_seconds` (default 180s); pass either to `run()` to widen both for large datasets. The summary endpoint does not return run state, so `RunSummary` exposes row counts only. A run passes only when the completed summary has no failed, error, or pending rows. `failed_rows` counts rows whose criteria were scored and did not meet their threshold, so a gate that ignored it would exit 0 on a run where every row failed its judge. Evaluation keys must be unique because every call creates a new evaluation with `POST`. @@ -240,7 +242,7 @@ warning and ignored. ### Process-wide defaults -`configure` sets behavior for the process. It does not initialize the LaunchDarkly client; call `init_client` for that. Handlers built without an explicit `capture_content` read this value when they run, so one startup call covers every handler. An explicit `True` or `False` on a handler factory or convenience wrapper overrides it. Call `configure` from process startup, not from request data. It can be replaced until the first AI invocation; after that only the same options are accepted. `shutdown()` restores the built-in default, which is off. +`configure` sets behavior for the process. It does not initialize the LaunchDarkly client; call `init_client` for that. Handlers built without an explicit `capture_content` read this value when they run, so one startup call covers every handler. An explicit `True` or `False` on a handler factory or convenience wrapper overrides it. Call `configure` from process startup, not from request data. It can be replaced until the first AI invocation; after that only the same options are accepted. `shutdown()` restores the built-in default, which is off. Offline evaluation spans are the exception: they export the transcript even when this default is off. See [Run an evaluation from code](#run-an-evaluation-from-code-experimental). ```python from launchdarkly_ai_server import SDKOptions, configure, init_client, shutdown diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/spans.py b/packages/client/src/launchdarkly_ai_server/evaluations/spans.py index 87647403..0a65391d 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/spans.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/spans.py @@ -11,8 +11,11 @@ ``evaluate`` span, a ``chat`` span for the judge call. These are written only when the handler opened no SDK span of its own, so a call is never traced twice. -These spans export to the customer's project on the SDK key, the copy that -always carries content. The ``$ld:ai:offline-evals:*`` events never do. +These spans export to the customer's project on the SDK key. They carry the +transcript even though ``capture_content`` defaults to false: the conversation +view is not useful without it. That is scoped to the spans this module writes. +Handler spans outside an evaluation still follow the resolved flag. The +``$ld:ai:offline-evals:*`` events never carry the transcript. """ from __future__ import annotations @@ -53,7 +56,8 @@ TRACER_NAME = "@launchdarkly/ai-server" ROW_SPAN_NAME = "launchdarkly.dataset_row" -#: The customer-project copy always carries content; see the module docstring. +#: Always on. Does not read ``capture_content`` or :func:`effective_capture_content`. +#: Handler spans pass the resolved flag into the same helpers and stay gated. _CAPTURE = True LD_EVALUATION_RUN_ID = "launchdarkly.evaluation_run.id" diff --git a/packages/client/tests/test_evaluations_spans.py b/packages/client/tests/test_evaluations_spans.py index aea3b4b0..db0327ca 100644 --- a/packages/client/tests/test_evaluations_spans.py +++ b/packages/client/tests/test_evaluations_spans.py @@ -7,6 +7,7 @@ import json from collections.abc import Callable, Iterator +from time import time_ns from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -19,6 +20,14 @@ from opentelemetry.trace import StatusCode from opentelemetry.util._once import Once +from launchdarkly_ai_server import SDKOptions, configure +from launchdarkly_ai_server.content import ( + ToolDefinitionInput, + set_input_content_attributes, + set_output_content_attributes, + set_tool_call_content_attributes, + text_message, +) from launchdarkly_ai_server.conversation import ( GEN_AI_CONVERSATION_ID, ConversationIdSpanProcessor, @@ -27,12 +36,21 @@ from launchdarkly_ai_server.evaluations.spans import ( LD_EVALUATION_RUN_ID, LD_EVALUATION_RUN_ROW_INDEX, + record_chat, + record_evaluation_result, + record_tool_calls, + start_row_span, ) from launchdarkly_ai_server.experimental.evaluations import ( EvalTool, Judge, init_evaluations, ) +from launchdarkly_ai_server.lifecycle import ( + _reset_for_testing, + effective_capture_content, +) +from launchdarkly_ai_server.trajectory import ToolInvocation _RUN_ID = "run-id" _ROW_INDEX = 3 @@ -654,3 +672,136 @@ async def handler( assert criterion["status"] == "ERROR" assert "verdict" not in criterion assert "score" not in criterion + + +def test_offline_spans_export_content_when_capture_content_is_false( + span_export: tuple[InMemorySpanExporter, list[str]], +) -> None: + """The built-in default is false. Evaluation spans still carry the transcript. + + A span written the way a normal handler writes one, through the shared + content helpers and the resolved default, carries none of it. + """ + exporter, _flushed = span_export + _reset_for_testing() + try: + assert SDKOptions().capture_content is False + configure(options=SDKOptions(capture_content=False)) + assert effective_capture_content(None) is False + + started = time_ns() + row = start_row_span("run-id", 0) + record_chat( + row, + config={ + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + "instructions": "Answer about {{topic}}", + "messages": [{"role": "user", "content": "earlier {{topic}}"}], + "tools": { + "lookup": { + "name": "lookup", + "description": "look up a place", + "parameters": {"type": "object"}, + } + }, + }, + user_input="Where is Paris?", + variables={"topic": "capitals"}, + started_ns=started, + ended_ns=started + 1_000, + result={ + "output": "France", + "usage": {"input_tokens": 3, "output_tokens": 1}, + }, + tool_calls=[ + ToolInvocation( + name="lookup", + arguments='{"q":"Paris"}', + result="France", + started_ns=started, + ended_ns=started + 100, + ) + ], + ) + record_tool_calls( + row, + [ + ToolInvocation( + name="lookup", + arguments='{"q":"Paris"}', + result="France", + started_ns=started, + ended_ns=started + 100, + ) + ], + ) + evaluate = otel_trace.get_tracer("@launchdarkly/ai-server").start_span( + "evaluate accuracy" + ) + record_evaluation_result(evaluate, "accuracy", 0.9, "matches the capital") + evaluate.end() + row.end() + + by_name = spans_by_name(exporter) + chat_attributes = by_name["chat gpt-4o"][0].attributes or {} + assert "Answer about capitals" in str( + chat_attributes["gen_ai.system_instructions"] + ) + rendered_input = str(chat_attributes["gen_ai.input.messages"]) + assert "Where is Paris?" in rendered_input + assert "earlier capitals" in rendered_input + assert "France" in str(chat_attributes["gen_ai.output.messages"]) + assert "lookup" in str(chat_attributes["gen_ai.tool.definitions"]) + tool_attributes = by_name["execute_tool lookup"][0].attributes or {} + assert tool_attributes["gen_ai.tool.call.arguments"] == '{"q":"Paris"}' + assert tool_attributes["gen_ai.tool.call.result"] == "France" + explanation = next( + event.attributes + for event in by_name["evaluate accuracy"][0].events + if event.name == "gen_ai.evaluation.result" + ) + assert explanation is not None + assert explanation["gen_ai.evaluation.explanation"] == "matches the capital" + + capture = effective_capture_content(None) + assert capture is False + handler_span = otel_trace.get_tracer("@launchdarkly/ai-server").start_span( + "invoke_agent" + ) + set_input_content_attributes( + handler_span, + capture, + system_instructions="secret instructions", + messages=[text_message("user", "secret input")], + tool_definitions=[ + ToolDefinitionInput(name="lookup", description="secret tool") + ], + ) + set_output_content_attributes( + handler_span, capture, [text_message("assistant", "secret output")] + ) + set_tool_call_content_attributes( + handler_span, capture, arguments='{"q":"secret"}', result="secret result" + ) + handler_span.end() + exported = next( + span + for span in exporter.get_finished_spans() + if span.name == "invoke_agent" + ) + handler_attributes = exported.attributes or {} + for key in ( + "gen_ai.system_instructions", + "gen_ai.input.messages", + "gen_ai.output.messages", + "gen_ai.tool.definitions", + "gen_ai.tool.call.arguments", + "gen_ai.tool.call.result", + "gen_ai.prompt.0.content", + "gen_ai.completion.0.content", + ): + assert key not in handler_attributes + assert not any("secret" in str(value) for value in handler_attributes.values()) + finally: + _reset_for_testing() From f8a96f30d6ab68a889c7a1bccb82539c50b37635 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 16:04:57 +0000 Subject: [PATCH 8/8] feat(client): stamp launchdarkly.project_id as the SDK key Keep highlight.project_id set to the same SDK key so older readers still match. The new name is the SDK key, not the project's BSON id. Co-authored-by: christopher.schmitz --- packages/client/README.md | 4 +++- .../src/launchdarkly_ai_server/lifecycle.py | 19 +++++++++++++++---- .../client/tests/test_evaluations_spans.py | 7 ++++++- packages/client/tests/test_lifecycle.py | 8 +++++--- 4 files changed, 29 insertions(+), 9 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index 7edf2d1e..29ad18d9 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -32,7 +32,7 @@ No code changes are required — `init_client()` detects the packages at runtime | Variable | Required | Description | |---|---|---| -| `LD_SDK_KEY` | Yes | LaunchDarkly server-side SDK key | +| `LD_SDK_KEY` | Yes | LaunchDarkly server-side SDK key. `init_client` stamps it as the OTel resource attributes `launchdarkly.project_id` and `highlight.project_id` (the SDK key, not the project's BSON id) | | `LD_BASE_URI` | No | Override the LaunchDarkly polling base URI (e.g. for staging) | | `LD_STREAM_URI` | No | Override the streaming URI | | `LD_EVENTS_URI` | No | Override the events URI | @@ -240,6 +240,8 @@ warning and ignored. | `serviceName` | `LD_SERVICE_NAME` | OTel `service.name` resource attribute | | `environment` | `LD_ENVIRONMENT` | `deployment.environment` resource attribute | +`init_client` also stamps `launchdarkly.project_id` and `highlight.project_id` on the tracer resource. Both are the SDK key from `sdkKey` / `LD_SDK_KEY`, not the project's BSON id. `launchdarkly.project_id` is the name; `highlight.project_id` is the same value. + ### Process-wide defaults `configure` sets behavior for the process. It does not initialize the LaunchDarkly client; call `init_client` for that. Handlers built without an explicit `capture_content` read this value when they run, so one startup call covers every handler. An explicit `True` or `False` on a handler factory or convenience wrapper overrides it. Call `configure` from process startup, not from request data. It can be replaced until the first AI invocation; after that only the same options are accepted. `shutdown()` restores the built-in default, which is off. Offline evaluation spans are the exception: they export the transcript even when this default is off. See [Run an evaluation from code](#run-an-evaluation-from-code-experimental). diff --git a/packages/client/src/launchdarkly_ai_server/lifecycle.py b/packages/client/src/launchdarkly_ai_server/lifecycle.py index d334311e..1db61b61 100644 --- a/packages/client/src/launchdarkly_ai_server/lifecycle.py +++ b/packages/client/src/launchdarkly_ai_server/lifecycle.py @@ -113,8 +113,11 @@ def _setup_telemetry(sdk_key: str, options: InitClientOptions | None = None) -> if OTel packages are not installed (graceful degradation). This function: - - Stamps ``service.name``, ``highlight.project_id``, and (when set) - ``deployment.environment`` resource attributes, mirroring the TS SDK. + - Stamps ``service.name``, ``launchdarkly.project_id``, + ``highlight.project_id``, and (when set) ``deployment.environment``. + Both project id attributes are the SDK key, not the project's BSON id. + ``launchdarkly.project_id`` is the name; ``highlight.project_id`` is the + same value. The TypeScript SDK still stamps only ``highlight.project_id``. - Registers W3C trace context and baggage propagators. - Configures GZIP compression on the OTLP exporter. """ @@ -151,6 +154,10 @@ def _setup_telemetry(sdk_key: str, options: InitClientOptions | None = None) -> resource_attrs: dict[str, str] = { "service.name": opts.get("serviceName") or os.environ.get("LD_SERVICE_NAME", "python-sdk"), + # Same SDK key under both names. launchdarkly.project_id is the + # name; highlight.project_id stays so older readers still match. + # Neither is the project's BSON id. + "launchdarkly.project_id": sdk_key, "highlight.project_id": sdk_key, } environment = opts.get("environment") or os.environ.get("LD_ENVIRONMENT") @@ -228,6 +235,9 @@ async def init_client( - ``environment`` (``LD_ENVIRONMENT``): the ``deployment.environment`` resource attribute. + The SDK key is also stamped as ``launchdarkly.project_id`` and + ``highlight.project_id``. Both are the SDK key, not the project's BSON id. + Idempotent: later calls return the existing client and ignore every option. An option not listed above is logged as a warning and ignored. @@ -358,8 +368,9 @@ def _flush_tracer_provider() -> None: """Export spans sitting on the provider ``init_client`` already registered. Does not build an exporter, replace the endpoint, or invent a project id. - The resource attribute ``highlight.project_id`` stays the SDK key that - ``_setup_telemetry`` stamped. A provider without ``force_flush`` — the API + The resource attributes ``launchdarkly.project_id`` and + ``highlight.project_id`` stay the SDK key that ``_setup_telemetry`` + stamped. A provider without ``force_flush`` — the API proxy, when the SDK extra is absent — is a no-op. """ try: diff --git a/packages/client/tests/test_evaluations_spans.py b/packages/client/tests/test_evaluations_spans.py index db0327ca..0f7833d6 100644 --- a/packages/client/tests/test_evaluations_spans.py +++ b/packages/client/tests/test_evaluations_spans.py @@ -85,7 +85,11 @@ def span_export() -> Iterator[tuple[InMemorySpanExporter, list[str]]]: exporter = InMemorySpanExporter() provider = TracerProvider( resource=Resource.create( - {"service.name": "python-sdk", "highlight.project_id": "sdk-key"} + { + "service.name": "python-sdk", + "launchdarkly.project_id": "sdk-key", + "highlight.project_id": "sdk-key", + } ) ) provider.add_span_processor(ConversationIdSpanProcessor()) @@ -316,6 +320,7 @@ async def handler( ] for span in spans: assert span.resource is not None + assert span.resource.attributes["launchdarkly.project_id"] == "sdk-key" assert span.resource.attributes["highlight.project_id"] == "sdk-key" assert_row_attributes(span.attributes) assert span.status.status_code == StatusCode.OK diff --git a/packages/client/tests/test_lifecycle.py b/packages/client/tests/test_lifecycle.py index 50a7e733..5bf3bb49 100644 --- a/packages/client/tests/test_lifecycle.py +++ b/packages/client/tests/test_lifecycle.py @@ -624,11 +624,13 @@ def _spy_create(attributes: dict | None = None) -> Resource: return created_resources, mock_exporter_cls - def test_stamps_highlight_project_id(self) -> None: - """highlight.project_id resource attribute must equal the resolved SDK key.""" + def test_stamps_project_id_attributes(self) -> None: + """Both project id resource attributes equal the resolved SDK key.""" created_resources, _ = self._setup_with_mock_exporter("my-sdk-key") assert any( - r.get("highlight.project_id") == "my-sdk-key" for r in created_resources + r.get("launchdarkly.project_id") == "my-sdk-key" + and r.get("highlight.project_id") == "my-sdk-key" + for r in created_resources ) def test_registers_w3c_propagators(self) -> None: