trace

trace

Trace data model for agentic eval runs.

Classes

TurnTrace dataclass

TurnTrace(
    turn_index: int,
    input_tokens: int = 0,
    output_tokens: int = 0,
    tool_result_tokens: int = 0,
    tools_called: List[str] = list(),
    tool_latencies_s: Dict[str, float] = dict(),
    wall_clock_s: float = 0.0,
    error: Optional[str] = None,
    gpu_energy_joules: Optional[float] = None,
    cpu_energy_joules: Optional[float] = None,
    gpu_power_avg_watts: Optional[float] = None,
    cpu_power_avg_watts: Optional[float] = None,
    cost_usd: Optional[float] = None,
    action_energy_breakdown: Optional[
        List[Dict[str, Any]]
    ] = None,
    tool_calls: List[Dict[str, Any]] = list(),
)

Per-turn telemetry data.

QueryTrace dataclass

QueryTrace(
    query_id: str,
    workload_type: str,
    query_text: str = "",
    response_text: str = "",
    turns: List[TurnTrace] = list(),
    total_wall_clock_s: float = 0.0,
    completed: bool = False,
    timed_out: bool = False,
    query_gpu_energy_joules: Optional[float] = None,
    query_cpu_energy_joules: Optional[float] = None,
    query_gpu_power_avg_watts: Optional[float] = None,
    query_cpu_power_avg_watts: Optional[float] = None,
    is_resolved: Optional[bool] = None,
    query_mbu_avg_pct: Optional[float] = None,
    query_mbu_max_pct: Optional[float] = None,
    error: Optional[str] = None,
    error_kind: Optional[str] = None,
)

Per-query aggregate telemetry.

Attributes

total_tokens property

total_tokens: int

Total tokens (input + output) across all turns.

avg_gpu_power_watts property

avg_gpu_power_watts: Optional[float]

Mean GPU power across turns; falls back to query-level power.

avg_cpu_power_watts property

avg_cpu_power_watts: Optional[float]

Mean CPU power across turns; falls back to query-level power.

throughput_tokens_per_sec property

throughput_tokens_per_sec: Optional[float]

Output tokens per second; None if zero tokens or zero time.

energy_per_token_joules property

energy_per_token_joules: Optional[float]

GPU energy per output token; None if no energy data or zero tokens.

Methods:

save_jsonl

save_jsonl(path: Path) -> None

Append this trace as a JSONL line.

Source code in src/diapason/evals/core/trace.py

def save_jsonl(self, path: Path) -> None:
    """Append this trace as a JSONL line."""
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "a") as f:
        f.write(json.dumps(self.to_dict()) + "\n")

load_jsonl classmethod

load_jsonl(path: Path) -> List[QueryTrace]

Load traces from a JSONL file.

Source code in src/diapason/evals/core/trace.py

@classmethod
def load_jsonl(cls, path: Path) -> List[QueryTrace]:
    """Load traces from a JSONL file."""
    traces = []
    with open(path) as f:
        for line in f:
            line = line.strip()
            if line:
                traces.append(cls.from_dict(json.loads(line)))
    return traces

to_hf_dataset staticmethod

to_hf_dataset(traces: List[QueryTrace]) -> Any

Convert a list of QueryTrace objects to a HuggingFace Dataset.

Returns: A datasets.Dataset with one row per trace.

Source code in src/diapason/evals/core/trace.py

@staticmethod
def to_hf_dataset(traces: List[QueryTrace]) -> Any:
    """Convert a list of QueryTrace objects to a HuggingFace Dataset.

    Returns:
        A datasets.Dataset with one row per trace.
    """
    from datasets import Dataset

    rows = []
    for trace in traces:
        rows.append(
            {
                "query_id": trace.query_id,
                "workload_type": trace.workload_type,
                "query_text": trace.query_text,
                "response_text": trace.response_text,
                "num_turns": trace.num_turns,
                "total_input_tokens": trace.total_input_tokens,
                "total_output_tokens": trace.total_output_tokens,
                "total_tool_calls": trace.total_tool_calls,
                "total_wall_clock_s": trace.total_wall_clock_s,
                "total_gpu_energy_joules": trace.total_gpu_energy_joules,
                "total_cpu_energy_joules": trace.total_cpu_energy_joules,
                "total_tokens": trace.total_tokens,
                "total_cost_usd": trace.total_cost_usd,
                "avg_gpu_power_watts": trace.avg_gpu_power_watts,
                "avg_cpu_power_watts": trace.avg_cpu_power_watts,
                "throughput_tokens_per_sec": trace.throughput_tokens_per_sec,
                "energy_per_token_joules": trace.energy_per_token_joules,
                "completed": trace.completed,
                "timed_out": trace.timed_out,
                "is_resolved": trace.is_resolved,
                "query_mbu_avg_pct": trace.query_mbu_avg_pct,
                "query_mbu_max_pct": trace.query_mbu_max_pct,
                "trace_json": json.dumps(trace.to_dict()),
            }
        )
    return Dataset.from_list(rows)