Runnable 接口与链式调用机制
Runnable 是 LangChain JS 的统一执行协议:Prompt、Model、Parser、Retriever 等都实现它,从而可用同一套 invoke / stream / batch 组合。
源码:@langchain/core/runnables。
1. 接口语义
| 能力 | 说明 |
|---|---|
invoke(input, config?) | 单步执行 |
batch(inputs, config?) | 并行/批量(可配 maxConcurrency) |
stream(input, config?) | 输出为 AsyncIterable chunk |
streamEvents | v2 事件流,用于调试与 UI |
输入输出类型由泛型约束;链式后类型由 pipe 推断串联。
2. Runnable 基类能力
- config 合并:子调用继承 callbacks、metadata。
- 可组合:
pipe返回新 Runnable,不修改原实例。 - 序列化:部分 Runnable 支持 JSON 配置导出(视类型而定)。
3. pipe(LCEL)
mermaid
flowchart LR
I[Input] --> A[Runnable A]
A --> B[Runnable B]
B --> C[Runnable C]
C --> O[Output]a.pipe(b).pipe(c) 等价于「先 A 再 B 再 C」。
规则:上一步输出类型应匹配下一步输入(或 Runnable 内做适配)。
常见链:ChatPromptTemplate → ChatModel → StrOutputParser。
4. RunnableLike
函数、{ invoke } 对象、Runnable 实例都可转为 Runnable(RunnableLambda),便于嵌入自定义逻辑而仍享受 stream/batch。
5. RunnableConfig
| 字段 | 用途 |
|---|---|
callbacks | 生命周期钩子、LangSmith |
tags / metadata | 过滤与归因 |
runName | 追踪树节点名 |
signal | Abort 取消 |
maxConcurrency | batch 并发上限 |
6. 重试与降级
| API | 场景 |
|---|---|
withRetry | 429、网络抖动 |
withFallbacks | 主模型失败切备用模型 |
在链的某一环包装即可,不必改内部实现。
7. 实践
- 链不宜过长而无观测:给关键节点设
runName。 - 需要中间结果时用
RunnablePassthrough.assign或 LangGraph。 - 流式 UI 消费最后一环的
stream,或streamEvents看全链路。
8. 总结
Runnable + pipe = LangChain 的函数式组合核心;读源码从 RunnableSequence 与 pipe 实现入手。