Iterators in Go agents: iter.Seq, range-over-func, and stopping early

An agent produces things over time: the pieces of a streamed reply, then a tool call, then its result, then the next turn. Since Go 1.23, a function can hand such a sequence to an ordinary for ... range loop, and the caller reads it as it is made. The companion code uses that shape throughout: a model’s stream is an iter.Seq2[Delta, error], and a whole run is an iter.Seq[Event]. This page covers how these iterators work and the patterns an agent needs, with code from the companion repository, where it is tested; the streaming guide shows the server-sent events reader built the same way.

An iterator is a function

The iter package defines two types:

• iter.Seq[V] is func(yield func(V) bool): a function that calls yield once for each value.
• iter.Seq2[K, V] is func(yield func(K, V) bool): the same with two values per step, such as a value and an error.

for v := range seq { ... } calls seq with a yield function made from the loop’s body. Each call to yield runs the body once. When the body finishes normally, yield returns true and the iterator goes on; when the body leaves the loop, with break, return, or a panic, yield returns false, and the iterator must stop. Calling yield again after it returned false is a bug the runtime catches: the program panics with “range function continued iteration after function for loop body returned false”.

The iterator runs in the caller’s goroutine, and nothing happens until the caller ranges over it. There is no channel to close and no goroutine to stop, unless the iterator starts one itself.

Writing one: a whole reply as a stream

The agent’s loop reads model replies as streams. A model that returns a whole reply at once, such as a scripted model in a test, is adapted with a few lines:

// Streamed turns a Model into a StreamModel whose stream holds the
// whole reply at once, so scripted models and saved replies work with
// the streaming loop.
func Streamed(m Model) StreamModel { return streamed{m} }

type streamed struct{ m Model }

func (s streamed) Stream(ctx context.Context,
	req Request) iter.Seq2[Delta, error] {
	return func(yield func(Delta, error) bool) {
		reply, err := s.m.Generate(ctx, req)
		if err != nil {
			yield(Delta{Usage: &reply.Usage}, err)
			return
		}
		msg := reply.Message
		ds := []Delta{{Text: msg.Content,
			ReasoningDetails: msg.ReasoningDetails}}
		for i, c := range msg.ToolCalls {
			ds = append(ds, Delta{Call: &CallDelta{Index: i, ID: c.ID,
				Name: c.Name, Arguments: string(c.Arguments)}})
		}
		ds = append(ds, Delta{FinishReason: reply.FinishReason,
			Usage: &reply.Usage})
		for _, d := range ds {
			if !yield(d, nil) {
				return
			}
		}
	}
}

Two habits are worth copying. An error is a step of its own, yield(Delta{...}, err), followed by return; the delta beside it carries the usage, since a failed call can still have cost something. And every later yield checks its result, returning as soon as the caller has stopped reading.

Reading one: values and errors

A caller of an iter.Seq2[Delta, error] checks the error on every step, the way it would check the error from any call:

// Collect assembles a whole reply from a stream, as the loop does, for
// code that wants the reply rather than its pieces, such as a test of
// an adapter's stream (chapter 11).
func Collect(stream iter.Seq2[Delta, error]) (Reply, error) {
	var a assembler
	for d, err := range stream {
		if err != nil { // the error's Delta may carry usage and ID
			if d.Usage != nil {
				a.reply.Usage = *d.Usage
			}
			id := cmp.Or(d.ID, a.reply.ID)
			return Reply{Usage: a.reply.Usage, ID: id}, err
		}
		if d.Reset { // a retry started over (chapter 12)
			a = assembler{}
			continue
		}
		if err := a.add(d); err != nil {
			return Reply{Usage: a.reply.Usage}, err
		}
	}
	return a.result(), nil
}

Collect assembles a whole reply from a stream, for code that wants the reply and not its pieces. It stops at the first error, keeping the usage the failed step reported, and it handles one more kind of step, a reset, which a retry sends when a stream starts over.

Wrapping one: middleware with cleanup

Because a stream is just a function, wrapping one is a function too. The cost meter wraps a model’s stream, passes each step through, and records what the call used:

