From b5b690f49fed748deedcf7f300e5188bb66171e1 Mon Sep 17 00:00:00 2001 From: Dylan O'Neill Date: Tue, 15 Sep 2026 16:20:14 -0700 Subject: [PATCH 1/2] feat(evaluations): preserve the tool trajectory for judges Handler packages record tool traffic onto spans and return only {output, usage}, so by the time a criterion ran the calls a row made on its way to that output were gone -- which made "did the agent call the right tool, in the right order, with the right arguments?" an unaskable question of an SDK-run evaluation that had just run the agent that answered it. The runner now records the trajectory itself, wrapping the caller's tool implementations once per row before handing them to the handler. Wrapping is what covers every handler package without changing any of them: a handler still resolves a tool by the key the model named and calls it. The trajectory reaches judges through message_history, interleaved between the row input and the generated output -- which is where it happened, and which is the variable every judge cloned from the AI Library's default templates already references, so a trajectory rubric needs no new judge template. There is deliberately no standalone trajectory variable: message_history is already the transcript variable, and a second overlapping one only invited a rubric to interpolate both and pay for the trajectory twice. A run with no observable tools adds no block, so judges authored before this read exactly the history they read before. Three properties are pinned by tests. The recorder observes and never intervenes: a wrapped tool returns and raises what the original did, and calls past the recording cap still execute and are only counted. A recorder belongs to one row, since rows generate concurrently against one shared tool map. And a tool result stays literal in the judge prompt -- it is a new injection surface, closed by the existing rule that the judge config is passed unrendered for the handler's single template pass. Native provider tools are passed through unwrapped and left out of the rendered "tools available" line: they execute inside the provider, so naming a tool whose use cannot be shown would invite a judge to conclude the model ignored it. Nothing about the trajectory is added to any event payload. Co-Authored-By: Claude Opus 5 --- packages/ai/README.md | 2 +- packages/client/README.md | 25 ++ .../evaluations/runner.py | 25 +- .../evaluations/trajectory.py | 256 +++++++++++++ packages/client/tests/test_evaluations_run.py | 338 ++++++++++++++++++ .../tests/test_evaluations_trajectory.py | 207 +++++++++++ 6 files changed, 851 insertions(+), 2 deletions(-) create mode 100644 packages/client/src/launchdarkly_ai_server/evaluations/trajectory.py create mode 100644 packages/client/tests/test_evaluations_trajectory.py diff --git a/packages/ai/README.md b/packages/ai/README.md index b1b47a70..829bb567 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -73,7 +73,7 @@ result = await evals.run( ) ``` -`LD_API_TOKEN` is required. Configure `LD_SDK_KEY` — or initialize your own client with `init_client(client=...)` — to emit one `$ld:ai:offline-evals:generation` event per generated row, plus one `$ld:ai:offline-evals:criterion` event per `(row, criterion)` when `criteria` are supplied, through the standard SDK event transport. The SDK reports scores; LaunchDarkly rules on them at ingest. A judge served by a different provider than `generation` needs a handler for it in `judge_handlers`. Use `LD_API_BASE_URI` for staging or local management API traffic; it is separate from the SDK delivery setting `LD_BASE_URI`. Evaluation-run links use the explicit `ui_base_uri` option or `LD_UI_BASE_URI` (for example, `https://ld-stg.launchdarkly.com` in staging), defaulting to `https://app.launchdarkly.com`. See the [core evaluations guide](../client/README.md#run-an-evaluation-from-code). +`LD_API_TOKEN` is required. Configure `LD_SDK_KEY` — or initialize your own client with `init_client(client=...)` — to emit one `$ld:ai:offline-evals:generation` event per generated row, plus one `$ld:ai:offline-evals:criterion` event per `(row, criterion)` when `criteria` are supplied, through the standard SDK event transport. The SDK reports scores; LaunchDarkly rules on them at ingest. A judge served by a different provider than `generation` needs a handler for it in `judge_handlers`. Each row's tool calls are recorded during generation and rendered into the judge's `{{message_history}}`, between the row input and the generated output, so a rubric can grade the tool trajectory as well as the final answer. Use `LD_API_BASE_URI` for staging or local management API traffic; it is separate from the SDK delivery setting `LD_BASE_URI`. Evaluation-run links use the explicit `ui_base_uri` option or `LD_UI_BASE_URI`, defaulting to `https://app.launchdarkly.com`; set it when the project is not in production, or a run created elsewhere still links to the production app. See the [core evaluations guide](../client/README.md#run-an-evaluation-from-code). --- diff --git a/packages/client/README.md b/packages/client/README.md index 1e50a5ec..92ca6b1c 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -116,6 +116,31 @@ result = await init_evaluations().run( **The SDK reports scores and never rules on them.** LaunchDarkly derives each row's verdict at ingest by comparing the score against the criterion's stored threshold and success direction, so pass/fail policy is one server-side implementation that applies to every SDK version and to runs already recorded. A judge's direction lives on its AI Config and is injected server-side, keeping the one input a verdict turns on server-attested; a `Scorer` has no LaunchDarkly-side config to read, so it declares its own `success_direction` (default `"higher_is_better"` — set `"lower_is_better"` for a scorer that counts something unwanted, like a regex hit count). +#### Judge the tool trajectory + +A judge is shown the tool calls the row made on the way to its output, so a rubric can grade *how* the agent answered and not only *what* it answered — whether it called the right tool, in the right order, with the right arguments, and how it handled a tool that failed. + +The harness records this itself: it wraps your tool implementations once per row before handing them to the handler, so every handler package is covered without changes and your tools still return and raise exactly what they did before. + +The trajectory is rendered into **`{{message_history}}`** — the row input, then the trajectory, then the generated output, then the formatting instructions, in that order. There is no separate trajectory variable: `message_history` is already the transcript variable every judge reads, and judges built from the AI Library's default templates reference it, so a trajectory rubric can be written against an existing judge template with no new placeholder. + +``` +Tools available: lookup_order, issue_refund +Tool calls made while producing the response, in order: +1. lookup_order + arguments: {"id":"A1"} + result: order A1 shipped 2026-08-02 +2. issue_refund + arguments: {"id":"A1","amount":19.99} + error: refund window closed +``` + +A row with tools that called none of them says so explicitly, which is the finding a tool-selection rubric most needs. A run with no tools adds no block at all, so judges written before trajectories existed read exactly the history they read before. + +Two limits keep a trajectory from spending the judge's context window: at most 50 recorded calls per row and 2000 characters per rendered argument bag or result, with anything beyond either reported as a count or marked truncated. Calls past the limit still execute — truncation drops the record, never the work. A `NativeTool` runs inside the provider, so no local wrapper sees it; such a tool is left out of the trajectory and out of the "Tools available" line, since naming a tool whose use cannot be shown would invite a judge to conclude the model ignored it. + +A tool result is now judge-prompt input. It stays literal for the same reason the generated output does: the judge config is handed to the handler unrendered and the handler makes exactly one template pass, so a `{{...}}` sequence coming back from a tool is never expanded into the judge prompt. + **Judges are independent AI Configs, so handlers are routed per judge.** A judge may resolve to a different provider or mode than `generation`, and a handler built for one provider cannot execute another's config. `handler` runs a judge when it provides for that judge's provider; pass handlers for any other providers in `judge_handlers`. Selection prefers a handler naming the judge's provider outright over a wildcard multi-provider adapter, and an agent-mode handler can serve a messages-mode judge with its messages collapsed into one instructions block. A plain callable that declares no `provides_for` routes itself, exactly as it already does for the generation config. Judges are resolved through flag delivery, and handlers are matched to them, **before** any evaluation records are created — a missing judge or one no handler covers fails the run up front rather than after the generation spend. After that point a criterion failure never aborts the run: an unparseable judge response, an out-of-range score, a raising handler or scorer, and a row whose generation errored each become a per-criterion `ERROR` event with a cause code (`invalid_judge_output`, `invalid_score`, `handler_raised`, `scorer_raised`, `generation_incomplete`) and a top-level `errorMessage`. Event *delivery* is different: the backend needs one result per `(row, criterion)` to finish row accounting, so if tracking a criterion event fails, every remaining result is still attempted and flushed and then `run()` raises — rather than polling to its timeout with the cause hidden. diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index 164ea57d..6738d68c 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -35,6 +35,7 @@ LDJudgeCriterionEventPayload, TokenUsage, ) +from .trajectory import TrajectoryRecorder, render_row_trajectory, row_fields from .types import ( DatasetRef, DatasetRow, @@ -544,11 +545,16 @@ async def _run_rows( async def invoke(row: DatasetRow) -> dict[str, Any]: await controller.acquire(config["provider"]["name"]) + # One recorder per row, not one per run: rows are generated + # concurrently against the same tool map, so a shared recorder + # would splice one row's tool calls into another's trajectory. + recorder = TrajectoryRecorder() + row_tool_handlers = recorder.wrap(tool_handlers) started = datetime.now(UTC) started_clock = time.perf_counter() try: result = await handler( - config, row.input, tool_handlers, dict(row.variables) + config, row.input, row_tool_handlers, dict(row.variables) ) if not isinstance(result, Mapping): raise TypeError("handler result must be a mapping") @@ -564,6 +570,7 @@ async def invoke(row: DatasetRow) -> dict[str, Any]: "generated_at": completed.isoformat().replace("+00:00", "Z"), "latency_ms": round((time.perf_counter() - started_clock) * 1000), "status": "COMPLETE", + **row_fields(recorder), } usage = result.get("usage") if isinstance(usage, Mapping): @@ -583,6 +590,9 @@ async def invoke(row: DatasetRow) -> dict[str, Any]: "latency_ms": round((time.perf_counter() - started_clock) * 1000), "status": "ERROR", "error": {"code": 5001, "message": f"handler raised: {error}"}, + # The calls that ran before the handler raised are what + # explain why it raised, so an errored row records them too. + **row_fields(recorder), } finally: controller.release() @@ -696,6 +706,12 @@ def _judge_variables( ground_truth = parse_template(ground_truth, variables) elif expected is not None: ground_truth = str(expected) + # The tool calls the row made on its way to `output`, recorded during + # generation (evaluations.trajectory). It sits between the input and the + # output in message_history because that is where it happened: a judge + # reading the history sees the request, what the agent did about it, and + # what it finally answered, in order. + trajectory = render_row_trajectory(row_result) # message_history carries FORMATTING_INSTRUCTIONS the same way the # online path builds it (judges.run_judges), because that -- not the # standalone formatting_instructions variable below -- is what every @@ -703,6 +719,12 @@ def _judge_variables( # relevance, toxicity, and any judge cloned from them) actually # references. A judge authored before this variable existed must keep # getting scored without edits. + # + # The trajectory goes here and nowhere else. It was briefly also + # exposed as a standalone tool_trajectory variable, which bought + # nothing: this is already the transcript variable every judge reads, + # and two overlapping variables only invited a rubric to interpolate + # both and pay for the trajectory twice. variables.update( { "input": row_result.get("input") or "", @@ -711,6 +733,7 @@ def _judge_variables( str(value) for value in ( row_result.get("input"), + trajectory, output, FORMATTING_INSTRUCTIONS, ) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/trajectory.py b/packages/client/src/launchdarkly_ai_server/evaluations/trajectory.py new file mode 100644 index 00000000..10ad15d5 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/trajectory.py @@ -0,0 +1,256 @@ +"""Tool-call trajectory capture for the generation phase of an SDK-run evaluation. + +A judge can only grade what it is shown. Handler packages record tool traffic +onto OpenTelemetry spans and return only ``{output, usage}``, so by the time a +criterion ran, the calls a row made on its way to that output were gone -- +which made "did the agent call the right tools, in the right order, with the +right arguments?" an unaskable question of an SDK-run evaluation, even though +the evaluation had just run the agent that answered it. + +The runner therefore records the trajectory itself, by wrapping the caller's +tool implementations once per row before handing them to the handler. Wrapping +is what makes this work with every handler package without changing any of +them: a handler looks a tool up by its key and calls it, exactly as before. + +Three properties are load-bearing. + +**The recorder observes; it never intervenes.** A wrapped tool returns what the +original returned and raises what the original raised. A row whose trajectory +hits :data:`MAX_RECORDED_TOOL_CALLS` still executes every remaining call -- +truncation drops the *record*, never the work, because an evaluation that +changed the agent's behavior would no longer be evaluating the agent. + +**A recorder belongs to one row.** ``_run_rows`` runs rows concurrently against +one shared tool map, so a single shared recorder would splice one row's calls +into another row's trajectory and hand the judge a conversation that never +happened. + +**Only observable tools are described.** A ``NativeTool`` is executed inside the +provider, so no local wrapper ever sees it and its calls cannot appear in the +trajectory. Such a tool is therefore left out of the rendered "tools available" +line as well: naming a tool whose use is invisible would let a judge conclude +the model ignored a tool it may well have called. +""" + +from __future__ import annotations + +import inspect +import json +from collections.abc import Callable, Mapping +from dataclasses import dataclass, replace +from typing import Any + +from ..types import NativeTool + +#: How many tool calls one row's trajectory records. A trajectory is +#: interpolated into a judge prompt, so an agent that loops over a large tool +#: result set would otherwise spend the judge's context window -- and its +#: budget -- on the tail of a trajectory the judge stopped reading. Calls past +#: the limit still execute and are reported as a count. +MAX_RECORDED_TOOL_CALLS = 50 + +#: How many characters one rendered argument bag or tool result contributes. +#: Bounds a single tool that returns a whole document, for the same reason. +MAX_RECORDED_VALUE_CHARS = 2000 + +_TRUNCATION_SUFFIX = "… (truncated)" + +ToolImplementation = Callable[..., Any] | NativeTool + + +@dataclass(frozen=True) +class ToolInvocation: + """One tool call made while generating a row, with how it turned out. + + ``result`` and ``error`` are mutually exclusive: a call that raised has no + result, and a call that returned has no error. Both are ``None`` on a call + that is still in flight, which is only observable from inside the wrapper. + """ + + name: str + arguments: Any = None + result: Any = None + error: str | None = None + + +class TrajectoryRecorder: + """Records one row's tool calls, in the order the calls were started. + + A slot is reserved when a call starts and filled in when it finishes, so + tools a handler runs concurrently keep their start order rather than being + reordered by which of them returned first. + """ + + def __init__(self, limit: int = MAX_RECORDED_TOOL_CALLS) -> None: + self._limit = limit + self._invocations: list[ToolInvocation] = [] + self._omitted = 0 + self._observable: list[str] = [] + + @property + def invocations(self) -> list[ToolInvocation]: + """The recorded calls, oldest first.""" + return list(self._invocations) + + @property + def omitted(self) -> int: + """How many calls executed past the recording limit.""" + return self._omitted + + @property + def observable_tools(self) -> list[str]: + """Keys of the tools this recorder can actually observe being called.""" + return list(self._observable) + + def wrap( + self, tool_handlers: Mapping[str, ToolImplementation] + ) -> dict[str, ToolImplementation]: + """Return ``tool_handlers`` with each callable recording into this row. + + Keys are preserved exactly: a handler resolves a tool by the key the + model named, so renaming one here would break the lookup. + """ + wrapped: dict[str, ToolImplementation] = {} + # Rebuilt rather than appended to, so re-wrapping a map does not report + # the same tool as available twice. + self._observable = [] + for name, implementation in tool_handlers.items(): + if isinstance(implementation, NativeTool) or not callable(implementation): + # Provider-executed, or already invalid and reported as such by + # tool resolution. Either way there is nothing local to observe, + # so pass the value through rather than replacing it with a + # wrapper the handler would treat differently. + wrapped[name] = implementation + continue + self._observable.append(name) + wrapped[name] = self._record(name, implementation) + return wrapped + + def _record(self, name: str, original: Callable[..., Any]) -> Callable[..., Any]: + async def wrapper(*args: Any, **kwargs: Any) -> Any: + slot = self._reserve(name, _call_arguments(args, kwargs)) + try: + result = original(*args, **kwargs) + if inspect.isawaitable(result): + result = await result + except Exception as error: + self._complete(slot, error=f"{error}") + raise + self._complete(slot, result=result) + return result + + return wrapper + + def _reserve(self, name: str, arguments: Any) -> int | None: + if len(self._invocations) >= self._limit: + self._omitted += 1 + return None + self._invocations.append(ToolInvocation(name=name, arguments=arguments)) + return len(self._invocations) - 1 + + def _complete( + self, slot: int | None, *, result: Any = None, error: str | None = None + ) -> None: + if slot is None: + return + self._invocations[slot] = replace( + self._invocations[slot], result=result, error=error + ) + + +def row_fields(recorder: TrajectoryRecorder) -> dict[str, Any]: + """The trajectory keys a generated-row record carries. + + Paired with :func:`render_row_trajectory` so one module owns both halves of + the record's shape: a key renamed here without its reader being updated + would silently render every row's trajectory as empty, which reads exactly + like an agent that called no tools. + """ + return { + "tool_calls": recorder.invocations, + "tool_calls_omitted": recorder.omitted, + "observable_tools": recorder.observable_tools, + } + + +def render_row_trajectory(row_result: Mapping[str, Any]) -> str: + """Render the trajectory carried by a generated-row record.""" + return render_trajectory( + row_result.get("tool_calls") or [], + observable_tools=row_result.get("observable_tools") or [], + omitted=int(row_result.get("tool_calls_omitted") or 0), + ) + + +def render_trajectory( + invocations: list[ToolInvocation], + *, + observable_tools: list[str] | None = None, + omitted: int = 0, +) -> str: + """Render a row's trajectory as the text a judge reads. + + Returns ``""`` when there was nothing observable to report, so the caller + can skip the block entirely rather than telling a judge about tools in a + run that had none. + + The empty trajectory of a row that *did* have tools is reported explicitly: + "this agent called nothing" is the finding a judge grading tool selection + most needs, and an omitted block would read as a run without tools. + """ + available = list(observable_tools or []) + if not available and not invocations: + return "" + + lines: list[str] = [] + if available: + lines.append(f"Tools available: {', '.join(available)}") + if not invocations: + lines.append("No tool calls were made while producing the response.") + return "\n".join(lines) + + lines.append("Tool calls made while producing the response, in order:") + for position, invocation in enumerate(invocations, start=1): + lines.append(f"{position}. {invocation.name}") + lines.append(f" arguments: {_render_value(invocation.arguments)}") + if invocation.error is not None: + lines.append(f" error: {_render_value(invocation.error)}") + else: + lines.append(f" result: {_render_value(invocation.result)}") + if omitted > 0: + lines.append(f"({omitted} further tool call(s) were made but not recorded.)") + return "\n".join(lines) + + +def _call_arguments(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: + """Normalize how a handler passed a tool its arguments. + + Every handler package in this SDK calls a tool with the model's argument + bag as one positional mapping, so that is the shape worth preserving + verbatim; the rest are recorded structurally rather than guessed at. + """ + if len(args) == 1 and not kwargs: + return args[0] + if kwargs and not args: + return dict(kwargs) + if not args and not kwargs: + return None + return {"args": list(args), "kwargs": dict(kwargs)} + + +def _render_value(value: Any) -> str: + if value is None: + return "" + if isinstance(value, str): + return _truncate(value) + try: + rendered = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + except (TypeError, ValueError): + rendered = str(value) + return _truncate(rendered) + + +def _truncate(text: str) -> str: + if len(text) <= MAX_RECORDED_VALUE_CHARS: + return text + return text[:MAX_RECORDED_VALUE_CHARS] + _TRUNCATION_SUFFIX diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index c7ca278b..d7a06b21 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -310,6 +310,17 @@ async def test_complete_run_with_zero_failed_and_error_rows_passes( assert event["emittedAt"].endswith("Z") assert datetime.fromisoformat(event["emittedAt"]).tzinfo is not None assert {"input", "expected_output", "metadata", "variables"}.isdisjoint(event) + # The captured tool trajectory reaches LaunchDarkly only inside the prompt a + # judge was shown, never as a generation wire field the backend has not + # specified. + assert { + "toolCalls", + "tool_calls", + "toolTrajectory", + "tool_trajectory", + "observableTools", + "observable_tools", + }.isdisjoint(event) emit_logs = [ record.getMessage() for record in caplog.records @@ -2249,3 +2260,330 @@ async def handler( assert result.passed is True assert max_in_flight == 2 + + +def tool_run_transport(*, rows: int = 1) -> SequencedTransport: + """Transport for a run that resolves one tool before its dataset.""" + return SequencedTransport( + [ + response(200, {"key": "lookup_order", "version": 4, "schema": {}}), + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [ + { + "rowIndex": index, + "input": f"Question {index}", + "expectedOutput": "Answer", + } + for index in range(rows) + ], + total=rows, + ), + ), + response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + { + "statusCounts": { + "total": rows, + "passed": rows, + "error": 0, + "pending": 0, + } + }, + ), + ] + ) + + +@pytest.mark.asyncio +async def test_tool_trajectory_reaches_the_judge_via_message_history( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + """The calls a row made are what let a judge grade its tool use. + + Handler packages return only {output, usage}, so without the runner + recording the trajectory itself a judge sees the answer and nothing about + how the agent arrived at it. + """ + transport = tool_run_transport() + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + seen: dict[str, str] = {} + + def lookup_order(args: dict[str, Any]) -> str: + return f"order {args['id']} shipped" + + 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", ""): + seen["message_history"] = variables["message_history"] + # The trajectory lives in message_history and nowhere else: this is + # already the transcript variable every judge reads, so a second + # overlapping variable only invited a rubric to pay for the + # trajectory twice. + assert "tool_trajectory" not in variables + return {"output": '{"score": 1, "reasoning": "used the right tool"}'} + assert await tool_handlers["lookup_order"]({"id": "A1"}) == "order A1 shipped" + return {"output": "Your order shipped."} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + tools={"lookup_order": lookup_order}, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + history = seen["message_history"] + assert "Tools available: lookup_order" in history + assert '1. lookup_order\n arguments: {"id":"A1"}' in history + assert "result: order A1 shipped" in history + # The trajectory sits between the request and the answer, because that is + # where it happened: a judge reading the history sees the question, what the + # agent did about it, then what it replied. + assert history.index("Question 0") < history.index("Tools available") + assert history.index("Tools available") < history.index("Your order shipped.") + assert history.index("Your order shipped.") < history.index( + "Your response MUST be in valid JSON" + ) + + +@pytest.mark.asyncio +async def test_each_row_gets_only_its_own_tool_trajectory( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + """Rows generate concurrently against one shared tool map. + + A recorder shared across rows would splice row 0's calls into row 1's + trajectory and hand the judge a conversation that never happened. + """ + import asyncio + + transport = tool_run_transport(rows=2) + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + histories: dict[str, str] = {} + both_started = asyncio.Barrier(2) + + def lookup_order(args: dict[str, Any]) -> str: + return f"order {args['id']}" + + 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", ""): + histories[str(user_input)] = variables["message_history"] + return {"output": '{"score": 1, "reasoning": "ok"}'} + row = str(user_input).split()[-1] + # Interleave the two rows' tool calls so a shared recorder would be + # caught rather than merely be possible. + await both_started.wait() + await tool_handlers["lookup_order"]({"id": row}) + return {"output": f"answered {row}"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + tools={"lookup_order": lookup_order}, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + concurrency=2, + ) + + assert result.passed is True + assert '{"id":"0"}' in histories["answered 0"] + assert '{"id":"1"}' not in histories["answered 0"] + assert '{"id":"1"}' in histories["answered 1"] + assert '{"id":"0"}' not in histories["answered 1"] + + +@pytest.mark.asyncio +async def test_a_row_that_called_no_tools_says_so_to_the_judge( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + """A judge grading tool selection needs to see the tool that went unused.""" + transport = tool_run_transport() + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + seen: dict[str, str] = {} + + 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", ""): + seen["message_history"] = variables["message_history"] + return {"output": '{"score": 0, "reasoning": "should have looked it up"}'} + return {"output": "I do not know."} + + await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + tools={"lookup_order": lambda args: "unused"}, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert ( + "Tools available: lookup_order\n" + "No tool calls were made while producing the response." + ) in seen["message_history"] + + +@pytest.mark.asyncio +async def test_a_run_without_tools_leaves_message_history_unchanged( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + """Judges authored before trajectories existed must read the same history. + + With no observable tools there is nothing to report, so no trajectory block + is added rather than one saying no tools were called. + """ + transport = judge_run_transport() + accuracy_judge_variation(monkeypatch) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + seen: dict[str, str] = {} + + 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", ""): + seen["message_history"] = variables["message_history"] + return {"output": '{"score": 1, "reasoning": "ok"}'} + return {"output": "generated"} + + await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert seen["message_history"].startswith("Question A\n\ngenerated\n\n") + assert "Tools available" not in seen["message_history"] + + +@pytest.mark.asyncio +async def test_tool_result_placeholders_are_not_expanded_into_the_judge_prompt( + monkeypatch: pytest.MonkeyPatch, + stub_sdk_client: MagicMock, +) -> None: + """A tool result is now judge-prompt input, so it is an injection surface. + + It stays literal for the same reason the generated output does: the judge + config is handed over unrendered and the handler makes exactly one template + pass, so a substituted value is never rescanned for placeholders. + """ + from launchdarkly_ai_server import parse_template + + transport = tool_run_transport() + + async def fake_extract_variation( + key: str, context: dict[str, Any] + ) -> dict[str, Any]: + return { + "config": { + "provider": {"name": "OpenAI"}, + "model": {"name": "gpt-4o"}, + "instructions": "Judge this history: {{message_history}}", + }, + "meta": {"variationKey": "default", "version": 12}, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.runner.extract_variation", + fake_extract_variation, + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=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 this history" in config.get("instructions", ""): + rendered = parse_template(config["instructions"], variables) + assert "result: {{expected_output}} leaked?" in rendered + assert "Answer leaked?" not in rendered + return {"output": '{"score": 1, "reasoning": "ok"}'} + await tool_handlers["lookup_order"]({"id": "A1"}) + return {"output": "done"} + + result = await evals.run( + project_key="proj", + key="support-qa", + dataset="golden", + handler=handler, + tools={"lookup_order": lambda args: "{{expected_output}} leaked?"}, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + criteria=[Judge(key="$ld:ai:judge:accuracy")], + ) + + assert result.passed is True + + +@pytest.mark.asyncio +async def test_a_failed_row_keeps_the_calls_made_before_the_handler_raised() -> None: + """The trajectory of a row that errored is what explains why it errored.""" + from launchdarkly_ai_server.evaluations.api import LDApiClient + from launchdarkly_ai_server.evaluations.runner import EvaluationsRunner + + runner = EvaluationsRunner( + LDApiClient(api_token="token", transport=failing_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]: + await tool_handlers["lookup_order"]({"id": "A1"}) + raise RuntimeError("model refused") + + results = await runner._run_rows( + [DatasetRow(row_index=0, input="Question")], + handler, + {"provider": {"name": "OpenAI"}, "model": {"name": "gpt-4o"}}, + {"lookup_order": lambda args: "shipped"}, + 1, + ) + + assert results[0]["status"] == "ERROR" + assert [invocation.name for invocation in results[0]["tool_calls"]] == [ + "lookup_order" + ] + assert results[0]["tool_calls"][0].result == "shipped" diff --git a/packages/client/tests/test_evaluations_trajectory.py b/packages/client/tests/test_evaluations_trajectory.py new file mode 100644 index 00000000..db90192c --- /dev/null +++ b/packages/client/tests/test_evaluations_trajectory.py @@ -0,0 +1,207 @@ +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest + +from launchdarkly_ai_server.evaluations.trajectory import ( + MAX_RECORDED_VALUE_CHARS, + ToolInvocation, + TrajectoryRecorder, + render_trajectory, +) +from launchdarkly_ai_server.types import NativeTool + + +@pytest.mark.asyncio +async def test_wrapped_tool_returns_what_the_original_returned() -> None: + def lookup(args: dict[str, Any]) -> str: + return f"order {args['id']}" + + recorder = TrajectoryRecorder() + wrapped = recorder.wrap({"lookup": lookup}) + + assert await wrapped["lookup"]({"id": "A1"}) == "order A1" + assert recorder.invocations == [ + ToolInvocation(name="lookup", arguments={"id": "A1"}, result="order A1") + ] + + +@pytest.mark.asyncio +async def test_wrapped_async_tool_is_awaited() -> None: + async def lookup(args: dict[str, Any]) -> str: + await asyncio.sleep(0) + return f"order {args['id']}" + + recorder = TrajectoryRecorder() + wrapped = recorder.wrap({"lookup": lookup}) + + assert await wrapped["lookup"]({"id": "A1"}) == "order A1" + assert recorder.invocations[0].result == "order A1" + + +@pytest.mark.asyncio +async def test_wrapped_tool_reraises_and_records_the_failure() -> None: + """The recorder observes; a tool that failed must still fail its caller. + + Swallowing the exception here would turn a broken tool into a silent one and + let the agent's error handling go unevaluated. + """ + + def refund(args: dict[str, Any]) -> str: + raise RuntimeError("gateway timeout") + + recorder = TrajectoryRecorder() + wrapped = recorder.wrap({"refund": refund}) + + with pytest.raises(RuntimeError, match="gateway timeout"): + await wrapped["refund"]({"id": "A1"}) + + assert recorder.invocations == [ + ToolInvocation(name="refund", arguments={"id": "A1"}, error="gateway timeout") + ] + + +@pytest.mark.asyncio +async def test_concurrent_calls_keep_their_start_order() -> None: + """Order is call order, not completion order. + + A judge asked whether the agent called `search` before `refund` is reading a + sequence, so a trajectory reordered by which tool happened to return first + would answer a different question than the one asked. + """ + started: dict[str, asyncio.Event] = {"slow": asyncio.Event()} + + async def slow(args: dict[str, Any]) -> str: + started["slow"].set() + await asyncio.sleep(0.02) + return "slow done" + + async def fast(args: dict[str, Any]) -> str: + await started["slow"].wait() + return "fast done" + + recorder = TrajectoryRecorder() + wrapped = recorder.wrap({"slow": slow, "fast": fast}) + + slow_task = asyncio.create_task(wrapped["slow"]({})) + await started["slow"].wait() + await wrapped["fast"]({}) + await slow_task + + assert [invocation.name for invocation in recorder.invocations] == ["slow", "fast"] + + +@pytest.mark.asyncio +async def test_calls_past_the_limit_still_execute_but_are_only_counted() -> None: + calls: list[int] = [] + + def append(args: dict[str, Any]) -> str: + calls.append(args["n"]) + return "ok" + + recorder = TrajectoryRecorder(limit=2) + wrapped = recorder.wrap({"append": append}) + for n in range(5): + await wrapped["append"]({"n": n}) + + # Every call ran: truncation bounds the record, never the agent's behavior. + assert calls == [0, 1, 2, 3, 4] + assert len(recorder.invocations) == 2 + assert recorder.omitted == 3 + + +@pytest.mark.asyncio +async def test_native_tools_pass_through_unwrapped_and_undescribed() -> None: + """A provider-executed tool is invisible, so it is not advertised either. + + Listing it as available while never being able to show a call to it would + let a judge conclude the model ignored a tool it may well have used. + """ + native = NativeTool("WebSearch") + recorder = TrajectoryRecorder() + wrapped = recorder.wrap({"web_search": native, "lookup": lambda args: "ok"}) + + assert wrapped["web_search"] is native + assert recorder.observable_tools == ["lookup"] + + +@pytest.mark.asyncio +async def test_keyword_arguments_are_recorded() -> None: + def lookup(**kwargs: Any) -> str: + return "ok" + + recorder = TrajectoryRecorder() + wrapped = recorder.wrap({"lookup": lookup}) + await wrapped["lookup"](id="A1") + + assert recorder.invocations[0].arguments == {"id": "A1"} + + +def test_render_lists_available_tools_calls_arguments_and_results() -> None: + rendered = render_trajectory( + [ + ToolInvocation(name="lookup", arguments={"id": "A1"}, result="shipped"), + ToolInvocation( + name="refund", arguments={"id": "A1"}, error="gateway timeout" + ), + ], + observable_tools=["lookup", "refund"], + ) + + assert rendered == ( + "Tools available: lookup, refund\n" + "Tool calls made while producing the response, in order:\n" + "1. lookup\n" + ' arguments: {"id":"A1"}\n' + " result: shipped\n" + "2. refund\n" + ' arguments: {"id":"A1"}\n' + " error: gateway timeout" + ) + + +def test_render_reports_an_empty_trajectory_when_tools_were_available() -> None: + """ "Called nothing" is the finding a tool-selection judge most needs.""" + rendered = render_trajectory([], observable_tools=["lookup"]) + + assert rendered == ( + "Tools available: lookup\nNo tool calls were made while producing the response." + ) + + +def test_render_is_empty_when_there_was_nothing_observable() -> None: + assert render_trajectory([], observable_tools=[]) == "" + + +def test_render_reports_omitted_calls() -> None: + rendered = render_trajectory( + [ToolInvocation(name="lookup", arguments=None, result="ok")], + observable_tools=["lookup"], + omitted=3, + ) + + assert "(3 further tool call(s) were made but not recorded.)" in rendered + + +def test_render_truncates_an_oversized_value() -> None: + rendered = render_trajectory( + [ToolInvocation(name="fetch", arguments={}, result="x" * 5000)], + observable_tools=["fetch"], + ) + + assert f" result: {'x' * MAX_RECORDED_VALUE_CHARS}… (truncated)" in rendered + + +def test_render_serializes_unserializable_values_without_raising() -> None: + class Opaque: + def __str__(self) -> str: + return "" + + rendered = render_trajectory( + [ToolInvocation(name="fetch", arguments={"k": Opaque()}, result=Opaque())], + observable_tools=["fetch"], + ) + + assert "" in rendered From c69920dce6bf1d19c92f649cd16a52abc95b7dd9 Mon Sep 17 00:00:00 2001 From: Dylan O'Neill Date: Thu, 17 Sep 2026 14:33:23 -0700 Subject: [PATCH 2/2] feat(judges): one message_history for every judge path The trajectory reached only the offline evaluations judge. The two online paths built their own message_history and neither included it, so the same judge grading the same response saw a different conversation depending on which path reached it -- and a trajectory rubric silently degraded to grading prose when run online. They had already drifted before the trajectory made it visible: offline row input + trajectory + output + format block inline user input + + output + format block deferred + output + format block The deferred path carried no input at all, so a background judge graded a response with no request beside it. judge_scoring.build_message_history is now the only place a history is built, in the module that already owns the {score, reasoning} contract for the same reason. All three paths call it, and a test asserts the inline and deferred paths produce byte-identical output for one row. Capture online happens in execute_and_track and execute_and_stream, which return the rendered trajectory alongside response and track_data. client.py and the two per-node graph.py judge runs thread it through. JudgeTask gains user_input and trajectory -- plain strings, since every field on it has to survive pickling to a worker thread. Recording is composed *inside* wrap_tool_handlers, on the original tool map, so the recorder still sees a NativeTool as a NativeTool and skips it. Wrapping the tracked map instead would have recorded the sync callable stub that wrapper substitutes for a native tool, showing a judge a call with an empty result while the provider's real result stayed invisible. Both paths now treat natives identically, and $ld:ai:tool_call still fires underneath -- both asserted. trajectory.py moves from evaluations/ to the package root: it is no longer evaluations-specific. A graph-level judge deliberately gets no trajectory. It grades a final answer produced across several nodes, and splicing their trajectories would describe a conversation that never happened. Co-Authored-By: Claude Opus 5 --- packages/client/README.md | 20 + .../src/launchdarkly_ai_server/client.py | 4 + .../evaluations/runner.py | 40 +- .../src/launchdarkly_ai_server/graph.py | 2 + .../launchdarkly_ai_server/judge_scoring.py | 47 ++- .../src/launchdarkly_ai_server/judges.py | 27 +- .../src/launchdarkly_ai_server/tracking.py | 36 +- .../{evaluations => }/trajectory.py | 34 +- .../src/launchdarkly_ai_server/types.py | 14 + .../tests/test_judge_message_history.py | 374 ++++++++++++++++++ ...tions_trajectory.py => test_trajectory.py} | 2 +- 11 files changed, 552 insertions(+), 48 deletions(-) rename packages/client/src/launchdarkly_ai_server/{evaluations => }/trajectory.py (86%) create mode 100644 packages/client/tests/test_judge_message_history.py rename packages/client/tests/{test_evaluations_trajectory.py => test_trajectory.py} (99%) diff --git a/packages/client/README.md b/packages/client/README.md index 92ca6b1c..c80298f5 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -122,6 +122,8 @@ A judge is shown the tool calls the row made on the way to its output, so a rubr The harness records this itself: it wraps your tool implementations once per row before handing them to the handler, so every handler package is covered without changes and your tools still return and raise exactly what they did before. +**This is not specific to offline evaluations.** Online judges — both the inline ones sampled by `config().invoke()` and the deferred ones you run from a `JudgeTask` on a background thread — are shown the same trajectory, built by the same function. See [Judges see one conversation](#judges-see-one-conversation). + The trajectory is rendered into **`{{message_history}}`** — the row input, then the trajectory, then the generated output, then the formatting instructions, in that order. There is no separate trajectory variable: `message_history` is already the transcript variable every judge reads, and judges built from the AI Library's default templates reference it, so a trajectory rubric can be written against an existing judge template with no new placeholder. ``` @@ -137,6 +139,24 @@ Tool calls made while producing the response, in order: A row with tools that called none of them says so explicitly, which is the finding a tool-selection rubric most needs. A run with no tools adds no block at all, so judges written before trajectories existed read exactly the history they read before. +#### Judges see one conversation + +All three judge paths build `{{message_history}}` through a single function, `judge_scoring.build_message_history`: + +| Path | Entry point | +| --- | --- | +| Online, inline | `config().invoke()` → `run_judges` | +| Online, deferred | `config(skip_judges=True).invoke()` → `run_judge(task, handlers)` on your own thread | +| Offline | `init_evaluations().run(criteria=[Judge(...)])` | + +Each one is the input, then the tool trajectory, then the output, then the `{score, reasoning}` format block, with empty parts skipped. A judge therefore grades the same conversation wherever it runs, which is what makes a rubric portable between a production sample and a dataset replay. + +They did not always agree, and that is why this is a single function now: each path used to join its own history. The offline one carried the row input, the inline one carried the user input, and the deferred one carried **neither** — so a deferred judge graded a response with no request beside it. `JudgeTask` gained `user_input` and `trajectory` to close that. + +For the deferred path those two fields travel on the task, which stays picklable — the trajectory crosses as the rendered string, not the structured record. + +A **graph-level** judge (`graph_judge`) gets no trajectory: it grades a final answer produced across several nodes, and splicing their trajectories together would describe a conversation that never happened. Per-node judges inside a graph do get their own node's. + Two limits keep a trajectory from spending the judge's context window: at most 50 recorded calls per row and 2000 characters per rendered argument bag or result, with anything beyond either reported as a count or marked truncated. Calls past the limit still execute — truncation drops the record, never the work. A `NativeTool` runs inside the provider, so no local wrapper sees it; such a tool is left out of the trajectory and out of the "Tools available" line, since naming a tool whose use cannot be shown would invite a judge to conclude the model ignored it. A tool result is now judge-prompt input. It stays literal for the same reason the generated output does: the judge config is handed to the handler unrendered and the handler makes exactly one template pass, so a `{{...}}` sequence coming back from a tool is never expanded into the judge prompt. diff --git a/packages/client/src/launchdarkly_ai_server/client.py b/packages/client/src/launchdarkly_ai_server/client.py index f64422db..f9f598f2 100644 --- a/packages/client/src/launchdarkly_ai_server/client.py +++ b/packages/client/src/launchdarkly_ai_server/client.py @@ -123,6 +123,8 @@ async def invoke( handlers=resolved_handler_list, llm_response=llm_str, base_track_data=track_data, + user_input=user_input, + trajectory=result.get("trajectory", ""), ) return ProviderResponse( response=parsed_response, @@ -137,6 +139,7 @@ async def invoke( handler=handler, handlers=resolved_handler_list, user_input=user_input, + trajectory=result.get("trajectory", ""), llm_response=llm_str, base_track_data=track_data, tool_handlers=resolved_tools, @@ -215,6 +218,7 @@ async def _stream_events( handler=handler, handlers=resolved_handler_list, user_input=user_input, + trajectory=done_event.get("trajectory", ""), llm_response=done_event.get("response", ""), base_track_data=track_data, tool_handlers=resolved_tools, diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index 6738d68c..c6da9eca 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -14,10 +14,16 @@ from ..judge_scoring import ( FORMATTING_INSTRUCTIONS, + build_message_history, numeric_score, parse_judge_response, ) from ..lifecycle import extract_variation +from ..trajectory import ( + TrajectoryRecorder, + render_row_trajectory, + row_fields, +) from ..types import NativeTool from ..utils import ( collapse_messages_to_instructions, @@ -35,7 +41,6 @@ LDJudgeCriterionEventPayload, TokenUsage, ) -from .trajectory import TrajectoryRecorder, render_row_trajectory, row_fields from .types import ( DatasetRef, DatasetRow, @@ -712,32 +717,21 @@ def _judge_variables( # reading the history sees the request, what the agent did about it, and # what it finally answered, in order. trajectory = render_row_trajectory(row_result) - # message_history carries FORMATTING_INSTRUCTIONS the same way the - # online path builds it (judges.run_judges), because that -- not the - # standalone formatting_instructions variable below -- is what every - # judge built from the AI Library's default templates (accuracy, - # relevance, toxicity, and any judge cloned from them) actually - # references. A judge authored before this variable existed must keep - # getting scored without edits. - # - # The trajectory goes here and nowhere else. It was briefly also - # exposed as a standalone tool_trajectory variable, which bought - # nothing: this is already the transcript variable every judge reads, - # and two overlapping variables only invited a rubric to interpolate - # both and pay for the trajectory twice. + # Built by the shared builder, not inline here: this path and both + # online paths must show a judge the same conversation, and they did + # not while each one joined its own. The trajectory goes into + # message_history and nowhere else -- it is already the transcript + # variable every judge cloned from the AI Library's default templates + # reads, so a second overlapping variable only invited a rubric to + # interpolate both and pay for the trajectory twice. variables.update( { "input": row_result.get("input") or "", "response_to_evaluate": output if output is not None else "", - "message_history": "\n\n".join( - str(value) - for value in ( - row_result.get("input"), - trajectory, - output, - FORMATTING_INSTRUCTIONS, - ) - if value + "message_history": build_message_history( + user_input=row_result.get("input"), + trajectory=trajectory, + output=output, ), "expected_output": expected if expected is not None else "", "ground_truth_context": ( diff --git a/packages/client/src/launchdarkly_ai_server/graph.py b/packages/client/src/launchdarkly_ai_server/graph.py index e9088e71..2398cdfc 100644 --- a/packages/client/src/launchdarkly_ai_server/graph.py +++ b/packages/client/src/launchdarkly_ai_server/graph.py @@ -225,6 +225,7 @@ async def run_node( base_track_data=result["track_data"], tool_handlers=tool_handlers, graph_key=key, + trajectory=result.get("trajectory", ""), ) if from_node: @@ -379,6 +380,7 @@ def _fn(*a: Any, **kw: Any) -> str: base_track_data=result["track_data"], tool_handlers=tool_handlers, graph_key=key, + trajectory=result.get("trajectory", ""), ) next_node = nodes.get(chosen[0]) if chosen else None diff --git a/packages/client/src/launchdarkly_ai_server/judge_scoring.py b/packages/client/src/launchdarkly_ai_server/judge_scoring.py index 8375b413..2c93298b 100644 --- a/packages/client/src/launchdarkly_ai_server/judge_scoring.py +++ b/packages/client/src/launchdarkly_ai_server/judge_scoring.py @@ -1,10 +1,19 @@ -"""Shared scoring contract for LaunchDarkly AI Judge invocations. +"""Shared contract for LaunchDarkly AI Judge invocations. -Both judge execution paths — the online path (``judges.run_judges``, sampled -per invocation) and the offline evaluations path (``evaluations.runner``) — -prompt a judge model for the same ``{"score": <0-1>, "reasoning": }`` -JSON shape and must parse it the same way. This module owns that contract so -the two paths cannot drift. +Three judge execution paths exist — the online inline path +(``judges.run_judges``, sampled per invocation), the online deferred path +(``judges.run_judge``, from a ``JudgeTask`` on a background thread), and the +offline evaluations path (``evaluations.runner``). All three prompt a judge +model for the same ``{"score": <0-1>, "reasoning": }`` JSON shape, and +all three must show the judge the same conversation. This module owns both +halves of that contract so the paths cannot drift. + +They did drift. Each path built ``message_history`` with its own inline join: +the offline one carried the row input, the inline online one carried the user +input, and the deferred one carried neither -- a judge grading the same +response saw a different conversation depending on which path reached it. The +trajectory landing in only one of the three is what made that visible. +:func:`build_message_history` is now the only place it is built. """ from __future__ import annotations @@ -27,6 +36,32 @@ ) +def build_message_history( + *, + user_input: Any = None, + trajectory: Any = None, + output: Any = None, +) -> str: + """The conversation a judge is shown, as the ``message_history`` variable. + + Ordered the way it happened: what was asked, what the agent did about it, + what it answered, and finally how to format the verdict. Empty parts are + skipped, so a run with no tools produces exactly the history it produced + before trajectories existed and a judge authored against it is unaffected. + + ``FORMATTING_INSTRUCTIONS`` is appended here rather than by each caller, + because every judge built from the AI Library's default templates + references ``{{message_history}}`` and not ``{{formatting_instructions}}`` + -- a judge that stopped being told the JSON shape would start returning + prose, and every one of its results would become an invalid-output error. + """ + return "\n\n".join( + str(part) + for part in (user_input, trajectory, output, FORMATTING_INSTRUCTIONS) + if part + ) + + def numeric_score(score: Any) -> float | None: """Return ``score`` as a float only when it already is a finite number. diff --git a/packages/client/src/launchdarkly_ai_server/judges.py b/packages/client/src/launchdarkly_ai_server/judges.py index 5878d6b1..bf9f256a 100644 --- a/packages/client/src/launchdarkly_ai_server/judges.py +++ b/packages/client/src/launchdarkly_ai_server/judges.py @@ -7,7 +7,7 @@ from .conversation import with_judge_evaluation from .judge_scoring import ( - FORMATTING_INSTRUCTIONS, + build_message_history, numeric_score, parse_judge_response, ) @@ -53,10 +53,16 @@ async def run_judges( base_track_data: TrackData, tool_handlers: dict[str, Callable[..., Any] | NativeTool] | None = None, graph_key: str | None = None, + trajectory: str = "", ) -> dict[str, JudgeResult]: """ Runs any judges configured on ``config['judgeConfiguration']`` against the produced output. Each judge is itself a tracked AI call. + + ``trajectory`` is the rendered tool-call trajectory of the invocation being + judged, from ``execute_and_track``. It defaults to empty so a caller that + has none -- a graph-level judge over several nodes, for instance -- is + unchanged, and so is a judge for a config with no tools. """ from .lifecycle import extract_variation from .tracking import execute_and_track @@ -145,8 +151,10 @@ async def run_judges( else judge_ai_config ) - message_history = "\n\n".join( - filter(None, [user_input, llm_response, FORMATTING_INSTRUCTIONS]) + message_history = build_message_history( + user_input=user_input, + trajectory=trajectory, + output=llm_response, ) async with with_judge_evaluation(judge_key) as record_evaluation: @@ -208,6 +216,8 @@ async def build_judge_tasks( handlers: list[ProviderHandler] | None = None, llm_response: str, base_track_data: TrackData, + user_input: str | None = None, + trajectory: str = "", ) -> list[JudgeTask]: """ Resolves all judges configured on ``config['judgeConfiguration']`` into @@ -306,6 +316,8 @@ async def build_judge_tasks( judge_config=judge_ai_config, judge_meta=judge_meta, actual_output=llm_response, + user_input=user_input, + trajectory=trajectory, user_context=user_context, judge_provider=judge_provider, judge_mode=judge_mode, @@ -374,8 +386,13 @@ def _matches(h: ProviderHandler) -> bool: else task.judge_config ) - message_history = "\n\n".join( - filter(None, [task.actual_output, FORMATTING_INSTRUCTIONS]) + # user_input and trajectory come off the task rather than being omitted: + # this path used to build a history with neither, so a judge grading the + # same response saw a different conversation than the inline path did. + message_history = build_message_history( + user_input=task.user_input, + trajectory=task.trajectory, + output=task.actual_output, ) async with with_judge_evaluation(task.config_key) as record_evaluation: diff --git a/packages/client/src/launchdarkly_ai_server/tracking.py b/packages/client/src/launchdarkly_ai_server/tracking.py index 639ff681..fc99a45e 100644 --- a/packages/client/src/launchdarkly_ai_server/tracking.py +++ b/packages/client/src/launchdarkly_ai_server/tracking.py @@ -7,6 +7,7 @@ from collections.abc import AsyncGenerator, Callable from typing import Any +from .trajectory import TrajectoryRecorder, render_trajectory from .types import ( NATIVE_TOOL_KEY, AiConfigRep, @@ -138,7 +139,16 @@ async def execute_and_track( client = get_client() ld_ctx = to_ld_context(client, user_context) - tracked_tool_handlers = wrap_tool_handlers(tool_handlers, ld_ctx, track_data) + # Recording is composed *inside* the tracking wrapper, on the original map, + # so the recorder still sees a NativeTool as a NativeTool and skips it. If + # it wrapped the tracked map instead it would see the callable stub + # wrap_tool_handlers substitutes for a native tool, and would show a judge + # a tool call with an empty result while the provider's real result stayed + # invisible. One recorder per invocation, since invocations run concurrently. + recorder = TrajectoryRecorder() + tracked_tool_handlers = wrap_tool_handlers( + recorder.wrap(tool_handlers or {}), ld_ctx, track_data + ) merged_variables: dict[str, Any] = { **(variables or {}), "ldContext": {**user_context}, @@ -172,7 +182,19 @@ async def execute_and_track( raw_output = result.get("output") response = raw_output if raw_output is not None else "" - return {"usage": usage, "response": response, "track_data": track_data} + return { + "usage": usage, + "response": response, + "track_data": track_data, + # Rendered here rather than returned structurally: the one consumer is + # message_history, and a JudgeTask has to stay picklable for the + # background path. + "trajectory": render_trajectory( + recorder.invocations, + observable_tools=recorder.observable_tools, + omitted=recorder.omitted, + ), + } async def execute_and_stream( @@ -219,7 +241,10 @@ async def execute_and_stream( client = get_client() ld_ctx = to_ld_context(client, user_context) - tracked_tool_handlers = wrap_tool_handlers(tool_handlers, ld_ctx, track_data) + recorder = TrajectoryRecorder() + tracked_tool_handlers = wrap_tool_handlers( + recorder.wrap(tool_handlers or {}), ld_ctx, track_data + ) merged_variables: dict[str, Any] = { **(variables or {}), "ldContext": {**user_context}, @@ -275,4 +300,9 @@ async def execute_and_stream( "response": full_text, "usage": usage, "track_data": track_data, + "trajectory": render_trajectory( + recorder.invocations, + observable_tools=recorder.observable_tools, + omitted=recorder.omitted, + ), } diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/trajectory.py b/packages/client/src/launchdarkly_ai_server/trajectory.py similarity index 86% rename from packages/client/src/launchdarkly_ai_server/evaluations/trajectory.py rename to packages/client/src/launchdarkly_ai_server/trajectory.py index 10ad15d5..67764c69 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/trajectory.py +++ b/packages/client/src/launchdarkly_ai_server/trajectory.py @@ -1,4 +1,4 @@ -"""Tool-call trajectory capture for the generation phase of an SDK-run evaluation. +"""Tool-call trajectory capture, shared by both judge paths. A judge can only grade what it is shown. Handler packages record tool traffic onto OpenTelemetry spans and return only ``{output, usage}``, so by the time a @@ -7,10 +7,16 @@ right arguments?" an unaskable question of an SDK-run evaluation, even though the evaluation had just run the agent that answered it. -The runner therefore records the trajectory itself, by wrapping the caller's -tool implementations once per row before handing them to the handler. Wrapping -is what makes this work with every handler package without changing any of -them: a handler looks a tool up by its key and calls it, exactly as before. +Both judge paths therefore record the trajectory themselves, by wrapping the +caller's tool implementations before handing them to the handler: once per row +in the offline evaluations runner, and once per invocation in +``tracking.execute_and_track``. Wrapping is what makes this work with every +handler package without changing any of them: a handler looks a tool up by its +key and calls it, exactly as before. + +The recorded trajectory reaches a judge through ``message_history``, built by +:func:`judge_scoring.build_message_history` -- one function for both paths, so +an online judge and an offline one are shown the same shape. Three properties are load-bearing. @@ -20,16 +26,24 @@ truncation drops the *record*, never the work, because an evaluation that changed the agent's behavior would no longer be evaluating the agent. -**A recorder belongs to one row.** ``_run_rows`` runs rows concurrently against -one shared tool map, so a single shared recorder would splice one row's calls -into another row's trajectory and hand the judge a conversation that never -happened. +**A recorder belongs to one invocation.** The offline runner generates rows +concurrently against one shared tool map, so a single shared recorder would +splice one row's calls into another row's trajectory and hand the judge a +conversation that never happened. The same holds for concurrent online +invocations, which is why ``execute_and_track`` builds its own per call. **Only observable tools are described.** A ``NativeTool`` is executed inside the provider, so no local wrapper ever sees it and its calls cannot appear in the trajectory. Such a tool is therefore left out of the rendered "tools available" line as well: naming a tool whose use is invisible would let a judge conclude the model ignored a tool it may well have called. + +Online, ``tracking.wrap_tool_handlers`` does turn a ``NativeTool`` into a +callable tracking stub, so a native call *is* locally observable there -- but it +is still skipped, and deliberately. The stub returns nothing, so recording it +would show a judge a tool call with an empty result while the provider's real +result stayed invisible. Recording is therefore composed *inside* that wrapper, +on the original map, so both paths see natives identically. """ from __future__ import annotations @@ -40,7 +54,7 @@ from dataclasses import dataclass, replace from typing import Any -from ..types import NativeTool +from .types import NativeTool #: How many tool calls one row's trajectory records. A trajectory is #: interpolated into a judge prompt, so an agent that loops over a large tool diff --git a/packages/client/src/launchdarkly_ai_server/types.py b/packages/client/src/launchdarkly_ai_server/types.py index f3505364..56ff58fa 100644 --- a/packages/client/src/launchdarkly_ai_server/types.py +++ b/packages/client/src/launchdarkly_ai_server/types.py @@ -299,6 +299,20 @@ class JudgeTask: variables: dict[str, Any] | None = None """Optional extra template variables for the judge prompt.""" evaluation_metric_key: str | None = None + user_input: str | None = None + """The input that produced ``actual_output``. + + Carried so this path builds the same ``message_history`` as the inline one + (:func:`judge_scoring.build_message_history`). It was previously absent, + which meant a deferred judge was shown the response with no request beside + it. + """ + trajectory: str = "" + """The rendered tool-call trajectory of the invocation being judged. + + A plain string, not the structured record, because every field here has to + stay picklable for the worker thread. + """ """LD metric key to track the score against.""" diff --git a/packages/client/tests/test_judge_message_history.py b/packages/client/tests/test_judge_message_history.py new file mode 100644 index 00000000..56900f52 --- /dev/null +++ b/packages/client/tests/test_judge_message_history.py @@ -0,0 +1,374 @@ +"""One message_history for every judge path. + +The three paths -- online inline (``run_judges``), online deferred +(``run_judge`` from a ``JudgeTask``), and offline evaluations -- each used to +join their own. They disagreed, so a judge grading the same response saw a +different conversation depending on which path reached it. These tests hold +them to :func:`judge_scoring.build_message_history`. +""" + +from __future__ import annotations + +import pickle +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import launchdarkly_ai_server.lifecycle as lifecycle_module +from launchdarkly_ai_server import JudgeTask, ProviderHandler, run_judge, run_judges +from launchdarkly_ai_server.judge_scoring import ( + FORMATTING_INSTRUCTIONS, + build_message_history, +) +from launchdarkly_ai_server.tracking import execute_and_track +from launchdarkly_ai_server.trajectory import TrajectoryRecorder, render_trajectory + +CONTEXT = {"kind": "user", "key": "u1"} + +JUDGE_CONFIG = { + "model": {"name": "gpt-4"}, + "provider": {"name": "TestProvider"}, + "instructions": "Judge it", +} + + +@pytest.fixture +def mock_ld_client() -> Any: + client = MagicMock() + client.track = MagicMock() + client.flush = AsyncMock() + client.close = AsyncMock() + client.variation = AsyncMock(return_value=None) + lifecycle_module._set_client_for_testing(client) + yield client + lifecycle_module._reset_for_testing() + + +def capturing_judge_handler(seen: list[dict[str, Any]]) -> ProviderHandler: + async def fn( + config: Any, + user_input: Any, + tool_handlers: Any, + variables: Any, + history: Any = None, + ) -> dict[str, Any]: + seen.append(dict(variables or {})) + return { + "output": '{"score": 1, "reasoning": "ok"}', + "usage": {"input_tokens": 1, "output_tokens": 1}, + } + + return ProviderHandler(fn=fn, provides_for=("TestProvider", "messages")) # type: ignore[arg-type] + + +def judged_config() -> dict[str, Any]: + return { + "model": {"name": "gpt-4"}, + "provider": {"name": "TestProvider"}, + "instructions": "hi", + "judgeConfiguration": {"judges": [{"key": "judge-1", "samplingRate": 1}]}, + } + + +@pytest.fixture +def judge_variation(monkeypatch: pytest.MonkeyPatch) -> None: + async def fake_extract_variation(key: str, context: Any) -> dict[str, Any]: + return { + "config": dict(JUDGE_CONFIG), + "meta": {"variationKey": "v", "version": 1}, + } + + monkeypatch.setattr( + "launchdarkly_ai_server.lifecycle.extract_variation", fake_extract_variation + ) + + +# ─── the builder itself ────────────────────────────────────────────────────── + + +def test_builder_orders_the_conversation_and_appends_the_format_block() -> None: + history = build_message_history(user_input="Q", trajectory="T", output="A") + + assert history == f"Q\n\nT\n\nA\n\n{FORMATTING_INSTRUCTIONS}" + + +def test_builder_skips_empty_parts() -> None: + """A run with no tools produces the history it produced before trajectories. + + This is what keeps a judge authored before this feature scoring unchanged. + """ + assert build_message_history(user_input="Q", trajectory="", output="A") == ( + f"Q\n\nA\n\n{FORMATTING_INSTRUCTIONS}" + ) + assert build_message_history(output="A") == f"A\n\n{FORMATTING_INSTRUCTIONS}" + + +def test_builder_always_carries_the_format_block() -> None: + """Judges from the AI Library's templates read the JSON shape from here. + + A history that stopped carrying it would make every such judge return prose, + turning each result into an invalid-output error. + """ + assert FORMATTING_INSTRUCTIONS in build_message_history() + + +# ─── online: inline ────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_inline_judge_is_shown_the_trajectory( + mock_ld_client: Any, judge_variation: None +) -> None: + seen: list[dict[str, Any]] = [] + await run_judges( + config=judged_config(), + user_context=CONTEXT, + handler=capturing_judge_handler(seen), + user_input="Where is order A1?", + llm_response="It shipped.", + base_track_data={}, + trajectory="Tools available: lookup\n1. lookup\n result: shipped", + ) + + history = seen[0]["message_history"] + assert "1. lookup" in history + assert ( + history.index("Where is order A1?") + < history.index("1. lookup") + < history.index("It shipped.") + ) + + +@pytest.mark.asyncio +async def test_inline_judge_without_a_trajectory_is_unchanged( + mock_ld_client: Any, judge_variation: None +) -> None: + seen: list[dict[str, Any]] = [] + await run_judges( + config=judged_config(), + user_context=CONTEXT, + handler=capturing_judge_handler(seen), + user_input="Q", + llm_response="A", + base_track_data={}, + ) + + assert seen[0]["message_history"] == build_message_history( + user_input="Q", output="A" + ) + + +# ─── online: deferred ──────────────────────────────────────────────────────── + + +def deferred_task(**overrides: Any) -> JudgeTask: + fields: dict[str, Any] = { + "config_key": "judge-1", + "judge_config": dict(JUDGE_CONFIG), + "judge_meta": {"variationKey": "v", "version": 1}, + "actual_output": "It shipped.", + "user_context": CONTEXT, + "judge_provider": "TestProvider", + "judge_mode": "messages", + "collapse_messages": False, + "parent_track_data": {}, + } + fields.update(overrides) + return JudgeTask(**fields) + + +@pytest.mark.asyncio +async def test_deferred_judge_is_shown_the_input_and_the_trajectory( + mock_ld_client: Any, +) -> None: + """This path carried neither before, so it graded a response in isolation.""" + seen: list[dict[str, Any]] = [] + task = deferred_task( + user_input="Where is order A1?", + trajectory="Tools available: lookup\n1. lookup\n result: shipped", + ) + + await run_judge(task, [capturing_judge_handler(seen)]) + + history = seen[0]["message_history"] + assert ( + history.index("Where is order A1?") + < history.index("1. lookup") + < history.index("It shipped.") + ) + + +@pytest.mark.asyncio +async def test_deferred_and_inline_agree_on_the_same_row( + mock_ld_client: Any, judge_variation: None +) -> None: + """The point of the shared builder: same inputs, byte-identical history.""" + inline_seen: list[dict[str, Any]] = [] + deferred_seen: list[dict[str, Any]] = [] + trajectory = "Tools available: lookup\n1. lookup\n result: shipped" + + await run_judges( + config=judged_config(), + user_context=CONTEXT, + handler=capturing_judge_handler(inline_seen), + user_input="Where is order A1?", + llm_response="It shipped.", + base_track_data={}, + trajectory=trajectory, + ) + await run_judge( + deferred_task(user_input="Where is order A1?", trajectory=trajectory), + [capturing_judge_handler(deferred_seen)], + ) + + assert inline_seen[0]["message_history"] == deferred_seen[0]["message_history"] + + +def test_judge_task_stays_picklable_with_the_new_fields() -> None: + """JudgeTask crosses a thread or IPC boundary, so it must stay primitives.""" + task = deferred_task(user_input="Q", trajectory="T") + + assert pickle.loads(pickle.dumps(task)).trajectory == "T" + + +# ─── online: capture ───────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_execute_and_track_records_the_trajectory(mock_ld_client: Any) -> None: + def lookup(args: dict[str, Any]) -> str: + return f"order {args['id']} shipped" + + async def handler( + config: Any, + user_input: Any, + tool_handlers: Any, + variables: Any, + history: Any = None, + ) -> dict[str, Any]: + await tool_handlers["lookup"]({"id": "A1"}) + return {"output": "It shipped.", "usage": {}} + + result = await execute_and_track( + config_key="c", + config={"model": {"name": "m"}, "provider": {"name": "TestProvider"}}, + meta={"variationKey": "v", "version": 1}, + user_context=CONTEXT, + handler=handler, # type: ignore[arg-type] + user_input="Where is order A1?", + tool_handlers={"lookup": lookup}, + ) + + assert "Tools available: lookup" in result["trajectory"] + assert '1. lookup\n arguments: {"id":"A1"}' in result["trajectory"] + assert "result: order A1 shipped" in result["trajectory"] + + +@pytest.mark.asyncio +async def test_a_native_tool_is_not_recorded_online_either( + mock_ld_client: Any, +) -> None: + """wrap_tool_handlers makes a native tool a callable stub, so it *is* + locally observable online -- but the stub returns nothing while the + provider's real result stays invisible, so recording it would show a judge + a call with an empty result. Both paths skip it identically. + """ + from launchdarkly_ai_server import NativeTool + + async def handler( + config: Any, + user_input: Any, + tool_handlers: Any, + variables: Any, + history: Any = None, + ) -> dict[str, Any]: + # Not awaited: wrap_tool_handlers substitutes a *sync* zero-arg stub + # for a native tool (§3.8), unlike the async wrapper it gives a real + # callable. That asymmetry is pre-existing. + tool_handlers["web_search"]({"q": "x"}) + return {"output": "done", "usage": {}} + + result = await execute_and_track( + config_key="c", + config={"model": {"name": "m"}, "provider": {"name": "TestProvider"}}, + meta={"variationKey": "v", "version": 1}, + user_context=CONTEXT, + handler=handler, # type: ignore[arg-type] + user_input="q", + tool_handlers={"web_search": NativeTool("WebSearch")}, + ) + + assert result["trajectory"] == "" + + +@pytest.mark.asyncio +async def test_tool_tracking_still_fires_under_the_recorder( + mock_ld_client: Any, +) -> None: + """Recording is composed inside wrap_tool_handlers, so §3.8 still works.""" + + async def handler( + config: Any, + user_input: Any, + tool_handlers: Any, + variables: Any, + history: Any = None, + ) -> dict[str, Any]: + await tool_handlers["lookup"]({"id": "A1"}) + return {"output": "ok", "usage": {}} + + await execute_and_track( + config_key="c", + config={"model": {"name": "m"}, "provider": {"name": "TestProvider"}}, + meta={"variationKey": "v", "version": 1}, + user_context=CONTEXT, + handler=handler, # type: ignore[arg-type] + user_input="q", + tool_handlers={"lookup": lambda args: "shipped"}, + ) + + tool_events = [ + call.args + for call in mock_ld_client.track.call_args_list + if call.args[0] == "$ld:ai:tool_call" + ] + assert len(tool_events) == 1 + assert tool_events[0][2]["toolKey"] == "lookup" + + +@pytest.mark.asyncio +async def test_a_run_with_no_tools_reports_no_trajectory(mock_ld_client: Any) -> None: + async def handler( + config: Any, + user_input: Any, + tool_handlers: Any, + variables: Any, + history: Any = None, + ) -> dict[str, Any]: + return {"output": "ok", "usage": {}} + + result = await execute_and_track( + config_key="c", + config={"model": {"name": "m"}, "provider": {"name": "TestProvider"}}, + meta={"variationKey": "v", "version": 1}, + user_context=CONTEXT, + handler=handler, # type: ignore[arg-type] + user_input="q", + ) + + assert result["trajectory"] == "" + + +def test_the_recorder_renders_identically_for_both_paths() -> None: + """Both paths call render_trajectory, so one fixture pins the shape.""" + recorder = TrajectoryRecorder() + recorder.wrap({"lookup": lambda args: "shipped"}) + + assert render_trajectory( + recorder.invocations, + observable_tools=recorder.observable_tools, + omitted=recorder.omitted, + ) == ( + "Tools available: lookup\nNo tool calls were made while producing the response." + ) diff --git a/packages/client/tests/test_evaluations_trajectory.py b/packages/client/tests/test_trajectory.py similarity index 99% rename from packages/client/tests/test_evaluations_trajectory.py rename to packages/client/tests/test_trajectory.py index db90192c..4c4328bb 100644 --- a/packages/client/tests/test_evaluations_trajectory.py +++ b/packages/client/tests/test_trajectory.py @@ -5,7 +5,7 @@ import pytest -from launchdarkly_ai_server.evaluations.trajectory import ( +from launchdarkly_ai_server.trajectory import ( MAX_RECORDED_VALUE_CHARS, ToolInvocation, TrajectoryRecorder,