Day 06 / 共 20 天 · 第 2 周编排引擎

Runnable 与 4 种流式范式

进入框架心脏。Runnable 是编排产物的统一执行抽象，有 4 种调用方式。理解"组件只实现一个、框架自动补齐四个"这个 Eino 最精妙的设计——是读懂后面一切的钥匙。

📍 你在整门课的位置 · 第 2 周编排引擎（共 4 周 · 20 天）

D6 流式范式→ D7 Chain 链→ D8 Graph 图→ D9 编译执行→ D10 Workflow

L01

为什么要一个统一抽象

🤔 痛点：一个节点只会流式吐，下个节点只要完整值，怎么办？模型天生"边想边吐字"（只会 Stream），但你下一个节点（比如存数据库、算个分数）只要一个完整结果（要 Invoke）。难道每对相邻节点你都要手写"把流拼成完整值"或"把完整值拆成流"的胶水？换个组件又重写一遍——这正是没有统一抽象时的噩梦。

💡 本质：一套 Runnable 接口抹平所有组件差异，缺的范式框架自动补 Eino 让任何组件、任何编排产物编译后都变成同一个 Runnable，都有相同的 4 个方法。组件作者只实现对它最自然的 1-2 种（模型实现 Stream/Invoke 就够），其余范式框架靠 schema 里注册的"拼接/包装函数"自动补齐。于是相邻节点的流/非流不匹配，框架替你转，不用你写一行胶水。

Day 04 你见过：ChatModel 有 Generate/Stream，Tool 有 InvokableRun/StreamableRun，各组件方法名都不一样。可编排引擎要把它们"当成统一的东西"连起来跑——总不能给每种组件写一套连接逻辑吧？

🎯 Runnable 就是那个"统一插头"：任何组件、任何编排产物（Chain/Graph/Workflow），编译后都变成一个 Runnable，都有相同的 4 个方法可调。编排引擎只跟 Runnable 打交道。

类比：各种电器插头形状不一，但都转成"国标插头"后就能插同一个插座。Runnable 就是 Eino 的"国标插头"——它抹平了各组件的差异。compose/runnable.go 是本页的主角文件。

🍜 用「餐厅」来理解本讲（后面统一用这个类比） Runnable 就像生活中餐厅的"统一点餐单"：后厨有中餐师傅、西餐师傅、甜点师，手艺各不同，但你（顾客）永远只用同一张点餐单下单，服务员统一对接——你不用管菜是谁做的、怎么做的。
4 种范式就像 4 种上菜方式：一次端齐（Invoke）、边做边一道道上（Stream）、你分批递食材厨房收齐做一道（Collect）、食材流水线进成品流水线出（Transform）。
自动补齐就像贴心服务员：后厨只会"边做边上"，但你想要"一次端齐"，服务员就在窗口帮你等齐了再一起端上桌。

L02

Runnable 接口（真实代码）

compose/runnable.go:32 的真实定义——4 个方法，泛型 [I, O]（输入类型/输出类型，编译期定死）：

type Runnable[I, O any] interface {
    Invoke(ctx, input I, opts ...Option) (output O, err error)
    Stream(ctx, input I, opts ...Option) (output *schema.StreamReader[O], err error)
    Collect(ctx, input *schema.StreamReader[I], opts ...Option) (output O, err error)
    Transform(ctx, input *schema.StreamReader[I], opts ...Option) (output *schema.StreamReader[O], err error)
}

I = 输入类型，O = 输出类型（泛型，编译期就确定，所以类型安全）。四个方法名字不同，区别只在"输入是不是流、输出是不是流"。

泛型 [I, O] 是什么？（Go 小白） Runnable[string, *Message] 意思是"输入是 string、输出是 *Message 的可运行体"。泛型让同一套 Runnable 代码能用于任意输入输出类型，且编译器会帮你检查类型对不对——你不会把一个"输入 int"的 Runnable 误当"输入 string"用。

L03

4 范式怎么区分：看输入输出是不是流

4 个方法按"输入是不是流 × 输出是不是流"两两组合（源码注释里的比喻很形象）：

Invoke 单值 → 单值

ping ⇒ pong。普通调用：给一句，等一个完整答案。

Stream 单值 → 流

