Python SDK 接入指南

智作工坊 SpeedPix Python SDK，提供简洁易用的 AI 图像生成和处理工作流接口。

查看 API 文档

• API概述

授权

获取 API 凭据

1. 创建应用

2. 创建应用获取 app_key 和 app_secret

调用 OpenAPI

准备 Python 环境

# 确保 Python 3.8+
python --version

# 创建虚拟环境（可选）
python -m venv venv
source venv/bin/activate  # macOS/Linux

配置环境变量

# 设置 API 凭据（推荐方式）
export SPEEDPIX_APP_KEY="your-app-key"
export SPEEDPIX_APP_SECRET="your-app-secret"

export SPEEDPIX_ENDPOINT="https://your-endpoint.com"  # 可选

安装依赖

# 使用 pip 安装
pip install speedpix

# 或使用 uv（推荐）
uv add speedpix

准备一个已经发布的工作流

在智作工坊控制台中：

参考 工作流管理

1. 创建或选择一个 AI 工作流，可以将以下示例工作流拖入工作流编辑器

   1. 这是一个将输入商品图变成白底图的工作流 ，SDK 测试案例.json该工作流会自动从用户输入图片抽取商品主体，然后根据用户输入文本生成商品白底图

   2. 示范参数：

      输入文本	输入图片	输出结果
      测试文本

2. 发布工作流并获取 workflow_id

   1. 发布时选择 #1 号节点的 image 字段以及 #6 号节点的 text 作为输入字段

      

   2. 选择 #10 号节点的 images 作为输出字段

      

   3. 点击提交发布，此时会产生一个 v1 版本，点击右侧别名管理，新建一个 main 别名，映射到 v1 版本，后续我们可以使用该别名调用该工作流，如果需要更新版本，只用在这里切换别名即可，不用代码层变更。

      

3. 记录工作流的输入、输出格式要求

复制示例代码

最简单方式（推荐）

from speedpix import Client

# 初始化客户端（自动从环境变量读取配置）
client = Client()

# 运行 AI 工作流

try:
    output = client.run(
        workflow_id="your_workflow_id",
        input={
            "text": "一幅美丽的山水画",
            "image": "./input.jpg",  # 本地文件路径，SDK会自动上传并转换为URL
        },
        alias_id="main",  # 版本别名，默认为 "main"
        # return_temp_files=True,  # 是否返回临时文件
    )
    output['images']['url'].save("output.png")  # 保存输出图片
    # output['subject_image']['url'].save("subject_image.png")  # 保存输出图片
    # output['mask_image']['url'].save("mask_image.png")  # 保存输出图片
except Exception as e:
    print("发生错误:", e)
    output = None

异步方式

import asyncio
from speedpix import Client


async def main():
    client = Client()  # 自动从环境变量读取配置

    # 异步运行工作流
    try:
        output = await client.async_run(
            workflow_id="your_workflow_id",
            input={
                "text": "一幅美丽的山水画",
                "image": "./input.jpg",  # 本地文件路径，SDK会自动上传并转换为URL
            },
            alias_id="main",
        )

        await output["images"]["url"].async_save("async_result.png")
    except Exception as e:
        print("发生错误:", e)
        output = None


asyncio.run(main())

全局函数方式

import speedpix

try:
    # 使用全局函数（需要先设置环境变量）
    output = speedpix.run(
        workflow_id="your_workflow_id",
        input={
            "text": "一幅美丽的山水画",
            "image": "./input.jpg",  # 本地文件路径，SDK会自动上传并转换为URL
        },
        alias_id="main",
    )

    output["images"]["url"].save("global_output.png")  # 保存输出图片
except Exception as e:
    print("发生错误:", e)
    output = None

执行测试

# 运行测试脚本
python quickstart.py

# 验证安装
python -c "import speedpix; print('SpeedPix SDK 安装成功！')"

# 检查生成的文件
ls -la result.png

资源配置

共享算力 vs 独享资源

智作工坊支持两种资源类型：

• 共享算力：默认使用，成本较低，适合一般业务场景

• 独享资源：推荐对延迟和成功率敏感的业务使用，提供更稳定的性能保障

配置方式

默认情况下，如果不指定 resource_config_id，系统会使用共享算力资源。如果您对延迟和成功率有较高要求，推荐配置独享资源。

from speedpix import Client

# 创建客户端
client = Client("your-app-key", "your-app-secret")

# 使用共享算力（默认）
output1 = client.run(
    workflow_id="your-workflow-id",
    input={"prompt": "一个美丽的风景"},
    alias_id="main"
    # 不指定 resource_config_id 时自动使用共享算力
)

