Eino Streaming and Callbacks: Correctness and Observability

TL;DR

Eino streaming is a typed data-flow mechanism, while callbacks are lifecycle hooks. Neither is automatically safe or complete. In v0.9.21, a StreamReader is single-consumer and read-once; copy it before reading when several consumers need the data, close every reader, and handle Recv errors in the consumer. Callback handlers receive private stream copies but must close them, must not mutate shared input/output, and cannot rely on ordering across handlers.

Key Takeaways

• Invoke, Stream, Collect, and Transform describe four distinct input/output modes.
• Stream conversion may trade lower first-output latency for buffering, memory, and cancellation complexity.
• A callback stream is an observation copy, but observation can still block or leak resources.
• OnError does not cover errors that appear later during StreamReader.Recv.
• Trace design should capture identities, timings, counts, and policy outcomes without defaulting to full prompts.

Four Invocation Modes

The official stream programming guide defines four Runnable modes:

Eino may adapt a node that implements one mode to a graph invoked through another. A value can become a one-item stream; a stream can be concatenated into a value. Fan-out may copy a stream, and fan-in may merge streams. These are observable runtime choices, not syntax sugar.

StreamReader Ownership

The pinned schema/stream.go defines the ownership rules:

1. A StreamReader is read once; only one goroutine should call Recv.
2. If multiple consumers need the stream, call Copy before any Recv.
3. Close each reader exactly once, including after io.EOF or an early return.
4. A producer should close its StreamWriter.
5. Buffer capacity is a limit on queued items, not a complete backpressure strategy.

If one copied reader stops consuming and does not close, the source may remain blocked or retained. Test slow consumers, early disconnects, cancellation, and oversized chunks.

Streaming Does Not Always Improve the Product

Streaming is useful when the client can render partial output safely. It may be unsuitable when:

• a JSON document must validate atomically before display;
• moderation or policy checks require the complete candidate;
• a Tool Call must be complete before execution;
• partial text would reveal sensitive data;
• the transport cannot express typed partial failure;
• retrying would duplicate user-visible output.

Measure time to first safe byte, time to last byte, cancellation latency, buffered bytes, error rate after first byte, and duplicated content after reconnect. Raw time to first token is not enough.

Callback Timing and Scope

The Callback manual defines five timings:

Handlers can be global, attached to one invocation, or designated to nodes. Global registration is not concurrency-safe and should happen once during initialization. RunInfo can be nil for standalone calls without injected metadata.

Different handlers have no guaranteed order and do not share a context chain. A single handler may use the context returned by its own start hook in its own end hook.

Runnable Stream with a Callback

This complete program targets Eino Core v0.9.21. It creates an explicit Graph node so the callback can be designated by node key, consumes the callback's private stream copy, and closes both readers.

package main

import (
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"sync/atomic"

	"github.com/cloudwego/eino/callbacks"
	"github.com/cloudwego/eino/compose"
	"github.com/cloudwego/eino/schema"
)

func main() {
	ctx := context.Background()
	var observed atomic.Int64

	handler := callbacks.NewHandlerBuilder().
		OnEndWithStreamOutputFn(func(
			ctx context.Context,
			info *callbacks.RunInfo,
			output *schema.StreamReader[callbacks.CallbackOutput],
		) context.Context {
			defer output.Close()
			for {
				_, err := output.Recv()
				if errors.Is(err, io.EOF) {
					break
				}
				if err != nil {
					break
				}
				observed.Add(1)
			}
			return ctx
		}).
		Build()

	graph := compose.NewGraph[string, string]()
	source := compose.StreamableLambda(func(ctx context.Context, input string) (*schema.StreamReader[string], error) {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		return schema.StreamReaderFromArray(strings.Fields(input)), nil
	})
	if err := graph.AddLambdaNode("word_stream", source); err != nil {
		panic(err)
	}
	if err := graph.AddEdge(compose.START, "word_stream"); err != nil {
		panic(err)
	}
	if err := graph.AddEdge("word_stream", compose.END); err != nil {
		panic(err)
	}

	runnable, err := graph.Compile(ctx)
	if err != nil {
		panic(err)
	}
	reader, err := runnable.Stream(
		ctx,
		"Eino streams safely",
		compose.WithCallbacks(handler).DesignateNode("word_stream"),
	)
	if err != nil {
		panic(err)
	}
	defer reader.Close()

	var words []string
	for {
		word, err := reader.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			panic(err)
		}
		words = append(words, word)
	}
	fmt.Printf("%s; observed=%d\n", strings.Join(words, " "), observed.Load())
}

Expected output:

Eino streams safely; observed=3

This example is intentionally small. A production callback should avoid synchronous network export on the request path and should define a bounded queue and drop or backpressure policy.

Trace Data Model

Capture enough data to reproduce behavior without collecting everything:

OpenTelemetry's Generative AI semantic conventions are a moving specification area. Pin the semantic-convention version and keep an internal schema so telemetry migrations are reviewable.

Failure Tests

• Consumer disconnects before the first chunk.
• Consumer stops after one chunk and closes its reader.
• Producer emits an error after several successful chunks.
• One callback copy is slow or fails.
• Context is canceled while a provider or Tool is blocked.
• A response is retried after partial bytes were delivered.
• Redaction fails or the exporter is unavailable.
• Multiple handlers observe the same node without relying on order.

Frequently Asked Questions

Does Stream always reduce time to first token?

No. It changes the delivery contract. Provider buffering, moderation, network proxies, application buffering, and client rendering can eliminate the expected benefit.

Can I read a StreamReader in two goroutines?

No. Call Copy before the first read and give each consumer one reader. Close every copy.

Does a callback receive the original stream?

Streaming callbacks receive a private copy. They must close it and must not mutate shared input or output values.

Where should stream errors be recorded?

Record errors from Recv in the consumer. OnError covers a component returning an error, not errors that surface later inside the stream.

Should callbacks export full prompts and tool results?

Not by default. Prefer hashes, sizes, model and schema revisions, timing, usage, policy outcomes, and redacted identifiers. Full payload capture needs an explicit purpose, access policy, retention, and deletion path.

Primary Sources

• Eino v0.9.21 stream source
• Eino Stream Programming Essentials
• Eino Callback User Manual
• Eino v0.9.21 callback contract
• Go context package
• OpenTelemetry Generative AI semantic conventions

Related Reading

• Eino Orchestration
• Eino ADK Guide
• Agent Observability
• Latency glossary
• Agent Trajectory glossary