// Wrap returns a ModelWare that records each attempt's usage in m. It
// goes inside the retries, so failed attempts are recorded too.
func (m *Meter) Wrap(next StreamModel) StreamModel {
	return StreamFunc(func(ctx context.Context,
		req Request) iter.Seq2[Delta, error] {
		return func(yield func(Delta, error) bool) {
			var c Metered
			defer func() { m.record(NameFrom(ctx), req, c) }()
			for d, err := range next.Stream(ctx, req) {
				if d.Usage != nil {
					c.Usage = *d.Usage
				}
				c.ID = cmp.Or(d.ID, c.ID)
				c.Failed = err != nil
				if !yield(d, err) {
					return
				}
			}
		}
	})
}

The recording is in a defer, so it runs whether the stream ended, failed, or the caller stopped reading partway. That is the iterator’s version of a Close method: cleanup in a deferred call inside the iterator runs when the caller’s loop ends, for whatever reason, without the caller doing anything.

Stopping early

A whole run is an iterator of events, so a caller that only wants the result reads until the last event:

// Run runs the agent loop on a Model that doesn't stream: it is
// RunStream over Streamed(m), so there is one loop for both.
func Run(ctx context.Context, m Model, tools []Tool,
	msgs []Message, lim Limits) (Result, error) {
	return RunStream(ctx, Streamed(m), tools, msgs, lim)
}

// RunStream runs the loop and returns when it ends, for callers that
// don't need to see the run as it happens.
func RunStream(ctx context.Context, m StreamModel, tools []Tool,
	msgs []Message, lim Limits) (Result, error) {
	for ev := range Events(ctx, m, tools, msgs, lim) {
		if ev.Kind == StoppedEvent {
			return ev.Result, ev.Err
		}
	}
	return Result{}, errNoStop // Events always ends with StoppedEvent
}

return inside the loop ends the range, and yield returns false inside the run. The run’s iterator cancels its own context when that happens, through a deferred cancel, so breaking out of the loop stops the model call in progress. Stopping a run is as simple as leaving the loop that reads it, and TestBreakOutOfEvents checks that it does.

Iterators with goroutines inside

An iterator can start goroutines, as long as it stops them before it returns. The fan-in that merges several runs’ events into one iterator does exactly that, and the goroutines page walks through how it releases every goroutine when the caller stops reading. That is the trade-off between the two ways of producing values over time: an iterator with no goroutine inside can’t leak one, while a channel always needs a goroutine on the other end, and someone responsible for stopping it.

The standard library’s iterators

Go 1.23 added iterators across the standard library, and they combine with the ones you write:

• maps.Keys(m) and maps.Values(m) iterate over a map.
• slices.Collect(seq) gathers an iterator into a slice, and slices.Sorted(seq) gathers and sorts it.
• slices.All, slices.Values, and slices.Backward iterate over a slice.
• iter.Pull turns an iterator into a next function, for code that has to step through two sequences at once.

The agent’s tool lookup uses two of them in one line, slices.Sorted(maps.Keys(byName)), to list the real tools in a stable order for the error the model reads; the errors page shows it.

Mistakes to avoid

• Calling yield after it returned false. The runtime panics. Check its result on every call, and return when it is false.
• Ignoring the error in an iter.Seq2[T, error]. It is a step like any other; check it before using the value.
• Cleanup after the loop instead of in a defer. Code after the last yield doesn’t run when the caller breaks out; a deferred call does.
• Ranging over the same iterator twice. An iterator like Events runs its work each time it is ranged over: ranging twice runs the agent twice. Keep the results, not the iterator.
• A goroutine inside an iterator, left running. If an iterator starts goroutines, it stops them and waits for them before it returns.

Try it

From the code directory of a clone of the companion repository:

$ go test ./ch06/agent -run 'TestBreakOutOfEvents|TestEventsOrder|TestStreamErrorDropsPartialReply' -v

1. In ch06/agent/stream_test.go, change watchedStream to call yield three times without checking its result, run TestBreakOutOfEvents, and read the panic.
2. Write a filter, a function that takes an iter.Seq[Event] and returns one holding only the tool-call events, and range over it in place of Events.
3. Range over Events twice with the same arguments and a scripted model, and count the model calls.