Traversal is an AI SRE agent that autonomously triages alerts, traces root causes via causal inference, and remediates production incidents without waking up a human. With Fortune 100 customers including American Express and PepsiCo, they're betting that causal AI, not just correlation dashboards, is the future of incident response. Replicability score: 68/100.
TL;DR
Traversal builds an AI SRE agent that autonomously triages alerts, traces root causes using causal inference across a real-time graph of your infrastructure, and remediates production incidents without waking up a human. Fortune 100 customers including American Express and PepsiCo trust it at petabyte scale.
Build difficulty
Complete replication guide — install as a slash command or rules file
Claude's Corner# Build Your Own AI SRE Agent (Traversal Clone)
## 7-Step Developer Guide
---
### Step 1: Design the Production World Model (Graph Database)
**Goal:** Build a causal graph of your entire infrastructure as a living, queryable data structure.
**DB Schema (PostgreSQL + pgvector + recursive CTEs):**
```sql
-- Core node table: every infrastructure entity
CREATE TABLE infra_nodes (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
external_id TEXT NOT NULL, -- k8s pod name, service name, host FQDN, etc.
node_type TEXT NOT NULL, -- 'pod', 'service', 'host', 'database', 'queue', 'container', 'ingress', etc.
namespace TEXT,
cluster TEXT,
metadata JSONB DEFAULT '{}', -- labels, annotations, resource limits
health_status TEXT DEFAULT 'unknown',
last_seen_at TIMESTAMPTZ DEFAULT now(),
created_at TIMESTAMPTZ DEFAULT now(),
UNIQUE(external_id, node_type, cluster)
);
-- Directed causal edges between nodes
CREATE TABLE infra_edges (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
source_id UUID REFERENCES infra_nodes(id) ON DELETE CASCADE,
target_id UUID REFERENCES infra_nodes(id) ON DELETE CASCADE,
edge_type TEXT NOT NULL, -- 'calls', 'depends_on', 'runs_on', 'stores_in', 'routes_to'
weight NUMERIC DEFAULT 1.0, -- traffic volume or dependency strength
latency_p99_ms NUMERIC,
error_rate NUMERIC,
metadata JSONB DEFAULT '{}',
updated_at TIMESTAMPTZ DEFAULT now()
);
-- Time-series metric snapshots per node
CREATE TABLE metric_snapshots (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
node_id UUID REFERENCES infra_nodes(id) ON DELETE CASCADE,
ts TIMESTAMPTZ NOT NULL,
cpu_percent NUMERIC,
memory_percent NUMERIC,
latency_p50_ms NUMERIC,
latency_p99_ms NUMERIC,
error_rate NUMERIC,
request_rate NUMERIC,
custom_metrics JSONB DEFAULT '{}'
);
CREATE INDEX ON metric_snapshots (node_id, ts DESC);
-- Deployments / change events
CREATE TABLE change_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
node_id UUID REFERENCES infra_nodes(id) ON DELETE CASCADE,
event_type TEXT NOT NULL, -- 'deployment', 'config_change', 'scale_event', 'restart'
description TEXT,
diff JSONB,
triggered_by TEXT,
occurred_at TIMESTAMPTZ NOT NULL,
created_at TIMESTAMPTZ DEFAULT now()
);
-- Incidents (root cause + resolution)
CREATE TABLE incidents (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
title TEXT NOT NULL,
severity TEXT, -- 'critical', 'high', 'medium', 'low'
status TEXT DEFAULT 'open', -- 'open', 'investigating', 'resolved'
root_cause_node_id UUID REFERENCES infra_nodes(id),
root_cause_summary TEXT,
rca_confidence NUMERIC, -- 0.0 - 1.0
remediation_applied TEXT,
ttd_seconds INTEGER, -- time to detect
ttr_seconds INTEGER, -- time to resolve
started_at TIMESTAMPTZ NOT NULL,
resolved_at TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT now()
);
-- Knowledge bank: runbooks + historical context
CREATE TABLE knowledge_entries (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
entry_type TEXT NOT NULL, -- 'runbook', 'past_incident', 'post_mortem', 'procedure'
title TEXT,
content TEXT,
tags TEXT[],
embedding VECTOR(1536), -- pgvector embedding for semantic search
metadata JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX ON knowledge_entries USING ivfflat (embedding vector_cosine_ops);
```
**Implementation notes:**
- Use TimescaleDB or partition `metric_snapshots` by time if ingesting high-volume metrics
- Neo4j is a viable alternative for the graph layer if you prefer native graph traversal
- Keep the graph synced in real-time using Kubernetes watch APIs + WebSocket listeners
---
### Step 2: Build Agentless Data Collectors
**Goal:** Pull metrics and topology from existing infrastructure without installing agents.
**Kubernetes Collector (Python):**
```python
from kubernetes import client, watch, config
import asyncio
import psycopg2
class K8sCollector:
def __init__(self, kubeconfig_path=None):
if kubeconfig_path:
config.load_kube_config(kubeconfig_path)
else:
config.load_incluster_config() # in-cluster service account
self.v1 = client.CoreV1Api()
self.apps_v1 = client.AppsV1Api()
async def sync_topology(self, db_conn):
"""Sync pods, services, deployments into infra_nodes + infra_edges."""
# Collect all pods
pods = self.v1.list_pod_for_all_namespaces()
for pod in pods.items:
upsert_node(db_conn, {
'external_id': pod.metadata.name,
'node_type': 'pod',
'namespace': pod.metadata.namespace,
'cluster': pod.metadata.cluster_name or 'default',
'metadata': {
'labels': pod.metadata.labels,
'node_name': pod.spec.node_name,
'containers': [c.name for c in pod.spec.containers],
'status': pod.status.phase,
'restart_count': sum(
cs.restart_count for cs in (pod.status.container_statuses or [])
)
}
})
# Collect services and wire edges
services = self.v1.list_service_for_all_namespaces()
for svc in services.items:
svc_id = upsert_node(db_conn, {
'external_id': svc.metadata.name,
'node_type': 'service',
'namespace': svc.metadata.namespace,
'metadata': {'selector': svc.spec.selector}
})
# Wire service -> pods via selector matching
if svc.spec.selector:
for pod in pods.items:
if matches_selector(pod.metadata.labels, svc.spec.selector):
upsert_edge(db_conn, svc_id, pod_id_map[pod.metadata.name], 'routes_to')
async def watch_events(self, db_conn):
"""Stream real-time change events."""
w = watch.Watch()
for event in w.stream(self.v1.list_event_for_all_namespaces):
obj = event['object']
if obj.reason in ['BackOff', 'Failed', 'OOMKilling', 'Unhealthy']:
emit_change_event(db_conn, obj)
```
**Additional collectors to build:**
- **Prometheus scraper**: Pull metrics from `/metrics` endpoints every 15s
- **Cloud provider APIs**: AWS CloudWatch, GCP Monitoring, Azure Monitor
- **Database metrics**: pg_stat_activity, Redis INFO, MySQL performance_schema
- **ArgoCD webhook**: Capture deployment events in real-time
---
### Step 3: Build the Causal Inference Engine
**Goal:** Distinguish root causes from correlated symptoms using causal graph analysis.
The key insight: most monitoring tools use correlation (Pearson/Spearman). You need causation (Granger, PC algorithm, or structural causal models).
**Causal Discovery (Python, using causallearn):**
```python
import numpy as np
from causallearn.search.ConstraintBased.PC import pc
from causallearn.utils.cit import fisherz
class CausalSearchEngine:
def __init__(self, db_conn, lookback_minutes=60):
self.db = db_conn
self.lookback = lookback_minutes
def get_metric_matrix(self, incident_ts, node_ids):
"""Pull aligned time-series for all nodes in the blast radius."""
# Returns shape: (time_steps, n_nodes)
rows = self.db.execute("""
SELECT node_id, ts, error_rate, latency_p99_ms, cpu_percent
FROM metric_snapshots
WHERE node_id = ANY(%s)
AND ts BETWEEN %s - interval '%s minutes' AND %s
ORDER BY ts ASC
""", [node_ids, incident_ts, self.lookback, incident_ts])
return pivot_to_matrix(rows)
def find_root_cause(self, incident, blast_radius_nodes):
"""Run PC algorithm to discover causal structure, then identify root node."""
X = self.get_metric_matrix(incident.started_at, blast_radius_nodes)
# PC algorithm for causal graph discovery
cg = pc(X, alpha=0.05, indep_test=fisherz)
# Find nodes with no incoming causal edges (potential root causes)
G = cg.G.graph
root_candidates = [
i for i in range(G.shape[0])
if G[:, i].sum() == 0 # no parents
]
# Score candidates by: (anomaly severity) × (causal distance to symptoms)
scored = self.score_candidates(root_candidates, blast_radius_nodes, incident)
return scored[0] # top candidate
def score_candidates(self, candidates, all_nodes, incident):
"""Combine anomaly score with graph centrality for final ranking."""
scores = []
for c in candidates:
node = all_nodes[c]
anomaly_score = self.compute_anomaly_score(node, incident.started_at)
change_recency = self.recent_change_score(node, incident.started_at)
scores.append((node, anomaly_score * 0.6 + change_recency * 0.4))
return sorted(scores, key=lambda x: x[1], reverse=True)
```
**Blast radius discovery** (recursive CTE to find all downstream nodes):
```sql
WITH RECURSIVE blast_radius AS (
SELECT id, 0 AS depth FROM infra_nodes WHERE id = :alert_node_id
UNION ALL
SELECT e.target_id, br.depth + 1
FROM infra_edges e
JOIN blast_radius br ON e.source_id = br.id
WHERE br.depth < 5
)
SELECT DISTINCT n.*, br.depth
FROM blast_radius br
JOIN infra_nodes n ON n.id = br.id
ORDER BY br.depth;
```
---
### Step 4: Build the Alert Intelligence Layer
**Goal:** Reduce alert noise and prioritize by business impact, not just technical severity.
```python
from anthropic import Anthropic
import numpy as np
client = Anthropic()
class AlertIntelligence:
def __init__(self, db_conn, knowledge_bank):
self.db = db_conn
self.kb = knowledge_bank
def triage_alert(self, alert):
"""Classify alert: noise / duplicate / real incident / cascade."""
# 1. Check if this alert is a known cascade symptom
if self.is_downstream_symptom(alert):
return {'action': 'suppress', 'reason': 'downstream_of_known_incident'}
# 2. Semantic similarity to recent alerts (dedup)
similar = self.kb.find_similar_alerts(alert['message'], top_k=5, hours=2)
if similar and similar[0]['score'] > 0.92:
return {'action': 'group', 'parent_incident_id': similar[0]['incident_id']}
# 3. Score business impact
impact = self.score_business_impact(alert)
return {
'action': 'escalate' if impact > 0.7 else 'monitor',
'impact_score': impact,
'suggested_priority': 'P1' if impact > 0.85 else 'P2'
}
def score_business_impact(self, alert):
"""Use LLM to assess business impact given service criticality context."""
service_context = self.get_service_criticality(alert['node_id'])
# Prompt caching for the system context (static across many alerts)
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=256,
system=[{
"type": "text",
"text": "You are an SRE impact assessor. Score alerts 0.0-1.0 by business impact.",
"cache_control": {"type": "ephemeral"}
}],
messages=[{
"role": "user",
"content": f"Alert: {alert['message']}\nService tier: {service_context['tier']}\nDependent services: {service_context['dependents']}\nReturn only a JSON: {{\"score\": 0.0-1.0, \"reason\": \"...\"}}"
}]
)
return parse_json(response.content[0].text)['score']
```
---
### Step 5: Build the LLM Reasoning & RCA Generation Layer
**Goal:** Generate human-readable root cause summaries, remediation suggestions, and post-mortem drafts.
```python
from anthropic import Anthropic
client = Anthropic()
class RCAEngine:
def generate_rca(self, incident, causal_chain, knowledge_context):
"""Generate a full RCA report with remediation steps."""
# Build causal chain description
chain_description = self.format_causal_chain(causal_chain)
# Retrieve relevant runbooks via semantic search
runbooks = knowledge_bank.search(
f"{incident.title} {causal_chain.root_node.node_type}",
top_k=3
)
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=[
{
"type": "text",
"text": """You are an expert SRE generating incident root cause analyses.
Be precise, technical, and actionable. Reference specific services and metrics.""",
"cache_control": {"type": "ephemeral"} # cache across incidents
},
{
"type": "text",
"text": f"Relevant runbooks:\n{format_runbooks(runbooks)}",
"cache_control": {"type": "ephemeral"} # cache per incident session
}
],
messages=[{
"role": "user",
"content": f"""Incident: {incident.title}
Severity: {incident.severity}
Started: {incident.started_at}
Causal chain discovered:
{chain_description}
Metrics at time of incident:
{format_metrics(incident.metrics_snapshot)}
Recent changes (30min window):
{format_changes(incident.recent_changes)}
Generate:
1. Root cause summary (2-3 sentences, specific)
2. Impact assessment
3. Immediate remediation steps (ordered)
4. Prevention recommendations
5. Post-mortem action items"""
}]
)
return parse_rca_response(response.content[0].text)
def format_causal_chain(self, chain):
"""Convert graph path to readable chain: A → B → C → symptom"""
nodes = chain.path
return " → ".join([
f"{n.external_id} ({n.node_type}, {n.anomaly_description})"
for n in nodes
])
```
---
### Step 6: Build the Self-Healing Automation Engine
**Goal:** Execute remediation actions safely, with configurable autonomy levels and circuit breakers.
```python
import asyncio
from enum import Enum
class AutonomyLevel(Enum):
SUGGEST = "suggest" # AI suggests, human approves
AUTO_LOW = "auto_low" # Auto-execute safe actions (restarts, scaling)
AUTO_HIGH = "auto_high" # Auto-execute all actions including rollbacks
class SelfHealingEngine:
def __init__(self, k8s_client, db_conn, autonomy_level=AutonomyLevel.SUGGEST):
self.k8s = k8s_client
self.db = db_conn
self.autonomy = autonomy_level
self.circuit_breaker = CircuitBreaker(max_actions_per_hour=20)
async def remediate(self, incident, rca_result):
"""Execute or queue remediation based on autonomy level."""
actions = self.plan_remediation(rca_result)
for action in actions:
if not self.circuit_breaker.allow(action):
await self.escalate_to_human(incident, action, "circuit_breaker_tripped")
continue
if self.autonomy == AutonomyLevel.SUGGEST:
await self.notify_oncall(incident, action)
elif self.autonomy == AutonomyLevel.AUTO_LOW and action.is_safe:
await self.execute(action, incident)
elif self.autonomy == AutonomyLevel.AUTO_HIGH:
await self.execute(action, incident)
else:
await self.notify_oncall(incident, action)
def plan_remediation(self, rca):
"""Map root cause type to remediation actions."""
playbook = {
'pod_oom': [RestartPodAction, ScaleDeploymentAction],
'connection_pool_exhausted': [RestartServiceAction, ScalePoolAction],
'bad_deployment': [RollbackDeploymentAction],
'database_slow': [KillLongRunningQueriesAction, ScaleReplicasAction],
'disk_full': [CompressLogsAction, EscalateToHumanAction],
}
return [ActionClass(rca) for ActionClass in playbook.get(rca.root_cause_type, [EscalateToHumanAction])]
async def execute(self, action, incident):
"""Execute a remediation action and record the result."""
try:
result = await action.run(self.k8s)
self.db.execute("""
INSERT INTO remediation_log (incident_id, action_type, result, executed_at)
VALUES (%s, %s, %s, now())
""", [incident.id, action.type, result])
return result
except Exception as e:
await self.escalate_to_human(incident, action, str(e))
raise
class RestartPodAction:
is_safe = True
type = 'restart_pod'
async def run(self, k8s):
k8s.v1.delete_namespaced_pod(
name=self.pod_name,
namespace=self.namespace
) # K8s will reschedule automatically
return {'status': 'pod_deleted', 'reschedule': 'automatic'}
```
---
### Step 7: Deploy, Harden, and Integrate
**Goal:** Make it production-grade: enterprise auth, BYOC option, observability of the observer.
**Architecture (Docker Compose for local dev, Kubernetes for prod):**
```yaml
# docker-compose.yml
services:
api:
build: ./api
environment:
DATABASE_URL: postgresql://postgres:pass@db:5432/sre_agent
ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY}
AUTONOMY_LEVEL: suggest
ports: ["8000:8000"]
collector:
build: ./collector
environment:
KUBECONFIG: /etc/kube/config
METRICS_INTERVAL_SECONDS: "15"
DATABASE_URL: postgresql://postgres:pass@db:5432/sre_agent
volumes:
- ~/.kube:/etc/kube:ro
causal-engine:
build: ./causal-engine
environment:
DATABASE_URL: postgresql://postgres:pass@db:5432/sre_agent
LOOKBACK_MINUTES: "60"
db:
image: timescale/timescaledb:latest-pg16
environment:
POSTGRES_PASSWORD: pass
POSTGRES_DB: sre_agent
volumes: [db_data:/var/lib/postgresql/data]
redis:
image: redis:7-alpine # for alert dedup + caching
```
**BYOC deployment checklist:**
- Deploy entirely within customer VPC, no data egress to your cloud
- Use customer's own Anthropic API key (or self-hosted model via Bedrock/Vertex)
- All data stays in customer's database
- Provide Helm chart for K8s deployment
- Audit logs for every AI action taken
**Key API endpoints to build:**
```
POST /api/alerts/ingest # receive alerts from PagerDuty/OpsGenie webhooks
GET /api/incidents # list active incidents with RCA status
GET /api/incidents/:id/rca # full RCA report for an incident
POST /api/incidents/:id/remediate # trigger remediation (with autonomy gate)
GET /api/graph/nodes # live infrastructure graph
GET /api/graph/blast-radius/:id # upstream/downstream of a node
POST /api/knowledge/ingest # add runbook or post-mortem to knowledge bank
GET /api/knowledge/search # semantic search across knowledge bank
WebSocket /ws/incidents # real-time incident stream
```
**Monitoring the monitor:**
- Track your own false positive rate per customer
- Alert on RCA confidence < 0.5 (escalate to human immediately)
- A/B test causal algorithms against ground-truth labels from confirmed incidents
- Build a feedback loop: when engineers override AI remediation, capture why
---
## Tech Stack Summary
| Layer | Technology |
|-------|-----------|
| Graph DB | PostgreSQL + recursive CTEs (or Neo4j for graph-native) |
| Time-series | TimescaleDB / InfluxDB |
| Vector search | pgvector |
| Causal inference | causallearn (PC algorithm) + custom scoring |
| LLM reasoning | Claude claude-sonnet-4-6 via Anthropic SDK (with prompt caching) |
| K8s integration | kubernetes-client (Python) |
| API | FastAPI (Python) or Go for latency-sensitive paths |
| Queue | Redis Streams or Kafka for high-volume alert ingestion |
| Auth | JWT + OAuth2 for enterprise SSO |
| Deployment | Helm chart for BYOC, Docker Compose for dev |