Export metrics

Every Noēsis episode produces structured metrics you can export to dashboards, alerting systems, and analysis tools.

Available metrics

The summary.json file contains these metrics:

Metric	Description
`success`	1 for success, 0 for failure
`plan_count`	Number of planning iterations
`act_count`	Number of tool/adapter invocations
`reflect_count`	Number of reflection passes
`veto_count`	Number of policy vetoes
`latencies.first_action_ms`	Time to first action
`plan_adherence`	How closely execution matched the plan
`tool_coverage`	Percentage of planned tools actually used

Reading metrics

CLI

noesis insight ep_abc123 -j | jq '.metrics'

Python

import noesis as ns

episode_id = ns.last()
summary = ns.summary.read(episode_id)
metrics = summary.get("metrics", {})

print(f"Success: {metrics.get('success')}")
print(f"Actions: {metrics.get('act_count')}")
print(f"Vetoes: {metrics.get('veto_count')}")

Export to JSON files

Export all recent episodes to a JSON file:

import json
import noesis as ns


def export_metrics(output_path: str, limit: int = 100):
    """Export metrics from recent episodes to JSON."""
    episodes = ns.list_runs(limit=limit)
    
    metrics_data = []
    for ep in episodes:
        summary = ns.summary.read(ep["episode_id"])
        metrics_data.append({
            "episode_id": ep["episode_id"],
            "timestamp": ep["timestamp"],
            "task": summary.get("task"),
            "metrics": summary.get("metrics", {}),
        })
    
    with open(output_path, "w") as f:
        json.dump(metrics_data, f, indent=2)
    
    return len(metrics_data)


# Export last 100 episodes
count = export_metrics("./metrics_export.json")
print(f"Exported {count} episodes")

Export to CSV

For spreadsheet analysis:

import csv
import noesis as ns


