Skip to content

README.md

Latest commit

 

History

History
297 lines (208 loc) · 10 KB

README.md

File metadata and controls

297 lines (208 loc) · 10 KB
Raw

agentdrain Package

Drain-style log template miner and anomaly detector for structured agent pipeline events — groups log lines into similarity clusters and produces scored anomaly reports.

The agentdrain package implements the Drain log template mining algorithm adapted for analyzing structured agent pipeline events. It is used for anomaly detection in agentic workflow runs.

The package also embeds default trained weights under data/, so callers can bootstrap a Coordinator with pre-trained clusters instead of starting from an empty model.

Overview

Drain is an online log parsing algorithm that groups log lines into clusters based on token similarity. Each cluster has a template — a tokenized log pattern where variable tokens are replaced with a wildcard (<*>). When a new log line arrives, Drain finds the most similar existing cluster or creates a new one.

In GitHub Agentic Workflows, agentdrain processes AgentEvent records emitted by pipeline stages (e.g. "plan", "tool_call", "finish") to:

  1. Build a model of normal behavior by training on events from successful runs.
  2. Detect anomalies in new runs by comparing events against the learned model.

Public API

Types

Config

Tuning parameters for the Drain miner.

Field Type Default Description
Depth int 4 Parse-tree depth
SimThreshold float64 0.4 Minimum similarity score to match a cluster
MaxChildren int 100 Maximum children per tree node
ParamToken string "<*>" Wildcard inserted at variable positions
RareClusterThreshold int 2 Clusters with Size ≤ this value are flagged as rare
MaskRules []MaskRule (see below) Regex substitutions applied before tokenization
ExcludeFields []string ["session_id", "trace_id", "span_id", "timestamp"] Event fields excluded from flattening

Use DefaultConfig() for production-ready defaults.

MaskRule

A regex-based substitution applied to log lines before tokenization to normalize variable content.

type MaskRule struct {
    Name        string // Human-readable identifier
    Pattern     string // Regular expression
    Replacement string // Substitution string
}

Default mask rules normalize UUIDs, session IDs, numeric values, URLs, quoted strings, and timestamps.

AgentEvent

A structured event from an agent pipeline stage.

type AgentEvent struct {
    Stage  string            // e.g. "plan", "tool_call", "finish"
    Fields map[string]string // Key-value pairs from the log line
}

Cluster

A group of log lines that share the same template.

type Cluster struct {
    ID       int      // Unique identifier
    Template []string // Tokenized template with wildcards
    Size     int      // Number of lines assigned to this cluster
    Stage    string   // Pipeline stage that generated this cluster
}

MatchResult

Returned after processing a log line.

type MatchResult struct {
    ClusterID  int      // Matched or newly created cluster ID
    Template   string   // Space-joined template string
    Params     []string // Actual token values at wildcard positions
    Similarity float64  // Fraction of non-wildcard tokens that matched exactly
    Stage      string   // Pipeline stage of the matched cluster
}

AnomalyReport

Describes anomalies detected for a log line.

type AnomalyReport struct {
    IsNewTemplate bool    // Line produced a brand-new log cluster
    LowSimilarity bool    // Best match score was below SimThreshold
    RareCluster   bool    // Matched cluster has been seen ≤ RareClusterThreshold times
    AnomalyScore  float64 // Weighted composite score in [0, 1]
    Reason        string  // Human-readable anomaly description
}

Core Components

Miner

The single-stage Drain miner. Processes one pipeline stage at a time.

cfg := agentdrain.DefaultConfig()
miner, err := agentdrain.NewMiner(cfg)

// Process a raw log line (training + matching in one step)
result, err := miner.Train(rawLogLine)

// Training phase — call for structured events from known-good runs
result, err := miner.TrainEvent(evt)

// Analysis phase — call for events to check for anomalies
result, report, err := miner.AnalyzeEvent(evt)

// Inspect clusters
clusters := miner.Clusters()

Persistence

// Save miner state to JSON
data, err := miner.SaveJSON()

// Restore miner state from JSON
err = miner.LoadJSON(data)

Coordinator

Manages a separate Miner per pipeline stage, routing events to the correct miner.

stages := []string{"plan", "tool_call", "finish"}
coord, err := agentdrain.NewCoordinator(cfg, stages)

// Load default trained weights
err = coord.LoadDefaultWeights()

// Training phase — route events from known-good runs to stage miners
result, err := coord.TrainEvent(evt)

// Analyze an event
result, report, err := coord.AnalyzeEvent(evt)

// Access all clusters across all stages
allClusters := coord.AllClusters()

// Save/restore snapshots
snapshots, err := coord.SaveSnapshots()
err = coord.LoadSnapshots(snapshots)

