nat.plugins.phoenix.scripts.export_trajectory_to_phoenix.atif_trajectory_exporter

ATIF trajectory-to-span converter.

See README.md in this directory for usage guidance and span hierarchy details.

Attributes

logger

Classes

ATIFTrajectorySpanExporter

Converts complete ATIF trajectories to NAT Span objects.

Functions

_iso_to_epoch(→ float)	Convert an ISO 8601 timestamp to Unix epoch seconds.
_new_trace_id(→ int)	Generate a random 128-bit trace ID for a new trace.
_is_terminal_agent_step(→ bool)	True for agent steps that represent a final answer (no tool_calls).
_per_toolcall_meta(...)	Pull per-tool-call ancestry + invocation out of each tool_call's
_topo_sort_indices(→ list[int])	Return indices in topological order (parents before children).
_message_to_str(→ str)	Normalise an ATIF message field to a plain string.

Module Contents

logger

_iso_to_epoch(timestamp: str) → float

Convert an ISO 8601 timestamp to Unix epoch seconds.

_new_trace_id() → int

Generate a random 128-bit trace ID for a new trace.

_is_terminal_agent_step(step: dict[str, Any]) → bool

True for agent steps that represent a final answer (no tool_calls).

_per_toolcall_meta(

tool_calls: list[dict[str, Any]],

) → tuple[list[nat.atif.AtifAncestry], list[nat.atif.AtifInvocationInfo | None]]

Pull per-tool-call ancestry + invocation out of each tool_call’s own extra dict. ATIF v1.7 colocates this metadata with the tool_call rather than as parallel arrays on the parent step.

Tool calls missing extra.ancestry are skipped from the ancestry list; the invocation list is index-aligned with the ancestry list.

_topo_sort_indices(

ancestries: list[nat.atif.AtifAncestry],

) → list[int]

Return indices in topological order (parents before children).

_message_to_str(message: Any) → str

Normalise an ATIF message field to a plain string.

class ATIFTrajectorySpanExporter(span_prefix: str | None = None)

Bases: nat.observability.mixin.serialize_mixin.SerializeMixin

Converts complete ATIF trajectories to NAT Span objects.

Parameters

span_prefixstr, optional

Prefix for span attribute keys. Defaults to the NAT_SPAN_PREFIX environment variable, or "nat".

_span_prefix = None

convert(

trajectory_data: dict[str, Any],

) → list[nat.data_models.span.Span]

Convert an ATIF trajectory dict to a list of Span objects.

Parameters

trajectory_datadict

ATIF trajectory as a dict (parsed JSON). Must contain at least session_id, agent, and steps.

Returns

list[Span]

Flat list of Span objects. The first element is always the root WORKFLOW span.

_create_llm_span(

step: dict[str, Any],

step_extra: nat.atif.AtifStepExtra,

trace_id: int,

session_id: str,

span_lookup: dict[str, nat.data_models.span.Span],

root_fn_id: str,

) → nat.data_models.span.Span

Create an LLM span from an agent step.

_create_agent_spans(

step: dict[str, Any],

step_extra: nat.atif.AtifStepExtra,

trace_id: int,

session_id: str,

span_lookup: dict[str, nat.data_models.span.Span],

root_fn_id: str,

delegation_refs: dict[str, str],

) → list[nat.data_models.span.Span]

Create LLM span + child tool spans from an agent step with tool_calls.

_create_function_span(

step: dict[str, Any],

step_extra: nat.atif.AtifStepExtra,

trace_id: int,

session_id: str,

span_lookup: dict[str, nat.data_models.span.Span],

root_fn_id: str,

) → nat.data_models.span.Span

Create a FUNCTION span from a system step (no LLM, no tool_calls).

_create_system_tool_spans(

step: dict[str, Any],

step_extra: nat.atif.AtifStepExtra,

trace_id: int,

session_id: str,

span_lookup: dict[str, nat.data_models.span.Span],

root_fn_id: str,

delegation_refs: dict[str, str],

) → list[nat.data_models.span.Span]

Create a FUNCTION parent span + TOOL child spans from a system step.

_create_tool_span(

ancestry: nat.atif.AtifAncestry,

invocation: nat.atif.AtifInvocationInfo | None,

tool_name: str,

tool_args: dict[str, Any],

tool_output: str | None,

trace_id: int,

session_id: str,

span_lookup: dict[str, nat.data_models.span.Span],

) → nat.data_models.span.Span

Build a TOOL span for a single tool call.

_convert_subagent(

sub_traj: dict[str, Any],

parent_trace_id: int,

parent_span_lookup: dict[str, nat.data_models.span.Span],

parent_session_id: str,

delegation_refs: dict[str, str],

) → list[nat.data_models.span.Span]

Process a subagent trajectory recursively.

Subagent spans share the parent trace ID so they appear in the same Phoenix trace. The subagent’s root WORKFLOW span is linked as a child of the delegating tool span when the reference can be resolved.

_make_span(

name: str,

function_id: str,

function_name: str,

parent_id: str | None,

parent_name: str | None,

event_type_str: str,

span_kind: nat.data_models.span.SpanKind,

trace_id: int,

start_epoch: float,

end_epoch: float | None,

session_id: str,

span_lookup: dict[str, nat.data_models.span.Span],

framework: str | None = None,

) → nat.data_models.span.Span

Build a Span with standard NAT attributes.

static _trajectory_time_bounds(

steps: list[dict[str, Any]],

) → tuple[float, float]

Find the earliest and latest timestamps across all steps.

Prefers extra.invocation epoch timestamps (authoritative) and only falls back to ISO step.timestamp fields when no invocation timestamps are available.