Sequential Workflow Deep Dive: State Consistency, Error Propagation, and Performance Optimization in Pipeline Orchestration

The Sequential workflow is the most basic and commonly used pattern in an Agent Team. It looks simple—A runs, then B, then C—but in production, pipeline orchestration involves complex issues such as state consistency, error propagation, and backpressure control. This post dives into the architectural design and engineering practice of Sequential workflows.

Core Principles of Pipeline Architecture

Data Flow Model

A Sequential workflow is essentially an implementation of the Pipe-Filter Pattern:

Input Data
    │
    ▼
┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  Filter A   │───►│  Filter B   │───►│  Filter C   │
│ (Data Fetch)│    │(Data Transform)│  │(Data Output)│
└─────────────┘    └─────────────┘    └─────────────┘
    │                    │                    │
    ▼                    ▼                    ▼
 RawData           ProcessedData         FinalResult

Each Agent is both a consumer (receiving upstream output) and a producer (generating downstream input). The key to this design is defining a clear Data Contract.

Data Contract Design

// Standard format for pipeline data
type PipelineData struct {
    // Version control - ensures compatibility
    SchemaVersion string `json:"schema_version"`

    // Business data
    Payload interface{} `json:"payload"`

    // Metadata - used for tracing and debugging
    Metadata PipelineMetadata `json:"metadata"`

    // Context passing
    Context map[string]interface{} `json:"context"`

    // Error information (if any)
    Error *PipelineError `json:"error,omitempty"`
}

type PipelineMetadata struct {
    StepID        string    `json:"step_id"`
    StepName      string    `json:"step_name"`
    Timestamp     time.Time `json:"timestamp"`
    LatencyMs     int64     `json:"latency_ms"`
    TokenUsed     int       `json:"token_used"`
    RetryCount    int       `json:"retry_count"`
    PreviousSteps []string  `json:"previous_steps"`
}

type PipelineError struct {
    Code      string `json:"code"`
    Message   string `json:"message"`
    Step      string `json:"step"`
    Retriable bool   `json:"retriable"`
}

Design Principles:

Schema Versioning: use version numbers to remain compatible with old code when data formats change.
Immutable Payload: each step generates a new Payload without modifying upstream data.
Complete Metadata: record execution details of each step to make troubleshooting easier.
Error Propagation: error information travels through the pipeline so downstream steps can decide how to handle it.

Production-Grade Sequential Implementation

Core Structures

type SequentialWorkflow struct {
    steps        []Step                    // step list
    state        *WorkflowState            // workflow state
    retryPolicy  *RetryPolicy              // retry policy
    errorHandler ErrorHandler              // error handler
    middleware   []Middleware              // middleware chain
    observers    []Observer                // observers
    tokenBudget  *TokenBudget              // token budget
    timeout      time.Duration             // total timeout
}

type Step struct {
    Name        string
    Agent       *agent.Agent
    Condition   ConditionFunc             // execution condition
    Transformer DataTransformer           // data transformer
    Timeout     time.Duration             // per-step timeout
    RetryPolicy *RetryPolicy              // per-step retry policy
    OnError     ErrorAction               // error handling action
}

// Error handling strategies
type ErrorAction int
const (
    ErrorActionFail     ErrorAction = iota // fail immediately
    ErrorActionSkip                        // skip current step
    ErrorActionRetry                       // retry current step
    ErrorActionFallback                    // use fallback logic
    ErrorActionContinue                    // continue (log the error)
)

Execution Engine

