agentdrain

package
v0.87.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 18, 2026 License: MIT Imports: 13 Imported by: 0

README

agentdrain Package

Drain-style log template mining and anomaly scoring for structured agent pipeline events.

Overview

The agentdrain package implements an online log-template miner inspired by the Drain algorithm and adapts it to AgentEvent records emitted by agentic workflow stages. It converts structured events into deterministic token streams, normalizes variable values with regex-based masking, groups similar events into clusters, and returns MatchResult values describing the matched template, extracted parameters, and similarity score.

The package supports two related workflows: training on known-good events and analyzing new events for anomalies. Miner manages a single stream of events, while Coordinator manages one Miner per stage so templates from plan, tool_call, finish, and other stages do not interfere with each other. Miner state can be serialized with Snapshot/SnapshotCluster, and coordinators can bootstrap from embedded default weights via LoadDefaultWeights.

The public API is intentionally small: event flattening and tokenization helpers, configurable masking, a concurrent miner, stage-aware coordination, and anomaly scoring. Internally, the package uses a parse tree and cluster store, but those remain unexported implementation details.

Public API

Types
Type Kind Description
AgentEvent struct Structured event with a Stage and key/value Fields used as miner input.
AnomalyDetector struct Evaluates MatchResult values and produces AnomalyReport values using similarity and rarity thresholds.
AnomalyReport struct Describes anomaly flags, normalized score, and human-readable reason text.
Cluster struct Represents a mined template cluster with ID, tokenized template, size, and optional stage.
Config struct Configures parse-tree depth, similarity threshold, wildcard token, masking rules, rarity threshold, and excluded fields.
Coordinator struct Owns one Miner per stage and provides stage-aware training, analysis, and persistence.
MaskRule struct Regex substitution rule applied before tokenization.
Masker struct Compiled ordered set of MaskRule values that normalizes log lines.
MatchResult struct Reports the cluster ID, rendered template, extracted params, similarity, and stage for a processed event.
Miner struct Concurrent Drain-style miner for one event stream.
Snapshot struct Serializable representation of a miner's config, clusters, and next cluster ID.
SnapshotCluster struct Serializable representation of one cluster inside a Snapshot.
Functions
Function Signature Description
(*AnomalyDetector).Analyze func (d *AnomalyDetector) Analyze(result *MatchResult, isNew bool, cluster *Cluster) *AnomalyReport Scores a match result and cluster context, producing anomaly flags, a normalized score, and reason text.
(*Coordinator).AllClusters func (c *Coordinator) AllClusters() map[string][]Cluster Returns a snapshot of clusters for every registered stage.
(*Coordinator).AnalyzeEvent func (c *Coordinator) AnalyzeEvent(evt AgentEvent) (*MatchResult, *AnomalyReport, error) Routes an event to its stage miner and returns both the match result and anomaly report.
(*Coordinator).LoadDefaultWeights func (c *Coordinator) LoadDefaultWeights() error Loads embedded default weights from data/default_weights.json unless the embedded file is empty or {}.
(*Coordinator).LoadSnapshots func (c *Coordinator) LoadSnapshots(snapshots map[string][]byte) error Restores per-stage miner snapshots, creating new stage miners when snapshots reference previously unknown stages.
(*Coordinator).LoadWeightsJSON func (c *Coordinator) LoadWeightsJSON(data []byte) error Restores all stage miners from a combined JSON document produced by SaveWeightsJSON.
(*Coordinator).SaveSnapshots func (c *Coordinator) SaveSnapshots() (map[string][]byte, error) Serializes each stage miner independently as JSON.
(*Coordinator).SaveWeightsJSON func (c *Coordinator) SaveWeightsJSON() ([]byte, error) Serializes all stage snapshots into one combined JSON blob suitable for embedding as default weights.
(*Coordinator).TrainEvent func (c *Coordinator) TrainEvent(evt AgentEvent) (*MatchResult, error) Routes an event to the miner for evt.Stage and updates that miner.
(*Masker).Mask func (m *Masker) Mask(line string) string Applies all configured masking rules in order.
(*Miner).AnalyzeEvent func (m *Miner) AnalyzeEvent(evt AgentEvent) (*MatchResult, *AnomalyReport, error) Performs inference, trains on the event, and returns both the resulting match and anomaly report.
(*Miner).Clusters func (m *Miner) Clusters() []Cluster Returns a snapshot of all known clusters.
(*Miner).LoadJSON func (m *Miner) LoadJSON(data []byte) error Replaces miner state from a JSON snapshot and rebuilds the parse tree.
(*Miner).SaveJSON func (m *Miner) SaveJSON() ([]byte, error) Serializes miner state to JSON.
(*Miner).Train func (m *Miner) Train(line string) (*MatchResult, error) Trains the miner on a raw line after masking and tokenization.
(*Miner).TrainEvent func (m *Miner) TrainEvent(evt AgentEvent) (*MatchResult, error) Flattens an AgentEvent, trains on it, and propagates the event stage onto the result and cluster.
DefaultConfig func DefaultConfig() Config Returns the built-in production defaults, including masking rules and excluded fields.
FlattenEvent func FlattenEvent(evt AgentEvent, excludeFields []string) string Converts an event into a deterministic space-separated key=value token string with stage= first when present.
NewAnomalyDetector func NewAnomalyDetector(simThreshold float64, rareClusterThreshold int) (*AnomalyDetector, error) Validates thresholds and constructs an anomaly detector.
NewCoordinator func NewCoordinator(cfg Config, stages []string) (*Coordinator, error) Creates a stage-aware coordinator with one miner per supplied stage.
NewMasker func NewMasker(rules []MaskRule) (*Masker, error) Compiles regex mask rules into a reusable masker.
NewMiner func NewMiner(cfg Config) (*Miner, error) Constructs a miner with compiled mask rules, a fresh parse tree, and an empty cluster store.
StageSequence func StageSequence(events []AgentEvent) string Returns the stages from a slice of events as a single space-separated string.
Tokenize func Tokenize(line string) []string Splits a line on whitespace.
Constants
Constant Type Value Description
AnomalyMaxScore untyped numeric constant 2.0 Maximum raw anomaly weight before normalization into the [0,1] score range.
AnomalyWeightLow untyped numeric constant 0.7 Weight added when a known cluster matches below the similarity threshold.
AnomalyWeightNew untyped numeric constant 1.0 Weight added when analysis creates a brand-new template cluster.
AnomalyWeightRare untyped numeric constant 0.3 Weight added when the matched cluster size is at or below the rare-cluster threshold.

