Nemotron-AIQ Agentic Safety Dataset

{
    "trace_id": str,           # OTel traceid for the run
    "attack_snapshot": dict,   # Nested dict of attack metadata for the run 
    "trace": list,             # List of OTEL spans (each span is a dict)
}

{
    "id": str,
    "template": {
        "topic": str,
        "name": str,                    # Template name (e.g., "ACTION_INCOMPLETE_2")
        "injection_node": str,          # Where attack injects
        "attack_objective": str,        # Clear attack objective
        "evaluation_nodes": [str],      # List of evaluation points
        "evaluation_method": str,       # "JUDGE", etc.
        "severity": str,                # "LOW", "MEDIUM", "HIGH"
        "risk_categories": [str],       # Varies by evaluation type
        "attack_category": str,
        # ... other standard fields 
    },
    "attack": {
        "injection_string": str,
        "template_name": str,
        "injection_string_source": str,  
        # ... other attack metadata
    },
    "result": {
        "trace_id": str,
        "success": bool,                  # Overall success flag
        "score": {str: float},            # Per-node risk scores (0.0-1.0)
                                          # Example: {"summarize_sources": 0.391, "reflect_on_summary": 0.296}
        "evaluation_strings": {str: str}, # Full outputs per evaluation node
        "injection_function_id": [{...}], # Detailed injection points
        "evaluation_function_id": {...},  # Nested evaluation structure
        # ... other result fields
    }
}

{
    "id": str,
    "template": {
        "topic": str,
        "organization": str,
        "num_queries": int,
        "llm_name": str,
        # ... other standard fields
    },
    "attack": {
        "injection_string": str,      # same as topic which is the direct injection
        "attack_type": str,           # "direct"
        "attack_category": str,       # "content_safety"
        "attack_subcategory": str,
        "persona_top": str,
        "domain": str,
    },
    "result": {
        "trace_id": str,
        "success": bool,              # True if attack propagated to final workflow span
        "attack_success_rate": float, # (spans with attack) / (total spans)
        "function_id": [...],         # List of ALL spans where attacks were detected
    }
}

[
    {
        "name": str,                    # Span name
        "span_kind": str,               # CHAIN, LLM, TOOL
        "parent_id": str,               # Parent span ID (null for root)
        "start_time": str,              # Timestamp (format varies by provider)
        "end_time": str,
        "status_code": str,
        "context.span_id": str,         # Span ID
        "context.trace_id": str,        # Trace ID
        "attributes.output.value": str, # Output content
        "attributes.input.value": str,  # Input content
        # ... many other OTEL attributes (flattened with dot notation)
    },
    ...
]

from datasets import load_dataset

# Load safety config (both splits)
safety_dataset = load_dataset("nvidia/Nemotron-AIQ-Agentic-Safety-Dataset-1.0", "safety")

# Load security config (both splits)  
security_dataset = load_dataset("nvidia/Nemotron-AIQ-Agentic-Safety-Dataset-1.0", "security")

# Example 1: Safety Data
print("=== Safety Data Example ===")
safety_with_defense = safety_dataset['with_defense']
example = safety_with_defense[0]

# attack_snapshot and trace are already dicts/lists (not JSON strings)
attack_snapshot = example['attack_snapshot']
traces = example['trace']

print(f"Trace ID: {example['trace_id']}")
print(f"Attack Category: {attack_snapshot['attack']['attack_category']}")
print(f"Attack Subcategory: {attack_snapshot['attack']['attack_subcategory']}")
print(f"Number of spans: {len(traces)}")

# Example 2: Security Data
print("\n=== Security Data Example ===")
security_with_defense = security_dataset['with_defense']
example = security_with_defense[0]

attack_snapshot = example['attack_snapshot']
traces = example['trace']

print(f"Trace ID: {example['trace_id']}")
print(f"Attack Template: {attack_snapshot['attack']['template_name']}")
print(f"Attack Category: {attack_snapshot['template']['lakera_attack_category']}")
print(f"Number of spans: {len(traces)}")
print(f"Evaluation nodes: {list(attack_snapshot['result']['score'].keys())}")

