diff --git a/packages/client/README.md b/packages/client/README.md index 924aefdf..81d1bf61 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -52,6 +52,8 @@ No code changes are required — `init_client()` detects the packages at runtime 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. +Each row is also exported as one trace on the tracer provider `init_client` registered: a `launchdarkly.dataset_row` root with the generation, its tool calls, and each criterion's `evaluate` span beneath it, all sharing the row's `gen_ai.conversation.id`. A provider handler traces its own call; for a plain function the runner writes the `chat` and `execute_tool` spans. Prompts, output, tool payloads, and judge explanations follow the same [`capture_content`](#process-wide-defaults) gate as every other span, so they are omitted unless capture is on. + Management API requests go to the `api_base_uri` option, then `LD_API_BASE_URI`, then `https://app.launchdarkly.com`; `LD_BASE_URI` is never used, because it may point at a relay proxy. Result links use `ui_base_uri`, then `LD_UI_BASE_URI`, then `https://app.launchdarkly.com`; this is independent of `LD_API_BASE_URI`. At either level a blank value counts as unset, and a trailing `/` is stripped. 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`. ```python diff --git a/packages/client/src/launchdarkly_ai_server/conversation.py b/packages/client/src/launchdarkly_ai_server/conversation.py index ba19a9bb..4218b5cf 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 @@ -82,6 +90,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, @@ -275,7 +330,7 @@ def record( 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. @@ -294,6 +349,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 d73e8191..ec6e88a5 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, Literal, cast, overload -from ..lifecycle import get_client, init_client +from ..lifecycle import _flush_tracer_provider, get_client, init_client from ..utils import normalize_mode from .api import ( DEFAULT_BASE_URI, @@ -482,14 +482,17 @@ async def run( ) raise config = self._runner._build_handler_config(run_generation, run_tools) - results = await self._runner._run_rows( - dataset_rows, - generation_handler, - config, - run_tool_handlers, - concurrency, - ) + row_spans: dict[int, Any] = {} try: + results = await self._runner._run_rows( + dataset_rows, + generation_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, @@ -504,6 +507,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, @@ -515,10 +520,19 @@ 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. A span + # that fails to end must not skip the event flush. + try: + for row_span in row_spans.values(): + row_span.end() + finally: + 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 e76bf0ea..cd9fb177 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -7,12 +7,17 @@ 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, cast +from opentelemetry import trace +from opentelemetry.trace import Status, StatusCode + from ..content import tool_result_string +from ..conversation import count_sdk_spans from ..judge_scoring import ( FORMATTING_INSTRUCTIONS, build_message_history, @@ -21,7 +26,11 @@ typesafe_judge_entries, without_output_format, ) -from ..lifecycle import extract_variation +from ..lifecycle import ( + _sdk_processor_is_active, + effective_capture_content, + extract_variation, +) from ..trajectory import ( TrajectoryRecorder, render_row_trajectory, @@ -51,6 +60,16 @@ LDJudgeCriterionEventPayload, TokenUsage, ) +from .spans import ( + bind_evaluation_row, + capture_for_call, + record_chat, + record_evaluation_result, + record_tool_calls, + set_config_model_identity, + set_row_attributes, + start_row_span, +) from .tools import ( EvalTool, ToolImplementation, @@ -212,6 +231,66 @@ 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], *, capture: bool | None = None +) -> 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``. + ``capture`` gates the reason on the span. The criterion event keeps it. + """ + 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, + capture=capture, + ) + 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().""" @@ -645,7 +724,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]: @@ -654,12 +741,46 @@ 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 and _sdk_processor_is_active(): + capture = capture_for_call(handler) + 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, + capture=capture, + ) + record_tool_calls( + row_span, recorder.invocations, capture=capture + ) if not isinstance(result, Mapping): raise TypeError("handler result must be a mapping") completed = datetime.now(UTC) @@ -680,8 +801,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, @@ -948,16 +1074,41 @@ async def _run_ld_judge_for_result( # parse_template pass, so ``{{...}}`` sequences inside generated output # or dataset values are never re-expanded into the judge prompt. variables = self._judge_variables(row, judge) + # Judges get no tools. The handler owns the single parse_template pass. + judge_variables = { + **variables, + "formatting_instructions": FORMATTING_INSTRUCTIONS, + } + evaluate_span = trace.get_current_span() + set_config_model_identity(evaluate_span, resolved.config) + started_ns = time.time_ns() + result: Any = None + raised: BaseException | None = None try: - result = await execution.handler( - dict(resolved.config), - variables["response_to_evaluate"], - {}, - { - **variables, - "formatting_instructions": FORMATTING_INSTRUCTIONS, - }, - ) + with count_sdk_spans() as observer: + try: + result = await execution.handler( + dict(resolved.config), + variables["response_to_evaluate"], + {}, + judge_variables, + ) + except Exception as error: + raised = error + raise + finally: + if observer.count == 0 and _sdk_processor_is_active(): + record_chat( + evaluate_span, + config=resolved.config, + user_input=variables["response_to_evaluate"], + variables=judge_variables, + started_ns=started_ns, + ended_ns=time.time_ns(), + result=result if isinstance(result, Mapping) else None, + error=raised, + capture=capture_for_call(execution.handler), + ) except Exception as error: return self._criterion_error_result( base, started_clock, "handler_raised", f"judge handler raised: {error}" @@ -1018,22 +1169,51 @@ 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, + 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, - 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) + capture = ( + capture_for_call(resolved_judges[criterion.key].handler) + if isinstance(criterion, Judge) + else effective_capture_content(None) + ) + _finish_criterion_span(span, event, capture=capture) + 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..14fbf559 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/spans.py @@ -0,0 +1,365 @@ +"""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 follow the +same content gate as every other span: the handler's explicit override when it +has one, otherwise the process default, which is off. An explicit ``True`` +exports the transcript on these spans too. 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 ..lifecycle import effective_capture_content +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" + +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 _resolved_capture(capture: bool | None) -> bool: + """An explicit bool wins. ``None`` is the process default.""" + return effective_capture_content(None) if capture is None else capture + + +def capture_for_call(handler: Any) -> bool: + """Content capture for one call. + + A handler exposes ``capture_content`` as its effective decision. A plain + function does not, so it follows the process default. + """ + flag = getattr(handler, "capture_content", None) + if isinstance(flag, bool): + return flag + return effective_capture_content(None) + + +def record_evaluation_result( + span: Any, + name: str, + score: float, + explanation: str | None = None, + *, + capture: bool | None = None, +) -> None: + """Write ``gen_ai.evaluation.result`` with no verdict and no score label. + + ``explanation`` is the criterion reason. It is written only when capture is + on for this call, the same gate as the online judge. The reason still + travels on the criterion event either way. This does not add a prompt, a + completion, or a tool payload. + """ + if not _resolved_capture(capture): + explanation = None + _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] = (), + capture: bool | None = None, +) -> 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) + resolved = _resolved_capture(capture) + if resolved: + system_instructions, messages = _prompt(config, user_input, variables) + set_input_content_attributes( + span, + resolved, + 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 and resolved: + set_output_content_attributes( + span, resolved, [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], + *, + capture: bool | None = None, +) -> 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") + resolved = _resolved_capture(capture) + if resolved: + set_tool_call_content_attributes( + span, + resolved, + 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 0d626f2d..f3c44a9d 100644 --- a/packages/client/src/launchdarkly_ai_server/lifecycle.py +++ b/packages/client/src/launchdarkly_ai_server/lifecycle.py @@ -396,6 +396,46 @@ def _release_otel_globals() -> None: logger.debug("Could not release the global OTel tracer provider", exc_info=True) +def _sdk_processor_is_active() -> bool: + """Whether the active tracer provider carries ``ConversationIdSpanProcessor``. + + Only then does ``count_sdk_spans`` see a handler's spans. Under a foreign + provider the count stays 0 even when the handler traced itself, so a + caller that falls back to writing spans on a 0 count would write them twice. + """ + if _owns_otel_globals: + return True + try: + from opentelemetry import trace + except ImportError: + return False + from .conversation import ConversationIdSpanProcessor + + # ponytail: reads the SDK TracerProvider's private processor list; a provider + # without it (proxy, vendor wrapper) reads as foreign, which only skips the fallback. + multi = getattr(trace.get_tracer_provider(), "_active_span_processor", None) + processors = getattr(multi, "_span_processors", ()) + return any(isinstance(p, ConversationIdSpanProcessor) for p in processors) + + +def _flush_tracer_provider() -> None: + """Export spans queued on the provider ``init_client`` registered. + + A provider without ``force_flush`` (the API proxy) 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 37044bac..e435172b 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, ) @@ -249,3 +251,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 435d0f4c..9f012252 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -3367,6 +3367,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..4dd806d3 --- /dev/null +++ b/packages/client/tests/test_evaluations_spans.py @@ -0,0 +1,897 @@ +"""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, create_handler +from launchdarkly_ai_server.conversation import ( + GEN_AI_CONVERSATION_ID, + ConversationIdSpanProcessor, +) +from launchdarkly_ai_server.evaluations.api 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.""" + yield from _install_provider(sdk_processor=True) + + +@pytest.fixture +def foreign_span_export() -> Iterator[tuple[InMemorySpanExporter, list[str]]]: + """A provider someone else registered, without the SDK processor.""" + yield from _install_provider(sdk_processor=False) + + +def _install_provider( + *, sdk_processor: bool +) -> Iterator[tuple[InMemorySpanExporter, list[str]]]: + 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", + } + ) + ) + if sdk_processor: + 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" + body = prompt or {"instructions": "Judge {{response_to_evaluate}}"} + mode = "messages" if "messages" in body else "agent" + return { + "config": { + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + **body, + }, + "meta": {"variationKey": "default", "version": 12, "mode": mode}, + } + + 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 declare(fn: Callable[..., Any], mode: str = "agent") -> Any: + """A plain function with the identity ``run()`` now requires for routing. + + The wrapper opens no span of its own, so the runner still describes the call. + """ + return create_handler(("OpenAI", mode), fn) + + +def lookup_order_tool() -> EvalTool: + return EvalTool( + key="lookup_order", + implementation=lambda args: _TOOL_RESULT, + schema={"type": "object"}, + description="Look up an order", + ) + + +@pytest.fixture +def capture_on() -> Iterator[None]: + """Transcript-shape cases run with capture on. The built-in default is off.""" + _reset_for_testing() + configure(options=SDKOptions(capture_content=True)) + yield + _reset_for_testing() + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("capture_on") +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], + history: list[dict[str, Any]] | None = None, + ) -> 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", + handlers=[declare(handler), declare(handler, "messages")], + tools=[lookup_order_tool()], + generation={ + "provider": "OpenAI", + "model": "gpt-4o", + "mode": "agent", + "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["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 +@pytest.mark.usefixtures("capture_on") +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], + history: list[dict[str, Any]] | None = None, + ) -> 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=declare(handler), + tools=[lookup_order_tool()], + generation={"provider": "OpenAI", "model": "gpt-4o", "mode": "agent"}, + 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], + history: list[dict[str, Any]] | None = None, + ) -> 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=declare(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", "mode": "agent"}, + 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_a_foreign_provider_gets_no_second_copy_of_handler_spans( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, + foreign_span_export: tuple[InMemorySpanExporter, list[str]], +) -> None: + # Without the SDK processor the span count stays 0 even though the handler + # traced itself, so the runner must not write its own chat spans too. + exporter, _flushed = foreign_span_export + 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") + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + history: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + with tracer.start_as_current_span("invoke_agent"): + with tracer.start_as_current_span("chat gpt-4o"): + pass + if "Judge" in config.get("instructions", ""): + return {"output": '{"score": 0.9, "reasoning": "ok"}'} + return {"output": "done"} + + await evals.run( + key="support-qa", + dataset="golden", + handler=declare(handler), + generation={"provider": "OpenAI", "model": "gpt-4o", "mode": "agent"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + by_name = spans_by_name(exporter) + # One from the generation and one from the judge, each written once. + assert len(by_name["chat gpt-4o"]) == 2 + assert len(by_name["invoke_agent"]) == 2 + + +@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], + history: list[dict[str, Any]] | None = None, + ) -> 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=declare(handler), + generation={"provider": "OpenAI", "model": "gpt-4o", "mode": "agent"}, + 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 + + +_CONTENT_KEYS = ( + "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", +) + + +def _write_captured_row(*, capture: bool | None = None) -> None: + started = time_ns() + row = start_row_span("run-id", 0) + chat_capture = {} if capture is None else {"capture": capture} + 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, + ) + ], + **chat_capture, + ) + record_tool_calls( + row, + [ + ToolInvocation( + name="lookup", + arguments='{"q":"Paris"}', + result="France", + started_ns=started, + ended_ns=started + 100, + ) + ], + **chat_capture, + ) + evaluate = otel_trace.get_tracer("@launchdarkly/ai-server").start_span( + "evaluate accuracy" + ) + record_evaluation_result( + evaluate, "accuracy", 0.9, "matches the capital", **chat_capture + ) + evaluate.end() + row.end() + + +def test_offline_spans_follow_the_capture_content_gate( + span_export: tuple[InMemorySpanExporter, list[str]], +) -> None: + """Evaluation spans omit the transcript by default and export it when capture is on. + + An explicit true on the call still writes it when the process default is false. + """ + exporter, _flushed = span_export + _reset_for_testing() + try: + assert SDKOptions().capture_content is False + assert effective_capture_content(None) is False + + _write_captured_row() + by_name = spans_by_name(exporter) + for attributes in ( + by_name["chat gpt-4o"][0].attributes or {}, + by_name["execute_tool lookup"][0].attributes or {}, + ): + for key in _CONTENT_KEYS: + assert key not in attributes + result = next( + event.attributes + for event in by_name["evaluate accuracy"][0].events + if event.name == "gen_ai.evaluation.result" + ) + assert result is not None + assert "gen_ai.evaluation.explanation" not in result + assert result["gen_ai.evaluation.name"] == "accuracy" + assert result["gen_ai.evaluation.score.value"] == 0.9 + assert not any( + "Paris" in str(value) + for value in (by_name["chat gpt-4o"][0].attributes or {}).values() + ) + + exporter.clear() + configure(options=SDKOptions(capture_content=True)) + assert effective_capture_content(None) is True + _write_captured_row() + 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" + + exporter.clear() + configure(options=SDKOptions(capture_content=False)) + _write_captured_row(capture=True) + kept = spans_by_name(exporter)["chat gpt-4o"][0].attributes or {} + assert "Where is Paris?" in str(kept["gen_ai.input.messages"]) + finally: + _reset_for_testing()