Metadata-Version: 2.4
Name: ei-pipe-sdk
Version: 0.2.8
Summary: Standalone SDK for ei-pipeline executors: ei_infer (model inference) and ei_ois (OIS object-storage transfer) workers
Requires-Python: <3.15,>=3.11
Description-Content-Type: text/markdown
Requires-Dist: arq>=0.26
Requires-Dist: redis>=8.0.1
Requires-Dist: structlog>=25.4.0
Requires-Dist: ois3-sdk-python>=3.2.31
Requires-Dist: pyyaml>=6.0.2
Provides-Extra: profile
Requires-Dist: fastapi>=0.115.0; extra == "profile"
Requires-Dist: memray>=1.14; extra == "profile"
Requires-Dist: py-spy>=0.4.0; extra == "profile"

# ei-pipe-sdk

面向算法同学的极简模型接入 SDK。把你的模型写成一个 `EiInfer` 的两个方法，ei-pipeline
就能调度它推理：**你不用关心数据怎么进来、结果怎么提交，也不用关心部署**。

## 安装

```bash
pip install ei-pipe-sdk
# 或先在仓库 ei_pipe_sdk/ 下本地构建
uv build && pip install dist/ei_pipe_sdk-*.whl
```

## 三步接入

```python
# my_model.py
from ei_infer import EiInfer, register


class MyModel(EiInfer):
    def init(self, model_cfg: dict | None = None) -> None:
        # 进程启动时调用一次：加载权重、tokenizer、选设备。失败会让 worker 启动即崩溃。
        self.model = load_weights(...)

    def inference(self, input_path: str, output_path: str, node_info: dict) -> dict:
        # input_path / output_path: worker 已恢复好的绝对路径
        #   = {data_root}/{service_id}/buckets/{time_bucket}/{pipeline_id}/<相对路径>
        # input_path 是输入（文件或目录）；output_path 是允许写结果文件的目录（可能为空）。
        # 是否落盘由模型自己决定；返回的 dict 会被 worker 回传给 pipeline。
        return {"detections": self.model(open(input_path))}


register("my-model", MyModel())   # 注册名 = 管道节点 params.name 的值
```

```bash
pip install ei-pipe-sdk
```

```python
# worker 入口：参数经构造参数显式给定（也可经配置文件，见下文），
# SDK 不读任何环境变量。每个注册模型各消费一条自己的队列。
from ei_infer import InferNode

node = InferNode(
    redis_url="redis://scheduler:6379/0",
    service_id="3fa2b1c9",  # 调度服务 h8；或经配置文件给 scheduler_url 解析
    data_root="/lpai/pvc/ei-autolabel-bd-ga-infer",  # PFS 挂载点裸路径
)
node.run()
```

## 协议

调度端（ei-pipeline 的 `ei_infer` 节点）会按统一协议给你发任务：

| 字段 | 类型 | 说明 |
|---|---|---|
| `name` | str | 模型注册名，由你 `register("name", ...)` 时命名 |
| `input_path` | str | 输入路径（相对 pipe 共享目录，如 `data/`），worker 恢复为绝对路径 |
| `output_path` | str? | 允许写结果文件的目录（相对 pipe 共享目录，可选） |
| `version` | str? | 版本提示，透传保留 |

worker 用 `node_info.service_id + time_bucket + pipeline_id` 与 `InferNode`
配置的 `data_root` 把 `input_path` / `output_path` 恢复成绝对路径
（`{data_root}/{service_id}/buckets/{time_bucket}/{pipeline_id}/<相对路径>`），
调用 `inference(input_path, output_path, node_info)`。
**是否把结果落盘由模型自己决定**，worker 不替模型落盘，只把模型返回的 dict 原样回传。
整个链路里**你的函数只看绝对路径 input_path / output_path 和 node_info**，其余（数据就位、
并发、轮询）全部由 SDK 兜底。

## 配置

参数一律显式给定（`InferNode` 构造参数，或经配置文件 + `build_node`），
**SDK 不读任何环境变量**。

构造参数 / 配置文件字段：

| 字段 | 默认 | 说明 |
|---|---|---|
| `redis_url` | 必填 | 消费队列的 Redis（与调度服务同一套） |
| `service_id` | 二选一 | 调度服务的 service_id（h8），静态给定时优先生效 |
| `scheduler_url` | 二选一 | 未给 `service_id` 时，向该调度请求 `GET /api/status/service-id` 解析（不可达每分钟重试） |
| `data_root` | 必填 | PFS 挂载点裸路径；构建时自动并入 `{service_id}/buckets` 隔离段 |
| `max_jobs` | `4` | 每模型队列的并发任务数（arq `max_jobs`） |
| `max_threads` | `max(4, max_jobs)` | 跑同步推理的线程池大小（仅构造参数） |

配置文件（yaml/json，`{{变量名}}` 占位由 `secrets` 映射在载入时注入）::

    service_id: '{{service_id}}'
    # scheduler_url: http://scheduler:8000
    redis_url: redis://localhost:6379/0
    data_root: /lpai/pvc/ei-autolabel-bd-ga-infer
    max_jobs: 4

```python
from ei_infer.worker import build_node

node = build_node("infer.yaml", secrets={"service_id": "3fa2b1c9"})
node.run()
```

完整示例见 `example/`（`infer_exa.py` + `run.sh`）。示例只需三个变量
（redis、data_root、service_id/scheduler_url），直接构造 `InferNode`，
不走配置文件；配置文件方式用于生产部署。

## 队列

每个注册模型一条队列，统一文法（详见 ei-pipeline `docs/arq_queue_naming.md`）：

```
arq:ei:{service_id}:ei_infer:{model_name}
```

worker 进程为每个注册模型起一个 arq Worker 消费对应队列；`service_id` 与
调度侧表名/Redis 前缀同源，隔离共用一套 Redis 的多个调度服务。

## 多模型

多个模型可在同一个 worker 进程里 `register` 多个名字；进程会为每个模型各起
一个 arq Worker，各自消费自己的队列，互不抢任务。

## 失败模式

- `inference` 返回非 dict → 任务失败（错误信息含模型名）。
- `name` 未注册 → 任务失败，节点错误信息会指出未知模型名。
- `init` 抛异常 → worker 启动即退出，方便在你自己的日志里早发现。

## 本地联调

```bash
# 起一个本地 redis 后（redis-server），在 example/ 下：
bash run.sh   # infer_exa.py（前台）+ ois_exa.py（后台）
# 每个注册模型一条队列：arq:ei:{service_id}:ei_infer:{model_name}
```

然后在 ei-pipeline 用 `ei_infer` 节点提交一条任务即可走通。