ping ⇒ stream。给一句，边生成边返回（打字机）。

Collect 流 → 单值

stream ⇒ pong。输入是流，收齐了给完整答案。

Transform 流 → 流

stream ⇒ stream。流进流出，一段段处理一段段产出。

4 种范式 = 「输入是不是流」×「输出是不是流」的 2×2 矩阵（runnable.go:33-36）

什么时候用哪个？ 大部分时候你用 Invoke（要完整结果）或 Stream（要打字机效果）。Collect/Transform 用于"输入本身就是流"的场景——比如上一个节点吐流、这个节点要接着处理。关键是：你不用关心组件内部实现了哪几个，任意一个都能调（下一讲揭秘）。对应地，你写自定义逻辑（Lambda）时也有这 4 种函数签名（types_lambda.go:27）。

L04

招牌魔法：实现 1 个，自动补齐 4 个

这是 Eino 最精髓的设计。源码注释直说（runnable.go:29）：

"we do downgrade compatibility for four data flow patterns, and can automatically connect components that only implement one or more methods. eg, if a component only implements Stream(), you can still call Invoke()..."

翻译：组件只实现了其中一两个方法，框架能自动"降级/升级"补齐另外几个。比如 ChatModel 只提供了 Generate（≈Invoke）和 Stream（component_to_graph_node.go:93，Collect/Transform 传 nil），但编译后你依然能对整张图调 Collect/Transform。

📝 最小例子：只实现了 Stream，却调它的 Invoke 假设某模型只会流式：Stream("北京天气?") 吐出 ["北","京","今","天","晴"] 一串 chunk。
你偏偏调 Invoke("北京天气?") 要完整值 —— 框架发现没有原生 Invoke，就自动"先 Stream 拿到那串 chunk，再用 ConcatMessages 拼接"，最终返回给你完整的 "北京今天晴"。
你的代码只写了一行 Invoke(...)，拼接的胶水框架全包了。

为什么这个设计牛？（大白话）想象你在搭流水线：模型天生是"流式"的（Stream），但你下一个节点（比如存数据库）只要"完整结果"（Invoke 式）。没有这个魔法，你就得手写"把模型的流拼成完整值"的胶水代码；有了它，框架自动帮你拼（用 Day 03 的 ConcatMessages）。反过来，一个只会返回完整值的组件，框架也能把它包成"单元素流"。结果：组件作者只写对它最自然的 1-2 种，用户四种随便调。

💬 小白 vs 老师 👶 小白：既然框架能自动补，那组件作者干脆四个方法都实现不就完了？
👨‍🏫 老师：那等于让每个厨师既练"一次端齐"又练"边做边上"，重复劳动还容易做得不一致。让作者只写最自然的（模型天生会 Stream），其余交给统一的"服务员"，省事又统一。
👶 小白：那自动补出来的，和作者亲手写的效果一样吗？
👨‍🏫 老师：功能一样，但"体验"可能降级——只有 Stream 却调 Invoke，得等所有 chunk 到齐拼好才返回：结果对，但没了"逐字"的即时感。想要真流式，链路每环都得原生支持流（见 L07）。

L05

补齐优先级表（真实代码）

自动补齐发生在 newRunnablePacker（runnable.go:345）。它是一张"优先级表"——补 Invoke 时，自己有就用自己，否则依次尝试用 Stream/Collect/Transform 转：

🅰️ 如果让你自己写，可能是这样（极简伪代码）

// 朴素想法：没有 Invoke？那就用 Stream 拼一下
func fillInvoke(s StreamFunc) InvokeFunc {
    return func(in) Out { return concat(s(in)) }  // 只想到用 Stream 兜底
}

问题：万一组件连 Stream 都没有、只有 Collect 或 Transform 呢？你的兜底就崩了。真实源码把"备胎"排成一条优先级链（Stream→Collect→Transform 逐个试），保证只要 4 个里有任意 1 个就能补出来 👇

// runnable.go:368 —— 补 Invoke
if i != nil {
    r.i = i                       // 自己有 Invoke，直接用
} else if s != nil {
    r.i = invokeByStream(s)       // 否则：用 Stream 转（拿到流再拼成完整值）
} else if c != nil {
    r.i = invokeByCollect(c)      // 或：用 Collect 转（把输入包成单元素流）
} else {
    r.i = invokeByTransform(t)    // 或：用 Transform 转
}
// Stream / Collect / Transform 各有类似的优先级链（runnable.go:378-406）

