Last updated 2026-05-17

The Streaming Model

Every call into llmrouter returns a *Stream. A Stream is a small handle around a buffered channel of chunks plus a context cancellation function. It deliberately looks and behaves like an idiomatic Go channel pipeline — read until closed, then check the terminal error.

Lifecycle

When you call CompletionStream, the provider does three things synchronously before returning to you:

Build the upstream HTTP request body.
Send the HTTP request and read response headers.
If the status is >= 400, read up to 1KB (OpenAI) or 8KB (Anthropic) of the body and return an *ErrUpstream. Otherwise, allocate the Stream, spawn a producer goroutine, and return.

After that point the producer goroutine owns the HTTP response body. It reads SSE frames, decodes each one into a llmrouter.Chunk, and pushes it into a buffered channel of size 16. Your code reads from that channel via Chunks(). When the producer finishes — whether on success, upstream error, network error, or context cancellation — it closes the channel and stores the terminal error.

The consumer side is a one-liner pattern:

stream, err := p.CompletionStream(ctx, req)
if err != nil {
    return err // could not even open the stream
}
for chunk := range stream.Chunks() {
    // ... use chunk ...
}
if err := stream.Err(); err != nil {
    return err // stream opened but terminated abnormally
}

The Stream type itself is small:

type Stream struct {
    chunks chan Chunk
    cancel context.CancelFunc
    errMu  chan struct{} // closed when err is final
    err    error
}

func (s *Stream) Chunks() <-chan Chunk { return s.chunks }
func (s *Stream) Err() error           { <-s.errMu; return s.err }
func (s *Stream) Cancel()              { /* ... */ }

Note. Err() blocks until the producer finishes. That's intentional — it lets you write the idiomatic "range, then check error" pattern without races. Don't call Err() before Chunks() drains; you'll deadlock waiting for a goroutine you're starving.

The single-consumer invariant

Only one goroutine should read from Chunks(). The library does not synchronize multiple readers. If you fan out to N consumers and they all read from the same channel, you'll get one chunk per reader, scrambled in arrival order, and you'll think the library is broken.

If you need fan-out, do it on top:

package main

import (
    "context"
    "sync"

    "github.com/elloloop/llmrouter"
)

// fanOut copies chunks from one stream to N output channels. Closes all
// outputs when the input is drained. Each output channel is single-
// consumer; the fan-out goroutine is the single consumer of the source.
func fanOut(ctx context.Context, in *llmrouter.Stream, n int) []<-chan llmrouter.Chunk {
    outs := make([]chan llmrouter.Chunk, n)
    readOnly := make([]<-chan llmrouter.Chunk, n)
    for i := range outs {
        outs[i] = make(chan llmrouter.Chunk, 16)
        readOnly[i] = outs[i]
    }
    go func() {
        defer func() {
            for _, c := range outs {
                close(c)
            }
        }()
        for chunk := range in.Chunks() {
            var wg sync.WaitGroup
            for _, c := range outs {
                wg.Add(1)
                go func(ch chan llmrouter.Chunk) {
                    defer wg.Done()
                    select {
                    case ch <- chunk:
                    case <-ctx.Done():
                    }
                }(c)
            }
            wg.Wait()
        }
    }()
    return readOnly
}

Backpressure

The internal chunk channel is buffered at 16. While the buffer has space, the producer pushes chunks as fast as it can decode them off the wire. When the buffer is full, the producer blocks inside sendChunk:

// sendChunk delivers one chunk to the consumer; respects ctx so a
// disconnected consumer doesn't pin the producer.
func (s *Stream) sendChunk(ctx context.Context, c Chunk) bool {
    select {
    case s.chunks <- c:
        return true
    case <-ctx.Done():
        return false
    }
}

Two things to notice. First, this is real backpressure all the way up the pipe — if your consumer falls behind, the producer stops reading from the HTTP connection, which (at typical SSE buffer sizes) eventually stops the upstream from sending. Second, the producer races the send against ctx.Done(), so a consumer that disconnects mid-stream can't pin a goroutine on a full buffer waiting for someone who'll never read.

Cancellation

There are two ways to terminate a stream early:

Cancel the parent context. Call the cancel function returned by context.WithCancel or context.WithTimeout around the request context.
Call stream.Cancel(). Internally this cancels a child context derived from the request context. Equivalent to cancelling the parent for this one stream; it does not affect siblings.

What happens after cancellation:

The producer goroutine's next sendChunk picks ctx.Done() instead of the channel send. It returns false.
The producer calls hooks.Finish(ctx.Err()), which stores the error and closes the chunks channel.
The consumer's range loop exits. stream.Err() returns context.Canceled (or context.DeadlineExceeded if a deadline fired).
The underlying HTTP connection's Read returns because http.NewRequestWithContext wired the same context into the transport — no orphan TCP socket, no orphan tokens billed.

See Context & Cancellation for a fuller treatment.

The `Chunk` type

type Chunk struct {
    ID      string    `json:"id"`
    Object  string    `json:"object"`
    Created int64     `json:"created"`
    Model   string    `json:"model"`
    Choices []Choice  `json:"choices"`
    Usage   *Usage    `json:"usage,omitempty"`

    // Raw is the original wire-format JSON for this chunk.
    Raw json.RawMessage `json:"-"`
}

