Agent Pipeline Constructor

The Agent Pipeline Constructor is an AI-powered service that analyses input signals, detects patterns, generates pipeline definitions with confidence scoring, and refines them through user feedback.

Architecture

graph TD
    S[Input Signals] --> SA[SignalAnalyser]
    SA --> PD[PatternDetector]
    PD --> AR[AnalysisResult]
    AR --> PB[PipelineBuilder]
    PB --> CS[ConfidenceScorer]
    CS --> AG[AlternativeGenerator]
    AG --> GP[GeneratedPipelinePresentation]
    GP --> FB[UserFeedback]
    FB --> FP[FeedbackProcessor]
    FP --> GP2[Refined Pipeline]

Signal Types

Type	Content	What It Provides
metadata	Map of key-value pairs	Volume hints, format info, scheduling patterns
schema	Map of field→type	Field count, nesting depth, type diversity
sample	List of records	Null detection, duplicate detection, quality indicators
log	String content	Batch/streaming/CDC keywords, error patterns
config	Map of settings	Execution mode, parallelism hints

Pattern Detection

The PatternDetector uses weighted scoring from signal combinations:

Pattern	Trigger Keywords/Signals
batch_etl	batch, job, spark, schedule, cron
streaming	stream, kafka, kinesis, event, topic
cdc	cdc, binlog, debezium, change_data
hybrid	Multiple patterns with significant scores (>0.2 each)

Usage

Analyse Signals

AgentPipelineConstructor agent = new DefaultAgentPipelineConstructor();

AnalysisResult analysis = agent.analyse(List.of(
    new InputSignal("metadata", "catalog",
        Map.of("format", "parquet", "record_count", 500000), Map.of()),
    new InputSignal("schema", "users-table",
        Map.of("id", "integer", "name", "string", "email", "string", "created_at", "timestamp"), Map.of()),
    new InputSignal("log", "application",
        "2024-01-15 batch job spark processing 500000 records completed", Map.of()),
    new InputSignal("config", "pipeline",
        Map.of("mode", "batch", "schedule", "daily"), Map.of())
));

// analysis.patterns() → ["batch_etl"]
// analysis.suggestedStages() → ["source", "validation", "transformation", "publish"]
// analysis.dataCharacteristics() → {volume_estimate=medium, schema_complexity=simple, ...}
// analysis.confidence() → ConfidenceScore(overall=0.75, ...)

Generate Pipeline

GeneratedPipelinePresentation result = agent.generate(analysis);

Pipeline pipeline = result.pipeline();
// pipeline.name() → "batch-etl-pipeline"
// pipeline.executionMode() → BATCH
// pipeline.stages() → 4 stages with resolved capabilities and timeouts

String explanation = result.explanation();
// "Generated a batch pipeline with 4 stages based on detected patterns: batch_etl. Overall confidence: 75%."

Map<String, Double> scores = result.componentScores();
// {source=0.71, validation=0.71, transformation=0.56, publish=0.68}

List<Alternative> alternatives = result.alternatives();
// [{name="Streaming Pipeline", description="Process data in real-time...", tradeoffs=[...]}]

Refine with Feedback

// User wants to increase transform timeout
UserFeedback feedback = new UserFeedback(
    "modify", "transform", "Data is larger than expected",
    Map.of("timeout_minutes", 60)
);

GeneratedPipelinePresentation refined = agent.refine(generatedPipeline, feedback);
// Transform stage now has Duration.ofMinutes(60)

Feedback types:

Type	Effect
approve	No changes; confidence boosted
reject	Removes the targeted stage
modify	Adjusts timeout, capability, or pipeline-level settings
clarify	Adjusts execution mode or scales timeouts by multiplier

Validate

ValidationResult<Pipeline> result = agent.validate(pipeline);

if (!result.valid()) {
    result.violations().forEach(v -> System.err.println("  ✗ " + v));
}

Validation rules: - Pipeline must have at least one stage - Stage names must be unique - SOURCE stages must precede PUBLISH stages - CDC stages require CDC or HYBRID execution mode - All stages must have non-zero timeouts and capability references

Confidence Scoring

Confidence is computed at two levels:

Analysis phase (from signals):

Factor	Weight	Calculation
Pattern detection	50%	Strength of pattern signals
Signal coverage	30%	Distinct signal types / 4
Signal volume	20%	Total signals / 5

Generation phase (per component):

• Base confidence = analysis overall score
• Multiplied by stage type factor (SOURCE=0.95, TRANSFORMATION=0.75, CDC=0.80, etc.)
• Reduced further by schema complexity (complex=×0.8, moderate=×0.9)

Components below the 0.5 threshold generate ReviewItem entries requiring human verification.

Timeout Resolution

Timeouts are assigned based on stage type and data characteristics:

Stage Type	Base Timeout	Volume Multiplier	Complexity Multiplier
SOURCE	5 min	large/very_large: ×2	complex: ×2
TRANSFORMATION	10 min	large/very_large: ×2	complex: ×2
VALIDATION	2 min	—	complex: ×2
PUBLISH	5 min	large/very_large: ×2	—
CDC	15 min	—	—