Ai Detecting Anomalies

Build an AI Anomaly Detector

Build an AI anomaly detector with DSPy - define what normal looks like, score events for severity, route by risk level, and explain findings to human reviewers.

Step 1: Understand the detection task

Ask the user: 1. What events are you analyzing? (transactions, logins, API calls, server logs, user actions, etc.) 2. What does "normal" look like? (Do you have historical baselines? Average values? Known-good patterns?) 3. What counts as suspicious? (Frequency spikes, unusual amounts, geographic outliers, time-of-day mismatches, etc.) 4. What action should fire on detection? (Alert, block, escalate to human, log for review, etc.) 5. What false-positive tolerance do you have? (Low tolerance = only flag high-confidence anomalies; high tolerance = cast wide net)

The answers determine severity thresholds, routing logic, and how much baseline context to include.

When NOT to use AI anomaly detection

• High-volume numeric time series — millions of events/second with simple numeric signals (CPU, latency, request rate). Use statistical methods instead - z-score, EWMA, isolation forest, or Prometheus alerting rules.
• Simple threshold rules — "flag any transaction over $10,000" does not need an LM. Write a rule.
• Real-time sub-10ms requirements — LM calls add latency. Use rule-based pre-filters and only invoke the LM on candidates.
• When you have millions of events per second — LM calls cost money. Pre-filter with cheap heuristics, then use AI only on flagged candidates.

Step 2: Build baseline construction

Summarize historical "normal" behavior into a compact string the LM can reason against. A good baseline summary describes typical patterns, ranges, and context.

import dspy

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

class SummarizeBaseline(dspy.Signature):
    """Summarize the normal behavior pattern from historical event data.
    Produce a concise description of what typical events look like - include
    typical values, frequencies, time patterns, and common attributes."""
    historical_events: str = dspy.InputField(
        desc="Sample of recent normal events, formatted as JSON or CSV"
    )
    baseline_summary: str = dspy.OutputField(
        desc="Concise description of normal behavior - typical ranges, patterns, and context"
    )

baseline_summarizer = dspy.ChainOfThought(SummarizeBaseline)

For many use cases, the baseline summary can be constructed once per period (daily, hourly) and cached rather than recomputed per event.

Step 3: Build the anomaly scorer

The core signature takes an event and the baseline summary, and outputs a severity level with a specific explanation.

from typing import Literal

SEVERITY_LEVELS = ["normal", "low", "medium", "high", "critical"]

class ScoreAnomaly(dspy.Signature):
    """Analyze an event against the normal baseline and determine if it is anomalous.
    Consider all dimensions - amount, timing, location, frequency, sequence, and context.
    Severity reflects both how unusual the event is AND the potential impact if malicious."""
    event: str = dspy.InputField(
        desc="The event to analyze, as a JSON object or structured text"
    )
    baseline_summary: str = dspy.InputField(
        desc="Description of normal behavior for this type of event"
    )
    severity: Literal[tuple(SEVERITY_LEVELS)] = dspy.OutputField(
        desc="Anomaly severity - normal (expected), low (minor deviation), "
             "medium (notable deviation), high (strong anomaly signal), "
             "critical (likely fraud or attack, needs immediate action)"
    )
    explanation: str = dspy.OutputField(
        desc="Specific explanation citing concrete details - what exactly is unusual, "
             "how it deviates from the baseline, and what the risk is. "
             "Not vague ('looks suspicious') but specific ('transaction amount $4,800 "
             "is 12x the users 30-day average of $400, combined with a new device "
             "and 3am local time')."
    )
    anomaly_score: float = dspy.OutputField(
        desc="Confidence score from 0.0 (clearly normal) to 1.0 (clearly anomalous)"
    )

anomaly_scorer = dspy.ChainOfThought(ScoreAnomaly)

Step 4: Full detection pipeline module

Combine baseline construction and anomaly scoring into a single reusable module.

class AnomalyDetector(dspy.Module):
    def __init__(self):
        self.baseline_summarizer = dspy.ChainOfThought(SummarizeBaseline)
        self.scorer = dspy.ChainOfThought(ScoreAnomaly)
        self._cached_baseline = None

    def set_baseline(self, historical_events: str):
        """Pre-compute the baseline summary from historical data."""
        result = self.baseline_summarizer(historical_events=historical_events)
        self._cached_baseline = result.baseline_summary
        return self._cached_baseline

    def forward(self, event: str, baseline_summary: str = None):
        baseline = baseline_summary or self._cached_baseline
        if not baseline:
            raise ValueError("No baseline set. Call set_baseline() first or pass baseline_summary.")
        return self.scorer(event=event, baseline_summary=baseline)