Usage Examples

Examples below are taken from the package's spec tests and reflect the current public API.

cfg := agentdrain.DefaultConfig()
miner, err := agentdrain.NewMiner(cfg)
if err != nil {
	panic(err)
}

evt := agentdrain.AgentEvent{
	Stage:  "plan",
	Fields: map[string]string{"action": "start", "step": "1"},
}
result, err := miner.TrainEvent(evt)
if err != nil {
	panic(err)
}
fmt.Println(result.ClusterID)

cfg := agentdrain.DefaultConfig()
coord, err := agentdrain.NewCoordinator(cfg, []string{"plan", "tool_call", "finish"})
if err != nil {
	panic(err)
}

evt := agentdrain.AgentEvent{
	Stage:  "plan",
	Fields: map[string]string{"action": "evaluate", "step": "1"},
}
result, report, err := coord.AnalyzeEvent(evt)
if err != nil {
	panic(err)
}
fmt.Println(result.Stage, report.AnomalyScore)

flat := agentdrain.FlattenEvent(
	agentdrain.AgentEvent{
		Stage: "tool_call",
		Fields: map[string]string{
			"session_id": "abc-123",
			"action":     "start",
		},
	},
	[]string{"session_id"},
)
fmt.Println(flat)
// Output: stage=tool_call action=start

masker, err := agentdrain.NewMasker([]agentdrain.MaskRule{{
	Name:        "number_test",
	Pattern:     `\d+`,
	Replacement: "<NUM>",
}})
if err != nil {
	panic(err)
}
fmt.Println(masker.Mask("step 42 completed"))

Design Decisions

FlattenEvent is deterministic by design: it emits stage= first when present, sorts remaining field keys alphabetically, and omits explicitly excluded fields. This makes clustering stable across Go map iteration order and allows saved weights to remain reusable.

Miner.AnalyzeEvent performs inference before training, then trains on the same event and scores the resulting cluster with AnomalyDetector. The anomaly flags intentionally treat “new template” and “low similarity” as mutually exclusive so a brand-new cluster is not double-counted as both conditions.

Coordinator isolates miners by stage. This prevents templates from unrelated workflow phases from merging into the same cluster space and supports persistence as either per-stage snapshots or one combined weights document. The embedded default-weights mechanism provides an opt-in pre-trained baseline without exposing embedding details through additional API surface.

Dependencies

Internal package dependencies include pkg/logger for debug logging, pkg/setutil for exclusion-set membership checks, and pkg/sliceutil for collection helpers used while flattening events. The package also embeds data/default_weights.json for coordinator bootstrapping.

External dependencies for production code are limited to the Go standard library, including encoding/json, regexp, sort, strings, sync, and embed support.

Thread Safety

