企业官网建设流程全解析

• 人工智能
• 大模型
• MLOps
• LLMOps
• 模型推理服务
• 云原生
• 后端

本文基于 docs-gb/examples/pipeline-cyclic.md 展开，讲解如何在 Seldon Core 2 中部署一条"循环流水线"（Cyclic Pipeline）：从用户自定义的起始值开始逐次自增、直到达到 10 为止的计数器示例，并深入到 Operator CRD、Scheduler 校验与 Dataflow（Kafka Streams）层的源码实现，说明allowCycles、maxStepRevisits、joinWindowMs与 trigger 张量在循环控制流中的实际作用。读完本文，你可以掌握循环 Pipeline 的完整部署流程，并理解 Seldon Core 2 如何防止无限循环。

循环 Pipeline 概念与示例

常规推理流水线是 DAG（有向无环图）：数据从 input 单向流向 output。但在 agentic workflow、条件重放、状态累积等场景中，控制流需要"回到"前面的步骤——即出现环（cycle）。Seldon Core 2 通过 Pipeline CRD 中的两个字段支持这种能力：

• allowCycles: true：声明本 Pipeline 允许存在环，跳过 Scheduler 的无环校验；
• maxStepRevisits：单个请求内某一步骤可被重复访问的最大次数，用于兜底防止无限循环。

本文的示例是一个简单计数器：以用户指定的起始值开始，每轮经过 counter 模型加 1，只要未达 10 就通过反馈回路再次进入 counter，达到 10 后终止并输出最终结果。若起始值已大于 10，Pipeline 会立即终止而不进入循环。

准备工作

按 docs-gb/examples/pipeline-cyclic.md 的前置要求：

1. 确认已在命名空间seldon-mesh中完成 Seldon Core 2 安装；
2. 在包含示例清单的 samples 目录中执行后续命令（示例模型代码上传于远端gs://seldon-models/scv2/examples/cyclic-pipeline/存储路径，清单中直接引用该storageUri）；
3. 获取 Istio 入口 IP（将<INGRESS_IP>替换为实际值）：

ISTIO_INGRESS=$(kubectl get svc seldon-mesh -n seldon-mesh -o jsonpath='{.status.loadBalancer.ingress[0].ip}') echo "Seldon Core 2: http://$ISTIO_INGRESS"

提示：记录输出的 IP 地址，后续所有请求命令中的<INGRESS_IP>都用它替换。

模型实现

Counter 模型：产生数据与 trigger 两个输出张量

第一个模型是一个简单的计数器（MLServer 实现）。它输出两个张量：第一个是加 1 后的数值，第二个是一个内容为空、但名字为continue或stop的张量——这个"空"张量就是 trigger（触发张量），它不带数据、只带语义，决定数据流向反馈回路还是输出路径。关于 trigger 张量的概念可参考 Pipelines 入门文档。

from mlserver.model import MLModel from mlserver.codecs import NumpyCodec, StringCodec from mlserver.types import InferenceRequest, InferenceResponse from mlserver.logging import logger class Counter(MLModel): async def load(self) -> bool: self.ready = True return self.ready async def predict(self, payload: InferenceRequest) -> InferenceResponse: x = NumpyCodec.decode_input(payload.inputs[0]) + 1 message = "continue" if x.item() < 10 else "stop" return InferenceResponse( model_name=self.name, model_version=self.version, outputs=[ NumpyCodec.encode_output( name="output", payload=x ), StringCodec.encode_output( name=message, payload=[""] ), ] )

从源码结构看，trigger 的本质只是"输出张量的名字"：Pipeline 步骤声明triggers: [counter.outputs.stop]时，Scheduler 会按步骤名.输出张量名解析出触发源，由 Dataflow 侧的 trigger 变换拓扑（见 TriggerTransforms.kt）用 join 窗口判断触发条件是否满足，再决定是否放行数据流。

IdentityModel 模型：透传并引入微小延迟

第二个模型是恒等（identity）模型，把输入张量原样透传到输出，但引入一个可配置的delay（秒）。这个延迟对防止无限循环至关重要——因为 Kafka Streams 的 join 窗口行为可能导致不同轮次的消息被错误关联（更多细节可参考 Apache Kafka 官方文档）。

import asyncio from mlserver.logging import logger from mlserver import MLModel, ModelSettings from mlserver.types import ( InferenceRequest, InferenceResponse, ResponseOutput ) class IdentityModel(MLModel): def __init__(self, settings: ModelSettings): super().__init__(settings) self.params = settings.parameters self.extra = self.params.extra if self.params is not None else None self.delay = self.extra.get("delay", 0) async def load(self) -> bool: self.ready = True return self.ready async def predict(self, payload: InferenceRequest) -> InferenceResponse: if self.delay: await asyncio.sleep(self.delay) return InferenceResponse( model_name=self.name, model_version=self.version, outputs=[ ResponseOutput( name=request_input.name, shape=request_input.shape, datatype=request_input.datatype, parameters=request_input.parameters, data=request_input.data ) for request_input in payload.inputs ] )

