# AI Flow (Python)

基于 FastAPI 的 AI 工作流执行引擎，支持多种 AI 任务（图像处理、视频编辑、智能美颜等）。

## 项目结构

```
ai-flow-python/
├── app/
│   ├── main.py                    # FastAPI 应用入口
│   ├── config.py                  # 配置管理 (Pydantic Settings)
│   ├── controller/
│   │   └── flow_controller.py     # REST API 路由
│   ├── middleware/
│   │   └── api_key.py             # API Key 认证中间件
│   ├── repository/
│   │   ├── record_interface.py    # 存储抽象接口
│   │   ├── task_record.py         # 任务记录数据模型
│   │   ├── redis_store.py         # Redis 存储实现
│   │   └── local_store.py         # 本地文件存储实现
│   ├── service/
│   │   ├── engine/                # 工作流引擎
│   │   │   ├── workflow_engine.py
│   │   │   ├── execution_context.py
│   │   │   ├── parameter_resolver.py
│   │   │   └── workflow_result.py
│   │   ├── spec/                  # 工作流规范定义
│   │   │   ├── workflow.py
│   │   │   ├── step.py
│   │   │   └── workflow_loader.py
│   │   ├── task/                  # 任务执行系统
│   │   │   ├── base_task.py
│   │   │   ├── task_executor.py
│   │   │   ├── task_param.py
│   │   │   ├── task_result.py
│   │   │   └── impl/             # 15 个任务实现
│   │   └── template/
│   │       └── template_service.py
│   └── util/                      # 工具类
├── data/                          # 数据目录
├── resources/                     # 资源文件 (美颜/编辑器)
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
└── .env.example
```

## 快速开始

### 本地运行

```bash
# 安装依赖
pip install -r requirements.txt

# 复制环境变量配置
cp .env.example .env
# 编辑 .env 填入实际的 API Key

# 启动服务
uvicorn app.main:app --host 0.0.0.0 --port 8080 --reload
```

### Docker 运行

```bash
docker compose up -d
```

## API 接口

> 若启用了 API Key 认证（`AI_FLOW_API_AUTH_ENABLED=true`），除 `/health` 外所有接口需携带请求头：
> `Authorization: Bearer <your-api-key>`

### 接口总览

| 方法 | 路径 | 说明 |
|------|------|------|
| POST | /execute | 提交并执行工作流 |
| GET  | /result | 查询单个任务结果 |
| GET  | /tasks | 获取所有任务列表 |
| POST | /cleanup | 清理已完成/失败的过期任务 |
| GET  | /health | 健康检查 |
| GET  | /executor/status | 线程池状态 |
| GET  | /template/cache/status | 模板缓存状态 |
| POST | /template/cache/cleanup | 清理过期模板缓存 |

---

### POST /execute

提交工作流任务，异步执行。提交后立即返回 `taskId`，通过 `/result` 轮询结果或通过 Redis Pub/Sub（channel: `WORKFLOW_TASK`）订阅结果。

**请求体 (JSON)**

| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| template | string | 是 | 工作流模板地址（URL，支持 zip 包） |
| inputs | object | 否 | 工作流输入参数，键值对形式 |

**请求示例**

```bash
curl -X POST http://localhost:8080/execute \
  -H "Content-Type: application/json" \
  -d '{
    "template": "https://example.com/workflow-template.zip",
    "inputs": {
      "prompt": "a beautiful sunset",
      "image_url": "https://example.com/photo.jpg"
    }
  }'
```

**响应 — 成功**

```json
{
  "state": "successes",
  "taskId": "f47ac10b-58cc-4372-a567-0e02b2c3d479"
}
```

**响应 — 参数缺失**

```json
{
  "state": "failed",
  "message": "template parameter is required"
}
```

---

### GET /result

查询单个任务的执行结果。

**查询参数**

| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| taskId | string | 是 | 任务 ID（由 `/execute` 返回） |

**请求示例**

```bash
curl http://localhost:8080/result?taskId=f47ac10b-58cc-4372-a567-0e02b2c3d479
```

**响应 — 执行中**

```json
{
  "taskId": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
  "status": "processing",
  "message": "",
  "type": "workflow",
  "data": null,
  "request": {
    "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
    "inputs": {
      "images": [
        "https://aitori-jp.oss-accelerate.aliyuncs.com/img/upload/2026-03-12/f61e1be77e114c1e84b1dd9bdaa38e02.jpg"
      ]
    }
  },
  "createTime": 1773313868333,
  "updateTime": 1773313868500
}
```

**响应 — 成功**

```json
{
  "taskId": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
  "status": "successes",
  "message": "",
  "type": "workflow",
  "data": {
    "images": [
      "https://aitori-jp.oss-accelerate.aliyuncs.com/0000/yyyy-MM-dd/2db32a61-9859-4f36-94d8-76f19ca65515.png"
    ]
  },
  "request": {
    "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
    "inputs": {
      "images": [
        "https://aitori-jp.oss-accelerate.aliyuncs.com/img/upload/2026-03-12/f61e1be77e114c1e84b1dd9bdaa38e02.jpg"
      ]
    }
  },
  "createTime": 1773313868333,
  "updateTime": 1773313889665
}
```

**响应 — 失败**

```json
{
  "taskId": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
  "status": "failed",
  "message": "Failed to load workflow: template not found",
  "type": "workflow",
  "data": null,
  "request": {
    "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
    "inputs": {
      "images": [
        "https://aitori-jp.oss-accelerate.aliyuncs.com/img/upload/2026-03-12/f61e1be77e114c1e84b1dd9bdaa38e02.jpg"
      ]
    }
  },
  "createTime": 1773313868333,
  "updateTime": 1773313870000
}
```

**响应 — 任务不存在**

