Building a Go worker pool for concurrent LLM requests

A go worker pool concurrent llm requests pattern lets you saturate throughput without blowing up file descriptors or tripping provider rate limits. This tutorial builds a production-shaped dispatcher from scratch using goroutines and channels, then wires it to an OpenAI-compatible LLM endpoint.

Prerequisites

• Go 1.21+ installed (go version)
• An API key for an OpenAI-compatible endpoint. Set these env vars:

  export LLM_API_KEY="sk-..."
  export LLM_BASE_URL="https://api.openai.com/v1"

• Comfort with go run and basic channel semantics.

If you point the client at n4n.ai’s OpenAI-compatible endpoint, you get automatic fallback when a provider is rate-limited or degraded, which simplifies local retry logic. The code below stays provider-agnostic by reading LLM_BASE_URL.

Job and result types

Define the units flowing through the pool. Keep them explicit:

type Job struct {
    ID     int
    Prompt string
}

type Result struct {
    JobID  int
    Output string
    Err    error
}

A bounded jobs channel acts as the work queue; a bounded results channel collects completions.

A minimal OpenAI-compatible client

We avoid SDK bloat. A small http.Client wrapper handles chat completions:

type llmClient struct {
    apiKey  string
    baseURL string
    http    *http.Client
}

func newLLMClient(apiKey, baseURL string) *llmClient {
    return &llmClient{apiKey: apiKey, baseURL: baseURL, http: &http.Client{Timeout: 30 * time.Second}}
}

func (c *llmClient) Complete(ctx context.Context, prompt string) (string, error) {
    body := map[string]any{
        "model":    "gpt-4o-mini",
        "messages": []map[string]string{{"role": "user", "content": prompt}},
    }
    b, _ := json.Marshal(body)
    req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/chat/completions", bytes.NewReader(b))
    if err != nil {
        return "", err
    }
    req.Header.Set("Authorization", "Bearer "+c.apiKey)
    req.Header.Set("Content-Type", "application/json")

    resp, err := c.http.Do(req)
    if err != nil {
        return "", err
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        raw, _ := io.ReadAll(resp.Body)
        return "", fmt.Errorf("status %d: %s", resp.StatusCode, raw)
    }

    var out struct {
        Choices []struct {
            Message struct{ Content string `json:"content"` } `json:"message"`
        } `json:"choices"`
    }
    if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
        return "", err
    }
    if len(out.Choices) == 0 {
        return "", fmt.Errorf("no choices returned")
    }
    return out.Choices[0].Message.Content, nil
}

The worker pool dispatcher

Workers pull from jobs until it closes, then exit. The dispatcher seeds the queue and drains results.

func worker(id int, jobs <-chan Job, results chan<- Result, c *llmClient) {
    for job := range jobs {
        out, err := c.Complete(context.Background(), job.Prompt)
        results <- Result{JobID: job.ID, Output: out, Err: err}
    }
}

func main() {
    apiKey := os.Getenv("LLM_API_KEY")
    baseURL := os.Getenv("LLM_BASE_URL")
    if apiKey == "" || baseURL == "" {
        log.Fatal("LLM_API_KEY and LLM_BASE_URL required")
    }

    const numWorkers = 5
    jobs := make(chan Job, 10)
    results := make(chan Result, 10)
    client := newLLMClient(apiKey, baseURL)

    for i := 0; i < numWorkers; i++ {
        go worker(i, jobs, results, client)
    }

    prompts := []string{
        "Say hi in one word",
        "Name three Go keywords",
        "Explain channels briefly",
    }
    for i, p := range prompts {
        jobs <- Job{ID: i, Prompt: p}
    }
    close(jobs)

    for i := 0; i < len(prompts); i++ {
        r := <-results
        if r.Err != nil {
            fmt.Printf("job %d error: %v\n", r.JobID, r.Err)
            continue
        }
        fmt.Printf("job %d: %s\n", r.JobID, r.Output)
    }
}

Running it: expected output

go run main.go

Sample output (content varies by model):

job 0: Hi
job 1: func, chan, go
job 2: Channels synchronize goroutines via typed pipes.

If a provider returns a 429, you’ll see job N error: status 429: .... With a gateway that performs automatic fallback, that error rate drops sharply without code changes.

Why a bounded go worker pool concurrent llm requests matters

Spawning one goroutine per prompt looks innocent in a test script. In a service, it means unbounded concurrent TLS connections, memory per in-flight request, and a fast path to IP-level throttling. A go worker pool concurrent llm requests design caps parallelism at numWorkers and turns overflow into buffered queue backpressure. The jobs channel buffer (10 above) is your shock absorber; size it to your burst tolerance, not your total dataset.

Tune numWorkers from real limits: provider RPM, local CPU for JSON parsing, and ulimit -n. Start at 5–10 and measure.

Adding streaming with goroutines and channels

The same go worker pool concurrent llm requests core extends to token streaming. Instead of returning a string, the client yields a token channel. Workers fan tokens into a shared aggregator.

SSE client sketch

func (c *llmClient) Stream(ctx context.Context, prompt string) (<-chan string, <-chan error) {
    tokens := make(chan string)
    errc := make(chan error, 1)

    go func() {
        defer close(tokens)
        body := map[string]any{
            "model":    "gpt-4o-mini",
            "stream":   true,
            "messages": []map[string]string{{"role": "user", "content": prompt}},
        }
        b, _ := json.Marshal(body)
        req, _ := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/chat/completions", bytes.NewReader(b))
        req.Header.Set("Authorization", "Bearer "+c.apiKey)
        req.Header.Set("Content-Type", "application/json")

        resp, err := c.http.Do(req)
        if err != nil {
            errc <- err
            return
        }
        defer resp.Body.Close()

        sc := bufio.NewScanner(resp.Body)
        for sc.Scan() {
            line := sc.Text()
            if !strings.HasPrefix(line, "data: ") {
                continue
            }
            payload := strings.TrimPrefix(line, "data: ")
            if payload == "[DONE]" {
                return
            }
            var chunk struct {
                Choices []struct {
                    Delta struct{ Content string `json:"content"` } `json:"delta"`
                } `json:"choices"`
            }
            if json.Unmarshal([]byte(payload), &chunk) != nil {
                continue
            }
            if len(chunk.Choices) > 0 {
                if t := chunk.Choices[0].Delta.Content; t != "" {
                    tokens <- t
                }
            }
        }
        if err := sc.Err(); err != nil {
            errc <- err
        }
    }()
    return tokens, errc
}

Pool adjustments for streaming

Workers no longer return a single Result. They forward tokens to a global tokenSink chan string and signal completion per job:

func streamWorker(id int, jobs <-chan Job, done chan<- int, sink chan<- string, c *llmClient) {
    for job := range jobs {
        tokens, errc := c.Stream(context.Background(), job.Prompt)
        for t := range tokens {
            sink <- t
        }
        if err := <-errc; err != nil {
            sink <- fmt.Sprintf("[job %d error: %v]", job.ID, err)
        }
        done <- job.ID
    }
}

The main goroutine selects on sink for live output and done to track completion. This preserves bounded concurrency while streaming partial results—exactly what a responsive LLM UI needs.

Closing the loop

You now have a typed job queue, fixed-parallelish workers, and a clean seam for either batch completions or streaming. Replace the mock prompts with a database cursor or Kafka consumer, and the pool stays identical. The moment you need per-token metering or provider cache hints, those belong in the llmClient headers, not the pool logic.