Miner is safe for concurrent use. It protects mutable state with an internal sync.RWMutex; training and load operations take the write lock, while cluster snapshots and JSON save operations take the read lock.

Coordinator is also safe for concurrent use. It protects its stage-to-miner map with its own sync.RWMutex and relies on each contained Miner for per-stage concurrency control.

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

Documentation

Index

Constants

View Source 
const (
	AnomalyWeightNew  = 1.0
	AnomalyWeightLow  = 0.7
	AnomalyWeightRare = 0.3
	AnomalyMaxScore   = 2.0
)

Scoring weights used by Analyze. Exported so tests can reference them directly and stay in sync with production logic at compile time.

Variables

This section is empty.

Functions

func FlattenEvent 

func FlattenEvent(evt AgentEvent, excludeFields []string) string

FlattenEvent converts an AgentEvent into a deterministic string suitable for template mining. Field keys are sorted alphabetically; fields listed in excludeFields are omitted. The result looks like: 

stage=tool_call key1=val1 key2=val2

func StageSequence 

func StageSequence(events []AgentEvent) string

StageSequence converts a slice of AgentEvents into a space-separated string of their stage names, e.g. "plan tool_call tool_result finish".

func Tokenize 

func Tokenize(line string) []string

Tokenize splits a log line on whitespace and returns the individual tokens.

Types

type AgentEvent 

type AgentEvent struct {
	// Stage identifies the pipeline stage (e.g., "plan", "tool_call", "finish").
	Stage string
	// Fields contains the key-value pairs parsed from the log line.
	Fields map[string]string
}

AgentEvent is a structured log event emitted by an agent pipeline stage.

type AnomalyDetector 

type AnomalyDetector struct {
	// contains filtered or unexported fields
}

AnomalyDetector evaluates match results and produces AnomalyReports.

func NewAnomalyDetector 

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

NewAnomalyDetector creates an AnomalyDetector with the given thresholds.

func (*AnomalyDetector) Analyze 

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

Analyze produces an AnomalyReport for a match result.

  • isNew indicates the line created a brand-new cluster.
  • cluster is the cluster that was matched or created.

type AnomalyReport 

type AnomalyReport struct {
	// IsNewTemplate is true when the log line produced a brand-new log cluster.
	IsNewTemplate bool
	// LowSimilarity is true when the best match score was below the configured threshold.
	LowSimilarity bool
	// RareCluster is true when the matched cluster has been seen fewer times than the rare threshold.
	RareCluster bool
	// AnomalyScore is a weighted composite score in the range [0, 1].
	AnomalyScore float64
	// Reason is a human-readable description of all anomalies that were detected.
	Reason string
}

AnomalyReport describes anomalies detected for a log line.

type Cluster 

type Cluster struct {
	// ID is the unique cluster identifier.
	ID int
	// Template is the tokenized log template with wildcards at variable positions.
	Template []string
	// Size is the number of log lines that have been assigned to this cluster.
	Size int
	// Stage identifies which agent stage generated this cluster.
	Stage string
}

Cluster represents a group of log lines that share the same template.

type Config 

type Config struct {
	// Depth controls how many levels of the parse tree are used.
	Depth int
	// SimThreshold is the minimum similarity score (0–1) required to match an existing cluster.
	SimThreshold float64
	// MaxChildren limits the number of children per internal tree node.
	MaxChildren int
	// ParamToken is the wildcard string inserted where tokens differ across log lines.
	ParamToken string
	// RareClusterThreshold marks clusters with size ≤ this value as rare.
	RareClusterThreshold int
	// MaskRules are applied before tokenization to normalize variable parts of log lines.
	MaskRules []MaskRule
	// ExcludeFields lists AgentEvent field keys that are omitted when flattening events.
	ExcludeFields []string
}

Config holds tuning parameters for the Drain log template miner.

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig returns a Config pre-loaded with sensible production defaults.

type Coordinator 

type Coordinator struct {
	// contains filtered or unexported fields
}

Coordinator manages one Miner per agent pipeline stage.

func NewCoordinator 

func NewCoordinator(cfg Config, stages []string) (*Coordinator, error)

NewCoordinator creates a Coordinator with one Miner for each provided stage name.

func (*Coordinator) AllClusters 

func (c *Coordinator) AllClusters() map[string][]Cluster

AllClusters returns a map from stage name to the list of clusters in that miner.

func (*Coordinator) AnalyzeEvent 

func (c *Coordinator) AnalyzeEvent(evt AgentEvent) (*MatchResult, *AnomalyReport, error)

AnalyzeEvent routes the event to the correct stage miner and returns both the match result and an anomaly report.

func (*Coordinator) LoadDefaultWeights 

func (c *Coordinator) LoadDefaultWeights() error