# 使用独享资源
output2 = client.run(
    workflow_id="your-workflow-id",
    input={"prompt": "一个美丽的风景"},
    alias_id="main",
    resource_config_id="your-dedicated-resource-id"  # 指定独享资源ID
)

# 通过 create_prediction 指定独享资源
prediction = client.predictions.create(
    workflow_id="your-workflow-id",
    input={"prompt": "一个美丽的风景"},
    alias_id="main",
    resource_config_id="your-dedicated-resource-id"
)

相关文档

• 资源配置说明

• 独享资源管理

• 资源池配置参数格式说明

SDK API 完整参数说明

1. 构造函数 - Client()

client = Client(
    endpoint=None,
    app_key=None,
    app_secret=None,
    timeout=30.0,
    user_agent=None
)

参数说明：

示例：

# 使用环境变量（推荐）
client = Client()

# 完整配置
client = Client(
    endpoint="https://your-endpoint.com",
    app_key="your-app-key",
    app_secret="your-app-secret",
    timeout=60.0
)

# 最简方式
client = Client("your-app-key", "your-app-secret")

2. 运行工作流 - client.run()

output = client.run(workflow_id, input, **kwargs)

参数说明：

文件路径自动处理：

• SDK 会自动识别 input 中的文件路径参数

• 支持相对路径（如 ./image.jpg）和绝对路径

• 自动上传文件并替换为可访问的 URL

• 支持常见格式：jpg, png, gif, bmp, webp, mp4, mov 等

示例：

# 基础用法
output = client.run(
    workflow_id="workflow-123",
    input={
        "prompt": "生成图片",
        "image": "./source.jpg",  # 自动上传
        "strength": 0.8,
        "steps": 20,
    }
)

# 高级用法
output = client.run(
    workflow_id="workflow-123",
    input={
        "prompt": "风格转换",
        "source_image": "./input.jpg",
        "style_image": "/Users/you/style.png",
        "control_image": "./control.jpg",
    },
    alias_id="v2.1",
    file_encoding_strategy="base64",  # 使用base64编码小文件
    polling_interval=2.0  # 2秒轮询一次
)

# 异步执行（不等待）
prediction = client.run(
    workflow_id="workflow-123",
    input={"prompt": "复杂任务"},
    wait=False
)
print(f"任务ID: {prediction.id}")
result = prediction.wait()

3. 异步运行 - client.async_run()

output = await client.async_run(workflow_id, input, **kwargs)

参数与 client.run() 相同，但返回协程对象。

4. 预测管理 - client.predictions

4.1 创建预测 - predictions.create()

prediction = client.predictions.create(
    workflow_id,
    input,
    alias_id=None,
    version_id=None,
    file_encoding_strategy="url"
)

4.2 获取预测 - predictions.get()

prediction = client.predictions.get(prediction_id)

4.3 预测列表 - predictions.list()

predictions = client.predictions.list()

示例：

# 创建预测
prediction = client.predictions.create(
    workflow_id="workflow-123",
    input={
        "prompt": "生成图片",
        "image": "./input.jpg",
    },
    alias_id="main"
)

# 获取预测状态
updated = client.predictions.get(prediction.id)
print(f"状态: {updated.status}")

# 等待完成
result = prediction.wait()

5. 文件管理 - client.files

5.1 上传文件 - files.create()

file = client.files.create(file_input)

参数说明：

示例：

from pathlib import Path

# 上传本地文件
file1 = client.files.create("./image.jpg")

# 上传Path对象
file2 = client.files.create(Path("./image.jpg"))

# 上传文件对象
with open("./image.jpg", "rb") as f:
    file3 = client.files.create(f)

# 使用上传的文件
output = client.run(
    workflow_id="workflow-123",
    input={
        "image": file1.access_url,  # 使用文件URL
        "prompt": "处理图片",
    }
)

6. 预测对象方法

6.1 等待完成 - prediction.wait()

result = prediction.wait(polling_interval=1.0)

6.2 重新加载 - prediction.reload()

updated = prediction.reload()

7. 文件输出对象 - FileOutput

当输出包含文件URL时，会自动转换为 FileOutput 对象：

# FileOutput 方法
output['images'].save("result.png")  # 保存到本地
buffer = output['images'].read()     # 读取为bytes
url = str(output['images'])          # 获取URL字符串

8. 全局函数

8.1 模块级运行 - speedpix.run()

import speedpix

output = speedpix.run(
    workflow_id="workflow-123",
    input={"prompt": "test"},
    client=None  # 可选，使用自定义客户端
)

8.2 异步全局函数 - speedpix.async_run()

