流水线运行时与处理器

相关源码文件

以下文件被用作生成此 wiki 页面的上下文：

本页面详述 RoboFrame 中执行推理流水线的核心组件。它覆盖流水线工厂、构成流水线的阶段与处理器、管理其执行的执行器、抽象模型交互的统一运行时句柄，以及促进数据流动的编解码器与绑定。

流水线工厂

inference_service.pipeline.factory 模块负责构建和校验后端与流水线实例。它是创建 InferencePipeline 对象的中心点，后者封装给定模型部署的整个推理过程。

工厂使用 RuntimeAssemblerRegistry 管理各类模型会话与执行器的创建。它通过提供专用会话构建器支持不同后端类型，包括 LeRobot Torch、Ascend OM、HMM、RKNN 与 Hisilicon src/inference_service/inference_service/pipeline/factory.py:68-134。

关键函数与类：

工厂还处理为 PI0.5 与 SmolVLA 等模型创建专用执行器，推导其拓扑并加载配置 src/inference_service/inference_service/pipeline/factory.py:35-49。

来源：src/inference_service/inference_service/pipeline/factory.py:1-194

阶段

推理流水线由有序的 InferenceStage 对象组成，每个对象表示一个独立操作。这些阶段作用于持有请求数据与中间值的 StageFrame。

inference_service.pipeline.stages 模块定义这些阶段的接口与具体实现。

关键接口：

具体阶段实现：

迭代阶段执行

        graph TD
    A[IterativeStage.execute] --> B{Initialize State Adapter};
    B --> C{Loop through IterationStep};
    C --> D{"Prepare Step (IterationStateAdapter)"};
    D --> E{Execute Body Stages};
    E --> F{"Update State (IterationStateAdapter)"};
    F --> C;
    C -- Loop ends --> G{Finalize State Adapter};
    G --> H[Return];

Title: Iterative Stage Execution Flow 来源：src/inference_service/inference_service/pipeline/stages.py:78-107

来源：src/inference_service/inference_service/pipeline/stages.py:1-175

处理器

处理器负责推理前的输入观测变换与模型输出的后处理。对 LeRobot 模型，它们由 _ProcessorPair 及其视图管理。

inference_service.pipeline.processors 模块提供 create_lerobot_processor_views 函数，为已编译的 LeRobot 部署实例化前处理与后处理器 src/inference_service/inference_service/pipeline/processors.py:145-147。

关键类：

来源：src/inference_service/inference_service/pipeline/processors.py:1-147, src/inference_service/tests/test_pipeline_processors.py:17-60, src/inference_service/tests/test_pipeline_processors.py:63-88, src/inference_service/tests/test_pipeline_processors.py:90-121

执行器

执行器负责编排流水线内各阶段的执行。它们管理组件生命周期并处理错误上报。

inference_service.pipeline.executor 模块定义基类 ComponentModelExecutor 及其具体实现 SequentialModelExecutor。

关键类：

执行器数据流

        graph TD
    A[ModelRequest] --> B{SequentialModelExecutor.execute};
    B --> C[ExecutionContext];
    B --> D[StageFrame Creation];
    D -- Initial values --> E[StageFrame];
    E --> F{For each InferenceStage};
    F --> G["Stage.execute(frame, deadline)"];
    G -- Updates frame --> E;
    F -- All stages complete --> H{"ResultAdapter.adapt(frame)"};
    H --> I[Pipeline Result];
    G -- Error --> J{"ResultAdapter.adapt_error(error)"};
    J --> K[Pipeline Error];

Title: Sequential Model Executor Data Flow 来源：src/inference_service/inference_service/pipeline/executor.py:73-96

来源：src/inference_service/inference_service/pipeline/executor.py:1-167

统一运行时句柄

ModelRuntimeHandle 在底层模型运行时之上提供统一门面，抽象掉不同后端与执行策略的复杂性。它确保策略请求与结果遵循一致的契约。

inference_service.pipeline.runtime 模块定义 InferencePipeline 类，该类在本地执行中使用 ModelRuntimeHandle。

关键类：

来源：src/inference_service/inference_service/pipeline/runtime.py:1-182

编解码器与绑定

编解码器与绑定对于在应用使用的语义张量表示与模型后端期望的运行时专用格式之间进行转换至关重要。

inference_service.codecs.bindings 模块定义 manifest 驱动的转换如何发生。

关键类与函数：

执行计划

inference_service.codecs.execution 模块负责构建与校验有序的多模块执行计划。这对由多个需按特定顺序执行、可能在彼此间传递中间张量的子模型（角色）组成的模型尤为重要。

关键类与函数：

执行计划构建

        graph TD
    A[Start build_execution_plan] --> B{Validate role names and bindings};
    B -- Invalid --> E[ExecutionPlanError];
    B -- Valid --> C[Identify internal tensor producers and consumers];
    C --> D{Validate host and device links};
    D -- Inconsistent --> E;
    D -- Consistent --> F[Construct ExecutionPlan];
    F --> G[Return ExecutionPlan];

Title: Execution Plan Building Process 来源：src/inference_service/inference_service/codecs/execution.py:104-178

来源：src/inference_service/inference_service/codecs/bindings.py:1-167, src/inference_service/inference_service/codecs/execution.py:1-178