detector = AnomalyDetector()

# Set baseline once from recent history
detector.set_baseline("""
Recent 30-day transactions:
- Average amount: $412, std dev: $180, max: $1,200
- Typical locations: New York, Chicago, LA
- Typical hours: 9am-9pm local time
- Devices: 1-2 known devices per user
- Frequency: 3-8 transactions/week per user
""")

result = detector(event='{"amount": 4800, "location": "Lagos", "hour": 3, "device": "unknown"}')
print(f"Severity: {result.severity}")
print(f"Score: {result.anomaly_score:.2f}")
print(f"Explanation: {result.explanation}")

Step 5: Severity scoring with confidence-based routing

Route events automatically based on severity, and escalate only when confidence is high enough.

def route_anomaly(result) -> dict:
    """Route a scored anomaly to the appropriate action."""
    routing = {
        "normal":   {"action": "dismiss",  "notify": False, "block": False},
        "low":      {"action": "log",      "notify": False, "block": False},
        "medium":   {"action": "queue",    "notify": True,  "block": False},
        "high":     {"action": "alert",    "notify": True,  "block": False},
        "critical": {"action": "escalate", "notify": True,  "block": True},
    }
    route = routing[result.severity]

    # Downgrade routing if confidence is low
    if result.anomaly_score < 0.6 and result.severity in ("high", "critical"):
        route = routing["medium"]  # reduce to queue for human review
        route["confidence_downgraded"] = True

    return {**route, "severity": result.severity, "score": result.anomaly_score}

Severity to action mapping

Adjust these thresholds based on your false-positive tolerance.

Step 6: Explanation generation for human reviewers

Explanations are only useful if they are specific. Force the LM to cite concrete numbers and deviations by adding a dedicated explanation signature for high-severity events.

class ExplainAnomaly(dspy.Signature):
    """Generate a reviewer-ready explanation of why this event is anomalous.
    Write for a human analyst who needs to decide quickly. Cite specific numbers,
    list each anomalous dimension separately, and state the risk clearly."""
    event: str = dspy.InputField(desc="The flagged event")
    baseline_summary: str = dspy.InputField(desc="Normal behavior baseline")
    severity: str = dspy.InputField(desc="Assigned severity level")
    reviewer_explanation: str = dspy.OutputField(
        desc="Bullet-point explanation for human reviewer: what deviates, by how much, "
             "and what action is recommended. Must cite specific values from the event."
    )

explainer = dspy.ChainOfThought(ExplainAnomaly)

# Only invoke for high/critical — saves cost
if result.severity in ("high", "critical"):
    detail = explainer(
        event=event,
        baseline_summary=baseline,
        severity=result.severity,
    )
    print(detail.reviewer_explanation)

Step 7: Alert routing by confidence

Batch multiple events together (same user, same time window) before scoring to give the LM session-level context.

import json

class ScoreSession(dspy.Signature):
    """Analyze a sequence of events from the same user session for anomalies.
    Consider the pattern across events, not just individual events in isolation.
    Rapid-fire actions, escalating amounts, or device switches mid-session
    are signals invisible when events are scored individually."""
    session_events: str = dspy.InputField(
        desc="JSON array of events from the same user/session, in chronological order"
    )
    baseline_summary: str = dspy.InputField(desc="Normal behavior baseline for this user")
    severity: Literal[tuple(SEVERITY_LEVELS)] = dspy.OutputField()
    explanation: str = dspy.OutputField(
        desc="What pattern across the session is anomalous, not just individual events"
    )
    anomaly_score: float = dspy.OutputField()

session_scorer = dspy.ChainOfThought(ScoreSession)

def score_user_window(events: list[dict], baseline: str, window_minutes: int = 15):
    """Group events into time windows and score as sessions."""
    # Sort by timestamp, group into windows
    events_json = json.dumps(events, indent=2)
    return session_scorer(session_events=events_json, baseline_summary=baseline)

Step 8: Evaluate and optimize

from dspy.evaluate import Evaluate

# Build a labeled dataset - events with known ground truth
# label: 0 = normal, 1 = anomalous
labeled_events = [
    dspy.Example(
        event='{"amount": 4800, "location": "Lagos", "hour": 3, "device": "unknown"}',
        baseline_summary="Avg $400, NY/Chicago, 9am-9pm, known devices",
        severity="critical",
    ).with_inputs("event", "baseline_summary"),
    dspy.Example(
        event='{"amount": 380, "location": "New York", "hour": 14, "device": "iPhone-known"}',
        baseline_summary="Avg $400, NY/Chicago, 9am-9pm, known devices",
        severity="normal",
    ).with_inputs("event", "baseline_summary"),
    # ... more examples
]

