Seldon Core 2 Pipeline 实战指南：模型串联、数据流合并与条件分支的完整示例

Seldon Core 2 的 Pipeline 允许把多个模型编排成一张推理数据流图，用声明式的 YAML 描述数据如何从一个模型流向另一个模型、如何在多个输入之间做 join、以及如何依据张量触发条件分支。本文以官方 pipeline-examples 为骨架，结合 Pipeline CRD 源码 与调度器实现，完整讲解模型 Chaining、从 Pipeline 输入引用张量、Model Join、条件数据流、Pipeline Input Tensors 与 Trigger Joins 六类典型用法，读者读完即可在本地 Seldon Core 2 环境中复现全部示例，并理解 tensorMap、stepsJoin、triggers 等核心字段的底层语义。

前置条件与示例模型

所有示例需要在本地运行 Seldon Core 2 环境（文档原话为 "We assume you have Seldon Core 2 running locally"），并建议在仓库的 samples 目录下执行命令，因为文档中的 cat ./models/...、cat ./pipelines/... 均相对于该目录。对应的真实文件位于 samples/models 与 samples/pipelines。

示例主要使用以下两类模型：

• gs://seldon-models/triton/simple：一个 Triton TensorFlow 示例模型，接收两个输入张量 INPUT0 和 INPUT1，输出 OUTPUT0 = INPUT0 + INPUT1 以及 OUTPUT1 = INPUT0 - INPUT1（其原始源码与许可证来自 Triton Inference Server 官方示例模型仓库的 model_repository/simple 目录）。
• gs://seldon-models/scv2/samples/triton_23-03/ 下的一组 Triton Python 示例模型：conditional（根据 CHOICE 张量决定输出 OUTPUT0 还是 OUTPUT1）、add10（输入加 10）与 mul10（输入乘 10）。

这些模型的存储 URI 通过 storageUri 字段声明，运行时由 Seldon Core 2 负责从对象存储拉取模型文件并部署为可推理的模型服务。模型注册统一使用 seldon model 命令族，其 YAML 使用 mlops.seldon.io/v1alpha1 API 版本，requirements 字段声明模型运行所需的推理服务器能力（如 tensorflow、triton、python），memory 字段给出模型加载时预留的最小内存量（示例中为 100Ki）。

模型 Chaining：将一个模型的输出串联为另一个模型的输入

第一个场景演示最基础的串联：把 tfsimple1 的输出张量改名后送入 tfsimple2，构成两跳流水线。

加载两个基础模型

首先查看并加载模型定义（tfsimple1 与 tfsimple2 使用同一个 gs://seldon-models/triton/simple 模型，只是各自独立加载为两个模型实例）：

cat ./models/tfsimple1.yaml
cat ./models/tfsimple2.yaml

对应的模型定义如下：

apiVersion: mlops.seldon.io/v1alpha1
kind: Model
metadata:
  name: tfsimple1
spec:
  storageUri: "gs://seldon-models/triton/simple"
  requirements:
  - tensorflow
  memory: 100Ki
---
apiVersion: mlops.seldon.io/v1alpha1
kind: Model
metadata:
  name: tfsimple2
spec:
  storageUri: "gs://seldon-models/triton/simple"
  requirements:
  - tensorflow
  memory: 100Ki

加载并等待两个模型进入可用状态：

seldon model load -f ./models/tfsimple1.yaml
seldon model load -f ./models/tfsimple2.yaml
seldon model status tfsimple1 -w ModelAvailable | jq -M .
seldon model status tfsimple2 -w ModelAvailable | jq -M .

-w ModelAvailable 表示等待（watch）直到模型状态变为 ModelAvailable，这是后续 Pipeline 能否就绪的前提。

定义串联 Pipeline 并使用 tensorMap 改名

由于 tfsimple1 与 tfsimple2 输入输出的 shape 与数据类型兼容，可以直接把前者输出接到后者输入。但输出张量名（OUTPUT0/OUTPUT1）与第二个模型期望的输入张量名（INPUT0/INPUT1）不一致，因此需要用 tensorMap 把张量改名。Pipeline 的最终输出来自 tfsimple2：

cat ./pipelines/tfsimples.yaml

apiVersion: mlops.seldon.io/v1alpha1
kind: Pipeline
metadata:
  name: tfsimples