LoadDefaultWeights restores all stage miners from the embedded default weights file (pkg/agentdrain/data/default_weights.json). When the file is empty or contains only an empty JSON object the call is a no-op and returns nil.

Update the default 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.

func (*Coordinator) LoadSnapshots 

func (c *Coordinator) LoadSnapshots(snapshots map[string][]byte) error

LoadSnapshots restores each stage miner from the provided JSON bytes map. Stages that are not present in snapshots retain their current state.

func (*Coordinator) LoadWeightsJSON 

func (c *Coordinator) LoadWeightsJSON(data []byte) error

LoadWeightsJSON restores all stage miners from a combined JSON blob produced by SaveWeightsJSON.

func (*Coordinator) SaveSnapshots 

func (c *Coordinator) SaveSnapshots() (map[string][]byte, error)

SaveSnapshots serializes each stage miner's state and returns a map from stage name to JSON bytes.

func (*Coordinator) SaveWeightsJSON 

func (c *Coordinator) SaveWeightsJSON() ([]byte, error)

SaveWeightsJSON serializes all stage snapshots into a single combined JSON blob. The result can be written to pkg/agentdrain/data/default_weights.json and committed to embed it as the default starting weights for future runs.

func (*Coordinator) TrainEvent 

func (c *Coordinator) TrainEvent(evt AgentEvent) (*MatchResult, error)

TrainEvent routes the event to the miner responsible for evt.Stage. Returns an error when the stage has no associated miner.

type MaskRule 

type MaskRule struct {
	// Name is a human-readable identifier for the rule.
	Name string
	// Pattern is the regular expression to match.
	Pattern string
	// Replacement is the string substituted for each match.
	Replacement string
}

MaskRule describes a regex substitution applied to log lines before processing.

type Masker 

type Masker struct {
	// contains filtered or unexported fields
}

Masker applies a sequence of regex substitution rules to normalize log lines.

func NewMasker 

func NewMasker(rules []MaskRule) (*Masker, error)

NewMasker compiles the given MaskRules into a Masker ready for use. Returns an error if any pattern fails to compile.

func (*Masker) Mask 

func (m *Masker) Mask(line string) string

Mask applies all mask rules in order and returns the transformed line.

type MatchResult 

type MatchResult struct {
	// ClusterID is the ID of the matched or newly created cluster.
	ClusterID int
	// Template is the space-joined template string.
	Template string
	// Params holds the actual token values at wildcard positions.
	Params []string
	// Similarity is the fraction of non-wildcard positions that matched exactly.
	Similarity float64
	// Stage is the agent stage associated with the matched cluster.
	Stage string
}

MatchResult is returned after processing a log line through the miner.

type Miner 

type Miner struct {
	// contains filtered or unexported fields
}

Miner is a concurrent Drain-style log template miner. Use NewMiner to create an instance.

func NewMiner 

func NewMiner(cfg Config) (*Miner, error)

NewMiner creates a Miner from the given Config.

func (*Miner) AnalyzeEvent 

func (m *Miner) AnalyzeEvent(evt AgentEvent) (*MatchResult, *AnomalyReport, error)

AnalyzeEvent performs inference on the event, builds an AnomalyReport, and then calls TrainEvent to update the miner. Returns the match result and report.

func (*Miner) Clusters 

func (m *Miner) Clusters() []Cluster

Clusters returns a snapshot of all known clusters.

func (*Miner) LoadJSON 

func (m *Miner) LoadJSON(data []byte) error

LoadJSON restores miner state from JSON bytes produced by SaveJSON. The existing state is replaced; the parse tree is rebuilt from the snapshot.

func (*Miner) SaveJSON 

func (m *Miner) SaveJSON() ([]byte, error)

SaveJSON serializes the miner's current state to JSON bytes.

func (*Miner) Train 

func (m *Miner) Train(line string) (*MatchResult, error)

Train processes a raw log line, updates the miner state, and returns the match result. It is safe to call from multiple goroutines.

func (*Miner) TrainEvent 

func (m *Miner) TrainEvent(evt AgentEvent) (*MatchResult, error)

TrainEvent flattens the AgentEvent and calls Train.

type Snapshot 

type Snapshot struct {
	Config   Config            `json:"config"`
	Clusters []SnapshotCluster `json:"clusters"`
	NextID   int               `json:"next_id"`
}

Snapshot is the serializable representation of a Miner's state.

type SnapshotCluster 

type SnapshotCluster struct {
	ID       int      `json:"id"`
	Template []string `json:"template"`
	Size     int      `json:"size"`
	Stage    string   `json:"stage"`
}

SnapshotCluster is the serializable form of a single Cluster.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.