Parallel Workflow Deep Dive: Concurrency Control, Result Aggregation, and Consistency Guarantees

The Parallel workflow is the core pattern for improving throughput in an Agent Team. Unlike the pipelined Sequential workflow, Parallel uses the Fan-Out/Fan-In pattern to execute multiple Agents concurrently. However, concurrency brings complexity—race conditions, partial failures, result disorder, and resource contention—which requires systematic engineering solutions. This post dives into the architectural design and production practice of Parallel workflows.

Fan-Out/Fan-In Architecture Model

Core Principles

                    Input Data
                        │
                        ▼
                ┌───────────────┐
                │   Dispatcher   │ ← Task Dispatcher
                │   (Fan-Out)    │
                └───────┬───────┘
                        │
        ┌───────────────┼───────────────┐
        │               │               │
        ▼               ▼               ▼
   ┌─────────┐    ┌─────────┐    ┌─────────┐
   │ Agent A │    │ Agent B │    │ Agent C │
   │(concurrent)│  │(concurrent)│  │(concurrent)│
   └────┬────┘    └────┬────┘    └────┬────┘
        │               │               │
        └───────────────┼───────────────┘
                        │
                        ▼
                ┌───────────────┐
                │   Aggregator   │ ← Result Aggregator
                │   (Fan-In)     │
                └───────┬───────┘
                        │
                        ▼
                 Aggregated Result

Key Design Decisions:

Dispatch Strategy: round-robin, random, hash, or load-aware.
Concurrency Limit: prevent resource exhaustion.
Timeout Control: avoid a slow Agent dragging down the whole workflow.
Result Aggregation: all-wait, partial-wait, or timeout-truncate.
Error Handling: all-fail, partial-fail, or degrade.

Concurrency Model Selection

// Concurrency model enum
type ConcurrencyModel int
const (
    // Unlimited concurrency - suitable for few Agents (<5)
    ModelUnlimited ConcurrencyModel = iota

    // Fixed worker pool - suitable for medium scale (5-20)
    ModelFixedPool

    // Dynamic concurrency - automatically adjusted based on load
    ModelDynamic

    // Priority queue - important tasks run first
    ModelPriorityQueue
)

// Dynamic concurrency controller
type DynamicConcurrencyController struct {
    minWorkers     int
    maxWorkers     int
    currentWorkers int
    taskQueue      chan Task
    metrics        *ConcurrencyMetrics
    mu             sync.RWMutex
}

func (dcc *DynamicConcurrencyController) adjustWorkers() {
    dcc.mu.Lock()
    defer dcc.mu.Unlock()

    queueDepth := len(dcc.taskQueue)
    utilization := dcc.metrics.GetCPUUtilization()

    // Scale up: queue is backlogged and CPU utilization < 70%
    if queueDepth > dcc.currentWorkers*2 && utilization < 0.7 {
        newWorkers := min(dcc.currentWorkers*2, dcc.maxWorkers)
        dcc.scaleUp(newWorkers - dcc.currentWorkers)
        dcc.currentWorkers = newWorkers
    }

    // Scale down: queue is idle for 5 minutes
    if queueDepth == 0 && dcc.metrics.GetIdleDuration() > 5*time.Minute {
        newWorkers := max(dcc.currentWorkers/2, dcc.minWorkers)
        dcc.scaleDown(dcc.currentWorkers - newWorkers)
        dcc.currentWorkers = newWorkers
    }
}

Production-Grade Parallel Implementation

Core Structures

type ParallelWorkflow struct {
    agents          []*agent.Agent        // Agent list
    dispatcher      Dispatcher            // task dispatcher
    aggregator      Aggregator            // result aggregator
    concurrencyCtrl ConcurrencyController // concurrency controller
    timeout         time.Duration         // total timeout
    partialFailure  bool                  // allow partial failure
    resultTimeout   time.Duration         // per-result timeout
    errorStrategy   ErrorStrategy         // error handling strategy
    middleware      []ParallelMiddleware  // middleware
}

// Dispatcher interface
type Dispatcher interface {
    Dispatch(ctx context.Context, input string, agents []*agent.Agent) ([]*Task, error)
}

// Aggregator interface
type Aggregator interface {
    Aggregate(ctx context.Context, results []*AgentResult) (*WorkflowResult, error)
}