spec:
  steps:
    - name: tfsimple1
    - name: tfsimple2
      inputs:
      - tfsimple1
      tensorMap:
        tfsimple1.outputs.OUTPUT0: INPUT0
        tfsimple1.outputs.OUTPUT1: INPUT1
  output:
    steps:
    - tfsimple2

这里的两个关键语法点：

• inputs: [tfsimple1] 是简写形式，等价于引用 tfsimple1.outputs（该步骤的全部输出）。
• tensorMap 的键采用 步骤名.outputs.张量名 这样的限定名（fully qualified tensor name），值是该张量在目标步骤中要被映射成的输入张量名。从调度器源码看，tensorMap 最终会被 topicNamer.GetFullyQualifiedTensorMap(...) 转换为 Kafka topic 层面的完整限定映射（见 scheduler/pkg/kafka/dataflow/server.go 中 createTriggerSources/step update 的构建逻辑）。

加载并等待 Pipeline 就绪：

seldon pipeline load -f ./pipelines/tfsimples.yaml
seldon pipeline status tfsimples -w PipelineReady | jq -M .

seldon pipeline status 返回的 JSON 中可以看到：Pipeline 拥有 versions 数组（每个版本有独立的 uid、递增的 version 号），state.status 为 PipelineReady，state.modelsReady 为 true，表示依赖的模型均已就绪。调度器侧的 Pipeline 状态机定义在 scheduler/pkg/store/pipeline/pipeline.go 中（PipelineCreating → PipelineReady → ... → PipelineTerminated），示例返回中的 reason: "created pipeline" 即对应创建成功路径。

使用 HTTP 与 gRPC 两种模式推理

Pipeline 与模型一样支持 HTTP（默认）与 gRPC 两种推理模式。先看 HTTP 模式，输入两个 INT32 张量，每个为 [1,16] 形状：

seldon pipeline infer tfsimples \
    '{"inputs":[{"name":"INPUT0","data":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16],"datatype":"INT32","shape":[1,16]},{"name":"INPUT1","data":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16],"datatype":"INT32","shape":[1,16]}]}' | jq -M .

返回结果（OUTPUT0 为相加、OUTPUT1 为相减）：

{
  "model_name": "",
  "outputs": [
    {"name": "OUTPUT0", "data": [2,4,6,8,10,12,14,16,18,20,22,24,26,28,30,32], "shape": [1,16], "datatype": "INT32"},
    {"name": "OUTPUT1", "data": [2,4,6,8,10,12,14,16,18,20,22,24,26,28,30,32], "shape": [1,16], "datatype": "INT32"}
  ]
}

接着用 --inference-mode grpc 走 gRPC 推理。注意 gRPC 模式的请求体遵循 gRPC Inference 协议：张量数据放在 contents 字段中（如 int_contents），shape 数组元素为字符串，并且可以附带 model_name：

seldon pipeline infer tfsimples --inference-mode grpc \
    '{"model_name":"simple","inputs":[{"name":"INPUT0","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]},{"name":"INPUT1","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]}]}' | jq -M .

返回的 gRPC 格式输出：

{
  "outputs": [
    {"name": "OUTPUT0", "datatype": "INT32", "shape": ["1","16"], "contents": {"intContents": [2,4,6,8,10,12,14,16,18,20,22,24,26,28,30,32]}},
    {"name": "OUTPUT1", "datatype": "INT32", "shape": ["1","16"], "contents": {"intContents": [2,4,6,8,10,12,14,16,18,20,22,24,26,28,30,32]}}
  ]
}

用 pipeline inspect 调试数据流

seldon pipeline inspect 是调试 Pipeline 的利器：它读取 Kafka 中流水线各环节的 topic 数据，按 pipelineName.stepName.(inputs|outputs).tensorName 的粒度展示数据。默认显示最近一条经过 Pipeline 的数据（默认 offset 为最后一条消息）：

seldon pipeline inspect tfsimples

输出可以看到每个环节的完整数据（seldon.default.model.tfsimple1.inputs、seldon.default.model.tfsimple1.outputs、seldon.default.model.tfsimple2.inputs、seldon.default.model.tfsimple2.outputs、seldon.default.pipeline.tfsimples.inputs、seldon.default.pipeline.tfsimples.outputs）。可以清楚验证：

