Part of the series: Production-Grade Concurrent AI Systems in Go
→ Full code for this post: github.com/madmmas/go-concurrent-ai-systems/tree/part-11 → Diff from Part 10: compare/part-10...part-11 → Run it: go run ./cmd/news-processor -articles=6 inside arc-2-production/part-11-pipeline-stages
Part 10 gave each article three concurrent tasks inside one worker. The per-article time dropped from 1,500ms to roughly 700ms — the slowest of the three tasks rather than their sum.
But all workers are still configured identically, and every article still passes through a monolithic processArticle function that handles everything: fetch the content, clean it, generate embeddings, write a summary. That function has very different performance characteristics at each step.
Scraping a URL is IO-bound. You can fire 20 concurrent requests without breaking a sweat. Generating an embedding or summary is an LLM call — if your provider caps you at 5 requests per second, you might only want 3 workers on that step so you have headroom for retries. Running 20 embedding workers against a rate-limited API means you will spend more time getting 429 errors than doing useful work.
A single Workers configuration cannot express this. Part 11 introduces pipeline stages — each stage is its own concurrent pool, connected to the next by a channel.
The Architecture
Instead of one worker function per article, the pipeline is a chain of stages:
Articles
↓
[Scrape stage] workers=10 — IO-bound, cheap to parallelise
↓
[Clean stage] workers=5 — CPU-bound, normalise text
↓
[Embed stage] workers=3 — LLM call, rate-limited
↓
[Summarise stage] workers=3 — LLM call, rate-limited
↓
Results
Each stage reads from the channel produced by the previous stage and writes to a channel consumed by the next. All four stages run concurrently with each other — while scrape workers are fetching articles 3 and 4, embed workers are generating embeddings for articles 1 and 2, and summarise workers are writing summaries for article 0. The pipeline is full at all times, and no stage waits for another to finish its entire batch.
This is the same principle behind Unix pipes, Kafka consumer groups, and every production data pipeline you have ever used. Go channels make it idiomatic.
Configuring Stages
Each stage gets a name and a worker count:
type StageConfig struct {
Name string
Workers int
}
The pipeline constructor takes a slice of these, so the caller controls the concurrency at each stage:
func New(llm *simulator.LLMClient, timeout time.Duration, stages []StageConfig) *Pipeline
// DefaultStages returns production-realistic ratios
func DefaultStages() []StageConfig {
return []StageConfig{
{Name: "scrape", Workers: 10},
{Name: "clean", Workers: 5},
{Name: "embed", Workers: 3},
{Name: "summarise", Workers: 3},
}
}
The defaults reflect a real pattern: scrape gets the most workers because HTTP requests are cheap and you want to keep the downstream stages fed. LLM stages get fewer workers because they are expensive and rate-limited. Clean sits in between — it is CPU work, so it scales with available cores rather than network capacity.
These numbers are starting points, not facts. In production you tune them by measuring queue depth at each stage under load. If the embed channel is always full, your scrape workers are producing faster than embed can consume — either add embed workers, reduce scrape workers, or add a rate limiter (Part 14).
Chaining Stages with Channels
ProcessAll seeds the first channel, then chains each stage's output into the next stage's input:
func (p *Pipeline) ProcessAll(ctx context.Context, articles []model.Article) ([]model.AIResult, time.Duration) {
start := time.Now()
// Seed: close immediately so scrape workers know when to stop
scrapeIn := make(chan model.Article, len(articles))
for _, a := range articles {
scrapeIn <- a
}
close(scrapeIn)
// Chain: each call returns a channel, the next call reads from it
scrapeOut := p.runStage(ctx, "scrape", scrapeIn, p.scrape)
cleanOut := p.runStageResult(ctx, "clean", scrapeOut, p.clean)
embedOut := p.runStageResult(ctx, "embed", cleanOut, p.embed)
summaryOut := p.runStageResult(ctx, "summarise", embedOut, p.summarise)
var results []model.AIResult
for r := range summaryOut {
results = append(results, r)
}
return results, time.Since(start)
}
The chain builds up before any work starts — each runStage call spawns its goroutines and returns immediately with a channel. The goroutines are running, but they are all blocked waiting for input. The moment scrapeIn has data and gets closed, the cascade begins: scrape workers start sending to scrapeOut, clean workers start receiving from scrapeOut and sending to cleanOut, and so on down the chain.
How Each Stage Works
Every stage follows the same three-step pattern:
func (p *Pipeline) runStageResult(
ctx context.Context,
name string,
in <-chan model.AIResult,
fn func(context.Context, model.AIResult) (model.AIResult, error),
) <-chan model.AIResult {
out := make(chan model.AIResult, cap(in)+1)
cfg := p.stageConfig(name)
var wg sync.WaitGroup
for w := 0; w < cfg.Workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for result := range in {
if result.Err != nil {
out <- result // pass failures through unchanged
continue
}
articleCtx, cancel := context.WithTimeout(ctx, p.timeout)
updated, err := fn(articleCtx, result)
cancel()
if err != nil {
result.Err = err
out <- result
} else {
out <- updated
}
}
}()
}
go func() {
wg.Wait()
close(out)
}()
return out
}
Step 1: Spawn cfg.Workers goroutines, each ranging over the input channel.
Step 2: When the input channel closes and drains, all goroutines exit their range loops and call wg.Done().
Step 3: The closer goroutine calls close(out) after all workers exit — signalling the next stage that no more data is coming.
This close cascade is what makes the whole chain self-terminating. When scrapeIn closes, scrape workers finish draining and close scrapeOut. That causes clean workers to drain and close cleanOut, and so on until summaryOut closes and the final range summaryOut exits.
Error Propagation
Notice the if result.Err != nil check at the top of each worker's loop:
if result.Err != nil {
out <- result // pass through, skip this stage's work
continue
}
A failure in one stage does not stop the pipeline — it short-circuits the remaining stages for that article. The failed result travels through clean, embed, and summarise unchanged, arriving at the final collector with its error intact.
This keeps the result count equal to the article count regardless of failures. The collector always gets exactly len(articles) results. Some will have Err set; most will not.
The alternative — dropping failed results — would mean the caller cannot tell how many articles actually failed. Silent drops in a pipeline that processes thousands of articles per minute are how you end up with gaps you cannot explain.
What the Output Looks Like
Multi-stage pipeline: 4 articles
Stages: scrape(4) → clean(2) → embed(2) → summarise(2)
─────────────────────────────────────────────────────────
[scrape] article 1
[1] Scrape started (224ms)
[scrape] article 2
[2] Scrape started (425ms)
[scrape] article 3
[3] Scrape started (453ms)
[scrape] article 4
[4] Scrape started (434ms)
[1] Scrape completed
[clean] article 1
[embed] article 1
[1] Embed started (592ms)
[2] Scrape completed
[clean] article 2
[embed] article 2
[2] Embed started (629ms)
[1] Embed completed
[summarise] article 1
[1] Summarise started (264ms)
...
Duration : 2.337s
The key thing to see: article 1 reaches the embed stage while articles 2, 3, and 4 are still being scraped. The stages overlap. While article 1's summarise worker is writing a summary, article 3's scrape worker is still fetching content. This is the pipeline full and running at all four stages simultaneously.
Tuning Stage Workers in Practice
The right worker count for each stage is a measurement problem, not a calculation problem. That said, a few heuristics help when starting from zero.
IO-bound stages (scrape, any HTTP fetch) can run many workers. The goroutines spend almost all their time waiting for network responses, so CPU is never the constraint. Start high — 10 to 20 — and reduce if you hit connection pool limits or downstream rate limits.
CPU-bound stages (clean, parse, chunk) are bounded by available cores. runtime.NumCPU() is a reasonable starting ceiling. Running more workers than cores adds scheduling overhead without adding throughput.
LLM stages (embed, summarise) are bounded by your provider's rate limits and your retry budget. If your provider allows 10 requests per second and each call takes about 1 second, 8 workers keeps you under the limit while leaving headroom for the retry spikes that Part 13 introduces. Match worker count to sustainable throughput, not theoretical maximum.
The metric to watch is queue depth at each stage boundary — the number of items sitting in the channel between stages. A consistently full queue means the next stage cannot keep up. A consistently empty queue means the previous stage is the bottleneck. Your goal is stages running roughly at parity: no queue builds up, no stage idles waiting for work.
What's Next
The pipeline stages are now concurrent and independently tunable. But they share one important assumption: every call succeeds eventually. The simulator in Parts 10 and 11 uses DefaultProfile — zero failures.
Part 12 introduces errgroup, a standard pattern for running concurrent tasks where the first failure should cancel all siblings. Then Part 13 adds retries: when a stage call fails with a rate limit or server error, the article is not dropped — it is retried with exponential backoff before being passed to the dead letter queue.
Together, Parts 12 and 13 turn the pipeline from something that works under ideal conditions into something that handles the failure modes that are routine in production AI systems.
See you in Part 12.
This is Part 11 of the series "Production-Grade Concurrent AI Systems in Go." Read Part 10 — Fan-Out / Fan-In or continue to Part 12 — errgroup: When One Task Fails, Cancel the Rest.