trainset = labeled_events[:int(len(labeled_events) * 0.8)]
devset = labeled_events[int(len(labeled_events) * 0.8):]

def anomaly_metric(example, prediction, trace=None):
    """Measure exact severity match, or partial credit for adjacent severities."""
    if prediction.severity == example.severity:
        return 1.0
    # Adjacent severity is a partial match (e.g., predicted high vs actual critical)
    order = {s: i for i, s in enumerate(SEVERITY_LEVELS)}
    distance = abs(order[prediction.severity] - order[example.severity])
    return max(0.0, 1.0 - distance * 0.3)

evaluator = Evaluate(
    devset=devset,
    metric=anomaly_metric,
    num_threads=4,
    display_progress=True,
    display_table=5,
)
score = evaluator(anomaly_scorer)
print(f"Baseline accuracy: {score:.1f}%")

Optimize with BootstrapFewShot

optimizer = dspy.BootstrapFewShot(
    metric=anomaly_metric,
    max_bootstrapped_demos=4,
)
optimized_detector = optimizer.compile(anomaly_scorer, trainset=trainset)

# Re-evaluate
score = evaluator(optimized_detector)
print(f"Optimized accuracy: {score:.1f}%")

# Save
optimized_detector.save("anomaly_scorer.json")

False positive rate metric

def false_positive_rate(examples, predictions):
    """Compute FPR - normal events flagged as anomalous."""
    normals = [(e, p) for e, p in zip(examples, predictions) if e.severity == "normal"]
    if not normals:
        return 0.0
    flagged = sum(1 for _, p in normals if p.severity != "normal")
    return flagged / len(normals)

Key patterns

Gotchas

• Claude flags every unusual event as anomalous without baseline context — always provide a baseline_summary in the signature. Without it, the model has no reference point and defaults to flagging anything non-trivial.
• Claude outputs binary anomaly/not-anomaly instead of severity levels — use Literal with five graduated severity levels (normal, low, medium, high, critical) so downstream routing can take proportional action.
• Claude uses `dspy.Assert`/`dspy.Suggest` for severity validation — use dspy.Refine with a reward function that checks severity is one of the valid values and the explanation cites specific numbers.
• Claude generates vague explanations such as "this looks suspicious" or "unusual activity detected" — add an explicit desc on the explanation field requiring concrete deviations, specific values, and a risk statement.
• Claude processes events independently and misses session-level patterns — batch related events (same user, same time window) into a session before scoring. Account takeover and credential stuffing are only visible at the session level.

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Need fraud scores instead of severity categories? See /ai-scoring
• Measure and improve detection accuracy - see /ai-improving-accuracy
• Generate labeled training data for anomalies - see /ai-generating-data
• Add reasoning before severity classification - see /dspy-chain-of-thought
• Iterative refinement with feedback to fix wrong severity outputs - see /dspy-refine
• Sample N severity predictions and pick the best one - see /dspy-best-of-n
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Additional resources

• For worked examples (transaction fraud, user behavior, log anomalies), see examples.md

Anomaly Detection Examples

Example 1 - Transaction Fraud Detector

Scores payment events against a user's spending baseline and routes by fraud severity.

import dspy
from typing import Literal

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

SEVERITY_LEVELS = ["normal", "low", "medium", "high", "critical"]

class ScoreTransaction(dspy.Signature):
    """Analyze a payment transaction for fraud signals.
    Consider amount, location, device, time of day, and velocity.
    Compare against the users normal spending baseline."""
    transaction: str = dspy.InputField(
        desc="Transaction details as JSON - amount, merchant, location, device, timestamp"
    )
    user_baseline: str = dspy.InputField(
        desc="Users normal spending patterns - typical amounts, locations, devices, and hours"
    )
    severity: Literal[tuple(SEVERITY_LEVELS)] = dspy.OutputField(
        desc="Fraud severity - normal to critical"
    )
    explanation: str = dspy.OutputField(
        desc="Specific fraud signals citing exact values from the transaction and how they "
             "deviate from the baseline. Example - amount $4,800 is 12x the $400 average, "
             "device is new, and time is 3am outside normal 9am-9pm window."
    )
    anomaly_score: float = dspy.OutputField(
        desc="Fraud confidence from 0.0 (clearly legitimate) to 1.0 (clearly fraudulent)"
    )