读法：四个方法各有一条"自己没有就找别人代劳"的优先级链。只要组件至少实现了 4 个里的 1 个，其余 3 个都能被推导出来。

用一张矩阵直观看"从谁能推出谁"：

目标\来源	有 Invoke	有 Stream	有 Collect	有 Transform
要 Invoke	直接用	拼流	包单元素流	拼流
要 Stream	包成单块流	直接用	—	真流式
要 Collect	拼输入+Invoke	—	直接用	真流式
要 Transform	拼输入+包块流	—	—	直接用

L06

4 个方向的转换（血肉）

补齐靠的是四类转换函数（runnable.go:203-343）。理解这三种基本操作就懂了全部：

流 → 单值：concatStreamReader 把所有 chunk 拼成完整值（用 Day 03 的 ConcatMessages）。例：invokeByStream——先 Stream 拿流，再拼成单值。
单值 → 流：把单值包成"只有一个元素的流"——StreamReaderFromArray([]O{out})（runnable.go:243 的 streamByInvoke）。
单值喂给需要流的方法：把单值包成单元素流再喂进去，如 invokeByCollect（runnable.go:214）。

一张记忆图：
Invoke（要完整值）→ 可从任何流方法"拼接(concat)"降级得到。
Stream（要真流）→ 真流式优先来自 Transform 或 Stream 自身；用 Invoke 降级会退化成"单块流"（就一个大块，没了逐字效果）。
Collect → 先把输入流拼成单值，再走 Invoke。
Transform → 真流式；用 Invoke 降级 = 先拼输入、算完再包成单块流。

L07

真流式的代价：一个节点破功，全链凝固

魔法虽好，但有个重要认知（初学者必懂）：真正的"逐字流式"只有当链路上每个节点都提供了 Stream/Transform 时，才能端到端保持。任何一个节点只有 Invoke，流到那里就会被 concat 成整块。

为什么？ 假设链路 模型(Stream) → 过滤(只有 Invoke) → 输出。模型吐流，但过滤节点只会 Invoke（要完整输入）——框架只能先把模型的流拼成完整消息再喂给过滤。于是"逐字"效果在过滤这里就没了，后面即使想流也是"一整块"。所以想要端到端打字机效果，链路上每个节点都得是流式的。源码在 chain.go:264 特别提醒：想要真流式输出，请用 StreamableLambda/TransformableLambda 写你的自定义节点。

内部实现的一个小知识 Runnable 对外是泛型 [I,O]（类型安全），但内部执行时用 any（无类型）——composableRunnable（runnable.go:46）只保留 i(处理非流)和 t(处理流)两个函数 + 三个 reflect.Type 做类型校验。编译好的图最后又用 toGenericRunnable（runnable.go:411）包回强类型接口给你。外层强类型（好用、安全）、内层无类型（灵活、统一）——这是 Go 泛型 + 反射的经典配合。

L08

今日小结 + 动手

                🧠 今天你应该能回答
                Runnable 是什么？为什么需要它？（编排产物的统一执行插头）
4 种范式怎么区分？（输入是不是流 × 输出是不是流）
"实现 1 个补齐 4 个"怎么做到？（newRunnablePacker 的优先级表 + 转换函数）
三种基本转换？（流→单值拼接、单值→单块流、单值→单元素流）
为什么"一个非流节点会让全链凝固"？想要真流式怎么办？

            

✋ 动手：对着真实代码读一遍

# 1. Runnable 接口（L02）
sed -n '32,40p' compose/runnable.go

# 2. 补齐优先级表（L05，本页精华）
sed -n '345,409p' compose/runnable.go

# 3. 4 个方向的转换函数（L06）
sed -n '203,260p' compose/runnable.go

# 4. Lambda 的 4 种函数签名
sed -n '27,39p' compose/types_lambda.go

明天预告 · Day 07：懂了 Runnable，我们看最简单的编排——Chain（链式）：组件一个接一个串成流水线，自动连边、自动补到 END。它是 Graph 的简化封装。

← Day 05 完整旅程 Day 07 · Chain 链式编排 →