Pipeline 拓扑

该计数器 Pipeline 由三个模型步骤构成：counter、用于反馈回路的 identity-loop，以及用于输出的 identity-output。拓扑结构如下（counter 依据 trigger 张量把数据分别导向 identity-loop 或 identity-output）：

注意图中的环：counter -> identity-loop -> counter。counter 步骤有两个输入（外部 pipeline 输入与 identity-loop 输出），因此使用inputsJoinType: any——从 CRD 定义看（pipeline_types.go），any表示"任一输入先到即转发"，正好匹配"首次请求走外部输入、后续轮次走回路输入"的语义。

部署模型

三个模型的model-settings.json分别如下。

counter：

{ "name": "counter", "implementation": "model.Counter", "parameters": { "version": "v0.1.0" } }

identity-loop（注意extra.delay为 0.001 秒，即 1 毫秒）：

{ "name": "identity-loop", "implementation": "model.IdentityModel", "parameters": { "version": "v0.1.0", "extra": { "delay": 0.001 } } }

这 1 毫秒延迟是防止无限循环的关键之一，它与 counter 模型对所有输入类型应用的 join 窗口、以及 Pipeline 定义中 identity 模型配置的 join 窗口（joinWindowMs: 1）对齐。

identity-output（无延迟）：

{ "name": "identity-output", "implementation": "model.IdentityModel", "parameters": { "version": "v0.1.0" } }

对应的三个 Model CRD 清单（./models/counter.yaml、./models/identity-loop.yaml、./models/identity-output.yaml）：

apiVersion: mlops.seldon.io/v1alpha1 kind: Model metadata: name: counter spec: storageUri: "gs://seldon-models/scv2/examples/cyclic-pipeline/counter" requirements: - mlserver

apiVersion: mlops.seldon.io/v1alpha1 kind: Model metadata: name: identity-loop spec: storageUri: "gs://seldon-models/scv2/examples/cyclic-pipeline/identity-loop" requirements: - mlserver

apiVersion: mlops.seldon.io/v1alpha1 kind: Model metadata: name: identity-output spec: storageUri: "gs://seldon-models/scv2/examples/cyclic-pipeline/identity-output" requirements: - mlserver

用如下命令部署并等待就绪：

kubectl apply -f - --namespace=seldon-mesh <<EOF apiVersion: mlops.seldon.io/v1alpha1 kind: Model metadata: name: counter spec: storageUri: "gs://seldon-models/scv2/examples/cyclic-pipeline/counter" requirements: - mlserver EOF

kubectl apply -f - --namespace=seldon-mesh <<EOF apiVersion: mlops.seldon.io/v1alpha1 kind: Model metadata: name: identity-loop spec: storageUri: "gs://seldon-models/scv2/examples/cyclic-pipeline/identity-loop" requirements: - mlserver EOF

kubectl apply -f - --namespace=seldon-mesh <<EOF apiVersion: mlops.seldon.io/v1alpha1 kind: Model metadata: name: identity-output spec: storageUri: "gs://seldon-models/scv2/examples/cyclic-pipeline/identity-output" requirements: - mlserver EOF

kubectl wait --for condition=ready --timeout=300s model --all -n seldon-mesh

预期输出：

model.mlops.seldon.io/counter condition met model.mlops.seldon.io/identity-loop condition met model.mlops.seldon.io/identity-output condition met

Pipeline 清单：循环控制参数详解

模型就绪后部署 Pipeline，清单如下：

apiVersion: mlops.seldon.io/v1alpha1 kind: Pipeline metadata: name: counter-pipeline spec: allowCycles: true maxStepRevisits: 100 steps: - name: counter inputsJoinType: any inputs: - counter-pipeline.inputs - identity-loop.outputs - name: identity-output joinWindowMs: 1 inputs: - counter.outputs triggers: - counter.outputs.stop - name: identity-loop joinWindowMs: 1 inputs: - counter.outputs.output triggers: - counter.outputs.continue output: steps: - identity-output.outputs

关键参数说明（字段定义见 pipeline_types.go）：

joinWindowMs的生效逻辑可以从 Dataflow 源码确认：在 PipelineStep.kt 中，若步骤的joinWindowMillis非 0，会用它copy覆盖全局KafkaDomainParams.joinWindowMillis，因此本文清单中两个 identity 步骤的 1ms 窗口只作用于自身步骤。

部署命令与预期输出：

kubectl create -f counter-pipeline.yaml -n seldon-mesh

pipeline.mlops.seldon.io/counter-pipeline created

源码级原理：allowCycles 与 maxStepRevisits 的完整链路

1. CRD 到 Scheduler 的传递。Operator 将 Pipeline CRD 转换为 Scheduler 的内部表示时，会原样带上这两个字段（pipeline_types.go 的AsSchedulerPipeline）；对应 protobuf 定义见 scheduler.proto 与 chainer.proto。