func (sw *SequentialWorkflow) Execute(ctx context.Context, input string) (*WorkflowResult, error) {
    // 1. Create a context with timeout
    ctx, cancel := context.WithTimeout(ctx, sw.timeout)
    defer cancel()

    // 2. Initialize pipeline data
    data := &PipelineData{
        SchemaVersion: "1.0",
        Payload:       input,
        Metadata: PipelineMetadata{
            StepID:    generateStepID(),
            Timestamp: time.Now(),
        },
        Context: make(map[string]interface{}),
    }

    // 3. Execute the step chain
    for i, step := range sw.steps {
        select {
        case <-ctx.Done():
            return nil, fmt.Errorf("workflow timeout at step %d (%s): %w", i, step.Name, ctx.Err())
        default:
        }

        // Check execution condition
        if step.Condition != nil && !step.Condition(data) {
            sw.notifyObservers(&StepEvent{
                StepIndex: i,
                StepName:  step.Name,
                Action:    "skipped",
            })
            continue
        }

        // Execute the step (with retry and error handling)
        result, err := sw.executeStep(ctx, step, data)
        if err != nil {
            handled, newData := sw.handleStepError(step, data, err)
            if !handled {
                return nil, fmt.Errorf("step %d (%s) failed: %w", i, step.Name, err)
            }
            data = newData
            continue
        }

        data = result
    }

    return &WorkflowResult{
        Output:    data.Payload.(string),
        Metadata:  data.Metadata,
        TokenUsed: sw.tokenBudget.UsedTokens,
    }, nil
}

func (sw *SequentialWorkflow) executeStep(
    ctx context.Context,
    step Step,
    input *PipelineData,
) (*PipelineData, error) {
    // Apply middleware
    handler := sw.applyMiddleware(step.Agent.Run)

    // Data transformation
    agentInput := input.Payload.(string)
    if step.Transformer != nil {
        agentInput = step.Transformer.Transform(agentInput)
    }

    // Execution with retry
    var result string
    var err error

    policy := step.RetryPolicy
    if policy == nil {
        policy = sw.retryPolicy
    }

    for attempt := 0; attempt <= policy.MaxAttempts; attempt++ {
        stepCtx, cancel := context.WithTimeout(ctx, step.Timeout)

        start := time.Now()
        result, err = handler(stepCtx, agentInput)
        latency := time.Since(start)
        cancel()

        if err == nil {
            // Success: update token budget
            tokens := estimateTokens(result)
            sw.tokenBudget.Allocate(tokens)

            return &PipelineData{
                SchemaVersion: input.SchemaVersion,
                Payload:       result,
                Metadata: PipelineMetadata{
                    StepID:        generateStepID(),
                    StepName:      step.Name,
                    Timestamp:     time.Now(),
                    LatencyMs:     latency.Milliseconds(),
                    TokenUsed:     tokens,
                    RetryCount:    attempt,
                    PreviousSteps: append(input.Metadata.PreviousSteps, input.Metadata.StepID),
                },
                Context: mergeContext(input.Context, map[string]interface{}{
                    fmt.Sprintf("%s_output", step.Name): result,
                }),
            }, nil
        }

        // Failure: decide whether to retry
        if attempt < policy.MaxAttempts && policy.IsRetriable(err) {
            backoff := policy.CalculateBackoff(attempt)
            sw.notifyObservers(&StepEvent{
                StepName: step.Name,
                Action:   "retry",
                Attempt:  attempt + 1,
                Error:    err,
            })
            time.Sleep(backoff)
            continue
        }

        break
    }

    return nil, fmt.Errorf("step %s failed after %d attempts: %w", step.Name, policy.MaxAttempts, err)
}

Hands-On Scenario: News Writing Pipeline

Full Implementation

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/google/adk-go/agent"
    "github.com/google/adk-go/team"
    "github.com/google/adk-go/tool"
)

// NewsWritingPipeline news writing pipeline
type NewsWritingPipeline struct {
    workflow *team.SequentialWorkflow
}