```json
{
  "error": "task not found"
}
```

**status 状态说明**

| 值 | 说明 |
|----|------|
| submitted | 已提交，等待执行 |
| processing | 执行中 |
| successes | 执行成功 |
| failed | 执行失败 |

---

### GET /tasks

获取所有任务列表。

**请求示例**

```bash
curl http://localhost:8080/tasks
```

**响应**

```json
{
  "total": 2,
  "tasks": {
    "68710045-e95c-4bd9-9103-6853cdcfd4fd": {
      "task_id": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
      "status": "successes",
      "message": "",
      "type": "workflow",
      "data": {
        "images": [
          "https://aitori-jp.oss-accelerate.aliyuncs.com/0000/yyyy-MM-dd/2db32a61-9859-4f36-94d8-76f19ca65515.png"
        ]
      },
      "request": {
        "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
        "inputs": { "images": ["https://..."] }
      },
      "create_time": 1773313868333,
      "update_time": 1773313889665
    },
    "a1b2c3d4-5e6f-7890-abcd-ef1234567890": {
      "task_id": "a1b2c3d4-5e6f-7890-abcd-ef1234567890",
      "status": "processing",
      "message": "",
      "type": "workflow",
      "data": null,
      "request": {
        "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/another-template.zip",
        "inputs": { "images": ["https://..."] }
      },
      "create_time": 1773314000000,
      "update_time": 1773314001000
    }
  }
}
```

---

### POST /cleanup

清理超过指定时间的已完成（successes）和失败（failed）任务记录。

**请求体 (JSON)**

| 字段 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| maxAgeHours | int | 否 | 24 | 保留时长（小时），超过此时间的终态任务将被删除 |

**请求示例**

```bash
curl -X POST http://localhost:8080/cleanup \
  -H "Content-Type: application/json" \
  -d '{"maxAgeHours": 48}'
```

**响应**

```json
{
  "message": "Cleanup completed",
  "maxAgeHours": 48
}
```

---

### GET /health

健康检查接口，无需认证。

**请求示例**

```bash
curl http://localhost:8080/health
```

**响应**

```json
{
  "status": "UP",
  "service": "ai-flow",
  "timestamp": 1740600000000
}
```

---

### GET /executor/status

查看工作流线程池状态。

**请求示例**

```bash
curl http://localhost:8080/executor/status
```

**响应**

```json
{
  "maxWorkers": 8,
  "isShutdown": false
}
```

---

### GET /template/cache/status

查看模板缓存统计信息。

**请求示例**

```bash
curl http://localhost:8080/template/cache/status
```

**响应**

```json
{
  "cacheEnabled": true,
  "cacheDir": "data/templates",
  "cachedTemplates": 3,
  "cacheSizeBytes": 52428800,
  "cacheSizeMB": 50,
  "templateDirectories": 3
}
```

---

### POST /template/cache/cleanup

清理过期的模板缓存目录。

**请求体 (JSON)**

| 字段 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| maxAgeHours | int | 否 | 24 | 保留时长（小时），超过此时间的模板缓存将被删除 |

**请求示例**

```bash
curl -X POST http://localhost:8080/template/cache/cleanup \
  -H "Content-Type: application/json" \
  -d '{"maxAgeHours": 72}'
```

**响应**

```json
{
  "message": "Template cache cleanup completed",
  "maxAgeHours": 72
}
```

---

### Redis Pub/Sub 事件通知

当任务到达终态（`successes` / `failed`）时，若使用 Redis 存储后端且启用了发布功能，会自动将任务记录发布到 Redis channel。

| 配置项 | 默认值 | 说明 |
|--------|--------|------|
| AI_FLOW_REDIS_PUBLISH_ENABLED | true | 是否启用发布 |
| AI_FLOW_REDIS_PUBLISH_CHANNEL | WORKFLOW_TASK | channel 名称 |

**消息格式（JSON）**

与 `/result` 接口返回的 `TaskRecord` 结构一致：

```json
{
  "task_id": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
  "status": "successes",
  "message": "",
  "type": "workflow",
  "data": {
    "images": [
      "https://aitori-jp.oss-accelerate.aliyuncs.com/0000/yyyy-MM-dd/2db32a61-9859-4f36-94d8-76f19ca65515.png"
    ]
  },
  "request": {
    "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
    "inputs": { "images": ["https://..."] }
  },
  "create_time": 1773313868333,
  "update_time": 1773313889665
}
```

**订阅示例（Python）**

```python
import redis, json

r = redis.Redis(host="10.146.0.3", port=6379, password="xxx", decode_responses=True)
pubsub = r.pubsub()
pubsub.subscribe("WORKFLOW_TASK")

for message in pubsub.listen():
    if message["type"] == "message":
        record = json.loads(message["data"])
        print(f"Task {record['task_id']} -> {record['status']}")
```

## 支持的任务类型

| 类型 | 说明 |
|------|------|
| kontext_pro | Flux Kontext Pro AI 图像生成 |
| kontext_max | Flux Kontext Max AI 图像生成 |
| keling_image2video | 可灵图生视频 |
| keling_video2effect | 可灵视频特效 |
| nano_banana | Gemini 图像生成 |
| google_veo3 | Google Veo3 视频生成 |
| seedance_video | SeedDance 视频生成 |
| seedream_image | SeedDream 图像生成 |
| smart_beauty | 智能美颜 |
| image_editor | 图像编辑 |
| video_editor | 视频编辑 |
| file_download | 文件下载 |
| file_upload | 文件上传 (阿里云 OSS / Google Cloud Storage) |
| base64_encode | Base64 编码 |
| base64_decode | Base64 解码 |

## 配置说明

所有配置通过环境变量注入，前缀为 `AI_FLOW_`。详见 `.env.example`。