• tfsimple1.outputs 中 OUTPUT1 全为 0（即 INPUT0 - INPUT1 = 0，因为两次输入相同）；
• 经过 tensorMap 改名后，tfsimple2.inputs 收到的是 INPUT0 = [2,4,...]、INPUT1 = [0,0,...]，并附带了 rawInputContents 原始字节缓冲；
• 最终 pipeline.tfsimples.outputs 的 OUTPUT0 与 OUTPUT1 均为 [2,4,...]（相加得到 2 倍输入、相减得到 0，再经 tfsimple2 相加/相减后仍为 [2,4,...]）。

inspect 还支持 --format json 输出，配合 jq 可以只取某个环节的消息值，例如查看 Pipeline 入口的原始输入：

seldon pipeline inspect tfsimples --format json | jq -M .topics[0].msgs[0].value

CLI 帮助文档（docs-gb/cli/seldon_pipeline_inspect.md）显示 inspect 支持 --kafka-broker、--kafka-config-path、--offset、--request-id、-t/--truncate、-v/--verbose 等选项，其中表达式可以是 pipelineName、pipelineName.(inputs|outputs)、pipelineName.stepName.(inputs|outputs) 乃至进一步限定到单个 tensorName。

清理

seldon pipeline unload tfsimples
seldon model unload tfsimple1
seldon model unload tfsimple2

从输入引用的 Chaining：直接引用上游模型的输入张量

第二个场景与第一个的唯一区别在于 tfsimple2 的 inputs 不再引用 tfsimple1 的全部输出，而是混合引用了 tfsimple1 的输入张量与输出张量：

cat ./pipelines/tfsimples-input.yaml

apiVersion: mlops.seldon.io/v1alpha1
kind: Pipeline
metadata:
  name: tfsimples-input
spec:
  steps:
    - name: tfsimple1
    - name: tfsimple2
      inputs:
      - tfsimple1.inputs.INPUT0
      - tfsimple1.outputs.OUTPUT1
      tensorMap:
        tfsimple1.outputs.OUTPUT1: INPUT1
  output:
    steps:
    - tfsimple2

注意这里张量限定名的第二个路径段是 inputs 或 outputs：tfsimple1.inputs.INPUT0 引用的是 tfsimple1 接收到的输入张量 INPUT0（即 Pipeline 的原始输入），而 tfsimple1.outputs.OUTPUT1 引用其输出张量。也就是说，数据流可以指向图中任意步骤的输入或输出，而不只是"紧邻的上游输出"，这正是 Pipeline 作为有向数据流图的灵活性所在。

加载、等待就绪并推理：

seldon pipeline load -f ./pipelines/tfsimples-input.yaml
seldon pipeline status tfsimples-input -w PipelineReady | jq -M .
seldon pipeline infer tfsimples-input \
    '{"inputs":[{"name":"INPUT0","data":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16],"datatype":"INT32","shape":[1,16]},{"name":"INPUT1","data":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16],"datatype":"INT32","shape":[1,16]}]}' | jq -M .

这次由于 tfsimple2 的 INPUT0 直接来自原始输入（tfsimple1.inputs.INPUT0），INPUT1 来自 tfsimple1.outputs.OUTPUT1（相减结果，全 0），最终输出变为：

• OUTPUT0 = INPUT0 + INPUT1 = [1,2,...,16]
• OUTPUT1 = INPUT0 - INPUT1 = [1,2,...,16]

gRPC 模式的结果一致：

seldon pipeline infer tfsimples-input --inference-mode grpc \
    '{"model_name":"simple","inputs":[{"name":"INPUT0","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]},{"name":"INPUT1","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]}]}' | jq -M .

最后清理：

seldon pipeline unload tfsimples-input
seldon model unload tfsimple1
seldon model unload tfsimple2

Model Join：把两个模型的输出张量合并喂给第三个模型

第三个场景展示数据流合并（join）：tfsimple3 的输入来自 tfsimple1 与 tfsimple2 两个并行模型各一个输出张量，形成典型的扇形汇聚（fan-in）。

首先加载三个模型：

cat ./models/tfsimple1.yaml
echo "---"
cat ./models/tfsimple2.yaml
echo "---"
cat ./models/tfsimple3.yaml
seldon model load -f ./models/tfsimple1.yaml
seldon model load -f ./models/tfsimple2.yaml
seldon model load -f ./models/tfsimple3.yaml
seldon model status tfsimple1 -w ModelAvailable | jq -M .
seldon model status tfsimple2 -w ModelAvailable | jq -M .
seldon model status tfsimple3 -w ModelAvailable | jq -M .