func NewNewsWritingPipeline(model agent.Model) (*NewsWritingPipeline, error) {
    // 1. Search Agent - responsible for information gathering
    searchAgent, err := agent.New(agent.Config{
        Name:        "search-expert",
        Model:       model,
        Instruction: `You are a senior news researcher. Based on the user's topic, search and organize key information.
Requirements:
1. Collect information from at least 5 reliable sources.
2. Annotate sources and timestamps.
3. Distinguish facts from opinions.
4. Output structured data in JSON format.`,
        Tools: []tool.Tool{
            tool.NewSearchTool(),
            tool.NewWebScraper(),
        },
        Timeout: 30 * time.Second,
    })
    if err != nil {
        return nil, fmt.Errorf("create search agent: %w", err)
    }

    // 2. Writing Agent - responsible for content creation
    writerAgent, err := agent.New(agent.Config{
        Name:        "writing-expert",
        Model:       model,
        Instruction: `You are a senior tech editor. Based on the research provided, write a high-quality news article.
Requirements:
1. Attractive and accurate headline.
2. Lead paragraph summarizes core information.
3. Clear logic and well-structured paragraphs.
4. Cite sources.
5. Keep the length between 800 and 1200 words.`,
        Timeout: 45 * time.Second,
    })
    if err != nil {
        return nil, fmt.Errorf("create writer agent: %w", err)
    }

    // 3. Fact-checking Agent - responsible for accuracy validation
    factCheckerAgent, err := agent.New(agent.Config{
        Name:        "fact-checker",
        Model:       model,
        Instruction: `You are a fact-checking expert. Verify every factual statement in the article line by line.
Requirements:
1. Mark all statements that need verification.
2. Give a confidence rating (high/medium/low).
3. Suggest revisions for low-confidence content.
4. Output a fact-check report.`,
        Tools: []tool.Tool{
            tool.NewFactCheckTool(),
        },
        Timeout: 30 * time.Second,
    })
    if err != nil {
        return nil, fmt.Errorf("create fact checker agent: %w", err)
    }

    // 4. Editor Agent - responsible for final polishing
    editorAgent, err := agent.New(agent.Config{
        Name:        "editor-expert",
        Model:       model,
        Instruction: `You are the editor-in-chief. Revise the article based on the fact-check report and output the final version.
Requirements:
1. Correct all factual errors.
2. Refine language expression.
3. Ensure consistent formatting.
4. Add editorial notes explaining the changes.`,
        Timeout: 30 * time.Second,
    })
    if err != nil {
        return nil, fmt.Errorf("create editor agent: %w", err)
    }

    // Build the Sequential workflow
    workflow := team.NewSequentialWorkflow(
        team.WithStep(team.Step{
            Name:    "research",
            Agent:   searchAgent,
            Timeout: 30 * time.Second,
            RetryPolicy: &team.RetryPolicy{
                MaxAttempts: 3,
                Backoff:     team.ExponentialBackoff(time.Second, 10*time.Second),
            },
            OnError: team.ErrorActionFail, // search failure fails the whole workflow
        }),
        team.WithStep(team.Step{
            Name:    "writing",
            Agent:   writerAgent,
            Timeout: 45 * time.Second,
            Transformer: &ResearchToWritingTransformer{}, // data format conversion
            OnError: team.ErrorActionRetry,
        }),
        team.WithStep(team.Step{
            Name:    "fact-check",
            Agent:   factCheckerAgent,
            Timeout: 30 * time.Second,
            Condition: func(data *PipelineData) bool {
                // Only articles with factual claims need checking
                return data.Context["require_fact_check"] == true
            },
            OnError: team.ErrorActionContinue, // continue on check failure but flag warning
        }),
        team.WithStep(team.Step{
            Name:    "editing",
            Agent:   editorAgent,
            Timeout: 30 * time.Second,
            OnError: team.ErrorActionFallback,
            Fallback: func(input string) (string, error) {
                // Fallback: return the draft without polishing
                log.Printf("Editor failed, returning raw draft")
                return input, nil
            },
        }),
        team.WithTokenBudget(50000),            // total token budget
        team.WithTotalTimeout(3 * time.Minute), // total timeout
        team.WithObserver(&LoggingObserver{}),  // logging observer
        team.WithObserver(&MetricsObserver{}),  // metrics observer
    )

    return &NewsWritingPipeline{workflow: workflow}, nil
}

// ResearchToWritingTransformer converts research output into writing input
type ResearchToWritingTransformer struct{}

func (t *ResearchToWritingTransformer) Transform(input string) string {
    // Parse the JSON research result
    // Extract key information and generate a writing prompt
    return fmt.Sprintf(`Write a news article based on the following research findings:

%s

Please ensure:
1. Use all key information provided.
2. Maintain an objective and neutral tone.
3. Cite sources appropriately.`, input)
}

// LoggingObserver logging observer
type LoggingObserver struct{}

func (o *LoggingObserver) OnStepStart(step string, input interface{}) {
    log.Printf("[Pipeline] Step %s started", step)
}