// Save/restore coordinator weights as JSON
data, err := coord.SaveWeightsJSON()
err = coord.LoadWeightsJSON(data)

AnomalyDetector

Post-processes MatchResult values to produce an AnomalyReport.

NewAnomalyDetector(simThreshold float64, rareClusterThreshold int) (*AnomalyDetector, error)

Creates a new AnomalyDetector. Returns an error if simThreshold is outside [0,1] or rareClusterThreshold is negative.

(*AnomalyDetector).Analyze(result *MatchResult, isNew bool, cluster *Cluster) *AnomalyReport

Evaluates result and produces an AnomalyReport. isNew indicates that result created a brand-new cluster; cluster is the matched or newly created Cluster.

detector, err := agentdrain.NewAnomalyDetector(cfg.SimThreshold, cfg.RareClusterThreshold)
if err != nil {
    panic(err)
}
report := detector.Analyze(result, isNew, cluster)

Masker

Applies MaskRule substitutions to log lines before tokenization.

(*Masker).Mask(line string) string

Applies all compiled mask rules to line in order, returning the normalized string.

masker, err := agentdrain.NewMasker(cfg.MaskRules)
masked := masker.Mask(rawLine)

Utility Functions

FlattenEvent(evt AgentEvent, excludeFields []string) string

Converts an AgentEvent to a single string for tokenization, omitting fields listed in excludeFields. Fields are sorted for deterministic output.

Tokenize(line string) []string

Splits a log line into tokens on whitespace boundaries.

StageSequence(events []AgentEvent) string

Returns a space-separated string of the stages from a slice of events (e.g. "plan tool_call tool_result finish"). Useful for summarizing pipeline execution paths.

Snapshot / SnapshotCluster

Serializable representations of miner state used for persistence.

type Snapshot struct {
    Config   Config            // Miner configuration
    Clusters []SnapshotCluster // Serialized cluster list
    NextID   int               // Next cluster ID counter
}

type SnapshotCluster struct {
    ID       int      // Cluster identifier
    Template []string // Tokenized template with wildcards
    Size     int      // Number of lines assigned to cluster
    Stage    string   // Pipeline stage
}

These types are returned and consumed by SaveSnapshots / LoadSnapshots and are serialized as JSON.

Usage Examples

import "github.com/github/gh-aw/pkg/agentdrain"

// Create a coordinator with default config
cfg := agentdrain.DefaultConfig()
stages := []string{"plan", "tool_call", "finish"}
coord, err := agentdrain.NewCoordinator(cfg, stages)
if err != nil {
    panic(err)
}

// Load pre-trained weights
if err := coord.LoadDefaultWeights(); err != nil {
    panic(err)
}

// Analyze an incoming agent event
evt := agentdrain.AgentEvent{
    Stage:  "tool_call",
    Fields: map[string]string{"tool": "read_file", "path": "/workspace/main.go"},
}
result, report, err := coord.AnalyzeEvent(evt)
if err != nil {
    panic(err)
}
if report.IsNewTemplate {
    fmt.Printf("New behavior pattern detected (score=%.2f): %s\n",
        report.AnomalyScore, report.Reason)
}

Thread Safety

Both Miner and Coordinator are safe for concurrent use from multiple goroutines. All mutating operations are protected by an internal sync.RWMutex:

  • Miner.Train, Miner.TrainEvent, Miner.AnalyzeEvent — acquire a write lock
  • Miner.Clusters — acquires a read lock
  • Miner.SaveJSON / Miner.LoadJSON — read and write lock respectively
  • Coordinator.TrainEvent, Coordinator.AnalyzeEvent — delegate to per-stage Miner instances
  • Coordinator.AllClusters, Coordinator.SaveSnapshots — acquire a read lock on the coordinator map

Dependencies

Internal:

  • github.com/github/gh-aw/pkg/logger — debug logging
  • github.com/github/gh-aw/pkg/sliceutil — slice utilities for cluster management

Default Weights

The package embeds a set of default trained weights (in data/) via //go:embed. Call coord.LoadDefaultWeights() to initialize the coordinator with pre-trained cluster weights rather than starting cold.

Update embedded weights by running gh aw logs --train --output <dir> and copying the resulting drain3_weights.json to pkg/agentdrain/data/default_weights.json, then rebuilding the binary.

Design Notes

  • The Drain algorithm is O(n·d) per event, where n is the number of tokens and d is Depth.
  • SimThreshold of 0.4 means at least 40% of tokens must match exactly (excluding wildcards) for a line to join an existing cluster.
  • The Coordinator routes each AgentEvent to its stage-specific Miner so that templates from different stages do not interfere.
  • SaveJSON/LoadJSON serialize the parse tree and cluster list to enable persistence across workflow runs.

This specification is automatically maintained by the spec-extractor workflow.