def export_to_csv(output_path: str, limit: int = 100):
    """Export metrics to CSV for spreadsheet analysis."""
    episodes = ns.list_runs(limit=limit)
    
    fieldnames = [
        "episode_id",
        "timestamp",
        "task",
        "success",
        "plan_count",
        "act_count",
        "veto_count",
        "first_action_ms",
    ]
    
    with open(output_path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        
        for ep in episodes:
            summary = ns.summary.read(ep["episode_id"])
            metrics = summary.get("metrics", {})
            
            writer.writerow({
                "episode_id": ep["episode_id"],
                "timestamp": ep["timestamp"],
                "task": summary.get("task", "")[:100],
                "success": metrics.get("success", 0),
                "plan_count": metrics.get("plan_count", 0),
                "act_count": metrics.get("act_count", 0),
                "veto_count": metrics.get("veto_count", 0),
                "first_action_ms": metrics.get("latencies", {}).get("first_action_ms", 0),
            })


export_to_csv("./metrics.csv")

Export to Prometheus

Push metrics to Prometheus using the pushgateway:

from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
import noesis as ns


def push_to_prometheus(episode_id: str, gateway: str = "localhost:9091"):
    """Push episode metrics to Prometheus pushgateway."""
    summary = ns.summary.read(episode_id)
    metrics = summary.get("metrics", {})
    
    registry = CollectorRegistry()
    
    # Define gauges
    success = Gauge(
        "noesis_episode_success",
        "Episode success (1=success, 0=failure)",
        registry=registry,
    )
    actions = Gauge(
        "noesis_episode_actions",
        "Number of actions in episode",
        registry=registry,
    )
    vetoes = Gauge(
        "noesis_episode_vetoes",
        "Number of policy vetoes",
        registry=registry,
    )
    latency = Gauge(
        "noesis_first_action_latency_ms",
        "Latency to first action in milliseconds",
        registry=registry,
    )
    
    # Set values
    success.set(metrics.get("success", 0))
    actions.set(metrics.get("act_count", 0))
    vetoes.set(metrics.get("veto_count", 0))
    latency.set(metrics.get("latencies", {}).get("first_action_ms", 0))
    
    # Push to gateway
    push_to_gateway(gateway, job="noesis", registry=registry)


# Usage
episode_id = ns.run("my task", intuition=True)
push_to_prometheus(episode_id)

Export to Datadog

Send metrics to Datadog:

from datadog import initialize, statsd
import noesis as ns


def send_to_datadog(episode_id: str):
    """Send episode metrics to Datadog."""
    initialize(statsd_host="localhost", statsd_port=8125)
    
    summary = ns.summary.read(episode_id)
    metrics = summary.get("metrics", {})
    tags = [f"episode:{episode_id}"]
    
    # Send metrics
    statsd.gauge("noesis.success", metrics.get("success", 0), tags=tags)
    statsd.gauge("noesis.actions", metrics.get("act_count", 0), tags=tags)
    statsd.gauge("noesis.vetoes", metrics.get("veto_count", 0), tags=tags)
    statsd.histogram(
        "noesis.first_action_latency",
        metrics.get("latencies", {}).get("first_action_ms", 0),
        tags=tags,
    )


# Usage
episode_id = ns.run("my task", intuition=True)
send_to_datadog(episode_id)

Export to BigQuery

For large-scale analysis:

from google.cloud import bigquery
import noesis as ns


def export_to_bigquery(project_id: str, dataset: str, table: str, limit: int = 1000):
    """Export metrics to BigQuery for analysis."""
    client = bigquery.Client(project=project_id)
    table_id = f"{project_id}.{dataset}.{table}"
    
    episodes = ns.list_runs(limit=limit)
    rows = []
    
    for ep in episodes:
        summary = ns.summary.read(ep["episode_id"])
        metrics = summary.get("metrics", {})
        
        rows.append({
            "episode_id": ep["episode_id"],
            "timestamp": ep["timestamp"],
            "task": summary.get("task", ""),
            "success": metrics.get("success", 0),
            "plan_count": metrics.get("plan_count", 0),
            "act_count": metrics.get("act_count", 0),
            "veto_count": metrics.get("veto_count", 0),
            "first_action_ms": metrics.get("latencies", {}).get("first_action_ms", 0),
        })
    
    errors = client.insert_rows_json(table_id, rows)
    if errors:
        raise RuntimeError(f"BigQuery insert failed: {errors}")
    
    return len(rows)

Automated export hook

Run exports automatically after each episode:

import noesis as ns


class MetricsExporter:
    """Hook to export metrics after each episode."""
    
    def __init__(self, exporters: list):
        self.exporters = exporters
    
    def __call__(self, episode_id: str):
        for exporter in self.exporters:
            try:
                exporter(episode_id)
            except Exception as e:
                print(f"Export failed: {e}")


# Configure exporters
exporter = MetricsExporter([
    lambda ep: push_to_prometheus(ep),
    lambda ep: send_to_datadog(ep),
])

# Run with export
episode_id = ns.run("my task", intuition=True)
exporter(episode_id)

Grafana dashboard example

Create a Grafana dashboard with these queries:

-- Success rate over time
SELECT
  date_trunc('hour', timestamp) as time,
  avg(success) as success_rate
FROM noesis_episodes
GROUP BY 1
ORDER BY 1

-- Veto count by policy
SELECT
  policy_id,
  count(*) as veto_count
FROM noesis_events
WHERE phase = 'direction' AND status = 'blocked'
GROUP BY 1
ORDER BY 2 DESC

-- Latency percentiles
SELECT
  percentile_cont(0.50) WITHIN GROUP (ORDER BY first_action_ms) as p50,
  percentile_cont(0.95) WITHIN GROUP (ORDER BY first_action_ms) as p95,
  percentile_cont(0.99) WITHIN GROUP (ORDER BY first_action_ms) as p99
FROM noesis_episodes
WHERE timestamp > now() - interval '24 hours'

Best practices

Export asynchronously. Don’t block episode execution on metric exports—use background workers or queues.

Redact sensitive data. Tasks may contain PII or secrets. Truncate or hash before exporting.

Set retention policies. Define how long to keep detailed metrics vs. aggregates.

Get started

Tutorials

How-to guides

Explanation

Available metrics

Reading metrics

CLI

Python

Export to JSON files

Export to CSV

Export to Prometheus

Export to Datadog

Export to BigQuery

Automated export hook

Grafana dashboard example

Best practices

Next steps

Events reference

Summary reference

​Available metrics

​Reading metrics

​CLI

​Python

​Export to JSON files

​Export to CSV

​Export to Prometheus

​Export to Datadog

​Export to BigQuery

​Automated export hook

​Grafana dashboard example

​Best practices

​Next steps

Events reference

Summary reference

Available metrics

Reading metrics

CLI

Python

Export to JSON files

Export to CSV

Export to Prometheus

Export to Datadog

Export to BigQuery

Automated export hook

Grafana dashboard example

Best practices

Next steps