# Analyze spans within a trace
print("\n=== Analyzing Spans ===")
for i, span in enumerate(traces[:3]):  # Show first 3 spans
    print(f"\nSpan {i}:")
    print(f"  Name: {span['name']}")
    print(f"  Kind: {span['span_kind']}")
    print(f"  Span ID: {span['context.span_id']}")
    print(f"  Parent ID: {span.get('parent_id', 'None')}")
    
    # Access span inputs/outputs
    if 'attributes.input.value' in span:
        print(f"  Input preview: {span['attributes.input.value'][:100]}...")
    if 'attributes.output.value' in span:
        print(f"  Output preview: {span['attributes.output.value'][:100]}...")

from datasets import load_dataset

# Count attack successes (binary: did attack reach final span?)
def count_safety_successes(split):
    successes = 0
    for ex in split:
        attack_snapshot = ex['attack_snapshot']  # Already a dict
        if attack_snapshot['result'].get('success', False):
            successes += 1
    return successes, len(split)

# Calculate average attack propagation rate
def calculate_safety_attack_rate(split):
    """Calculate average proportion of spans affected by attacks across all traces."""
    total_rate = 0
    for ex in split:
        attack_snapshot = ex['attack_snapshot']  # Already a dict
        total_rate += attack_snapshot['result'].get('attack_success_rate', 0)
    return total_rate / len(split) if len(split) > 0 else 0

# Load safety data (both splits)
safety_dataset = load_dataset("nvidia/Nemotron-AIQ-Agentic-Safety-Dataset-1.0", "safety")
safety_without_defense = safety_dataset['without_defense']
safety_with_defense = safety_dataset['with_defense']

# Binary success metric (attack reached final workflow span)
without_defense_success, without_defense_total = count_safety_successes(safety_without_defense)
with_defense_success, with_defense_total = count_safety_successes(safety_with_defense)

# Continuous attack propagation metric
without_defense_propagation = calculate_safety_attack_rate(safety_without_defense)
with_defense_propagation = calculate_safety_attack_rate(safety_with_defense)

print("Safety Defense Effectiveness:")
print(f"Binary Success Rate:")
print(f"  Without defense: {100*without_defense_success/without_defense_total:.1f}%")
print(f"  With defense: {100*with_defense_success/with_defense_total:.1f}%")
print(f"\nAttack Propagation Rate:")
print(f"  Without defense: {100*without_defense_propagation:.1f}%")
print(f"  With defense: {100*with_defense_propagation:.1f}%")

from collections import defaultdict
from datasets import load_dataset

# Calculate average risk scores per evaluation node
def calculate_security_risk_scores(split):
    """Calculate average risk scores per node across all traces."""
    node_scores = defaultdict(list)
    
    for ex in split:
        attack_snapshot = ex['attack_snapshot']  # Already a dict
        scores = attack_snapshot['result']['score']
        for node, score in scores.items():
            node_scores[node].append(score)
    
    # Calculate averages
    avg_scores = {node: sum(scores)/len(scores) for node, scores in node_scores.items()}
    overall_avg = sum(avg_scores.values()) / len(avg_scores) if avg_scores else 0
    
    return avg_scores, overall_avg

# Load security data (both splits)
security_dataset = load_dataset("nvidia/Nemotron-AIQ-Agentic-Safety-Dataset-1.0", "security")
security_without_defense = security_dataset['without_defense']
security_with_defense = security_dataset['with_defense']

without_defense_scores, without_defense_avg = calculate_security_risk_scores(security_without_defense)
with_defense_scores, with_defense_avg = calculate_security_risk_scores(security_with_defense)

print("Security Guard Effectiveness:")
print(f"Without defense - Avg risk score: {without_defense_avg:.3f}")
print(f"With defense - Avg risk score: {with_defense_avg:.3f}")
print(f"\nPer-node scores (without defense):")
for node, score in without_defense_scores.items():
    print(f"  {node}: {score:.3f}")

@misc{ghosh2025safetysecurityframeworkrealworld,
      title={A Safety and Security Framework for Real-World Agentic Systems}, 
      author={Shaona Ghosh and Barnaby Simkin and Kyriacos Shiarlis and Soumili Nandi and Dan Zhao and Matthew Fiedler and Julia Bazinska and Nikki Pope and Roopa Prabhu and Daniel Rohrer and Michael Demoret and Bartley Richardson},
      year={2025},
      eprint={2511.21990},
      archivePrefix={arXiv},
      primaryClass={cs.LG},
      url={https://arxiv.org/abs/2511.21990}, 
}