scorer = dspy.ChainOfThought(ScoreTransaction)

# User baseline (built from 30-day history)
baseline = """
User spending baseline (30 days):
- Average transaction: $412, typical range $50-$800, max ever: $1,400
- Common merchants: Amazon, Whole Foods, Shell Gas, Netflix
- Locations: New York City, Brooklyn (home), occasional travel to Chicago
- Devices: iPhone 14 (primary), MacBook Pro (web)
- Active hours: 8am-11pm Eastern
- Frequency: 4-7 transactions per day
"""

# Test transactions
test_cases = [
    {
        "amount": 4800,
        "merchant": "Electronics Store",
        "location": "Lagos, Nigeria",
        "device": "Unknown Android",
        "hour_local": 3,
        "day": "Tuesday"
    },
    {
        "amount": 52,
        "merchant": "Whole Foods",
        "location": "Brooklyn, NY",
        "device": "iPhone 14",
        "hour_local": 18,
        "day": "Wednesday"
    },
    {
        "amount": 1100,
        "merchant": "Best Buy",
        "location": "New York City",
        "device": "MacBook Pro",
        "hour_local": 14,
        "day": "Saturday"
    },
]

import json

for tx in test_cases:
    result = scorer(transaction=json.dumps(tx), user_baseline=baseline)
    print(f"\nTransaction: ${tx['amount']} at {tx['merchant']}")
    print(f"Severity: {result.severity} (score: {result.anomaly_score:.2f})")
    print(f"Explanation: {result.explanation}")

    # Route by severity
    if result.severity == "critical":
        print("ACTION - Block transaction, send SMS alert")
    elif result.severity == "high":
        print("ACTION - Alert fraud team, hold for review")
    elif result.severity == "medium":
        print("ACTION - Queue for next analyst review")
    else:
        print("ACTION - Approve")

---

Example 2 - User Behavior Anomaly Detector

Detects account takeover attempts by analyzing login and usage patterns.

import dspy
from typing import Literal
import json

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

SEVERITY_LEVELS = ["normal", "low", "medium", "high", "critical"]

class ScoreUserSession(dspy.Signature):
    """Analyze a user session for account takeover or abuse signals.
    Consider login patterns, device changes, geographic jumps, feature usage,
    and action velocity. Session-level patterns matter as much as individual events."""
    session_events: str = dspy.InputField(
        desc="JSON array of session events in chronological order - logins, actions, API calls"
    )
    user_baseline: str = dspy.InputField(
        desc="This users normal behavior - typical devices, locations, usage patterns, and hours"
    )
    severity: Literal[tuple(SEVERITY_LEVELS)] = dspy.OutputField(
        desc="Anomaly severity - normal to critical"
    )
    explanation: str = dspy.OutputField(
        desc="What specific pattern across this session is anomalous. "
             "Cite the events and values that raised suspicion."
    )
    anomaly_score: float = dspy.OutputField(
        desc="Confidence score from 0.0 to 1.0"
    )
    risk_factors: list[str] = dspy.OutputField(
        desc="List of individual risk factors identified, e.g. ['new device', 'geographic jump', 'high velocity']"
    )

session_scorer = dspy.ChainOfThought(ScoreUserSession)

user_baseline = """
User normal behavior (90 days):
- Devices: MacBook Pro (home), iPhone 13 (mobile)
- Locations: San Francisco, CA (primary); occasional trips to NYC
- Login times: 8am-8pm Pacific, weekdays mostly
- Typical session - check dashboard, run 1-3 reports, update settings occasionally
- API calls per session: 10-50, spread over 20-60 minutes
- Password changes: 0 in 90 days
- 2FA method: authenticator app
"""

# Suspicious session - looks like account takeover
suspicious_session = [
    {"time": "2024-01-15T02:14:00Z", "event": "login_success", "device": "Unknown Windows PC",
     "ip": "185.220.101.42", "country": "Romania", "2fa": "bypass_attempted"},
    {"time": "2024-01-15T02:14:30Z", "event": "password_change", "device": "Unknown Windows PC"},
    {"time": "2024-01-15T02:14:45Z", "event": "email_change", "new_email": "backup9912@tempmail.com"},
    {"time": "2024-01-15T02:15:00Z", "event": "api_key_created", "scope": "full_access"},
    {"time": "2024-01-15T02:15:10Z", "event": "bulk_data_export", "records": 50000},
]

result = session_scorer(
    session_events=json.dumps(suspicious_session, indent=2),
    user_baseline=user_baseline
)