模型定义与前面相同，只是多了一个 tfsimple3。Join Pipeline 定义如下：

cat ./pipelines/tfsimples-join.yaml

apiVersion: mlops.seldon.io/v1alpha1
kind: Pipeline
metadata:
  name: join
spec:
  steps:
    - name: tfsimple1
    - name: tfsimple2
    - name: tfsimple3
      inputs:
      - tfsimple1.outputs.OUTPUT0
      - tfsimple2.outputs.OUTPUT1
      tensorMap:
        tfsimple1.outputs.OUTPUT0: INPUT0
        tfsimple2.outputs.OUTPUT1: INPUT1
  output:
    steps:
    - tfsimple3

要点：tfsimple3 同时声明了两个 inputs，并通过 tensorMap 把 tfsimple1.outputs.OUTPUT0 改名为 INPUT0、把 tfsimple2.outputs.OUTPUT1 改名为 INPUT1，从而对齐 tfsimple3 期望的输入张量名。tfsimple1 与 tfsimple2 没有依赖关系，因此会并行执行，两个输出到达后在 tfsimple3 汇合。

seldon pipeline load -f ./pipelines/tfsimples-join.yaml
seldon pipeline status join -w PipelineReady | jq -M .
seldon pipeline infer join --inference-mode grpc \
    '{"model_name":"simple","inputs":[{"name":"INPUT0","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]},{"name":"INPUT1","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]}]}' | jq -M .

推理结果：tfsimple1 的 OUTPUT0 = INPUT0+INPUT1 = [2,4,...,32] 与 tfsimple2 的 OUTPUT1 = INPUT0-INPUT1 = [0,0,...] 作为 tfsimple3 的输入，最终 OUTPUT0 与 OUTPUT1 均为 [2,4,...,32]——文档指出 "The outputs are the sequence 2,4,6... which conforms to the logic of this model"，即结果与模型的加/减逻辑完全吻合。

seldon pipeline unload join
seldon model unload tfsimple1
seldon model unload tfsimple2
seldon model unload tfsimple3

从 Pipeline CRD 源码 看，当步骤声明多个 inputs 时，默认的 join 语义是 inputsJoinType: inner（+kubebuilder:default=inner），即必须等所有输入数据都到达后才会触发该步骤；JoinType 的合法取值与语义为：

• inner：必须等所有输入的数据都到达（数据齐备才触发）；
• outer：窗口结束时，把到达的任意输入数据都送出去；
• any：第一个到达的输入数据立即转发。

上述枚举定义与注释见 operator/apis/mlops/v1alpha1/pipeline_types.go 中 JoinTypeInner/JoinTypeOuter/JoinTypeAny 三个常量，调度器侧对应的 JoinInner/JoinOuter/JoinAny 定义在 scheduler/pkg/store/pipeline/pipeline.go。

Conditional：基于张量值触发条件分支

第四个场景展示条件数据流：根据 conditional 模型的 CHOICE 输入张量决定走 add10 还是 mul10 分支，两个分支在 Pipeline 输出端用 stepsJoin: any 汇聚。

加载三个模型

cat ./models/conditional.yaml
echo "---"
cat ./models/add10.yaml
echo "---"
cat ./models/mul10.yaml
seldon model load -f ./models/conditional.yaml
seldon model load -f ./models/add10.yaml
seldon model load -f ./models/mul10.yaml
seldon model status conditional -w ModelAvailable | jq -M .
seldon model status add10 -w ModelAvailable | jq -M .
seldon model status mul10 -w ModelAvailable | jq -M .

模型定义：

apiVersion: mlops.seldon.io/v1alpha1
kind: Model
metadata:
  name: conditional
spec:
  storageUri: "gs://seldon-models/scv2/samples/triton_23-03/conditional"
  requirements:
  - triton
  - python
---
apiVersion: mlops.seldon.io/v1alpha1
kind: Model
metadata:
  name: add10
spec:
  storageUri: "gs://seldon-models/scv2/samples/triton_23-03/add10"
  requirements:
  - triton
  - python
