Skip to main content Skills Marktplatz Entdecken und erkunden Sie KI-Skills, die von der Community erstellt wurden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Prompt-Details anzeigen Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
npx skills add https://github.com/aws-samples/sample-oh-my-aidlcops --skill anomaly-detectionDer Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
ZIP herunterladen Herunterladen... Mehr aus diesem Repository
Verwandte Berufe SOC
Basierend auf der SOC-Berufsklassifikation
name anomaly-detection description CloudWatch 메트릭과 Prometheus 시계열 데이터에서 통계적 이상 징후를 자동 탐지하여 incident-response의 입력 소스로 제공한다. 베이스라인 학습(7일 이동 평균 + 3σ), 다변량 상관 분석, 계절성 보정을 수행하며 탐지된 anomaly를 severity 분류하여 알람을 생성한다. argument-hint [target-metric or agent-name] user-invocable true model claude-sonnet-4-6 allowed-tools Read,Grep,Bash,mcp__cloudwatch,mcp__prometheus ontology {"produces":["Incident"],"references":["Deployment","Agent"]} loop-interface {"accepts-from":[],"emits-to":["root-cause-analysis.anomaly","cost-governance.anomaly"],"emits-signals":["anomaly-detection.anomaly"]}
When to Use
Sub-Phase 2 (Observe)에서 프로덕션 메트릭의 이상 패턴을 사전 탐지할 때
incident-response의 알람 소스로 — CloudWatch/Prometheus 정적 임계값으로 잡히지 않는 이상 징후 탐지배포 직후 canary 메트릭의 비정상 변동을 조기 감지할 때
주기적 cron (5분 간격)으로 핵심 메트릭 모니터링
사용 제외:
정적 임계값으로 충분한 단순 알람 (CloudWatch Alarm 직접 사용)
베이스라인 데이터가 7일 미만인 신규 메트릭 (최소 7일 학습 필요)
일회성 스파이크가 정상인 배치 작업 메트릭 (이벤트 캘린더로 제외 처리)
Prerequisites
awslabs.cloudwatch-mcp-server==0.0.25 — 메트릭 조회 (@latest 금지, PyPI 버전 pin 필수).awslabs.prometheus-mcp-server==0.2.15 — 시계열 쿼리.베이스라인 설정: .omao/plans/observability/baselines/ 디렉토리 (본 skill이 자동 생성).
incident-response skill 연동 — anomaly 탐지 시 자동 알람 생성.autopilot-deploy 상태 파일 접근 — 배포 직후 감도 조정 (배포 후 15분간 sensitivity 하향).모니터링 대상 정의: .omao/plans/observability/monitoring-targets.yaml.
모니터링 대상 정의
targets:
- name: rag-qa-agent
metrics:
- source: prometheus
query: 'rate(agent_request_errors_total{service="rag-qa-agent"}[5m])'
display_name: "Error Rate"
baseline_window_days: 7
sensitivity: 3.0
- source: prometheus
query: 'histogram_quantile(0.99, rate(agent_request_duration_seconds_bucket{service="rag-qa-agent"}[5m]))'
display_name: "P99 Latency"
baseline_window_days: 7
sensitivity:
3.0
-
source:
cloudwatch
namespace:
"AWS/Bedrock"
metric_name:
"InvocationLatency"
dimensions:
ModelId:
"anthropic.claude-sonnet-4-6-v1:0"
display_name:
"Bedrock Invocation Latency"
baseline_window_days:
7
sensitivity:
3.0
-
source:
prometheus
query:
'rate(agent_token_usage_total{service="rag-qa-agent"}[5m])'
display_name:
"Token Usage"
baseline_window_days:
7
sensitivity:
2.5
correlations:
-
pair:
"Error Rate:P99 Latency"
expected_r:
0.7
description:
"에러율과 레이턴시는 양의 상관관계"
-
pair:
"Token Usage:P99 Latency"
expected_r:
0.5
description:
"토큰 사용량과 레이턴시는 중간 상관관계"
exclusions:
-
type:
scheduled
cron:
"0 3 * * *"
duration_min:
30
reason:
"Daily reindex batch job"
schedule:
interval_minutes:
5
baseline_refresh_hours:
6
탐지 알고리즘
1. Statistical Baseline (Adaptive 3σ) 7일 이동 평균과 표준편차를 기반으로 정상 범위를 정의합니다. 베이스라인은 6시간마다 자동 갱신됩니다.
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass, field
import json
import yaml
@dataclass
class Baseline :
metric_name: str
mean: float
std: float
upper_bound: float
lower_bound: float
sensitivity: float
window_days: int
computed_at: str
seasonal_profile: dict = field(default_factory=dict )
def compute_baseline (metric_data: list [float ], timestamps: list [str ],
sensitivity: float = 3.0 , window_days: int = 7 ) -> Baseline:
"""7일 이동 평균 + Nσ 기반 정상 범위 계산.
Args:
metric_data: 시계열 데이터 포인트
timestamps: ISO 8601 타임스탬프 목록
sensitivity: σ 배수 (기본 3.0)
window_days: 베이스라인 윈도우 (기본 7일)
Returns:
Baseline 객체 (정상 범위 포함)
"""
data = np.array(metric_data)
q1, q3 = np.percentile(data, [25 , 75 ])
iqr = q3 - q1
mask = (data >= q1 - 1.5 * iqr) & (data <= q3 + 1.5 * iqr)
clean_data = data[mask]
mean = float (np.mean(clean_data))
std = float (np.std(clean_data))
seasonal = {}
for ts, val in zip (timestamps, metric_data):
hour = datetime.fromisoformat(ts.replace("Z" , "+00:00" )).hour
dow = datetime.fromisoformat(ts.replace("Z" , "+00:00" )).weekday()
key = f"{dow} _{hour} "
seasonal.setdefault(key, []).append(val)
seasonal_profile = {k: float (np.mean(v)) for k, v in seasonal.items()}
return Baseline(
metric_name="" ,
mean=mean,
std=std,
upper_bound=mean + sensitivity * std,
lower_bound=max (0 , mean - sensitivity * std),
sensitivity=sensitivity,
window_days=window_days,
computed_at=datetime.utcnow().isoformat() + "Z" ,
seasonal_profile=seasonal_profile,
)
2. Seasonality Correction 시간대별/요일별 패턴을 보정하여 false positive를 줄입니다.
def seasonal_adjust (value: float , timestamp: str ,
seasonal_profile: dict , global_mean: float ) -> tuple [float , float ]:
"""계절성 프로파일 기반 보정.
편차 기반 보정: value - seasonal_mean + global_mean
(비율 기반 value/factor는 저트래픽 시간대에서 false critical을 유발하므로 사용 금지)
Returns:
(adjusted_value, seasonal_mean)
"""
dt = datetime.fromisoformat(timestamp.replace("Z" , "+00:00" ))
key = f"{dt.weekday()} _{dt.hour} "
seasonal_mean = seasonal_profile.get(key)
if seasonal_mean is None :
return value, global_mean
adjusted = value - seasonal_mean + global_mean
return adjusted, seasonal_mean
def is_in_exclusion_window (timestamp: str , exclusions: list [dict ] ) -> bool :
"""현재 시점이 제외 윈도우(배치 작업 등)에 해당하는지 확인."""
dt = datetime.fromisoformat(timestamp.replace("Z" , "+00:00" ))
for exc in exclusions:
if exc["type" ] == "scheduled" :
parts = exc["cron" ].split()
cron_min, cron_hour = int (parts[0 ]), int (parts[1 ])
if dt.hour == cron_hour and dt.minute >= cron_min:
if (dt.minute - cron_min) < exc["duration_min" ]:
return True
return False
3. Multi-variate Correlation Analysis 단일 메트릭 이상이 아닌, 관련 메트릭 간 상관관계 붕괴를 탐지합니다.
def correlation_check (metrics: dict [str , list [float ]],
expected_correlations: list [dict ],
window_size: int = 60 ) -> list [dict ]:
"""메트릭 간 상관관계 이탈 탐지.
Args:
metrics: {metric_name: [values]} 딕셔너리
expected_correlations: 기대 상관관계 정의 목록
window_size: 상관관계 계산 윈도우 (데이터 포인트 수)
Returns:
탐지된 상관관계 이상 목록
"""
anomalies = []
for corr_def in expected_correlations:
m1_name, m2_name = corr_def["pair" ].split(":" )
expected_r = corr_def["expected_r" ]
m1_data = metrics.get(m1_name, [])
m2_data = metrics.get(m2_name, [])
if len (m1_data) < window_size or len (m2_data) < window_size:
continue
recent_m1 = np.array(m1_data[-window_size:])
recent_m2 = np.array(m2_data[-window_size:])
if np.std(recent_m1) == 0 or np.std(recent_m2) == 0 :
continue
actual_r = float (np.corrcoef(recent_m1, recent_m2)[0 , 1 ])
deviation = abs (actual_r - expected_r)
if deviation > 0.3 :
anomalies.append({
"type" : "correlation_breakdown" ,
"pair" : corr_def["pair" ],
"description" : corr_def.get("description" , "" ),
"expected_r" : expected_r,
"actual_r" : actual_r,
"deviation" : deviation,
"severity" : "critical" if deviation > 0.5 else "warning" ,
"detected_at" : datetime.utcnow().isoformat() + "Z" ,
})
return anomalies
4. Anomaly Scoring & Classification @dataclass
class AnomalyEvent :
metric_name: str
timestamp: str
value: float
baseline_mean: float
baseline_std: float
sigma_deviation: float
severity: str
type : str
context: dict = field(default_factory=dict )
def classify_anomaly (value: float , baseline: Baseline,
timestamp: str ) -> AnomalyEvent | None :
"""단일 데이터 포인트의 이상 여부 판정 및 severity 분류."""
adjusted_value, seasonal_mean = seasonal_adjust(value, timestamp, baseline.seasonal_profile, baseline.mean)
adjusted_mean = baseline.mean
if baseline.std == 0 :
return None
sigma = abs (adjusted_value - adjusted_mean) / baseline.std
if sigma < 2.0 :
return None
if sigma >= 5.0 :
severity = "critical"
elif sigma >= baseline.sensitivity:
severity = "warning"
else :
severity = "info"
return AnomalyEvent(
metric_name=baseline.metric_name,
timestamp=timestamp,
value=value,
baseline_mean=baseline.mean,
baseline_std=baseline.std,
sigma_deviation=sigma,
severity=severity,
type ="point_anomaly" ,
context={
"seasonal_factor" : factor,
"adjusted_value" : adjusted_value,
"upper_bound" : baseline.upper_bound,
"lower_bound" : baseline.lower_bound,
},
)
Anomaly Severity 분류 Severity 기준 후속 조치 Critical 5σ 이상 이탈 또는 다변량 상관 붕괴 3건 이상 동시 incident-response SEV2 자동 호출Warning 3σ~5σ 이탈 또는 계절성 보정 후에도 이상 incident-response SEV3 큐에 적재Info 2σ~3σ 이탈 (계절성 보정 전) 로그 기록만, 주간 리뷰 큐
실행 흐름
Step 1: 메트릭 수집
TARGETS_FILE=".omao/plans/observability/monitoring-targets.yaml"
mcp__prometheus__query_range \
--query 'rate(agent_request_errors_total{service="rag-qa-agent"}[5m])' \
--start "$(date -u -d '-1 hour' +%s) " \
--end "$(date -u +%s) " \
--step 60
mcp__cloudwatch__get_metric_data \
--metric-name "InvocationLatency" \
--namespace "AWS/Bedrock" \
--dimensions '[{"Name":"ModelId","Value":"anthropic.claude-sonnet-4-6-v1:0"}]' \
--period 60 \
--start-time "$(date -u -d '-1 hour' +%Y-%m-%dT%H:%M:%SZ) " \
--end-time "$(date -u +%Y-%m-%dT%H:%M:%SZ) "
Step 2: 베이스라인 로드 또는 갱신 import os
import yaml
BASELINE_DIR = ".omao/plans/observability/baselines"
def load_or_refresh_baseline (metric_name: str , metric_config: dict ,
force_refresh: bool = False ) -> Baseline:
"""베이스라인 로드. 6시간 이상 경과 시 자동 갱신."""
baseline_file = os.path.join(BASELINE_DIR, f"{metric_name} .yaml" )
if os.path.exists(baseline_file) and not force_refresh:
with open (baseline_file) as f:
data = yaml.safe_load(f)
computed_at = datetime.fromisoformat(data["computed_at" ].replace("Z" , "+00:00" ))
age_hours = (datetime.utcnow() - computed_at.replace(tzinfo=None )).total_seconds() / 3600
if age_hours < 6 :
return Baseline(**data)
window_days = metric_config.get("baseline_window_days" , 7 )
historical_data, timestamps = fetch_historical_data(
metric_config, days=window_days
)
baseline = compute_baseline(
historical_data, timestamps,
sensitivity=metric_config.get("sensitivity" , 3.0 ),
window_days=window_days,
)
baseline.metric_name = metric_name
os.makedirs(BASELINE_DIR, exist_ok=True )
with open (baseline_file, "w" ) as f:
yaml.dump(vars (baseline), f, default_flow_style=False )
return baseline
Step 3: 이상 판정 및 이벤트 생성 def run_detection_cycle (targets_file: str ) -> list [AnomalyEvent]:
"""단일 탐지 사이클 실행."""
with open (targets_file) as f:
config = yaml.safe_load(f)
all_anomalies = []
for target in config["targets" ]:
target_metrics = {}
for metric_config in target["metrics" ]:
metric_name = metric_config["display_name" ]
now = datetime.utcnow().isoformat() + "Z"
if is_in_exclusion_window(now, target.get("exclusions" , [])):
continue
current_values, timestamps = fetch_recent_data(metric_config, minutes=5 )
if not current_values:
continue
target_metrics[metric_name] = current_values
baseline = load_or_refresh_baseline(metric_name, metric_config)
for val, ts in zip (current_values, timestamps):
anomaly = classify_anomaly(val, baseline, ts)
if anomaly and anomaly.severity != "info" :
all_anomalies.append(anomaly)
if target.get("correlations" ) and len (target_metrics) >= 2 :
corr_anomalies = correlation_check(
target_metrics, target["correlations" ]
)
for ca in corr_anomalies:
all_anomalies.append(AnomalyEvent(
metric_name=ca["pair" ],
timestamp=ca["detected_at" ],
value=ca["actual_r" ],
baseline_mean=ca["expected_r" ],
baseline_std=0 ,
sigma_deviation=0 ,
severity=ca["severity" ],
type ="correlation_breakdown" ,
context=ca,
))
return all_anomalies
Step 4: 알람 라우팅 import json
import os
ANOMALY_STATE_DIR = ".omao/state/anomaly"
def route_anomalies (anomalies: list [AnomalyEvent] ) -> dict :
"""탐지된 anomaly를 severity에 따라 라우팅."""
os.makedirs(ANOMALY_STATE_DIR, exist_ok=True )
routed = {"critical" : [], "warning" : [], "info" : []}
for anomaly in anomalies:
event_file = os.path.join(
ANOMALY_STATE_DIR,
f"{anomaly.timestamp.replace(':' , '-' )} -{anomaly.metric_name.replace(' ' , '_' )} .json"
)
with open (event_file, "w" ) as f:
json.dump(vars (anomaly), f, indent=2 , default=str )
routed[anomaly.severity].append(anomaly)
if routed["critical" ]:
trigger_incident_response(
severity="SEV2" ,
source="anomaly-detection" ,
anomalies=routed["critical" ],
)
if routed["warning" ]:
enqueue_incident(
severity="SEV3" ,
source="anomaly-detection" ,
anomalies=routed["warning" ],
)
return {
"total_detected" : len (anomalies),
"critical" : len (routed["critical" ]),
"warning" : len (routed["warning" ]),
"info" : len (routed["info" ]),
}
Step 5: 배포 컨텍스트 연동 autopilot-deploy의 최근 배포 이벤트와 anomaly 시점을 매핑하여 change-correlated anomaly를 식별합니다.
def check_deploy_context (anomaly: AnomalyEvent ) -> dict | None :
"""anomaly 시점에 최근 배포가 있었는지 확인."""
deploy_state_dir = ".omao/state/autopilot-deploy"
if not os.path.exists(deploy_state_dir):
return None
for state_file in sorted (os.listdir(deploy_state_dir), reverse=True ):
if not state_file.endswith(".json" ):
continue
with open (os.path.join(deploy_state_dir, state_file)) as f:
deploy = json.load(f)
deploy_time = datetime.fromisoformat(deploy.get("started_at" , "" ).replace("Z" , "+00:00" ))
anomaly_time = datetime.fromisoformat(anomaly.timestamp.replace("Z" , "+00:00" ))
delta_min = (anomaly_time - deploy_time).total_seconds() / 60
if 0 < delta_min < 15 :
return {
"correlated_deploy" : deploy.get("target" ),
"deploy_version" : deploy.get("version" ),
"time_after_deploy_min" : delta_min,
"recommendation" : "Consider rollback if anomaly persists" ,
}
return None
상태 관리
.omao/plans/observability/monitoring-targets.yaml — 모니터링 대상 정의.omao/plans/observability/baselines/${metric}.yaml — 메트릭별 베이스라인 (6시간마다 갱신).omao/state/anomaly/${timestamp}-${metric}.json — 탐지된 anomaly 이벤트.omao/state/anomaly/correlation-matrix.json — 메트릭 간 상관관계 매트릭스 (일간 갱신).omao/state/anomaly/summary.json — 최근 24시간 탐지 요약
Example Inputs/Outputs Input : /anomaly-detection rag-qa-agent
Output (anomaly detected) :
[14:05:00Z] Loading monitoring targets for: rag-qa-agent
[14:05:01Z] Metrics to check: 4 (Error Rate, P99 Latency, Bedrock Latency, Token Usage)
[14:05:01Z] Exclusion check: not in exclusion window
[14:05:02Z] Fetching recent data (5min window)...
[14:05:03Z] Baseline status:
Error Rate : loaded (age: 2h, mean=0.012, σ=0.004)
P99 Latency : loaded (age: 2h, mean=380ms, σ=45ms)
Bedrock Latency : loaded (age: 5h, mean=220ms, σ=30ms)
Token Usage : refreshing (age: 7h)...done
[14:05:05Z] Detection results:
Error Rate : 0.031 (4.8σ) → WARNING
P99 Latency : 890ms (11.3σ) → CRITICAL
Bedrock Latency : 245ms (0.8σ) → normal
Token Usage : 1250 (1.2σ) → normal
[14:05:05Z] Correlation check:
Error Rate:P99 Latency — expected r=0.7, actual r=0.92 (Δ=0.22) → normal
Token Usage:P99 Latency — expected r=0.5, actual r=-0.1 (Δ=0.60) → CRITICAL
[14:05:06Z] Deploy context: canary v2.3.1 deployed 8min ago → CHANGE-CORRELATED
[14:05:06Z] Summary: 2 critical, 1 warning, 0 info
[14:05:06Z] Action: Triggering incident-response SEV2
Source: anomaly-detection
Evidence: P99 Latency 11.3σ + correlation breakdown (Token:Latency)
Context: Correlated with deploy rag-qa-agent:v2.3.1 (8min ago)
[14:05:06Z] Events saved to .omao/state/anomaly/
[14:05:00Z] Loading monitoring targets for: rag-qa-agent
[14:05:03Z] Detection results:
Error Rate : 0.013 (0.3σ) → normal
P99 Latency : 395ms (0.3σ) → normal
Bedrock Latency : 218ms (0.1σ) → normal
Token Usage : 980 (0.5σ) → normal
[14:05:04Z] Correlation check: all within expected range
[14:05:04Z] Summary: 0 critical, 0 warning, 0 info
[14:05:04Z] No anomalies detected. Next check in 5 minutes.
Output (exclusion window) :
[03:05:00Z] Loading monitoring targets for: rag-qa-agent
[03:05:01Z] Exclusion check: IN EXCLUSION WINDOW
Reason: "Daily reindex batch job" (03:00-03:30 UTC)
[03:05:01Z] Skipping detection cycle. Next check at 03:35 UTC.
기존 스킬 연동 연동 대상 방향 설명 incident-response→ 출력 Critical/Warning anomaly를 알람으로 전달 autopilot-deploy← 입력 배포 이벤트 시점을 anomaly 컨텍스트로 수신 continuous-eval← 입력 품질 메트릭 regression을 anomaly 소스로 활용 cost-governance→ 출력 비용 메트릭 이상(Token Usage 급증)을 cost alert로 전달 root-cause-analysis→ 출력 Critical anomaly 발생 시 RCA 트리거
참고 자료
공식 문서
기술 블로그
관련 문서 (내부) 