2. Scheduler 侧的环检测开关。Scheduler 收到 Pipeline 后先做结构校验，环检测在 validate.go：checkForCycles从每个步骤出发做 DFS，一旦在visited集合中再次遇到已访问步骤即返回PipelineCycleErr；但当pv.AllowCycles为true时直接跳过整个环检测。也就是说，allowCycles: true是循环 Pipeline 能通过部署校验的前提。

3. Dataflow 侧的拓扑构建。Kafka Streams 拓扑构建在 Pipeline.kt：当metadata.allowCycles为真时，会额外注册一个内存键值状态存储visiting-counter-store，用于按请求ID:输出topic复合键记录每步的访问次数。

4. 访问计数器与错误分支。每个数据步骤（单源的 Chainer.kt 或多源 join 的 Joiner.kt）在allowCycles开启时都会挂上VisitingCounterProcessor，并把输出分流为两条支路：

// Processor.kt：VisitingCounterProcessor.process 核心逻辑 val newCount = (visitingCounterStore.get(compositeKey) ?: 0) if (newCount > maxStepRevisits) { // 写入错误消息并转发到 pipelineErrorTopic } else { context.forward(record) } visitingCounterStore.put(compositeKey, newCount + 1)

见 Processor.kt。超限时会生成ERROR: Max step revisits (N) exceeded for request <id> in topic <topic>消息，经由createVisitingCounterBranches的错误支路写入该 Pipeline 的 error topic，从而终止该请求在环中的继续流转。从实现看，计数器在请求记录到达 pipeline 输出 topic 时会被清理（同一文件 L45-L57），因此计数是"每请求"粒度而非全局的。

5. join 窗口与 trigger 的协作。Joiner对三种 join 类型使用不同窗口（Joiner.kt）：inner/outer使用JoinWindows.ofTimeDifferenceWithNoGrace(joinWindowMillis)，而any采用JoinWindows.of(Duration.ofMillis(1))的outerJoin链式合并（代码注释中引用了 KAFKA-10847 等上游行为）。这正是示例中要求 feedback 回路带 1ms 延迟、且步骤joinWindowMs也设为 1ms 的原因：让"下一轮进入 counter 的消息"落在窗口之外，避免相邻迭代被同一 join 窗口合并，保证计数逐次推进而不会跳变或死锁。

补充：maxStepRevisits的计数语义是"超过该值即报错"。测试用例中的默认值为allowCycles = false、maxStepRevisits = 0（PipelineStepTest.kt），即不允许任何重复访问；本文示例设为 100，与"计数器最多循环约 10 次即终止"的确定性行为留出了充足余量。对于控制流由 LLM 等外部因素决定、可能无法保证终止的 agentic workflow，这个参数尤其重要。

测试 Pipeline

向 pipeline 发送推理请求（起始值为 0），Pipeline 会逐步自增直到 10 并终止：

curl 方式：

curl -k http://<INGRESS_IP>:80/v2/models/counter-pipeline/infer \ -H "Host: seldon-mesh.inference.seldon" \ -H "Seldon-Model: counter-pipeline.pipeline" \ -H "Content-Type: application/json" \ -d '{ "inputs": [ { "name": "counter-pipeline.inputs", "datatype": "INT32", "shape": [1], "data": [0] } ] }' | jq -M .

seldon-cli 方式：

seldon pipeline infer counter-pipeline --inference-host <INGRESS_IP>:80\ '{"inputs":[{"name":"counter-pipeline.inputs","shape":[1],"datatype":"INT32","data":[0]}]}' | jq -M .

两种方式的响应一致：最终计数器值为 10，且带有stop标记，表明 Pipeline 已终止：

{ "model_name": "", "outputs": [ { "data": [ 10 ], "name": "output", "shape": [ 1, 1 ], "datatype": "INT32", "parameters": { "content_type": "np" } }, { "data": [ "" ], "name": "stop", "shape": [ 1, 1 ], "datatype": "BYTES", "parameters": { "content_type": "str" } } ] }

清理资源

删除 Pipeline 与三个模型：

kubectl delete pipeline counter-pipeline -n seldon-mesh

pipeline.mlops.seldon.io "counter-pipeline" deleted

kubectl delete model identity-loop -n seldon-mesh kubectl delete model identity-output -n seldon-mesh kubectl delete model counter -n seldon-mesh

model.mlops.seldon.io "identity-loop" deleted model.mlops.seldon.io "identity-output" deleted model.mlops.seldon.io "counter" deleted

小结

循环 Pipeline 在 Seldon Core 2 中并非概念上的支持，而是一条从 CRD、Scheduler 校验到 Kafka Streams 拓扑的完整链路：allowCycles关闭环检测并启用访问计数状态存储，maxStepRevisits通过VisitingCounterProcessor在超限时将请求转入错误 topic，joinWindowMs与 feedback 回路的小延迟则共同保证各轮次消息不被 join 窗口错误合并。掌握这三者的配合，你就能在 Core 2 上安全地构建带反馈回路的有状态工作流，例如本示例的计数器、条件重放乃至由 LLM 决定控制流的 agentic 场景。

• 人工智能
• 大模型
• MLOps
• LLMOps
• 模型推理服务
• 云原生
• 后端