// Agent execution result
type AgentResult struct {
    AgentName string
    Output    string
    Error     error
    Latency   time.Duration
    Tokens    int
    Timestamp time.Time
}

// Workflow result
type WorkflowResult struct {
    Output         string
    PartialResults map[string]*AgentResult
    SuccessCount   int
    FailureCount   int
    TotalLatency   time.Duration
    TokenUsage     int
}

Execution Engine

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

    // 2. Dispatch tasks
    tasks, err := pw.dispatcher.Dispatch(ctx, input, pw.agents)
    if err != nil {
        return nil, fmt.Errorf("dispatch failed: %w", err)
    }

    // 3. Execute concurrently
    resultChan := make(chan *AgentResult, len(tasks))
    var wg sync.WaitGroup

    for _, task := range tasks {
        wg.Add(1)
        go func(t *Task) {
            defer wg.Done()

            // Apply concurrency control
            if err := pw.concurrencyCtrl.Acquire(ctx); err != nil {
                resultChan <- &AgentResult{
                    AgentName: t.Agent.Name(),
                    Error:     fmt.Errorf("acquire concurrency slot: %w", err),
                }
                return
            }
            defer pw.concurrencyCtrl.Release()

            // Execute Agent
            result := pw.executeAgent(ctx, t)
            resultChan <- result
        }(task)
    }

    // 4. Wait for all tasks to complete or timeout
    go func() {
        wg.Wait()
        close(resultChan)
    }()

    // 5. Collect results
    results := make([]*AgentResult, 0, len(tasks))
    for result := range resultChan {
        results = append(results, result)
    }

    // 6. Aggregate results
    return pw.aggregator.Aggregate(ctx, results)
}

func (pw *ParallelWorkflow) executeAgent(ctx context.Context, task *Task) *AgentResult {
    start := time.Now()

    // Per-Agent timeout control
    agentCtx, cancel := context.WithTimeout(ctx, pw.resultTimeout)
    defer cancel()

    output, err := task.Agent.Run(agentCtx, task.Input)

    latency := time.Since(start)

    return &AgentResult{
        AgentName: task.Agent.Name(),
        Output:    output,
        Error:     err,
        Latency:   latency,
        Tokens:    estimateTokens(output),
        Timestamp: time.Now(),
    }
}

Result Aggregation Strategies

1. All-Wait Aggregation

Wait for all Agents to complete, regardless of success or failure:

type AllWaitAggregator struct {
    formatStrategy FormatStrategy
}

func (a *AllWaitAggregator) Aggregate(ctx context.Context, results []*AgentResult) (*WorkflowResult, error) {
    var output strings.Builder
    successCount := 0
    failureCount := 0
    totalTokens := 0
    maxLatency := time.Duration(0)

    partialResults := make(map[string]*AgentResult)

    for _, result := range results {
        partialResults[result.AgentName] = result

        if result.Error != nil {
            failureCount++
            output.WriteString(fmt.Sprintf("\n## %s (failed)\nError: %v\n",
                result.AgentName, result.Error))
        } else {
            successCount++
            output.WriteString(fmt.Sprintf("\n## %s\n%s\n",
                result.AgentName, result.Output))
            totalTokens += result.Tokens
        }

        if result.Latency > maxLatency {
            maxLatency = result.Latency
        }
    }

    // If there are failures and partial failure is not allowed, return an error
    if failureCount > 0 && !pw.partialFailure {
        return nil, fmt.Errorf("%d agents failed", failureCount)
    }

    return &WorkflowResult{
        Output:         output.String(),
        PartialResults: partialResults,
        SuccessCount:   successCount,
        FailureCount:   failureCount,
        TotalLatency:   maxLatency, // Parallel total latency = slowest Agent
        TokenUsage:     totalTokens,
    }, nil
}

2. Timeout-Truncate Aggregation

Return the results of Agents that completed within a specified time:

type TimeoutTruncateAggregator struct {
    waitTimeout time.Duration
}

