Eino 流式处理与 Callback：正确性和可观测性

核心摘要

Eino Streaming 是类型化数据流机制，Callback 是生命周期 Hook，两者都不会自动带来安全或完整观测。v0.9.21 中，StreamReader 是单 Consumer、单次读取对象；多个消费者必须在首次读取前复制，并关闭每个 Reader。Callback 获得私有 Stream Copy，但也必须关闭，不能修改共享输入输出，更不能依赖不同 Handler 的执行顺序。

核心要点

• Invoke、Stream、Collect 与 Transform 描述四种不同输入输出模式。
• Stream 转换可能以 Buffer、Memory 与 Cancellation 复杂度换取更早输出。
• Callback Stream 虽是观察副本，仍可能阻塞或泄漏资源。
• OnError 不覆盖后续 StreamReader.Recv 才暴露的错误。
• Trace 应优先记录身份、耗时、计数与 Policy Outcome，而不是默认保存完整 Prompt。

四种调用模式

官方流式编程指南定义了四种 Runnable 模式：

Eino 可以把 Node 已实现的模式适配为 Graph 请求的另一种模式：Value 可封装成单元素 Stream，Stream 可拼接成 Value；Fan-out 可能复制流，Fan-in 可能合并流。这些都是可观察的运行时行为，不只是语法糖。

StreamReader 所有权

固定版本的 schema/stream.go定义了所有权规则：

1. StreamReader 只读一次，只允许一个 goroutine 调用 Recv。
2. 多个 Consumer 需要同一数据时，在首次 Recv 前调用 Copy。
3. 每个 Reader 恰好关闭一次，包括 io.EOF 或提前返回场景。
4. Producer 应关闭自己的 StreamWriter。
5. Buffer Capacity 只是队列元素上限，不是完整背压策略。

如果某个副本停止消费又没有关闭，Source 可能阻塞或一直保留资源。必须测试 Slow Consumer、Early Disconnect、Cancellation 和 Oversized Chunk。

Streaming 不一定改善产品

Streaming 适合客户端能够安全渲染部分结果的场景。以下情况可能更适合等待完整结果：

• JSON 文档必须整体验证后才能展示；
• Moderation 或 Policy Check 需要完整候选；
• Tool Call 必须完整后才能执行；
• Partial Text 可能泄露敏感数据；
• Transport 无法表达类型化的 Partial Failure；
• Retry 会重复已经展示的内容。

应测量 First Safe Byte、Last Byte、Cancellation Latency、Buffered Bytes、首字节后的错误率与重连重复内容，而不只看 TTFT。

Callback 时机与作用域

Callback 官方手册定义五个时机：

Handler 可以全局注册、附着在单次调用，或指定到 Node。全局注册不是并发安全操作，应只在初始化阶段执行。Standalone 调用没有注入 Metadata 时，RunInfo 可能为空。

不同 Handler 之间没有保证顺序，也不会传递 Context Chain。同一个 Handler 可以把 Start Hook 返回的 Context 交给自己的 End Hook。

带 Callback 的可运行 Stream

下面程序以 Eino Core v0.9.21 为目标。它创建显式 Graph Node，按 Node Key 指定 Callback，消费 Callback 的私有 Stream Copy，并关闭两份 Reader。

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())
}

预期输出：

Eino streams safely; observed=3

该示例刻意保持简单。生产 Callback 不应在请求路径同步执行网络导出，还要定义有界 Queue 以及 Drop 或 Backpressure Policy。

Trace 数据模型

在不采集一切的前提下，保留足够复现行为的数据：

OpenTelemetry 的 Generative AI Semantic Conventions仍在演进。应固定语义约定版本，并维护内部 Schema，让 Telemetry 迁移可审阅。

故障测试

• Consumer 在首个 Chunk 前断开。
• Consumer 读取一个 Chunk 后停止并关闭 Reader。
• Producer 在若干成功 Chunk 后返回错误。
• 某个 Callback Copy 变慢或失败。
• Provider 或 Tool 阻塞时 Context 被取消。
• Partial Byte 已交付后发生 Retry。
• Redaction 失败或 Exporter 不可用。
• 多个 Handler 观察同一 Node，但不依赖执行顺序。

常见问题

Stream 一定会降低首 Token 延迟吗？

不会。它只改变交付契约；Provider Buffer、Moderation、Proxy、应用缓冲和客户端渲染都可能抵消预期收益。

可以让两个 goroutine 读取同一个 StreamReader 吗？

不可以。首次读取前调用 Copy，每个 Consumer 使用一个 Reader，并关闭所有副本。

Callback 收到的是原始 Stream 吗？

流式 Callback 收到私有副本。Handler 必须关闭它，也不能修改共享 Input 或 Output。

Stream Error 应在哪里记录？

在 Consumer 中记录 Recv 返回的错误。OnError 处理 Component 直接返回的错误，不覆盖后续流内错误。

Callback 应保存完整 Prompt 与 Tool Result 吗？

默认不保存。优先记录 Hash、Size、Model 与 Schema Revision、Timing、Usage、Policy Outcome 和脱敏 ID。完整 Payload 采集必须有明确 Purpose、Access Policy、Retention 与 Delete Path。

一手资料

• Eino v0.9.21 Stream 源码
• Eino 流式编程指南
• Eino Callback 用户手册
• Eino v0.9.21 Callback 契约
• Go context Package
• OpenTelemetry Generative AI Semantic Conventions

相关阅读

• Eino 编排详解
• Eino ADK 实战
• Agent 可观测性
• Latency 术语
• Agent Trajectory 术语