Wiring go channels llm token streaming into your backend lets you decouple token generation from consumption without blocking goroutines or leaking connections. This tutorial builds a minimal but production-shaped streaming client in Go that opens a Server-Sent Events (SSE) connection to an OpenAI-compatible endpoint, parses completion chunks, and pushes each token delta into a typed channel.
Prerequisites
- Go 1.21 or newer (
go version). - An API key for an OpenAI-compatible inference gateway. The single endpoint from n4n.ai fronts 240+ models and degrades gracefully on provider errors, but any compliant server works.
curlfor a one-shot sanity check before writing code.- Comfort with
go run, goroutines, andchansemantics.
Set the key in your shell:
export API_KEY="sk-..."
export BASE_URL="https://api.n4n.ai/v1/chat/completions"Step 1: Define the wire types
OpenAI-compatible streaming uses a JSON request with stream: true and returns newline-delimited data: frames. We only need a sliver of the response schema.
package main
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"strings"
"bufio"
)
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
}
type ChatRequest struct {
Model string `json:"model"`
Messages []Message `json:"messages"`
Stream bool `json:"stream"`
}
type StreamChunk struct {
Choices []struct {
Delta struct {
Content string `json:"content"`
} `json:"delta"`
FinishReason *string `json:"finish_reason"`
} `json:"choices"`
}These structs are intentionally thin. You can extend StreamChunk later to capture usage or system fingerprints.
Step 2: Open the HTTP request with streaming
The key detail: do not use ioutil.ReadAll. Stream the body. We build the request with a context so callers can cancel mid-stream.
func StreamTokens(ctx context.Context, baseURL, apiKey, model, prompt string) (<-chan string, <-chan error) {
tokenCh := make(chan string)
errCh := make(chan error, 1)
go func() {
defer close(tokenCh)
reqBody := ChatRequest{
Model: model,
Messages: []Message{{Role: "user", Content: prompt}},
Stream: true,
}
payload, err := json.Marshal(reqBody)
if err != nil {
errCh <- err
return
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL, bytes.NewReader(payload))
if err != nil {
errCh <- err
return
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+apiKey)
resp, err := http.DefaultClient.Do(req)
if err != nil {
errCh <- err
return
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
errCh <- fmt.Errorf("status %d: %s", resp.StatusCode, body)
return
}
// parsing continues below
scanAndForward(ctx, resp.Body, tokenCh, errCh)
}()
return tokenCh, errCh
}The function returns a receive-only token channel and a buffered error channel. Buffering errCh by 1 prevents the goroutine from blocking if the consumer has already exited.
Step 3: Parse SSE and push to the channel
SSE frames look like data: {json}\n\n. The terminal frame is data: [DONE]. We scan line by line and extract the content delta.
func scanAndForward(ctx context.Context, r io.Reader, tokenCh chan<- string, errCh chan<- error) {
scanner := bufio.NewScanner(r)
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data: ") {
continue
}
data := strings.TrimPrefix(line, "data: ")
if data == "[DONE]" {
break
}
var chunk StreamChunk
if err := json.Unmarshal([]byte(data), &chunk); err != nil {
continue // skip malformed frame, don't kill stream
}
if len(chunk.Choices) == 0 {
continue
}
content := chunk.Choices[0].Delta.Content
if content == "" {
continue
}
select {
case tokenCh <- content:
case <-ctx.Done():
return
}
}
if err := scanner.Err(); err != nil {
errCh <- err
}
}The select with ctx.Done() is what makes go channels llm token streaming cancellable: if the consumer disappears, the producer unwinds instead of hanging on a full channel.
Step 4: Consume the channel in main
A minimal driver prints tokens as they arrive and exits cleanly when the channel closes.
func main() {
ctx := context.Background()
baseURL := os.Getenv("BASE_URL")
apiKey := os.Getenv("API_KEY")
tokens, errs := StreamTokens(ctx, baseURL, apiKey, "gpt-3.5-turbo", "Write a haiku about goroutines")
for {
select {
case tok, ok := <-tokens:
if !ok {
fmt.Println("\n[stream closed]")
return
}
fmt.Print(tok)
case err := <-errs:
fmt.Println("\n[error]", err)
return
}
}
}Run it:
go run main.goExpected partial output (your model will vary):
Silent workers queue
Channels pass the whispered task
Concurrency flows
[stream closed]The token boundaries are not word-aligned; you will see substrings like Silent, workers, queue. That is expected with raw delta streaming.
Step 5: Fan-out and context cancellation
The real win for go channels llm token streaming is fan-out: one producer, many consumers (a logger, a websocket writer, a rate limiter). Here is a version that broadcasts to two consumers and respects SIGINT.
func main() {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
defer stop()
baseURL := os.Getenv("BASE_URL")
apiKey := os.Getenv("API_KEY")
tokens, errs := StreamTokens(ctx, baseURL, apiKey, "gpt-4o-mini", "Explain channels like I'm five")
// consumer 1: stdout
go func() {
for t := range tokens {
fmt.Print(t)
}
}()
// consumer 2: naive logger
go func() {
for t := range tokens {
_ = t // would write to disk or metrics
}
}()
select {
case err := <-errs:
fmt.Println("\n[error]", err)
case <-ctx.Done():
fmt.Println("\n[cancelled by signal]")
}
}This code has a bug: you cannot range over the same channel from two goroutines and expect both to see every token—channels are point-to-point. For true broadcast, use a fanOut helper that multiplexes to N channels, or adopt golang.org/x/sync/errgroup with a sync.Map of subscribers. The pattern stays the same: producer writes once, dispatcher copies to each subscriber channel inside a select per subscriber.
Step 6: Production hardening notes
A few things the tutorial glosses over that you will hit in real systems:
- Timeouts:
http.DefaultClienthas no timeout. SetTransportandClient.Timeout, or rely oncontextwithhttp.NewRequestWithContextas shown. - Backpressure: If a consumer is slow, the producer blocks on
tokenCh <- content. For a websocket frontend, buffer the channel (e.g.,make(chan string, 256)) and drop frames on overflow if freshness matters more than completeness. - Provider fallback: Gateways that honor client routing directives will retry across providers when one is rate-limited. Your code should treat a non-200 as retryable only if the gateway does not already do so.
- Token accounting: If you meter per-token usage, sum
len(content)or parse the finalusageframe if your gateway sends it. n4n.ai returns per-token metering on the billing side regardless of stream fragmentation.
Expected full output with cancellation
With the Step 5 code and a Ctrl-C after two seconds:
A channel is like a tube. You put blocks in one end, and a friend pulls them out the other. If the tube is empty, your friend waits. If it's full, you wait. That way you both share work without yelling!
[cancelled by signal]If you let it finish, the [cancelled by signal] line is replaced by [stream closed] from the producer’s deferred close.
Takeaways
The go channels llm token streaming pattern is three moving parts: an SSE scanner in a goroutine, a typed channel as the boundary, and a select on ctx.Done() for clean shutdown. Keep the producer dumb, push raw deltas, and let downstream consumers decide on buffering, broadcasting, or persistence. Once that boundary is clean, swapping the backend model or provider is a one-line config change, not a refactor.