---
apiVersion: mlops.seldon.io/v1alpha1
kind: Model
metadata:
  name: mul10
spec:
  storageUri: "gs://seldon-models/scv2/samples/triton_23-03/mul10"
  requirements:
  - triton
  - python

定义条件 Pipeline

文档假设 conditional 模型能输出 OUTPUT0 与 OUTPUT1 两个张量，但只会输出其中一个：当 CHOICE 张量为 0 时输出 OUTPUT0，否则输出 OUTPUT1。因此下游两个模型中只有一个会收到数据并实际运行：

cat ./pipelines/conditional.yaml

apiVersion: mlops.seldon.io/v1alpha1
kind: Pipeline
metadata:
  name: tfsimple-conditional
spec:
  steps:
  - name: conditional
  - name: mul10
    inputs:
    - conditional.outputs.OUTPUT0
    tensorMap:
      conditional.outputs.OUTPUT0: INPUT
  - name: add10
    inputs:
    - conditional.outputs.OUTPUT1
    tensorMap:
      conditional.outputs.OUTPUT1: INPUT
  output:
    steps:
    - mul10
    - add10
    stepsJoin: any

这里的关键点：

• mul10 只订阅 conditional.outputs.OUTPUT0，add10 只订阅 conditional.outputs.OUTPUT1；由于 conditional 只会发出其中一个张量，另一个分支永远收不到数据、不会运行。
• 输出端 output.steps 同时列出 mul10 与 add10，但用 stepsJoin: any 声明"任意一个步骤的数据先到就先作为 Pipeline 输出"。由于本例中两个分支只有一个运行，any join 保证了结果能立即返回而不会死等。

seldon pipeline load -f ./pipelines/conditional.yaml
seldon pipeline status tfsimple-conditional -w PipelineReady | jq -M .

验证两个分支

CHOICE = 0 时走 mul10，输入 [1,2,3,4] 被乘以 10 得到 [10,20,30,40]：

seldon pipeline infer tfsimple-conditional --inference-mode grpc \
 '{"model_name":"conditional","inputs":[{"name":"CHOICE","contents":{"int_contents":[0]},"datatype":"INT32","shape":[1]},{"name":"INPUT0","contents":{"fp32_contents":[1,2,3,4]},"datatype":"FP32","shape":[4]},{"name":"INPUT1","contents":{"fp32_contents":[1,2,3,4]},"datatype":"FP32","shape":[4]}]}' | jq -M .

{
  "outputs": [
    {"name": "OUTPUT", "datatype": "FP32", "shape": ["4"], "contents": {"fp32Contents": [10,20,30,40]}}
  ]
}

CHOICE = 1 时走 add10，输入 [1,2,3,4] 被加上 10 得到 [11,12,13,14]：

seldon pipeline infer tfsimple-conditional --inference-mode grpc \
 '{"model_name":"conditional","inputs":[{"name":"CHOICE","contents":{"int_contents":[1]},"datatype":"INT32","shape":[1]},{"name":"INPUT0","contents":{"fp32_contents":[1,2,3,4]},"datatype":"FP32","shape":[4]},{"name":"INPUT1","contents":{"fp32_contents":[1,2,3,4]},"datatype":"FP32","shape":[4]}]}' | jq -M .

{
  "outputs": [
    {"name": "OUTPUT", "datatype": "FP32", "shape": ["4"], "contents": {"fp32Contents": [11,12,13,14]}}
  ]
}

注意：示例请求中还传入了 INPUT0、INPUT1 两个张量，conditional 模型的内部逻辑会结合 CHOICE 决定发出哪个输出分支。

seldon pipeline unload tfsimple-conditional
seldon model unload conditional
seldon model unload add10
seldon model unload mul10

从实现层面看，stepsJoin: any 会经由 scheduler/pkg/store/pipeline/utils.go 的 proto 转换映射为 scheduler.PipelineOutput_ANY，并由 dataflow 控制器在输出端创建对应的 join 逻辑（见 scheduler/pkg/kafka/dataflow/server.go 中 JoinInner/JoinOuter/JoinAny 的分支处理），这正是"先到先出"语义的来源。

Pipeline Input Tensors：在不同步骤中访问 Pipeline 的输入张量

