Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions plugins/nemo-experimentalist/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,41 @@ effective inputs:
$NEMO agents experimentalist doctor
```

## Agent trace formats

The agent under test can emit traces as OTLP or ATIF. **OTLP is the default** —
skip this section unless your agent emits ATIF.

To use an ATIF-emitting agent:

1. Have the agent write its trajectory under its trace directory (`/app/traces`
in the Harbor task container) with a `.atif.json` suffix.
2. Select the format on the evaluator, in the profile's `experiment_config`:

```yaml
experiment_config:
evaluator:
trace_format: atif # otlp (default) | atif
```
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved

3. Run against a platform, which ATIF requires:

```bash
$NEMO agents experimentalist run --base-url https://<platform-host> ...
```

Experiment grouping, run counts, and evaluator scores in Studio behave the same
as for OTLP. ATIF traces do not carry per-step timing, so individual step
durations show as zero.

### Troubleshooting

- *"configured `trace_format='otlp'` matched no trace artifact, but atif
artifacts are present"* — set `trace_format: atif`.
- *"Cannot read ATIF trajectory from disk … this trace was never uploaded"* — the
run had no reachable platform, so the trace was never ingested. Supply
`--base-url` and a workspace.

## Run the Experimentalist locally

`nemo agents experimentalist run` runs the local Experimentalist loop. It evaluates
Expand Down
2 changes: 2 additions & 0 deletions plugins/nemo-experimentalist/artifacts/.gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
*
Comment thread
schuellc-nvidia marked this conversation as resolved.
Resolved
!.gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,11 @@ async def run(
context.n_input_tokens = metrics["n_input_tokens"]
context.n_output_tokens = metrics["n_output_tokens"]
context.n_cache_tokens = metrics["n_cache_tokens"]
except (FileNotFoundError, OSError, ValueError) as exc:
# RuntimeError covers the environment refusing to copy /app/traces at all, which
# is what happens when the agent died before writing any: the traces directory
# never exists. Letting that escape would mask the agent's own failure below,
# reporting a Docker copy error instead of the traceback that caused it.
except (FileNotFoundError, OSError, RuntimeError, ValueError) as exc:
metrics_error = f"Trace metrics unavailable: {type(exc).__name__}: {exc}"
logger.warning(metrics_error)
finally:
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""ATIF trajectory helpers for reading and uploading agent traces.

Kept separate from ``otlp.py``: the two formats share no representation, and
mixing an ATIF reader into a module named for OTLP would blur a clean boundary.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

from nemo_experimentalist_plugin.entities import ResourceRef

# ATIF carries agent identity on the trajectory; the Experimentalist carries it as
# OTLP-style span attributes. This maps one onto the other.
_AGENT_FIELD_BY_ATTR: tuple[tuple[str, str], ...] = (
("name", "gen_ai.agent.name"),
("version", "agent.version"),
("model_name", "gen_ai.request.model"),
)


def _load(ref: ResourceRef) -> dict[str, Any]:
"""Parse the ATIF trajectory a ResourceRef points at."""
path = Path(urlparse(ref.uri).path)
trajectory = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(trajectory, dict):
raise ValueError(f"ATIF trajectory is not a JSON object: {ref.uri}")
return trajectory


def read_session_id(ref: ResourceRef) -> str:
"""Return the trajectory's session_id, which Intake uses as the trace id.

Args:
ref(ResourceRef): Resource reference to the ATIF trajectory file.

Returns:
str: The trajectory session id.
"""
session_id = _load(ref).get("session_id")
if not isinstance(session_id, str) or not session_id:
raise ValueError(f"No session_id found in {ref.uri}")
return session_id


def build_ingest_payload(
ref: ResourceRef,
*,
experiment_id: str,
task_id: str,
agent_attrs: dict[str, str],
) -> dict[str, Any]:
"""Load a trajectory and stamp evaluation identity onto it for Intake ingest.

Agent fields the producer already set are left alone; ``agent_attrs`` only
fills blanks, so a well-behaved producer stays authoritative.

``extra.verifier_result`` is deliberately not injected: the backend already
creates ``evaluator_results`` rows per trial metric, and letting Intake derive
a second set from the trajectory would double-count them.

Args:
ref(ResourceRef): Resource reference to the ATIF trajectory file.
experiment_id(str): Evaluation name, used as ``evaluation_context.evaluation_id``.
task_id(str): Test case id, used as ``evaluation_context.test_case_id``.
agent_attrs(dict[str, str]): OTLP-style agent attributes used as fallbacks.

Returns:
dict[str, Any]: The ATIF ingest request body.
"""
trajectory = _load(ref)
trajectory["evaluation_context"] = {
"evaluation_id": experiment_id,
"test_case_id": task_id,
}
agent = trajectory.get("agent")
if not isinstance(agent, dict):
agent = {}
for field, attr in _AGENT_FIELD_BY_ATTR:
if not agent.get(field) and attr in agent_attrs:
agent[field] = agent_attrs[attr]
trajectory["agent"] = agent
return trajectory
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
from dataclasses import dataclass
from pathlib import Path
from types import ModuleType, TracebackType
from typing import Any, TypeAlias, TypedDict
from typing import Any, Literal, TypeAlias, TypedDict
from uuid import uuid4

from harbor.constants import MAIN_SERVICE_NAME
Expand Down Expand Up @@ -112,6 +112,7 @@ class HarborResourceSpec(TypedDict):
)
_TRACE_ARTIFACT_SOURCE = "/app/traces"
_TRACE_ARTIFACT_DESTINATION = "traces"
_ATIF_TRACE_SUFFIX = ".atif.json"
_SHELL_SYNTAX_TIMEOUT_SEC = 10.0
_AGENT_IMPORT_ROOT = "_nemo_experimentalist_eval_agents"
_IDENTIFIER_RE = re.compile(r"\W+")
Expand Down Expand Up @@ -188,6 +189,13 @@ class HarborEvaluatorConfig(EvaluatorConfig):
retry: RetryConfig = Field(default=RetryConfig(exclude_exceptions=set()))
import_path: str = Field(default="harbor_wrapper:WrappedAgent")
trace_dir: str = Field(default=_TRACE_ARTIFACT_SOURCE)
trace_format: Literal["otlp", "atif"] = Field(
default="otlp",
description=(
"Which trace artifact becomes the trial's trace. Both are still collected and "
"exposed in resources; this only selects the one that gets uploaded and analysed."
),
)


class HarborDependencyRuntime(DependencyRuntime):
Expand Down Expand Up @@ -646,7 +654,9 @@ def _trial_metrics(trial_dir: Path, trial_data: dict[str, Any], spec: MetricSpec
return metrics


def _trial_resources(trial_dir: Path) -> tuple[dict[str, ResourceRef], ResourceRef | None]:
def _trial_resources(
trial_dir: Path, *, trace_format: str = "otlp"
) -> tuple[dict[str, ResourceRef], ResourceRef | None]:
resources: dict[str, ResourceRef] = {
"trial_dir": ResourceRef(
uri=trial_dir.resolve().as_uri(),
Expand All @@ -656,7 +666,8 @@ def _trial_resources(trial_dir: Path) -> tuple[dict[str, ResourceRef], ResourceR
),
)
}
trace_ref = None
otlp_trace_ref: ResourceRef | None = None
atif_trace_ref: ResourceRef | None = None

for path in sorted(candidate for candidate in trial_dir.rglob("*") if candidate.is_file()):
relative_path = path.relative_to(trial_dir).as_posix()
Expand All @@ -680,8 +691,15 @@ def _trial_resources(trial_dir: Path) -> tuple[dict[str, ResourceRef], ResourceR
elif path.suffix == ".jsonl" and "traces" in Path(relative_path).parts[:-1]:
Comment thread
schuellc-nvidia marked this conversation as resolved.
Resolved
trace_relative_path = relative_path.removeprefix("artifacts/")
description = f"Agent execution trace JSONL for {trace_relative_path}."
trace_ref = trace_ref or ResourceRef(uri=uri, description=description)
resources[f"trace:{trace_relative_path}"] = ResourceRef(uri=uri, description=description)
ref = ResourceRef(uri=uri, description=description, metadata={"trace_format": "otlp"})
otlp_trace_ref = otlp_trace_ref or ref
resources[f"trace:{trace_relative_path}"] = ref
elif relative_path.endswith(_ATIF_TRACE_SUFFIX) and "traces" in Path(relative_path).parts[:-1]:
trace_relative_path = relative_path.removeprefix("artifacts/")
description = f"Agent execution ATIF trajectory for {trace_relative_path}."
ref = ResourceRef(uri=uri, description=description, metadata={"trace_format": "atif"})
atif_trace_ref = atif_trace_ref or ref
resources[f"trace:{trace_relative_path}"] = ref
elif _is_trial_log_path(relative_path):
description = _TRIAL_LOG_DESCRIPTIONS.get(relative_path)
if description is None and relative_path.startswith("agent/command-"):
Expand Down Expand Up @@ -709,7 +727,16 @@ def _trial_resources(trial_dir: Path) -> tuple[dict[str, ResourceRef], ResourceR
description=f"Collected Harbor artifact {relative_path}.",
)

return resources, trace_ref
selected = atif_trace_ref if trace_format == "atif" else otlp_trace_ref
other = otlp_trace_ref if trace_format == "atif" else atif_trace_ref
if selected is None and other is not None:
found = "otlp" if trace_format == "atif" else "atif"
logger.warning(
f"Trial {trial_dir.name}: configured trace_format='{trace_format}' matched no trace "
f"artifact, but {found} artifacts are present. This trial will have no trace — set "
f"trace_format='{found}' if the agent under test emits {found.upper()}."
)
return resources, selected


class HarborDataset(Dataset):
Expand Down Expand Up @@ -1261,6 +1288,8 @@ async def _run(self, agent: Path, dataset: Dataset, options: HarborEvaluatorConf
options_dict["job_name"] = options.job_name or f"{agent.name}-{dataset.id}"
import_path: str = options_dict.pop("import_path")
trace_dir: str = options_dict.pop("trace_dir", _TRACE_ARTIFACT_SOURCE)
# Harbor's JobConfig forbids unknown keys, so this must not survive into it.
trace_format: str = options_dict.pop("trace_format", "otlp")
options_dict["artifacts"] = _with_trace_artifact(options_dict.get("artifacts") or [], trace_dir)
force_rerun: bool = options_dict.pop("force_rerun", False)

Expand All @@ -1286,10 +1315,12 @@ async def _run(self, agent: Path, dataset: Dataset, options: HarborEvaluatorConf
finally:
_cleanup_scoped_imports(scoped_package)

trials = await self._trials_from_dir(job.job_dir, dataset.tasks)
trials = await self._trials_from_dir(job.job_dir, dataset.tasks, trace_format=trace_format)
return trials

async def _trials_from_dir(self, job_dir: Path, tasks: Sequence[Task]) -> Sequence[TrialResult]:
async def _trials_from_dir(
self, job_dir: Path, tasks: Sequence[Task], *, trace_format: str = "otlp"
) -> Sequence[TrialResult]:
task_map = {task.id: task for task in tasks}
trials: list[TrialResult] = []
for trial_dir in sorted(path for path in job_dir.iterdir() if path.is_dir()):
Expand All @@ -1310,7 +1341,7 @@ async def _trials_from_dir(self, job_dir: Path, tasks: Sequence[Task]) -> Sequen
metric_spec = metric_spec or _trial_metric_spec(trial_dir, trial_data)

exception_info = trial_data.get("exception_info")
resources, trace = _trial_resources(trial_dir)
resources, trace = _trial_resources(trial_dir, trace_format=trace_format)

trials.append(
TrialResult(
Expand Down
Loading
Loading