ID, Object, Created, Model — OpenAI Chat Completion chunk fields. Same on every chunk in a stream. Anthropic provides these as a synthesized chat completion id (chatcmpl-<uuid>) plus the model echoed from message_start.
Choices — typically a single-element slice in streaming. The interesting field is Choices[0].Delta.
Usage — token counts. Populated only on the final chunk. On OpenAI this requires stream_options.include_usage (the OpenAI provider forces this for you). On Anthropic the provider accumulates tokens from message_start and message_delta and emits them on the final chunk.
Raw — original JSON bytes. See Byte Passthrough.

The `Delta` type

type Delta struct {
    Role    string `json:"role,omitempty"`
    Content string `json:"content,omitempty"`
}

Role — set once, on the first chunk of a response (always "assistant" for chat completions). Subsequent chunks leave it empty. Don't concatenate Role across chunks; concatenate Content.
Content — the incremental token(s) emitted by the model. Concatenating all Content strings across all chunks gives you the full response text.

Finish reason normalization

OpenAI uses one vocabulary for finish_reason; Anthropic uses another (stop_reason). The Anthropic provider normalizes to the OpenAI set on the way out:

func mapStopReason(r string) string {
    switch r {
    case "end_turn", "stop_sequence":
        return "stop"
    case "max_tokens":
        return "length"
    case "tool_use":
        return "tool_calls"
    default:
        return "stop"
    }
}

So your consumer code can rely on a single set of values:

"stop" — natural end of response.
"length" — hit the max_tokens ceiling.
"tool_calls" — model emitted a tool/function call.
"content_filter" — moderation filter tripped (OpenAI only).

FinishReason is only set on the final chunk that contains it. Most chunks have it as the empty string.

Full example: token budget with early termination

The pattern below is how you'd implement a per-request budget cap. It counts emitted characters as a coarse stand-in for tokens, calls stream.Cancel() when the cap is hit, and returns a structured error to the caller:

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "strings"
    "time"

    "github.com/elloloop/llmrouter"
    "github.com/elloloop/llmrouter/providers/openai"
)

// ErrBudgetExceeded is returned when a stream is cut short by the
// caller because it would have used more than budgetChars characters.
var ErrBudgetExceeded = errors.New("budget exceeded")

// budgetedChat streams a completion and cancels early if the response
// would exceed budgetChars characters. Returns whatever text was
// produced before the cut-off, plus ErrBudgetExceeded if we cut early.
func budgetedChat(
    ctx context.Context,
    p llmrouter.Provider,
    req llmrouter.ChatRequest,
    budgetChars int,
) (string, error) {
    stream, err := p.CompletionStream(ctx, req)
    if err != nil {
        return "", err
    }

    var (
        out      strings.Builder
        consumed int
        capped   bool
    )
    for chunk := range stream.Chunks() {
        for _, c := range chunk.Choices {
            out.WriteString(c.Delta.Content)
            consumed += len(c.Delta.Content)
        }
        if consumed >= budgetChars && !capped {
            capped = true
            stream.Cancel()
            // Don't break — keep draining so the goroutine exits
            // cleanly and stream.Err() unblocks promptly.
        }
    }

    if streamErr := stream.Err(); streamErr != nil {
        if capped && errors.Is(streamErr, context.Canceled) {
            return out.String(), ErrBudgetExceeded
        }
        return out.String(), streamErr
    }
    return out.String(), nil
}

func main() {
    p, err := openai.New(llmrouter.WithAPIKey(os.Getenv("OPENAI_API_KEY")))
    if err != nil {
        log.Fatal(err)
    }

    ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    defer cancel()

    req := llmrouter.ChatRequest{
        Model: "gpt-4o-mini",
        Messages: []llmrouter.Message{
            llmrouter.TextMessage("user", "write a 500-word essay on goroutines"),
        },
    }

    text, err := budgetedChat(ctx, p, req, 200)
    fmt.Printf("---\n%s\n---\n", text)
    switch {
    case errors.Is(err, ErrBudgetExceeded):
        fmt.Println("(stopped at budget)")
    case err != nil:
        fmt.Println("error:", err)
    default:
        fmt.Println("(completed naturally)")
    }
}

Why drain after Cancel? If you break out of the range loop on cancel, the producer goroutine may still be blocked on a send into the buffered channel. Letting the loop keep draining until the channel is naturally closed guarantees the goroutine exits promptly and stream.Err() doesn't block waiting for it.

Next steps

Byte Passthrough — how Chunk.Raw lets gateways forward original SSE bytes verbatim.
Context & Cancellation — deeper coverage of cancellation, deadlines, and HTTP-handler patterns.
Error Handling — distinguishing config, upstream, network, and stream errors.

The Streaming Model

Lifecycle

The single-consumer invariant

Backpressure

Cancellation

The Chunk type

The Delta type

Finish reason normalization

Full example: token budget with early termination

Next steps

The `Chunk` type

The `Delta` type