第五个场景演示如何在任意步骤中直接引用 Pipeline 自身的输入张量。Pipeline 输入被建模为一个名为 <pipelineName>.inputs.<tensorName> 的虚拟数据源，任何步骤都可以订阅。

cat ./models/mul10.yaml
echo "---"
cat ./models/add10.yaml
seldon model load -f ./models/mul10.yaml
seldon model load -f ./models/add10.yaml
seldon model status mul10 -w ModelAvailable | jq -M .
seldon model status add10 -w ModelAvailable | jq -M .

Pipeline 定义：mul10 读取 Pipeline 输入的 INPUT0，add10 读取 Pipeline 输入的 INPUT1，两个步骤互不依赖、并行执行：

cat ./pipelines/pipeline-inputs.yaml

apiVersion: mlops.seldon.io/v1alpha1
kind: Pipeline
metadata:
  name: pipeline-inputs
spec:
  steps:
  - name: mul10
    inputs:
    - pipeline-inputs.inputs.INPUT0
    tensorMap:
      pipeline-inputs.inputs.INPUT0: INPUT
  - name: add10
    inputs:
    - pipeline-inputs.inputs.INPUT1
    tensorMap:
      pipeline-inputs.inputs.INPUT1: INPUT
  output:
    steps:
    - mul10
    - add10

这里 pipeline-inputs.inputs.INPUT0 的完整限定名语义为：Pipeline pipeline-inputs 的 inputs（入口张量集合）中的 INPUT0 张量。与 "Model Chaining from inputs" 一节中的 tfsimple1.inputs.INPUT0 相比，区别在于前者引用的是 Pipeline 的入口，后者引用的是某个模型步骤收到的输入。两者都通过 tensorMap 改名为下游模型的 INPUT 张量。

seldon pipeline load -f ./pipelines/pipeline-inputs.yaml
seldon pipeline status pipeline-inputs -w PipelineReady | jq -M .
seldon pipeline infer pipeline-inputs --inference-mode grpc \
    '{"model_name":"pipeline","inputs":[{"name":"INPUT0","contents":{"fp32_contents":[1,2,3,4]},"datatype":"FP32","shape":[4]},{"name":"INPUT1","contents":{"fp32_contents":[1,2,3,4]},"datatype":"FP32","shape":[4]}]}' | jq -M .

输出为两个并行的结果：mul10 对 INPUT0 乘 10 得到 [10,20,30,40]，add10 对 INPUT1 加 10 得到 [11,12,13,14]：

{
  "outputs": [
    {"name": "OUTPUT", "datatype": "FP32", "shape": ["4"], "contents": {"fp32Contents": [10,20,30,40]}},
    {"name": "OUTPUT", "datatype": "FP32", "shape": ["4"], "contents": {"fp32Contents": [11,12,13,14]}}
  ]
}

seldon pipeline unload pipeline-inputs
seldon model unload mul10
seldon model unload add10

Trigger Joins：用张量存在性控制步骤触发

最后一个场景展示 triggers（触发条件）：步骤的触发不仅可以由 inputs 的数据到达驱动，还可以由独立的 triggers 张量控制。这常用于"先满足某些前置条件，再执行推理"的模式。

模型与 Pipeline 定义

cat ./models/mul10.yaml
cat ./models/add10.yaml
seldon model load -f ./models/mul10.yaml
seldon model load -f ./models/add10.yaml
seldon model status mul10 -w ModelAvailable | jq -M .
seldon model status add10 -w ModelAvailable | jq -M .

Pipeline 定义（samples/pipelines/trigger-joins.yaml）：

cat ./pipelines/trigger-joins.yaml

apiVersion: mlops.seldon.io/v1alpha1
kind: Pipeline
metadata:
  name: trigger-joins
spec:
  steps:
  - name: mul10
    inputs:
    - trigger-joins.inputs.INPUT
    triggers:
    - trigger-joins.inputs.ok1
    - trigger-joins.inputs.ok2
    triggersJoinType: any
  - name: add10
    inputs:
    - trigger-joins.inputs.INPUT
    triggers:
    - trigger-joins.inputs.ok3
  output:
    steps:
    - mul10
    - add10
    stepsJoin: any

语义解读：

