Documentation
¶
Overview ¶
Package graphclustering provides anomaly detection integration for graph-clustering.
Package graphclustering provides the graph-clustering component for community detection.
Package graphclustering provides the graph-clustering component for community detection, structural analysis, and anomaly detection.
Overview ¶
The graph-clustering component performs community detection on the entity graph using Label Propagation Algorithm (LPA), computes structural indices (k-core, pivot distances), and detects anomalies within community contexts. Optionally enhances community descriptions using LLM.
Tier ¶
Tier: STATISTICAL (Tier 1) without LLM, SEMANTIC (Tier 2) with LLM enhancement. Not used in Structural (Tier 0) deployments.
Architecture ¶
graph-clustering is a Tier 1+ component. It polls ENTITY_STATES on a timer and runs community detection each cycle; it holds no live ENTITY_STATES watcher (the declared kv-watch input port is discovery metadata). Authoritative entity bytes are contract-validated at consume time, and a poisoned value latches a sticky whole-view reset-required projection state.
┌───────────────────┐
ENTITY_STATES ─────►│ │
(KV reads) │ graph-clustering ├──► COMMUNITY_INDEX (KV)
│ ├──► STRUCTURAL_INDEX (KV)
│ ├──► ANOMALY_INDEX (KV)
└─────────┬─────────┘
│ (reads)
┌───────────────┼───────────────┐
▼ ▼ ▼
OUTGOING_INDEX INCOMING_INDEX graph-embedding
(query path)
Features ¶
- Label Propagation Algorithm (LPA) for community detection
- Configurable detection interval and batch thresholds
- Optional LLM-based community summarization
- Structural index computation (k-core decomposition, pivot distances)
- Anomaly detection within community contexts
- Semantic gap detection via graph-embedding query path
Configuration ¶
The component is configured via JSON with the following structure:
{
"ports": {
"inputs": [
{"name": "entity_watch", "subject": "ENTITY_STATES", "type": "kv-watch"}
],
"outputs": [
{"name": "communities", "subject": "COMMUNITY_INDEX", "type": "kv"},
{"name": "structural", "subject": "STRUCTURAL_INDEX", "type": "kv"},
{"name": "anomalies", "subject": "ANOMALY_INDEX", "type": "kv"}
]
},
"detection_interval": "30s",
"batch_size": 100,
"enable_llm": true,
"min_community_size": 2,
"max_iterations": 100,
"enable_structural": true,
"pivot_count": 16,
"max_hop_distance": 10,
"enable_anomaly_detection": true,
"anomaly_config": {
"enabled": true,
"core_anomaly": {"enabled": true, "min_core_for_hub_analysis": 2},
"semantic_gap": {"enabled": true, "min_semantic_similarity": 0.7},
"virtual_edges": {
"auto_apply": {"enabled": false, "min_confidence": 0.95},
"review_queue": {"enabled": false, "min_confidence": 0.7, "max_confidence": 0.95}
}
}
}
Detection Cycle ¶
When triggered, the component runs through these phases:
- Community Detection (LPA) → COMMUNITY_INDEX
- Structural Computation (if enabled) → STRUCTURAL_INDEX
- Anomaly Detection (if enabled) → ANOMALY_INDEX
Scheduling ¶
Community detection runs on a timer, every detection_interval. (batch_size is accepted in config but batch-threshold triggering is not implemented.)
Port Definitions ¶
Inputs:
- KV reads: ENTITY_STATES - polled during each detection cycle (the kv-watch port declaration is discovery metadata; no watcher is held)
Outputs:
- KV bucket: COMMUNITY_INDEX - stores detected communities
- KV bucket: STRUCTURAL_INDEX - stores k-core levels and pivot distances
- KV bucket: ANOMALY_INDEX - stores detected anomalies
Usage ¶
Register the component with the component registry:
import graphclustering "github.com/c360studio/semstreams/processor/graph-clustering"
func init() {
graphclustering.Register(registry)
}
Dependencies ¶
Upstream (reads during detection):
- graph-ingest: writes ENTITY_STATES (authoritative entity state)
- graph-index: reads OUTGOING_INDEX and INCOMING_INDEX for graph structure
- graph-embedding: queries for similar entities via NATS request/reply
Downstream:
- graph-query: reads COMMUNITY_INDEX (community cache → GraphRAG / search_graph)
- graph-gateway: reads ANOMALY_INDEX for the optional inference-review API
- STRUCTURAL_INDEX is written here but currently has no production query consumer (read back only by the anomaly detectors in-memory and by e2e validation). Wiring k-core / pivot distance into search ranking is a tracked ADR-054 follow-up.
Package graphclustering query handlers ¶
Package graphclustering provides embedding-based similarity search for anomaly detection.
Package graphclustering provides structural analysis integration for graph-clustering.
Index ¶
- func CreateGraphClustering(rawConfig json.RawMessage, deps component.Dependencies) (component.Discoverable, error)
- func Register(registry *component.Registry) error
- type CommunityRequest
- type CommunityResponse
- type Component
- func (c *Component) ConfigSchema() component.ConfigSchema
- func (c *Component) DataFlow() component.FlowMetrics
- func (c *Component) Health() component.HealthStatus
- func (c *Component) Initialize() error
- func (c *Component) InputPorts() []component.Port
- func (c *Component) Meta() component.Metadata
- func (c *Component) OutputPorts() []component.Port
- func (c *Component) Start(ctx context.Context) error
- func (c *Component) Stop(timeout time.Duration) error
- type Config
- type EntityIDEdgesConfig
- type EntityRequest
- type EntityResponse
- type LevelRequest
- type LevelResponse
- type MembersRequest
- type MembersResponse
- type SemanticEdgesConfig
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CreateGraphClustering ¶
func CreateGraphClustering(rawConfig json.RawMessage, deps component.Dependencies) (component.Discoverable, error)
CreateGraphClustering is the factory function for creating graph-clustering components
Types ¶
type CommunityRequest ¶
type CommunityRequest struct {
ID string `json:"id"`
}
CommunityRequest is the request format for community query
type CommunityResponse ¶
type CommunityResponse struct {
Community *clustering.Community `json:"community"`
}
CommunityResponse is the response format for community query
type Component ¶
type Component struct {
// contains filtered or unexported fields
}
Component implements the graph-clustering processor
func (*Component) ConfigSchema ¶
func (c *Component) ConfigSchema() component.ConfigSchema
ConfigSchema returns the configuration schema
func (*Component) DataFlow ¶
func (c *Component) DataFlow() component.FlowMetrics
DataFlow returns current data flow metrics
func (*Component) Health ¶
func (c *Component) Health() component.HealthStatus
Health returns current health status
func (*Component) Initialize ¶
Initialize validates configuration and sets up ports (no I/O)
func (*Component) InputPorts ¶
InputPorts returns input port definitions
func (*Component) OutputPorts ¶
OutputPorts returns output port definitions
type Config ¶
type Config struct {
Ports *component.PortConfig `json:"ports" schema:"type:ports,description:Port configuration,category:basic"`
DetectionIntervalStr string `` /* 132-byte string literal not displayed */
BatchSize int `json:"batch_size" schema:"type:int,description:Event count threshold for triggering detection,category:basic"`
EnableLLM bool `` /* 167-byte string literal not displayed */
EnhancementWorkers int `` /* 133-byte string literal not displayed */
MinCommunitySize int `json:"min_community_size" schema:"type:int,description:Minimum number of entities to form a community,category:advanced"`
MaxIterations int `json:"max_iterations" schema:"type:int,description:Maximum iterations for LPA algorithm,category:advanced"`
// AllowUngatedReads permits community detection to run when graph-index's readiness
// status endpoint is unreachable (gh#474 Codex #4). Default false = FAIL-CLOSED:
// unknown readiness defers the cycle, so detection cannot derive communities from a
// partial INCOMING_INDEX during a cutover when the authoritative owner is
// crashed/restarting. Set true ONLY for a standalone deployment (or test) that runs
// clustering without a co-deployed graph-index handler; it MUST NOT be used during a
// format cutover.
AllowUngatedReads bool `` /* 169-byte string literal not displayed */
// Structural analysis (optional, enables anomaly detection)
EnableStructural bool `` /* 137-byte string literal not displayed */
PivotCount int `json:"pivot_count" schema:"type:int,description:Number of pivot nodes for distance indexing (default 16),category:advanced"`
MaxHopDistance int `json:"max_hop_distance" schema:"type:int,description:Maximum BFS traversal depth (default 10),category:advanced"`
// Anomaly detection (optional, requires EnableStructural)
EnableAnomalyDetection bool `` /* 134-byte string literal not displayed */
AnomalyConfig inference.Config `json:"anomaly_config" schema:"type:object,description:Configuration for anomaly detection,category:advanced"`
// Dependency startup configuration
StartupAttempts int `` /* 130-byte string literal not displayed */
StartupInterval int `` /* 134-byte string literal not displayed */
// EntityID virtual-edge synthesis (gh#461). Omit to keep the built-in
// defaults (sibling + system-peer edges ON); set include_* false to run
// community detection on explicit topology alone.
EntityIDEdges *EntityIDEdgesConfig `` /* 191-byte string literal not displayed */
// Semantic co-location virtual-edge synthesis (Epic B B2, ADR-086). Omit to
// keep the tier OFF (byte-identical to today). When enabled, mutual-kNN
// semantic edges join the detection vote and the structural edge caps are
// rebalanced so the new tier competes rather than being dominated.
SemanticEdges *SemanticEdgesConfig `` /* 167-byte string literal not displayed */
// contains filtered or unexported fields
}
Config holds configuration for graph-clustering component
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns a valid default configuration
func (*Config) ApplyDefaults ¶
func (c *Config) ApplyDefaults()
ApplyDefaults sets default values for configuration
func (*Config) DetectionInterval ¶
DetectionInterval returns the parsed detection interval duration
type EntityIDEdgesConfig ¶
type EntityIDEdgesConfig struct {
IncludeSiblings *bool `` /* 207-byte string literal not displayed */
IncludeSystemPeers *bool `` /* 153-byte string literal not displayed */
SiblingWeight float64 `json:"sibling_weight,omitempty" schema:"type:number,description:Edge weight for synthesized sibling edges (default 0.7)"`
MaxSiblings int `json:"max_siblings,omitempty" schema:"type:int,description:Max sibling neighbors synthesized per entity (default 10)"`
SystemPeerWeight float64 `` /* 128-byte string literal not displayed */
MaxSystemPeers int `json:"max_system_peers,omitempty" schema:"type:int,description:Max system-peer neighbors synthesized per entity (default 15)"`
}
EntityIDEdgesConfig is the operator-facing shape for community detection's EntityID virtual-edge synthesis (gh#461). The two toggles are pointers so the config is tri-state: nil (unset) resolves to the built-in default, and only an explicit true/false overrides it — omitting the block therefore preserves the current behavior (both ON) rather than silently disabling synthesis. Numeric fields default when zero (mirrors clustering.NewEntityIDProvider).
type EntityRequest ¶
EntityRequest is the request format for entity community query
type EntityResponse ¶
type EntityResponse struct {
EntityID string `json:"entity_id"`
Level int `json:"level"`
Community *clustering.Community `json:"community"`
}
EntityResponse is the response format for entity community query
type LevelRequest ¶
type LevelRequest struct {
Level int `json:"level"`
}
LevelRequest is the request format for level query
type LevelResponse ¶
type LevelResponse struct {
Level int `json:"level"`
Communities []*clustering.Community `json:"communities"`
Count int `json:"count"`
}
LevelResponse is the response format for level query
type MembersRequest ¶
type MembersRequest struct {
CommunityID string `json:"community_id"`
}
MembersRequest is the request format for members query
type MembersResponse ¶
type MembersResponse struct {
CommunityID string `json:"community_id"`
Members []string `json:"members"`
Count int `json:"count"`
}
MembersResponse is the response format for members query
type SemanticEdgesConfig ¶
type SemanticEdgesConfig struct {
EnableSemanticEdges bool `` /* 235-byte string literal not displayed */
SemanticSimilarityThreshold float64 `` /* 196-byte string literal not displayed */
SemanticMaxNeighbors int `` /* 179-byte string literal not displayed */
SemanticEdgeWeight float64 `` /* 199-byte string literal not displayed */
}
SemanticEdgesConfig is the operator-facing block for community detection's semantic co-location virtual-edge synthesis (Epic B B2, ADR-086). It is a strict-decoded block (see rejectUnknownSemanticEdgeKeys) so an operator's typo on the enable toggle fails loudly rather than silently leaving the feature off. Omitting the block keeps the tier OFF and the detection edge set exactly what it is today (the gh#461 default-preservation invariant).
Numeric fields default when zero (mirrors clustering.NewSemanticEdgeProvider); the starting values are EMPIRICAL (ADR-086), tuned against partition_colocation_mean — recorded as a starting point, not asserted final.