func (o *LoggingObserver) OnStepComplete(step string, result interface{}, latency time.Duration) {
    log.Printf("[Pipeline] Step %s completed in %v", step, latency)
}

func (o *LoggingObserver) OnStepError(step string, err error, attempt int) {
    log.Printf("[Pipeline] Step %s failed (attempt %d): %v", step, attempt, err)
}

// MetricsObserver metrics observer
type MetricsObserver struct {
    stepLatencies map[string][]time.Duration
}

func (o *MetricsObserver) OnStepComplete(step string, result interface{}, latency time.Duration) {
    o.stepLatencies[step] = append(o.stepLatencies[step], latency)
}

func (o *MetricsObserver) GetAverageLatency(step string) time.Duration {
    latencies := o.stepLatencies[step]
    if len(latencies) == 0 {
        return 0
    }
    var total time.Duration
    for _, l := range latencies {
        total += l
    }
    return total / time.Duration(len(latencies))
}

func main() {
    ctx := context.Background()

    pipeline, err := NewNewsWritingPipeline(model)
    if err != nil {
        log.Fatalf("Failed to create pipeline: %v", err)
    }

    result, err := pipeline.workflow.Execute(ctx, "What are today's tech news?")
    if err != nil {
        log.Fatalf("Pipeline failed: %v", err)
    }

    fmt.Printf("Final article:\n%s\n", result.Output)
    fmt.Printf("Total tokens used: %d\n", result.TokenUsed)
    fmt.Printf("Total latency: %v\n", result.Metadata.LatencyMs)
}

Error Handling Strategies in Detail

Error Classification and Handling Matrix

// Error classifier
type ErrorClassifier struct {
    rules []ClassificationRule
}

type ClassificationRule struct {
    Pattern  *regexp.Regexp
    Category ErrorCategory
    Severity ErrorSeverity
    Action   ErrorAction
}

type ErrorCategory string
const (
    ErrorCategoryNetwork    ErrorCategory = "network"    // network error
    ErrorCategoryRateLimit  ErrorCategory = "rate_limit" // rate limiting
    ErrorCategoryValidation ErrorCategory = "validation" // validation error
    ErrorCategoryTimeout    ErrorCategory = "timeout"    // timeout
    ErrorCategoryModel      ErrorCategory = "model"      // model error
    ErrorCategoryTool       ErrorCategory = "tool"       // tool error
)

type ErrorSeverity string
const (
    ErrorSeverityCritical ErrorSeverity = "critical" // fatal, must interrupt
    ErrorSeverityHigh     ErrorSeverity = "high"     // severe, recommend interrupt
    ErrorSeverityMedium   ErrorSeverity = "medium"   // moderate, can degrade
    ErrorSeverityLow      ErrorSeverity = "low"      // minor, can ignore
)

// Default classification rules
var DefaultRules = []ClassificationRule{
    {
        Pattern:  regexp.MustCompile(`(?i)timeout|deadline exceeded`),
        Category: ErrorCategoryTimeout,
        Severity: ErrorSeverityHigh,
        Action:   ErrorActionRetry,
    },
    {
        Pattern:  regexp.MustCompile(`(?i)rate limit|too many requests`),
        Category: ErrorCategoryRateLimit,
        Severity: ErrorSeverityMedium,
        Action:   ErrorActionRetry,
    },
    {
        Pattern:  regexp.MustCompile(`(?i)invalid|validation|bad request`),
        Category: ErrorCategoryValidation,
        Severity: ErrorSeverityCritical,
        Action:   ErrorActionFail,
    },
    {
        Pattern:  regexp.MustCompile(`(?i)connection refused|network error`),
        Category: ErrorCategoryNetwork,
        Severity: ErrorSeverityHigh,
        Action:   ErrorActionRetry,
    },
}

Circuit Breaker Pattern

type CircuitBreaker struct {
    state            CircuitState
    failureCount     int
    successCount     int
    lastFailureTime  time.Time
    threshold        int
    resetTimeout     time.Duration
    halfOpenMaxCalls int
    mu               sync.RWMutex
}

type CircuitState int
const (
    StateClosed    CircuitState = iota // normal state
    StateOpen                          // open state
    StateHalfOpen                      // half-open state
)

