workflow

ChenYu 1b7c7c73ba 修改超时时间		5 달 전
.idea	32701f63e8 first commit	7 달 전
app	5e11843548 Remove tracked Python bytecode artifacts	5 달 전
data	8b30848a6c 保留 gcs-credentials.json 的 git 跟踪	7 달 전
resources	32701f63e8 first commit	7 달 전
.env	1b7c7c73ba 修改超时时间	5 달 전
.env.example	327bb7b121 Improve workflow timeout handling and runtime safety	5 달 전
.gitignore	4efe418b1e 添加 *.tar.gz 到 .gitignore	7 달 전
Dockerfile	32701f63e8 first commit	7 달 전
README.md	87866f8a95 错误的时候添加了更多的信息	7 달 전
build.sh	0a64907712 修改build.sh	7 달 전
docker-compose.yml	32701f63e8 first commit	7 달 전
requirements.txt	32701f63e8 first commit	7 달 전

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

快速开始

本地运行

# 安装依赖
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 运行

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	否	工作流输入参数，键值对形式

请求示例

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"
    }
  }'

响应 — 成功

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

响应 — 参数缺失

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

GET /result

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

查询参数

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

请求示例

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

响应 — 执行中

{
  "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
}

响应 — 成功

{
  "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
}

响应 — 失败

{
  "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
}

响应 — 任务不存在

{
  "error": "task not found"
}

status 状态说明

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

GET /tasks

获取所有任务列表。

请求示例

curl http://localhost:8080/tasks

响应

{
  "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	保留时长（小时），超过此时间的终态任务将被删除

请求示例

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

响应

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

GET /health

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

请求示例

curl http://localhost:8080/health

响应

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

GET /executor/status

查看工作流线程池状态。

请求示例

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

响应

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

GET /template/cache/status

查看模板缓存统计信息。

请求示例

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

响应

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

POST /template/cache/cleanup

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

请求体 (JSON)

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

请求示例

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

响应

{
  "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 结构一致：

{
  "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）

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。

README.md

AI Flow (Python)

项目结构

快速开始

本地运行

Docker 运行

API 接口

接口总览

POST /execute

GET /result

GET /tasks

POST /cleanup

GET /health

GET /executor/status

GET /template/cache/status

POST /template/cache/cleanup

Redis Pub/Sub 事件通知

支持的任务类型

配置说明