func (a *TimeoutTruncateAggregator) Aggregate(ctx context.Context, results []*AgentResult) (*WorkflowResult, error) {
    ctx, cancel := context.WithTimeout(ctx, a.waitTimeout)
    defer cancel()

    collected := make([]*AgentResult, 0)
    resultChan := make(chan *AgentResult, len(results))

    // Send results into channel
    for _, r := range results {
        resultChan <- r
    }
    close(resultChan)

    // Collect as many as possible before timeout
    for {
        select {
        case result, ok := <-resultChan:
            if !ok {
                goto done
            }
            collected = append(collected, result)
        case <-ctx.Done():
            goto done
        }
    }

done:
    // Mark uncollected results as timed out
    collectedMap := make(map[string]bool)
    for _, r := range collected {
        collectedMap[r.AgentName] = true
    }

    for _, r := range results {
        if !collectedMap[r.AgentName] {
            collected = append(collected, &AgentResult{
                AgentName: r.AgentName,
                Error:     fmt.Errorf("aggregation timeout"),
            })
        }
    }

    return a.formatResult(collected)
}

3. Smart Aggregation

Select results based on quality:

type SmartAggregator struct {
    qualityThreshold float64
    selector         ResultSelector
}

func (a *SmartAggregator) Aggregate(ctx context.Context, results []*AgentResult) (*WorkflowResult, error) {
    // Filter successful results
    successful := make([]*AgentResult, 0)
    for _, r := range results {
        if r.Error == nil {
            successful = append(successful, r)
        }
    }

    if len(successful) == 0 {
        return nil, fmt.Errorf("all agents failed")
    }

    // Evaluate quality of each result
    scored := make([]*ScoredResult, len(successful))
    for i, r := range successful {
        score := a.evaluateQuality(r)
        scored[i] = &ScoredResult{
            Result: r,
            Score:  score,
        }
    }

    // Sort by quality
    sort.Slice(scored, func(i, j int) bool {
        return scored[i].Score > scored[j].Score
    })

    // Choose the highest-quality result or merge multiple high-quality results
    if scored[0].Score >= a.qualityThreshold {
        // Single result is good enough
        return &WorkflowResult{
            Output:       scored[0].Result.Output,
            SuccessCount: 1,
            TotalLatency: scored[0].Result.Latency,
        }, nil
    }

    // Merge top N results
    topResults := a.selectTopResults(scored, 3)
    merged := a.mergeResults(topResults)

    return &WorkflowResult{
        Output:       merged,
        SuccessCount: len(topResults),
    }, nil
}

func (a *SmartAggregator) evaluateQuality(result *AgentResult) float64 {
    // Multi-dimensional quality evaluation
    lengthScore := min(float64(len(result.Output))/1000.0, 1.0) // moderate length
    latencyScore := 1.0 / (1.0 + float64(result.Latency.Seconds())) // lower latency is better

    // Content quality (evaluated by LLM)
    contentScore := a.assessContentQuality(result.Output)

    // Weighted combination
    return 0.2*lengthScore + 0.3*latencyScore + 0.5*contentScore
}

Hands-On Scenario: Multi-Source Real-Time Analysis

Financial Data Analysis System

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"
)

// FinancialAnalysisSystem financial data analysis system
type FinancialAnalysisSystem struct {
    workflow *team.ParallelWorkflow
}

