diff --git a/packages/client/README.md b/packages/client/README.md index bbe0170f..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 | @@ -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`. @@ -238,9 +240,11 @@ 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. +`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/conversation.py b/packages/client/src/launchdarkly_ai_server/conversation.py index d3af0e9e..165d2eea 100644 --- a/packages/client/src/launchdarkly_ai_server/conversation.py +++ b/packages/client/src/launchdarkly_ai_server/conversation.py @@ -6,7 +6,13 @@ from __future__ import annotations -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 @@ -28,6 +34,8 @@ _CONV_KEY = otel_context.create_key("launchdarkly.gen_ai.conversation.id") _EVAL_KEY = otel_context.create_key("launchdarkly.judge.evaluation") +_STAMP_KEY = otel_context.create_key("launchdarkly.span.stamped_attributes") +_COUNTER_KEY = otel_context.create_key("launchdarkly.span.counter") @dataclass @@ -81,6 +89,53 @@ def _conversation_id_from(ctx: otel_context.Context | None) -> str | None: return value if isinstance(value, str) and value 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(_STAMP_KEY) + return value if isinstance(value, Mapping) else {} + + +@contextmanager +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. + """ + merged = {**_stamped_attributes_from(otel_context.get_current()), **attributes} + token = otel_context.attach(otel_context.set_value(_STAMP_KEY, merged)) + try: + yield + finally: + otel_context.detach(token) + + +@dataclass +class SpanCounter: + """Number of SDK spans started while :func:`count_sdk_spans` is bound.""" + + count: int = 0 + + +@contextmanager +def count_sdk_spans() -> Iterator[SpanCounter]: + """Count the spans this SDK's tracers start inside this block. + + Only the innermost block counts a span. Relies on + :class:`ConversationIdSpanProcessor`, which ``init_client`` registers. + """ + counter = SpanCounter() + token = otel_context.attach(otel_context.set_value(_COUNTER_KEY, counter)) + try: + yield counter + finally: + otel_context.detach(token) + + def _record_evaluation( span: Any, name: str, score: float, explanation: str | None = None ) -> None: @@ -256,7 +311,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 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. @@ -275,6 +330,20 @@ def on_start( if conv and _is_launchdarkly_span(span): set_conversation_id_if_absent(span, conv) + counter = otel_context.get_value(_COUNTER_KEY, ctx) + if isinstance(counter, SpanCounter) and _is_launchdarkly_span(span): + counter.count += 1 + + 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): + 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..8940da95 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,17 @@ 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, - ) + row_spans: dict[int, Any] = {} try: + results = await self._runner._run_rows( + dataset_rows, + handler, + config, + run_tool_handlers, + concurrency, + evaluation_run_id=evaluation_run.id, + row_spans=row_spans, + ) self._runner._emit_generation_events( client, project_key=self._project_key, @@ -449,6 +452,8 @@ async def run( run_criteria, resolved_judges, concurrency, + evaluation_run_id=evaluation_run.id, + row_spans=row_spans, ) self._runner._emit_evaluation_events( client, @@ -460,10 +465,16 @@ 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. + for row_span in row_spans.values(): + row_span.end() + 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..aba7091d 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -7,11 +7,16 @@ 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 count_sdk_spans from ..judge_scoring import ( FORMATTING_INSTRUCTIONS, build_message_history, @@ -47,6 +52,15 @@ LDJudgeCriterionEventPayload, 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 ( EvalTool, ToolImplementation, @@ -242,6 +256,62 @@ def record_rate_limit( del provider, retry_after +@contextmanager +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") + 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") + set_row_attributes(span, run_id, 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, + ) + span.set_status(Status(StatusCode.OK)) + return + 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: + span.set_attribute("error.type", code) + + class EvaluationsRunner: """Private API operations and orchestration used by EvaluationsModule.run().""" @@ -635,7 +705,15 @@ async def _run_rows( config: dict[str, Any], tool_handlers: dict[str, ToolImplementation], 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]: @@ -644,12 +722,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: - 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), + count_sdk_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) @@ -670,8 +778,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, @@ -947,16 +1060,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 count_sdk_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}" @@ -1011,23 +1147,47 @@ async def _run_criteria_for_results( criteria: list[Criterion], resolved_judges: Mapping[str, JudgeExecution], 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 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 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], - ) + 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): + with _criterion_span( + evaluation_run_id, + index, + criterion.criterion_type, + (row_spans or {}).get(index), + ) as span: + event = await score(row, criterion) + _finish_criterion_span(span, event) + return event finally: controller.release() 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..0a65391d --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/spans.py @@ -0,0 +1,333 @@ +"""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. 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 + +import json +import uuid +from collections.abc import Iterator, Mapping, Sequence +from contextlib import contextmanager +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, + _record_evaluation, + conversation_id, + stamp_span_attributes, +) +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" + +#: 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" +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) + + +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/src/launchdarkly_ai_server/lifecycle.py b/packages/client/src/launchdarkly_ai_server/lifecycle.py index ed767400..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. @@ -354,6 +364,28 @@ 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 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: + 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) + + async def shutdown() -> None: """ Shuts down the singleton client. Idempotent — safe to call multiple times diff --git a/packages/client/src/launchdarkly_ai_server/trajectory.py b/packages/client/src/launchdarkly_ai_server/trajectory.py index 071925db..002200e4 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 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) 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(), ) diff --git a/packages/client/tests/test_conversation.py b/packages/client/tests/test_conversation.py index 83b5f8bf..056b3235 100644 --- a/packages/client/tests/test_conversation.py +++ b/packages/client/tests/test_conversation.py @@ -14,7 +14,9 @@ GEN_AI_CONVERSATION_ID, ConversationIdSpanProcessor, conversation_id, + count_sdk_spans, set_conversation_id_if_absent, + stamp_span_attributes, with_judge_evaluation, ) @@ -229,3 +231,72 @@ 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 TestStampSpanAttributes: + """The core primitive names no attribute of its own; callers supply the set.""" + + 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() + 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 stamp_span_attributes({"example.key": "value"}): + span = foreign.start_span("POST") + span.end() + assert "example.key" not in (finished()[0].attributes or {}) + + 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() + 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_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..0f7833d6 --- /dev/null +++ b/packages/client/tests/test_evaluations_spans.py @@ -0,0 +1,812 @@ +"""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 time import time_ns +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 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, +) +from launchdarkly_ai_server.evaluations import HttpResponse +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 +_CONVERSATION_ID = "faa03308-8fe7-5805-8c07-f82a3dbdb3f1" +_TOOL_ARGUMENTS = '{"order_id": "A1"}' +_TOOL_RESULT = "shipped" +_TOOL_ARGUMENTS_RENDERED = '{"order_id":"A1"}' + + +@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", + "launchdarkly.project_id": "sdk-key", + "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, prompt: dict[str, Any] | None = None +) -> 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"}, + **(prompt or {"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) + + +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_row_is_one_trace_with_the_exchange_tools_and_judge( + 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, + { + "messages": [ + {"role": "system", "content": "Judge {{response_to_evaluate}}"}, + {"role": "user", "content": "Transcript:\n{{message_history}}"}, + ] + }, + ) + + 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 config.get("messages"): + 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, + 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.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["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 + + (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" + ) + 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) + + 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"} + ] + 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", + "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] + 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 + 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 + criterion = payloads[1] + assert criterion["score"] == 0.9 + 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( + 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]: + 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", + 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 + 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 +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) + # 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" + 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,) = 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" + 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 + + +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() 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: