Eino流式处理4大范式详解Invoke、Stream、Collect与Transform实战指南【免费下载链接】einoGo 语言编写的终极大型语言模型LLM应用开发框架强调简洁性、可扩展性、可靠性与有效性。项目地址: https://gitcode.com/CloudWeGo/einoEino 是 Go 语言编写的 LLM 应用开发框架其流式处理机制由 Invoke、Stream、Collect、Transform 四大范式组成。搞懂这 4 个方法你就掌握了 Eino 处理大模型流式输出的核心能力——无论是新手入门还是进阶编排本文都帮你一次讲透。为什么 LLM 应用绕不开流式处理大模型是边生成边吐字的ChatModel 会一块一块chunk实时输出消息而不是等全部生成完才返回。这在两个场景里尤其关键用户体验流式输出让回答打字机式呈现等待体感更短编排需求Chain / Graph 中串联的组件越来越多下游组件要能处理这些流动的数据块。Eino 的回答很干脆把流式处理下沉到框架层。一个编译后的 Graph 可以支持 4 种流式范式自由运行框架自动帮你拼接、装箱、合并、拷贝数据块组件自己实现哪套范式对你完全透明详见 README.md。先认识核心类型StreamReader理解 4 大范式前先认识流式数据的载体StreamReader[T]定义在 schema/stream.gosr, sw : schema.Pipestring // 创建一条容量为3的流 go func() { // 发送端写入数据块 defer sw.Close() for i : 0; i 10; i { sw.Send(i, nil) } }() defer sr.Close() // 接收端必须记得 Close for chunk, err : sr.Recv() { // 逐块读取 if errors.Is(err, io.EOF) { break } fmt.Println(chunk) }StreamReader还提供两个高频工具方法StreamReaderFromArray把一个普通数组装箱成流——非流式数据瞬间变流式StreamReaderWithConvert流式地把每个数据块转换/过滤成另一个类型比如丢掉空消息。另外别忘了StreamReader支持 Copy(n)把一条流复制成 n 条供并发节点或回调处理器各自消费。4 大范式一张表看懂输入输出组合核心接口 Runnable[I, O] 只有 4 个方法本质是输入是否流式 × 输出是否流式的 2×2 组合范式输入输出记忆口诀Invoke非流 I非流 Oping pongStream非流 I流式 StreamReader[O]ping 流Collect流式 StreamReader[I]非流 O流 pongTransform流式 StreamReader[I]流式 StreamReader[O]流 流源码注释也用了同样的比喻见 compose/runnable.goInvoke 是发一球回一球其余三种都带流。Invoke最经典的一次性调用传入完整输入等全部结果。适合不需要实时感知的调用方比如脚本、单元测试out, err : chain.Invoke(ctx, map[string]any{query: 你是谁})Stream流式输出打字机体验的关键输入是普通值输出是StreamReader[O]。这是 ChatModel 最自然的形态——消息逐块流出前端逐块渲染sr, err : chain.Stream(ctx, input) defer sr.Close() for { chunk, err : sr.Recv() if errors.Is(err, io.EOF) { break } render(chunk) // 逐块推送到前端 }Eino 的 ChatModel 组件接口就同时提供了Generate对应 Invoke和Stream对应 Stream两个方法。Collect流式输入一次性交付输入端已经在流动比如上游节点的输出流但你需要完整结果才能继续——典型如工具节点必须拿到完整消息才知道调哪个工具、传什么参数。Collect把流读干、拼接成完整值再返回拼接逻辑见 compose/stream_concat.go。Transform流进流出真正的边读边写Transform是唯一能边消费边产出的范式读一个输入块立刻可以吐一个输出块延迟最低。自定义组件做流式转发、流式过滤、流式改写时首选它。自动降级与升级你只实现一种4 种都能跑这是 Eino 流式机制最优雅的部分。组件只需实现 4 种范式中的任意一种框架会自动补齐其余三种。补齐规则定义在 newRunnablePacker按优先级互相转换你实现了框架如何补全其余InvokeStream 调用后把结果装箱成单块流Transform 先拼流再调用输出装箱StreamInvoke 读完输出流再拼接Transform 先拼输入流再调用CollectInvoke 把输入装箱成单块流再调用Stream 收集后把结果装箱TransformInvoke 拼输入流 拼输出流Stream 把输入装箱成单块流比如invokeByStream的套路就是先 Stream再把输出流 concat 成完整值见 compose/runnable.go反向的streamByInvoke则用StreamReaderFromArray把一次性结果伪装成流。由此还带来一个最优实现建议如果你能实现 Transform就尽量实现 Transform——它可以零损耗地驱动另外三种范式而只实现了 Invoke 的组件每次走流式路径都得多付一次装箱/拼流的开销。编排中自动发生的 4 件事当你在 Graph 里串联多个节点如 tool_call_graph 这类 ChatModel ToolsNode 的组合时框架会自动处理所有流式细节见 README.md拼接concatenate下游只要完整输入时自动把流块拼好再喂进去装箱box需要流但手上是非流数据时自动包成流合并merge多条上游流汇入同一节点时自动合并拷贝copy流扇出到多个下游或回调处理器时自动复制。也就是说图里每个节点用哪套范式是透明的你只管连线数据形态的转换全部交给框架——这正是 Runnable 注释里说的 downgrade compatibility for four data flow patterns。实战选型该怎么挑场景推荐范式理由面向用户的对话界面Stream逐块推送首字延迟最低后台批处理 / 单元测试Invoke代码最简拿到完整结果即可流上游结果做聚合统计、落库Collect需要完整数据才做下一步自定义组件做流式改写/转发/过滤Transform边读边写延迟与内存占用双低一个实用技巧如果自定义组件的中间结果类型没有默认拼接规则可以用 RegisterStreamChunkConcatFunc 注册拼接函数让Collect/Invoke路径也能正确工作。小结Eino 用Invoke / Stream / Collect / Transform4 个方法统一了输入/输出 × 流式/非流式的所有组合核心载体是 StreamReader[T]配合Pipe、FromArray、WithConvert、Copy即可应对绝大多数流式操作组件只需实现其中一种范式框架自动降级/升级补齐其余三种并在编排中自动完成拼接、装箱、合并与拷贝选型口诀给用户看选 Stream要完整结果选 Invoke/Collect边读边写选 Transform。掌握这 4 大范式后再去看 Eino 的 ReAct Agent 这类复杂编排流式数据的走向会一目了然。【免费下载链接】einoGo 语言编写的终极大型语言模型LLM应用开发框架强调简洁性、可扩展性、可靠性与有效性。项目地址: https://gitcode.com/CloudWeGo/eino创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考