Streaming tokens from more than one LLM at once is a practical way to hedge against latency spikes or compare outputs. The go select statement merge llm streams pattern lets you fan-in multiple channel sources into a single consumer without blocking on any one response. This post walks through a runnable implementation using goroutines, channels, and context cancellation that you can drop into a real gateway client.
Step 1: Define a common token type and mock sources
Before merging anything, agree on a single shape for a streamed token. A struct that carries the originating model and the text fragment keeps downstream code agnostic to which provider emitted it.
package main
import (
"context"
"fmt"
"math/rand"
"time"
)
type Token struct {
Model string
Text string
}
// mockStream simulates an LLM SSE feed by emitting tokens on a channel.
func mockStream(ctx context.Context, model string, tokens []string) <-chan Token {
out := make(chan Token)
go func() {
defer close(out)
for _, t := range tokens {
select {
case <-ctx.Done():
return
case out <- Token{Model: model, Text: t}:
time.Sleep(time.Duration(rand.Intn(30)+10) * time.Millisecond)
}
}
}()
return out
}The generator respects ctx: if the context is cancelled, the goroutine exits instead of leaking. The defer close(out) guarantees the consumer sees the channel close when the stream ends. In production you would replace the sleep loop with an http.Response.Body reader parsing SSE lines, but the channel contract stays identical.
Step 2: Merge two streams with a select statement
The simplest merge handles two channels. Go’s select blocks until one of its cases can proceed, so it naturally interleaves whichever provider responds first.
func mergeTwo(ctx context.Context, a, b <-chan Token) <-chan Token {
out := make(chan Token)
go func() {
defer close(out)
for a != nil || b != nil {
select {
case <-ctx.Done():
return
case t, ok := <-a:
if !ok {
a = nil // disable this case
continue
}
out <- t
case t, ok := <-b:
if !ok {
b = nil
continue
}
out <- t
}
}
}()
return out
}Setting a channel to nil removes it from the select rotation—a nil channel never becomes ready. That is the trick that lets a fixed select handle streams that close at different times. Without it, reading from a closed channel returns immediately with ok=false and would spin the loop.
Why not just spawn a goroutine per source?
You could launch one forwarder goroutine per input channel that all send to a shared out. That works, but it gives you no central place to prioritize or drop stale tokens. The go select statement merge llm streams approach keeps fan-in in one goroutine, making it easier to add timeout logic or per-model weighting later.
Step 3: Generalize to N streams with reflect.Select
Real systems rarely know the stream count at compile time. reflect.Select builds the same multiplexer dynamically.
func mergeN(ctx context.Context, channels []<-chan Token) <-chan Token {
out := make(chan Token)
go func() {
defer close(out)
cases := make([]reflect.SelectCase, len(channels)+1)
cases[0] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ctx.Done())}
for i, c := range channels {
cases[i+1] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(c)}
}
alive := len(channels)
for alive > 0 {
chosen, recv, ok := reflect.Select(cases)
if chosen == 0 {
return // context cancelled
}
if !ok {
cases[chosen] = reflect.SelectCase{Dir: reflect.SelectDefault} // disable
alive--
continue
}
out <- recv.Interface().(Token)
}
}()
return out
}When a source channel closes, we swap its case to SelectDefault, which is never chosen, effectively removing it. The loop terminates when alive hits zero or the context fires. This preserves the same semantics as mergeTwo but scales to any fan-in width.
If you are pulling from a gateway such as n4n.ai, which exposes one OpenAI-compatible endpoint addressing 240+ models, each model request returns its own SSE body; wrap each http.Response in a function like mockStream and pass the resulting channels to mergeN.
Step 4: Adapt real HTTP streams to channels
A minimal SSE adapter for an OpenAI-compatible endpoint looks like this:
func streamFromHTTP(ctx context.Context, model, url string) <-chan Token {
out := make(chan Token)
go func() {
defer close(out)
req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBufferString(`{"model":"`+model+`","stream":true}`))
resp, err := http.DefaultClient.Do(req)
if err != nil {
return
}
defer resp.Body.Close()
sc := bufio.NewScanner(resp.Body)
for sc.Scan() {
line := sc.Text()
if !strings.HasPrefix(line, "data: ") {
continue
}
// parse JSON, extract delta, send Token
select {
case <-ctx.Done():
return
case out <- Token{Model: model, Text: extractDelta(line)}:
}
}
}()
return out
}The select inside the scanner loop ensures cancellation propagates even while waiting to send. Use a buffered out (e.g., make(chan Token, 16)) if the consumer is slower than the network, otherwise the goroutine blocks on send until the consumer catches up—which is usually fine for token printing but matters for downstream aggregation.
Step 5: Consume the merged stream and verify
Wire it together in main:
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
s1 := mockStream(ctx, "model-a", []string{"The", " quick", " brown"})
s2 := mockStream(ctx, "model-b", []string{"fox", " jumps", " over"})
merged := mergeTwo(ctx, s1, s2)
for t := range merged {
fmt.Printf("%s: %s\n", t.Model, t.Text)
}
fmt.Println("merge complete")
}Run with go run main.go. You should see lines from both models interleaved in non-deterministic order, then merge complete printed exactly once. The process exits cleanly with no goroutine leak because every source channel is closed and the merge goroutine returns when both are nil (or when the 2-second context fires).
Verification checklist
- No
panic: send on closed channel—confirmdefer close(out)is only in the producer. - Output contains tokens from every input stream.
- Program terminates without
go exitwarnings; if you addruntime.NumGoroutine()before and after, the count returns to baseline. - Cancel the context early (short timeout) and confirm in-flight streams stop emitting.
Backpressure and buffering notes
An unbuffered merged channel couples producer and consumer speed. For a CLI that prints, that is fine. For a proxy that forwards to a browser over WebSocket, buffer the merged channel or run a separate pump goroutine so a slow network write does not stall all LLM connections. Remember that reflect.Select does not prioritize cases; if you need fairness, shuffle the case slice periodically or use a round-robin counter.
The go select statement merge llm streams pattern is deliberately small. It gives you one cancellation point, predictable closure semantics, and a single goroutine owning fan-in. Extend it with per-model weighting by sending to out only after a token passes a priority check, or duplicate the merge step to build a tree for hundreds of streams.