func (cb *CircuitBreaker) Execute(fn func() error) error {
    cb.mu.Lock()

    switch cb.state {
    case StateOpen:
        if time.Since(cb.lastFailureTime) > cb.resetTimeout {
            cb.state = StateHalfOpen
            cb.failureCount = 0
            cb.successCount = 0
        } else {
            cb.mu.Unlock()
            return fmt.Errorf("circuit breaker is open")
        }
    case StateHalfOpen:
        if cb.successCount+cb.failureCount >= cb.halfOpenMaxCalls {
            cb.mu.Unlock()
            return fmt.Errorf("circuit breaker half-open limit reached")
        }
    }

    cb.mu.Unlock()

    err := fn()

    cb.mu.Lock()
    defer cb.mu.Unlock()

    if err != nil {
        cb.failureCount++
        cb.lastFailureTime = time.Now()

        if cb.state == StateHalfOpen || cb.failureCount >= cb.threshold {
            cb.state = StateOpen
        }
        return err
    }

    cb.successCount++

    if cb.state == StateHalfOpen && cb.successCount >= cb.halfOpenMaxCalls {
        cb.state = StateClosed
        cb.failureCount = 0
    } else if cb.state == StateClosed {
        // Consecutive successes, reset counter
        if cb.successCount > cb.threshold {
            cb.failureCount = 0
        }
    }

    return nil
}

Performance Optimization Strategies

1. Pipeline Parallelism

When some steps in a Sequential workflow process independent data, micro-parallelism can be introduced:

func (sw *SequentialWorkflow) executeWithMicroParallelism(
    ctx context.Context,
    steps []Step,
    data *PipelineData,
) (*PipelineData, error) {
    // Detect groups of steps that can run in parallel
    groups := sw.detectParallelGroups(steps)

    currentData := data
    for _, group := range groups {
        if len(group) == 1 {
            // Single step, run sequentially
            result, err := sw.executeStep(ctx, group[0], currentData)
            if err != nil {
                return nil, err
            }
            currentData = result
        } else {
            // Multiple steps, run in parallel
            results := make([]*PipelineData, len(group))
            errs := make([]error, len(group))

            var wg sync.WaitGroup
            for i, step := range group {
                wg.Add(1)
                go func(idx int, s Step) {
                    defer wg.Done()
                    results[idx], errs[idx] = sw.executeStep(ctx, s, currentData)
                }(i, step)
            }
            wg.Wait()

            // Merge results
            currentData = sw.mergeParallelResults(currentData, results, errs)
        }
    }

    return currentData, nil
}

2. Result Caching

type ResultCache struct {
    backend CacheBackend
    ttl     time.Duration
    keyFunc func(string) string
}

func (c *ResultCache) GetOrExecute(
    ctx context.Context,
    key string,
    fn func() (string, error),
) (string, error) {
    cacheKey := c.keyFunc(key)

    // Try reading from cache
    if cached, err := c.backend.Get(cacheKey); err == nil {
        return cached, nil
    }

    // Execute and cache
    result, err := fn()
    if err != nil {
        return "", err
    }

    c.backend.Set(cacheKey, result, c.ttl)
    return result, nil
}

// Apply in Sequential
func (sw *SequentialWorkflow) executeStepWithCache(
    ctx context.Context,
    step Step,
    input *PipelineData,
) (*PipelineData, error) {
    if step.CacheConfig == nil {
        return sw.executeStep(ctx, step, input)
    }

    cacheKey := fmt.Sprintf("%s:%s", step.Name, hashInput(input.Payload.(string)))

    result, err := step.CacheConfig.GetOrExecute(ctx, cacheKey, func() (string, error) {
        data, err := sw.executeStep(ctx, step, input)
        if err != nil {
            return "", err
        }
        return data.Payload.(string), nil
    })

    if err != nil {
        return nil, err
    }

    return &PipelineData{
        Payload: result,
        // ... other fields
    }, nil
}

3. Backpressure Control

type BackpressureController struct {
    maxInflight int
    semaphore   chan struct{}
    queue       chan Task
}