• mul10：除了要有 trigger-joins.inputs.INPUT 数据外，还要求 Pipeline 输入中存在 ok1 或 ok2 张量（triggersJoinType: any 表示二者满足其一即可触发）；
• add10：要求 Pipeline 输入中存在 ok3 张量才能触发（未指定 triggersJoinType，默认 inner，本例只有一个 trigger，等价于必须存在 ok3）；
• 输出端 stepsJoin: any：哪个分支先产出结果就返回哪个。

从 CRD 源码看，triggers 是 PipelineStep 的可选字段（Triggers []string json:"triggers,omitempty"），triggersJoinType 与 inputsJoinType 共用 JoinType 枚举（inner/outer/any），调度器在构建 dataflow 时会用 createTriggerSources 把 trigger 引用转换为对应的 Kafka topic 源并设置 TriggersJoinTy（见 scheduler/pkg/kafka/dataflow/server.go）。

seldon pipeline load -f ./pipelines/trigger-joins.yaml
seldon pipeline status trigger-joins -w PipelineReady | jq -M .

验证任意触发与精确触发

请求 1：只带 ok1（不含 ok3），mul10 的 any 触发被满足，add10 因缺少 ok3 不触发，输出 [10,20,30,40]：

seldon pipeline infer trigger-joins --inference-mode grpc \
    '{"model_name":"pipeline","inputs":[{"name":"ok1","contents":{"fp32_contents":[1]},"datatype":"FP32","shape":[1]},{"name":"INPUT","contents":{"fp32_contents":[1,2,3,4]},"datatype":"FP32","shape":[4]}]}' | jq -M .

{
  "outputs": [
    {"name": "OUTPUT", "datatype": "FP32", "shape": ["4"], "contents": {"fp32Contents": [10,20,30,40]}}
  ]
}

请求 2：带 ok3（不含 ok1/ok2），add10 触发，输出 [11,12,13,14]：

seldon pipeline infer trigger-joins --inference-mode grpc \
    '{"model_name":"pipeline","inputs":[{"name":"ok3","contents":{"fp32_contents":[1]},"datatype":"FP32","shape":[1]},{"name":"INPUT","contents":{"fp32_contents":[1,2,3,4]},"datatype":"FP32","shape":[4]}]}' | jq -M .

{
  "outputs": [
    {"name": "OUTPUT", "datatype": "FP32", "shape": ["4"], "contents": {"fp32Contents": [11,12,13,14]}}
  ]
}

seldon pipeline unload trigger-joins
seldon model unload mul10
seldon model unload add10

Pipeline 编排的源码级机制小结

通过上述六个示例，可以对 Seldon Core 2 Pipeline 的编排机制形成完整认识：

1. 声明式图模型：Pipeline 的 spec 由 input（可选外部输入）、steps（步骤列表）、output（可选同步输出）、dataflow（如 cleanTopicsOnDelete）以及 allowCycles/maxStepRevisits（环形 Pipeline 支持）组成，完整字段定义见 operator/apis/mlops/v1alpha1/pipeline_types.go 中的 PipelineSpec。

2. 张量寻址语法：数据引用统一采用 <来源>.inputs|outputs.<张量名> 的限定名格式，来源可以是其他步骤（如 tfsimple1.outputs.OUTPUT0），也可以是 Pipeline 自身入口（如 pipeline-inputs.inputs.INPUT0）。tensorMap 用于把来源张量改名为目标步骤期望的输入张量名。

3. Join 语义：inputsJoinType、triggersJoinType、stepsJoin 共用 inner/outer/any 三种语义，分别对应"全部数据齐备才触发""窗口结束送出已到数据""先到先出"。调度器内部的 JoinType 定义与 proto 映射见 scheduler/pkg/store/pipeline/pipeline.go 与 scheduler/pkg/store/pipeline/utils.go。

4. 触发与数据分离：triggers 把"步骤是否运行"与"步骤吃什么数据"解耦，可独立于 inputs 声明触发条件，适合实现门控、前置校验等复杂控制流。

5. 可观测性：seldon pipeline inspect 从 Kafka topic 读取各环节数据（<pipeline>.<step>.inputs/outputs），是排查多步骤数据流问题（张量改名错误、join 不触发、分支未命中）的核心手段。

更多 Pipeline 相关的 CRD 示例（如 tfsimple-combined-step、tfsimple-combined-trigger、version-test-a 等）可以在 samples/pipelines 目录中找到；与本文配套的可执行 Notebook 版本见 samples/pipeline-examples.ipynb。