func NewFinancialAnalysisSystem(model agent.Model) (*FinancialAnalysisSystem, error) {
    // 1. Stock data Agent
    stockAgent, err := agent.New(agent.Config{
        Name:        "stock-analyst",
        Model:       model,
        Instruction: `You are a stock analysis expert. Analyze the technical indicators and trends of the given stock.
Output format:
- Current price and change
- Key technical indicators (MA, RSI, MACD)
- Short-term trend judgment (up/down/sideways)
- Risk warnings`,
        Tools: []tool.Tool{
            tool.NewStockPriceTool(),
            tool.NewTechnicalIndicatorTool(),
        },
        Timeout: 20 * time.Second,
    })
    if err != nil {
        return nil, err
    }

    // 2. News sentiment Agent
    newsAgent, err := agent.New(agent.Config{
        Name:        "news-analyst",
        Model:       model,
        Instruction: `You are a news sentiment analysis expert. Analyze market sentiment from recent related news.
Output format:
- News summary (latest 5)
- Sentiment score (-1 to +1)
- Key event impact analysis
- Sentiment trend changes`,
        Tools: []tool.Tool{
            tool.NewNewsSearchTool(),
            tool.NewSentimentAnalysisTool(),
        },
        Timeout: 25 * time.Second,
    })
    if err != nil {
        return nil, err
    }

    // 3. Macroeconomic Agent
    macroAgent, err := agent.New(agent.Config{
        Name:        "macro-analyst",
        Model:       model,
        Instruction: `You are a macroeconomic analyst. Analyze macro factors affecting the market.
Output format:
- Interest rate policy impact
- Inflation data interpretation
- Industry policy changes
- International market linkage`,
        Tools: []tool.Tool{
            tool.NewEconomicDataTool(),
            tool.NewPolicyTrackerTool(),
        },
        Timeout: 20 * time.Second,
    })
    if err != nil {
        return nil, err
    }

    // 4. Competitor analysis Agent
    competitorAgent, err := agent.New(agent.Config{
        Name:        "competitor-analyst",
        Model:       model,
        Instruction: `You are an industry competitive analyst. Analyze competitor dynamics in the same industry.
Output format:
- Recent moves of major competitors
- Market share changes
- Product/service comparison
- Competitive landscape assessment`,
        Tools: []tool.Tool{
            tool.NewCompetitorTrackerTool(),
        },
        Timeout: 20 * time.Second,
    })
    if err != nil {
        return nil, err
    }

    // Build the Parallel workflow
    workflow := team.NewParallelWorkflow(
        team.WithAgents(stockAgent, newsAgent, macroAgent, competitorAgent),
        team.WithConcurrencyLimit(4),              // max 4 concurrent
        team.WithTimeout(30 * time.Second),        // total 30s timeout
        team.WithResultTimeout(25 * time.Second),  // per-result 25s timeout
        team.WithPartialFailure(true),             // allow partial failure
        team.WithAggregator(&FinancialReportAggregator{}),
        team.WithDispatcher(&LoadAwareDispatcher{}),
    )

    return &FinancialAnalysisSystem{workflow: workflow}, nil
}

// FinancialReportAggregator financial report aggregator
type FinancialReportAggregator struct{}

func (a *FinancialReportAggregator) Aggregate(ctx context.Context, results []*AgentResult) (*WorkflowResult, error) {
    report := &FinancialReport{
        GeneratedAt: time.Now(),
        Sections:    make(map[string]string),
    }

    for _, result := range results {
        if result.Error != nil {
            report.Sections[result.AgentName] = fmt.Sprintf("[Data acquisition failed] %v", result.Error)
            continue
        }
        report.Sections[result.AgentName] = result.Output
    }

    // Generate comprehensive analysis
    synthesis := a.generateSynthesis(report.Sections)
    report.Synthesis = synthesis

    output := fmt.Sprintf(`# Financial Data Analysis Report
Generated at: %s

## Comprehensive Analysis
%s

## Detailed Data

### Stock Technical Analysis
%s

### News Sentiment Analysis
%s

### Macroeconomic Analysis
%s

### Competitor Analysis
%s

---
*Report generated by multi-Agent parallel analysis. Total time: %v*`,
        report.GeneratedAt.Format("2006-01-02 15:04:05"),
        report.Synthesis,
        report.Sections["stock-analyst"],
        report.Sections["news-analyst"],
        report.Sections["macro-analyst"],
        report.Sections["competitor-analyst"],
        calculateMaxLatency(results),
    )

    return &WorkflowResult{
        Output:       output,
        SuccessCount: countSuccesses(results),
        FailureCount: countFailures(results),
    }, nil
}

func (a *FinancialReportAggregator) generateSynthesis(sections map[string]string) string {
    // Generate comprehensive analysis based on each Agent's result
    // In practice, another LLM Agent can be called to summarize
    var synthesis strings.Builder
    synthesis.WriteString("Based on multi-dimensional analysis, the current market shows the following characteristics:\n\n")

    // Extract key signals
    signals := a.extractSignals(sections)
    for _, signal := range signals {
        synthesis.WriteString(fmt.Sprintf("- %s\n", signal))
    }

    return synthesis.String()
}

// LoadAwareDispatcher load-aware dispatcher
type LoadAwareDispatcher struct {
    loadBalancer *LoadBalancer
}