func NewBackpressureController(maxInflight, queueSize int) *BackpressureController {
    return &BackpressureController{
        maxInflight: maxInflight,
        semaphore:   make(chan struct{}, maxInflight),
        queue:       make(chan Task, queueSize),
    }
}

func (bc *BackpressureController) Submit(task Task) error {
    select {
    case bc.queue <- task:
        return nil
    default:
        return fmt.Errorf("queue full, backpressure applied")
    }
}

func (bc *BackpressureController) Start(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return
        case task := <-bc.queue:
            bc.semaphore <- struct{}{}
            go func(t Task) {
                defer func() { <-bc.semaphore }()
                t.Execute()
            }(task)
        }
    }
}

Common Questions in Depth

Q: How can Sequential latency be optimized?

A: Optimize at three levels:

Model level: use faster models (e.g., gemini-flash instead of gemini-pro), sacrificing a small amount of quality for speed.
Architecture level: introduce micro-parallelism to parallelize independent steps.
Infrastructure level: use model caches such as GPTCache to return cached results directly for common inputs.

Measured data: in a 4-step news writing pipeline, model downgrade + result caching reduced average latency from 45s to 18s.

Q: How do I handle incompatible data formats between steps?

A: Use the Adapter Pattern:

type DataAdapter interface {
    Convert(input interface{}) (interface{}, error)
    Validate(data interface{}) error
}

// Each step can configure input/output adapters
type Step struct {
    // ...
    InputAdapter  DataAdapter
    OutputAdapter DataAdapter
}

Q: How can long-running Sequential workflows prevent state loss?

A: Implement a checkpoint mechanism:

func (sw *SequentialWorkflow) executeWithCheckpoints(ctx context.Context, input string) error {
    checkpoint, err := sw.loadCheckpoint()
    if err == nil && checkpoint != nil {
        // Resume from checkpoint
        log.Printf("Resuming from checkpoint at step %d", checkpoint.StepIndex)
        sw.resumeFromCheckpoint(checkpoint)
    }

    for i := checkpoint.StepIndex; i < len(sw.steps); i++ {
        // Execute step
        result, err := sw.executeStep(ctx, sw.steps[i], data)

        // Save checkpoint
        sw.saveCheckpoint(&Checkpoint{
            StepIndex: i + 1,
            Data:      result,
            Timestamp: time.Now(),
        })

        if err != nil {
            return err
        }

        data = result
    }

    // Clear checkpoint after completion
    sw.clearCheckpoint()
    return nil
}

Next Steps

You now have a deep understanding of Sequential workflow pipeline orchestration. Next, explore the Parallel workflow—concurrent execution control, result aggregation, and consistency guarantees.

← Agent Team Architecture | Parallel Workflow →

Want to learn more Go ADK hands-on? Follow the “Mengshou Programming” channel for weekly updates on Go and AI programming.

Frequently Asked Questions

What scenarios are suitable for the Sequential workflow?

It is suitable for pipeline tasks with clear, fixed steps, such as search → write → review → publish.

How can Sequential latency be optimized?

Use faster models, introduce micro-parallelism for independent steps, use result caching, and model caches such as GPTCache.

How do I handle incompatible data formats between steps?

Use the Adapter Pattern: configure input/output adapters for each step to perform data transformation and validation.

How can long-running Sequential workflows prevent state loss?

Implement a checkpoint mechanism that periodically saves the step index and intermediate results, supporting resume after failure.

What is backpressure control in Sequential workflows?

When a downstream step is slower than upstream, bounded channels or token buckets slow the upstream send rate to avoid memory spikes.

Core Principles of Pipeline Architecture#

Data Flow Model#

Data Contract Design#

Production-Grade Sequential Implementation#

Core Structures#

Execution Engine#

Hands-On Scenario: News Writing Pipeline#

Full Implementation#

Error Handling Strategies in Detail#

Error Classification and Handling Matrix#

Circuit Breaker Pattern#

Performance Optimization Strategies#

1. Pipeline Parallelism#

2. Result Caching#

3. Backpressure Control#

Common Questions in Depth#

Q: How can Sequential latency be optimized?#

Q: How do I handle incompatible data formats between steps?#

Q: How can long-running Sequential workflows prevent state loss?#

Next Steps#