# ciel

**Repository Path**: kumaoko/ciel

## Basic Information

- **Project Name**: ciel
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-07-16
- **Last Updated**: 2026-07-16

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Trainflow

一个轻量训练框架适配层：用户保留已有训练函数或 shell 内容，只给函数增加
`@node` 描述，框架便会校验并组装训练 DAG。DAG 可编译成 Kedro Pipeline，实际运行时
每个节点动态注册为 Prefect task，整次训练由 MLflow 追踪。

## 已实现能力

- Kedro：`Pipeline.to_kedro()` 生成标准 `kedro.pipeline.Pipeline`。
- Prefect：`Pipeline.run()` 在运行时生成 flow 和 task；数据依赖由声明的输入输出决定，
  支持节点重试，不要求 Prefect 工程模板。
- MLflow：自动开启一个共享 run，启用支持框架的 `mlflow.autolog()`，自动记录传入的
  `params`、`hyperparameters` 或 `hparams`，捕获返回结果中的 `loss`、`*_loss`、
  `metrics`、`checkpoint*`，并扫描 checkpoint 目录。自定义训练循环也提供单行记录函数。
- 训练结构约束：阶段固定为 `load_data -> preprocess -> train -> evaluate -> export`；
  至少必须有 `load_data` 和 `train`。框架会检查重复输出、逆阶段依赖和环路。
- 少配置：不使用 YAML。连接信息只来自环境变量或一个 `Settings` 对象。
- 兼容旧代码：装饰器不会包装或改变函数行为；`shell_node()` 可以直接运行原 shell 命令，
  无需修改 shell 文件内容。

## 安装

需要 Python 3.10 或更高版本：

```bash
python3 -m venv .venv
. .venv/bin/activate
python3 -m pip install -e .
```

本项目声明了 Kedro、MLflow 和 Prefect 依赖。框架采用延迟导入，因此不安装这些依赖时，
仍可用 `run_local()` 或 CLI 的 `--local` 对 DAG 做快速调试。

## 最小改造

```python
from trainflow import metric, node

@node(stage="load_data", outputs="dataset")
def load_dataset():
    return existing_load_function()

@node(
    stage="train",
    inputs=("dataset", "hyperparameters"),
    outputs="train_result",
    retries=2,
)
def train(dataset, hyperparameters):
    result = existing_train_function(dataset, hyperparameters)
    # sklearn / PyTorch Lightning / Keras 等支持的训练接口由 MLflow 自动记录。
    # 只有自定义训练循环需要这一可选行：
    metric("loss", result.loss, step=result.epoch)
    return {"loss": result.loss, "model": result.model}
```

装饰器的 `inputs` 是函数参数名和上游数据名，`outputs` 是提供给下游节点的数据名。
多输出函数可返回与 `outputs` 同序的 tuple/list，或以输出名为 key 的 dict。

无需手写节点列表。CLI 会按模块中的源码声明发现所有 `@node`：

```bash
export MLFLOW_TRACKING_URI=http://mlflow:5000
export MLFLOW_EXPERIMENT_NAME=image-classifier
export PREFECT_API_URL=http://prefect:4200/api

trainflow run examples.basic_pipeline \
  --inputs '{"learning_rate": 0.01, "hyperparameters": {"batch_size": 32}}'
```

不连接服务的快速调试：

```bash
PYTHONPATH=src python3 -m trainflow.cli run examples.basic_pipeline \
  --local --inputs '{"learning_rate": 0.01}'
```

也可从 Python 调用：

```python
import examples.basic_pipeline as training
from trainflow import Pipeline

pipeline = Pipeline.discover(training, name="image-classifier")
kedro_pipeline = pipeline.to_kedro()
result = pipeline.run({"learning_rate": 0.01})
```

## 原有 shell 脚本

脚本内容不需要改，把原命令声明成节点即可：

```python
from trainflow import Pipeline, shell_node

load = shell_node("bash prepare_data.sh", stage="load_data", outputs="prepared")
train = shell_node(
    "bash train.sh", stage="train", inputs="prepared", outputs="exit_code", retries=1
)
Pipeline([load, train], name="legacy-training").run()
```

输入会同时以同名环境变量传入脚本。需要说明的是，如果仍直接执行 `bash train.sh`，进程并未
进入 Trainflow，因而不可能获得 Prefect/MLflow 能力；保持不变的是脚本内容，托管执行入口应改为
`trainflow run ...` 或上述 Python 入口。

## 配置项

| 环境变量 | 默认值 | 用途 |
| --- | --- | --- |
| `MLFLOW_TRACKING_URI` | `./mlruns` | MLflow 服务或本地存储地址 |
| `MLFLOW_EXPERIMENT_NAME` | `default` | 实验名称 |
| `PREFECT_API_URL` | 空（Prefect 默认） | Prefect Server/Cloud API 地址 |
| `TRAINFLOW_CHECKPOINT_DIR` | `./checkpoints` | 自动上传 checkpoint 的目录 |
| `TRAINFLOW_AUTOLOG` | `1` | 设为 `0` 关闭 MLflow autolog |

代码配置等价写法：

```python
from trainflow import Settings

settings = Settings(
    mlflow_tracking_uri="http://mlflow:5000",
    prefect_api_url="http://prefect:4200/api",
)
pipeline.run(settings=settings)
```

## 测试

基础逻辑只用标准库即可测试：

```bash
PYTHONPATH=src python3 -m unittest discover -s tests -v
```

完整的 Prefect、Kedro 和 MLflow 集成应在安装项目依赖后运行。