func (d *LoadAwareDispatcher) Dispatch(ctx context.Context, input string, agents []*agent.Agent) ([]*Task, error) {
    tasks := make([]*Task, len(agents))

    for i, agent := range agents {
        // Adjust input based on current Agent load
        load := d.loadBalancer.GetLoad(agent.Name())

        taskInput := input
        if load > 0.8 {
            // Under high load, simplify input to reduce processing
            taskInput = d.simplifyInput(input)
        }

        tasks[i] = &Task{
            Agent: agent,
            Input: fmt.Sprintf("Analysis target: %s\n\n%s", taskInput, agent.Instruction()),
        }
    }

    return tasks, nil
}

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

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

    result, err := system.workflow.Execute(ctx, "Kweichow Moutai (600519)")
    if err != nil {
        log.Fatalf("Analysis failed: %v", err)
    }

    fmt.Println(result.Output)
    fmt.Printf("\nSuccess rate: %d/%d\n", result.SuccessCount, result.SuccessCount+result.FailureCount)
}

Partial Failure Handling Strategies

Graceful Degradation Pattern

type DegradationStrategy struct {
    levels []DegradationLevel
}

type DegradationLevel struct {
    Name      string
    Condition func(*WorkflowResult) bool
    Action    func(*WorkflowResult) *WorkflowResult
}

// Three-level degradation strategy
var DefaultDegradationStrategy = &DegradationStrategy{
    levels: []DegradationLevel{
        {
            Name: "Level 1 - Data Completion",
            Condition: func(r *WorkflowResult) bool {
                return r.FailureCount > 0 && r.FailureCount <= len(r.PartialResults)/2
            },
            Action: func(r *WorkflowResult) *WorkflowResult {
                // Fill missing data with cache or defaults
                for name, result := range r.PartialResults {
                    if result.Error != nil {
                        cached := getCachedResult(name)
                        if cached != "" {
                            result.Output = cached
                            result.Error = nil
                            r.SuccessCount++
                            r.FailureCount--
                        }
                    }
                }
                return r
            },
        },
        {
            Name: "Level 2 - Simplified Output",
            Condition: func(r *WorkflowResult) bool {
                return r.SuccessCount > 0 && r.FailureCount > len(r.PartialResults)/2
            },
            Action: func(r *WorkflowResult) *WorkflowResult {
                // Keep only successful results and generate a simplified report
                var output strings.Builder
                output.WriteString("[Some services are unavailable; below is the available data]\n\n")
                for name, result := range r.PartialResults {
                    if result.Error == nil {
                        output.WriteString(fmt.Sprintf("## %s\n%s\n\n", name, result.Output))
                    }
                }
                r.Output = output.String()
                return r
            },
        },
        {
            Name: "Level 3 - Full Degradation",
            Condition: func(r *WorkflowResult) bool {
                return r.SuccessCount == 0
            },
            Action: func(r *WorkflowResult) *WorkflowResult {
                r.Output = "[System temporarily unavailable, please try again later]"
                return r
            },
        },
    },
}

Race Conditions and Data Safety

State Isolation

// Each Agent gets an isolated execution context
type IsolatedContext struct {
    AgentID    string
    Input      string
    State      map[string]interface{}
    CancelFunc context.CancelFunc
}

func (pw *ParallelWorkflow) executeWithIsolation(ctx context.Context, agent *agent.Agent, input string) (*AgentResult, error) {
    // Create isolated context
    isolatedCtx, cancel := context.WithCancel(ctx)
    defer cancel()

    // State isolation - deep copy to avoid sharing
    isolatedState := deepCopy(pw.sharedState)

    // Execute Agent
    result, err := agent.Run(isolatedCtx, input, agent.WithState(isolatedState))

    // Merge state (thread-safe)
    if err == nil {
        pw.mergeStateSafely(agent.Name(), isolatedState)
    }

    return result, err
}

func (pw *ParallelWorkflow) mergeStateSafely(agentName string, state map[string]interface{}) {
    pw.stateMu.Lock()
    defer pw.stateMu.Unlock()

    for key, value := range state {
        qualifiedKey := fmt.Sprintf("%s.%s", agentName, key)
        pw.sharedState[qualifiedKey] = value
    }
}

Performance Tuning Guide

1. Concurrency Tuning

// Calculate optimal concurrency based on CPU cores and model API rate limits
func calculateOptimalConcurrency() int {
    cpuCores := runtime.NumCPU()
    apiRateLimit := 100 // assume API rate limit 100 req/s
    agentCount := 4

    // Considering network I/O wait, concurrency can exceed CPU core count
    ioMultiplier := 3

    // Limited by API rate limit
    apiLimit := apiRateLimit / agentCount

    return min(cpuCores*ioMultiplier, apiLimit)
}

2. Connection Pool Optimization

type ModelConnectionPool struct {
    clients chan *ModelClient
    maxSize int
}