print(f"Severity: {result.severity}")
print(f"Anomaly score: {result.anomaly_score:.2f}")
print(f"Risk factors: {result.risk_factors}")
print(f"\nExplanation:\n{result.explanation}")

# Optimize with BootstrapFewShot given labeled session data
# trainset = [dspy.Example(...).with_inputs("session_events", "user_baseline"), ...]
# optimizer = dspy.BootstrapFewShot(metric=anomaly_metric, max_bootstrapped_demos=4)
# optimized = optimizer.compile(session_scorer, trainset=trainset)
# optimized.save("session_scorer.json")

---

Example 3 - Log Anomaly Detector

Flags unusual error patterns and security events in server logs.

import dspy
from typing import Literal
import json
from collections import defaultdict

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

SEVERITY_LEVELS = ["normal", "low", "medium", "high", "critical"]

class ScoreLogBatch(dspy.Signature):
    """Analyze a batch of server log entries for anomalies.
    Look for unusual error rates, unexpected endpoints, suspicious IPs,
    timing patterns, scanning behavior, and attack signatures."""
    log_batch: str = dspy.InputField(
        desc="JSON array of log entries from a 5-minute window - method, path, status, ip, latency"
    )
    baseline_summary: str = dspy.InputField(
        desc="Normal log patterns - typical error rates, common paths, expected IPs, usual latency"
    )
    severity: Literal[tuple(SEVERITY_LEVELS)] = dspy.OutputField(
        desc="Anomaly severity level"
    )
    explanation: str = dspy.OutputField(
        desc="What in these logs is anomalous. Cite specific paths, IPs, error rates, "
             "or patterns with concrete numbers."
    )
    anomaly_score: float = dspy.OutputField(desc="Confidence from 0.0 to 1.0")
    anomalous_entries: list[str] = dspy.OutputField(
        desc="List of the most suspicious individual log entries or patterns"
    )

log_scorer = dspy.ChainOfThought(ScoreLogBatch)

log_baseline = """
Normal log patterns (7-day baseline):
- Error rate: 0.3% of requests (mostly 404s from typos)
- Common paths: /api/v1/users, /api/v1/products, /health, /static/*
- Typical IPs: known CDN ranges, office IPs (10.0.0.0/8)
- Latency p99: 250ms, average: 45ms
- Request rate: 200-800 req/min during business hours
- No 401/403 spikes, no /admin access from external IPs
"""

# Suspicious log batch - looks like a scanning/injection attempt
suspicious_logs = [
    {"time": "02:31:00", "method": "GET", "path": "/admin/config.php", "status": 404, "ip": "185.220.101.42"},
    {"time": "02:31:01", "method": "GET", "path": "/wp-admin/", "status": 404, "ip": "185.220.101.42"},
    {"time": "02:31:01", "method": "GET", "path": "/.env", "status": 200, "ip": "185.220.101.42"},
    {"time": "02:31:02", "method": "GET", "path": "/api/v1/users?id=1 OR 1=1", "status": 500, "ip": "185.220.101.42"},
    {"time": "02:31:02", "method": "POST", "path": "/api/v1/login", "status": 401, "ip": "185.220.101.42"},
    {"time": "02:31:03", "method": "POST", "path": "/api/v1/login", "status": 401, "ip": "185.220.101.42"},
    {"time": "02:31:03", "method": "POST", "path": "/api/v1/login", "status": 401, "ip": "185.220.101.42"},
    {"time": "02:31:04", "method": "GET", "path": "/api/v1/products", "status": 200, "ip": "10.0.1.5"},
]

result = log_scorer(
    log_batch=json.dumps(suspicious_logs, indent=2),
    baseline_summary=log_baseline
)

print(f"Severity: {result.severity} (score: {result.anomaly_score:.2f})")
print(f"\nAnomalous entries:")
for entry in result.anomalous_entries:
    print(f"  - {entry}")
print(f"\nExplanation:\n{result.explanation}")

# Batch processing pipeline for streaming logs
def process_log_window(log_entries: list[dict], baseline: str, window_size: int = 50):
    """Score a rolling window of log entries."""
    # Group into batches of window_size
    for i in range(0, len(log_entries), window_size):
        batch = log_entries[i:i + window_size]
        result = log_scorer(
            log_batch=json.dumps(batch, indent=2),
            baseline_summary=baseline
        )
        if result.severity not in ("normal", "low"):
            yield {
                "window_start": batch[0]["time"],
                "window_end": batch[-1]["time"],
                "severity": result.severity,
                "score": result.anomaly_score,
                "explanation": result.explanation,
                "anomalous_entries": result.anomalous_entries,
            }