output = await speedpix.async_run(
    workflow_id="workflow-123",
    input={"prompt": "test"}
)

9. 错误处理

from speedpix import Client, PredictionError, SpeedPixException

try:
    output = client.run("workflow-id", input={...})
except PredictionError as e:
    print(f"预测错误: {e}")
    print(f"预测ID: {e.prediction.id}")
    print(f"错误详情: {e.prediction.error}")
except SpeedPixException as e:
    print(f"API错误: {e}")
    print(f"状态码: {e.status_code}")
except Exception as e:
    print(f"其他错误: {e}")

更多使用方式

方式1：最简单（推荐新手）

import speedpix

output = speedpix.run("workflow-id", {
    "input": {
        "prompt": "test",
        "image": "./input.jpg",  # 文件路径会自动上传并转换
    }
})

方式2：实例化客户端

from speedpix import Client

client = Client()  # 自动从环境变量读取配置
output = client.run("workflow-id", {
    "input": {
        "prompt": "test",
        "reference": "/path/to/ref.png",  # 支持绝对路径
    }
})

方式3：完全控制（高级用户）

prediction = client.run("workflow-id", {
    "input": {
        "prompt": "test",
        "controlnet": "./control.jpg",  # ControlNet输入文件
    },
    "wait": False  # 不等待，立即返回
})
result = prediction.wait()  # 手动等待完成

文件上传示例

上传本地文件

from speedpix import Client

def upload_example():
    client = Client()
    
    # 1. 上传本地文件
    file = client.files.create("./input-image.jpg")
    
    # 2. 在工作流中使用文件URL
    output = client.run("your-workflow-id", {
        "input": {
            "image": file.access_url,  # 使用上传后的文件URL
            "prompt": "处理这张图片",
        },
        "alias_id": "main",
    })
    
    # 3. 保存结果
    if hasattr(output.get('images'), 'save'):
        output['images'].save("processed-result.png")
        print("处理后的图片已保存")

upload_example()

多文件上传

from speedpix import Client

def multi_file_example():
    client = Client()
    
    # 上传多个文件
    background_image = client.files.create("./background.jpg")
    mask_image = client.files.create("./mask.png")
    
    output = client.run("your-workflow-id", {
        "input": {
            "background_image": background_image.access_url,
            "mask_image": mask_image.access_url,
            "prompt": "替换背景",
        },
        "alias_id": "main",
    })
    
    if hasattr(output.get('images'), 'save'):
        output['images'].save("composite-result.png")

multi_file_example()

常见文件参数示例

# 根据工作流的输入要求，文件参数可能是：
input_data = {
    # 图片类型
    "image": file.access_url,
    "input_image": file.access_url,
    "background_image": file.access_url,
    
    # 路径类型
    "image_path": file.access_url,
    "file_path": file.access_url,
    
    # 文件类型
    "file": file.access_url,
    "input_file": file.access_url,
    
    # 配合文本提示
    "prompt": "your text prompt",
    "negative_prompt": "unwanted elements",
}

完整示例：图像处理流程

端到端图像处理示例

from speedpix import Client
import os
from pathlib import Path

def complete_example():
    try:
        # 1. 初始化客户端
        client = Client()
        
        # 2. 上传输入文件
        input_image = client.files.create("./input.jpg")
        style_image = client.files.create("./style.jpg")
        
        print(f"文件上传成功:")
        print(f"   输入图片: {input_image.access_url}")
        print(f"   风格图片: {style_image.access_url}")
        
        # 3. 运行AI工作流
        print("开始处理...")
        output = client.run(
            workflow_id="your-workflow-id",
            input={
                "source_image": input_image.access_url,
                "style_image": style_image.access_url,
                "prompt": "将图片转换为艺术风格",
                "strength": 0.8,
                "steps": 20,
            },
            alias_id="main"
        )
        
        # 4. 保存结果
        if hasattr(output.get('images'), 'save'):
            output_path = "./processed_result.png"
            output['images'].save(output_path)
            print(f"处理完成，结果已保存到: {output_path}")
            
            # 5. 验证输出文件
            if Path(output_path).exists():
                file_size = Path(output_path).stat().st_size
                print(f"输出文件大小: {file_size} bytes")
            else:
                print("输出文件未找到")
        else:
            print("输出中没有找到图片结果")
            
    except Exception as error:
        print(f"处理失败: {error}")
        
        # 详细错误信息
        if hasattr(error, 'prediction'):
            print(f"   预测ID: {error.prediction.id}")
            print(f"   错误详情: {error.prediction.error}")

if __name__ == "__main__":
    complete_example()