func NewConnectionPool(size int) *ModelConnectionPool {
    pool := &ModelConnectionPool{
        clients: make(chan *ModelClient, size),
        maxSize: size,
    }

    // Warm up connections
    for i := 0; i < size; i++ {
        client, err := createModelClient()
        if err != nil {
            log.Printf("Failed to create client %d: %v", i, err)
            continue
        }
        pool.clients <- client
    }

    return pool
}

func (p *ModelConnectionPool) Acquire(ctx context.Context) (*ModelClient, error) {
    select {
    case client := <-p.clients:
        return client, nil
    case <-ctx.Done():
        return nil, ctx.Err()
    }
}

func (p *ModelConnectionPool) Release(client *ModelClient) {
    select {
    case p.clients <- client:
    default:
        // Pool is full, close connection
        client.Close()
    }
}

Common Questions in Depth

Q: What should I do when one Agent in a Parallel workflow is slow?

A: Three layers of protection:

Per-Agent timeout: set resultTimeout to prevent a single Agent from dragging down the whole workflow.
Circuit breaker: trigger circuit breaking for consistently failing Agents to fail fast.
Asynchronous fallback: return a placeholder for timed-out Agents, continue execution in the background, and update via callback when complete.

Q: How is result ordering guaranteed?

A: Use an ordered aggregator:

type OrderedAggregator struct {
    agentOrder []string
}

func (a *OrderedAggregator) Aggregate(ctx context.Context, results []*AgentResult) (*WorkflowResult, error) {
    // Arrange results in predefined order
    ordered := make([]*AgentResult, len(a.agentOrder))
    resultMap := make(map[string]*AgentResult)

    for _, r := range results {
        resultMap[r.AgentName] = r
    }

    for i, name := range a.agentOrder {
        ordered[i] = resultMap[name]
    }

    return a.formatOrdered(ordered)
}

Q: How do I avoid the thundering herd problem?

A: Introduce jitter and token-bucket rate limiting:

func (pw *ParallelWorkflow) executeWithJitter(ctx context.Context, tasks []*Task) {
    for i, task := range tasks {
        // Add random jitter to avoid simultaneous triggers
        jitter := time.Duration(rand.Intn(1000)) * time.Millisecond

        go func(t *Task, delay time.Duration) {
            time.Sleep(delay)
            pw.executeAgent(ctx, t)
        }(task, time.Duration(i)*100*time.Millisecond+jitter)
    }
}

Next Steps

You now have a deep understanding of Parallel workflow concurrency control and result aggregation. Next, explore the Loop workflow—iterative optimization, termination conditions, and convergence guarantees for looped execution patterns.

← Sequential Workflow | Loop 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 is the core pattern of the Parallel workflow?

The core is the Fan-Out/Fan-In pattern: distribute input to multiple Agents for concurrent execution, then aggregate the results.

What should I do when one Agent in a Parallel workflow is slow?

Set a per-Agent timeout, enable circuit breaking for consistently failing Agents, and use asynchronous fallback with background continuation for timed-out tasks.

How is result ordering guaranteed?

Use an ordered aggregator to arrange results according to the predefined Agent order, avoiding disorder caused by concurrency.

How do I avoid the thundering herd problem?

Introduce random jitter and token-bucket rate limiting when launching concurrent tasks to avoid triggering a flood of requests at the same time.

What scenarios are suitable for the Parallel workflow?

It is suitable for multiple independent, parallelizable subtasks, such as querying multiple data sources simultaneously.

Fan-Out/Fan-In Architecture Model#

Core Principles#

Concurrency Model Selection#

Production-Grade Parallel Implementation#

Core Structures#

Execution Engine#

Result Aggregation Strategies#

1. All-Wait Aggregation#

2. Timeout-Truncate Aggregation#

3. Smart Aggregation#

Hands-On Scenario: Multi-Source Real-Time Analysis#

Financial Data Analysis System#

Partial Failure Handling Strategies#

Graceful Degradation Pattern#

Race Conditions and Data Safety#

State Isolation#

Performance Tuning Guide#

1. Concurrency Tuning#

2. Connection Pool Optimization#

Common Questions in Depth#

Q: What should I do when one Agent in a Parallel workflow is slow?#

Q: How is result ordering guaranteed?#

Q: How do I avoid the thundering herd problem?#

Next Steps#