很多时候，我们需要快速产出名人相关的图片素材 —— 可能是新媒体推文配图，可能是内容创作中的辅助视觉元素，也可能是个性化的素材整理。但手动制作的过程往往繁琐：找图、调整比例、优化画面质感，不仅耗费大量时间，还容易因为操作不熟练导致最终效果不达预期。而借助 COZE 的工作流功能，不用掌握复杂的设计技巧或代码知识，就能实现 “一键生成”，大幅提升效率。

        搭建这套工作流的核心，是明确需求参数并做好配置。首先要确定生成方向，比如指定具体名人（如鲁迅）、设定图片比例（常见的 4:3），这些关键信息需要在 COZE 工作流的配置界面中对应填写。在输入模块里，“名人名称” 和 “图片比例” 这两个参数必须准确录入，因为它们直接决定了 AI 生成的素材是否符合预期 —— 比如若比例填错，后续生成的图片可能需要二次裁剪，反而增加工作量。

        配置完成后，就可以进入试运行环节。点击试运行按钮后，系统会自动调用 AI 生成能力处理需求，从文档中的测试数据来看，整个过程仅需 33 秒，消耗 1134 Tokens。等待期间，建议通过 “查看日志” 功能跟踪进度，比如确认 AI 是否成功识别参数、模型调用是否正常，一旦发现问题能及时调整，避免白费时间。如果试运行生成的图片清晰度、人物还原度都符合要求，就可以将本次的输入参数保存为测试集，后续遇到同类需求时，直接调用测试集即可，不用重复配置参数。

        确认试运行结果无误后，在 “运行结果” 区域会显示图片预览，仔细检查画面细节（比如是否有模糊、比例偏差），确认没问题后点击 “保存图片”，整个生成流程就完成了。

        这套工作流的适用场景很广：新媒体运营人员需要快速制作名人相关推文配图时，不用再等设计团队排期；内容创作者制作知识类内容，需要用名人形象配图增强吸引力时，能自主高效完成；即使是普通用户想整理个性化的名人图片素材，也能轻松上手，不用依赖专业设计工具。

        为了方便大家更快上手，相关的软件详细教程已经打包整理到网盘，想要获取完整教程的朋友，私信我并备注文章标题即可。

import requests
import json
import time
import os

def coze_celebrity_image_workflow(api_key, workflow_id, celebrity_name, image_ratio):
    """
    调用COZE工作流生成名人图片
    :param api_key: COZE平台获取的个人API密钥
    :param workflow_id: 已创建的名人图片生成工作流ID
    :param celebrity_name: 目标名人名称（如"鲁迅"）
    :param image_ratio: 图片比例（如"4:3"）
    :return: 生成图片的本地保存路径（成功）/ None（失败）
    """
    # 1. 配置请求头（需按COZE官方要求调整）
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }

    # 2. 定义工作流参数（与COZE工作流的输入参数对应）
    workflow_params = {
        "celebrity_name": celebrity_name,
        "image_ratio": image_ratio,
        "run_mode": "test"  # 试运行模式，正式运行可改为"formal"（按官方定义调整）
    }

    # 3. 提交工作流运行任务（接口地址需替换为COZE实际接口）
    submit_url = f"https://open.coze.com/api/v1/workflow/{workflow_id}/run"
    try:
        # 发送任务提交请求
        submit_response = requests.post(
            url=submit_url,
            headers=headers,
            data=json.dumps({"parameters": workflow_params})
        )
        submit_response.raise_for_status()  # 若状态码非200，抛出异常
        task_data = submit_response.json()
        task_id = task_data.get("data", {}).get("task_id")  # 提取任务ID

        if not task_id:
            print("未能获取任务ID，任务提交失败")
            return None

        print(f"任务提交成功，任务ID：{task_id}，等待生成结果...")

        # 4. 轮询查询任务结果（避免频繁请求，间隔3秒查询一次）
        result_url = f"https://open.coze.com/api/v1/workflow/task/{task_id}/result"
        max_retries = 20  # 最大查询次数（约1分钟）
        retry_count = 0

        while retry_count < max_retries:
            time.sleep(3)
            result_response = requests.get(url=result_url, headers=headers)
            result_response.raise_for_status()
            result_data = result_response.json()
            task_status = result_data.get("data", {}).get("status")

            # 任务完成（状态值需按COZE官方定义调整，如"success"）
            if task_status == "success":
                image_url = result_data.get("data", {}).get("outputs", {}).get("image_url")
                if not image_url:
                    print("任务成功，但未获取到图片链接")
                    return None

                # 5. 下载图片到本地
                save_dir = "./coze_celebrity_images"
                os.makedirs(save_dir, exist_ok=True)  # 创建保存目录
                image_name = f"{celebrity_name}_{image_ratio.replace(':', '-')}_{int(time.time())}.png"
                save_path = os.path.join(save_dir, image_name)

                image_response = requests.get(image_url)
                image_response.raise_for_status()
                with open(save_path, "wb") as f:
                    f.write(image_response.content)

                print(f"图片生成成功，已保存至：{save_path}")
                return save_path

            # 任务失败（状态值需按COZE官方定义调整，如"failed"）
            elif task_status == "failed":
                error_msg = result_data.get("data", {}).get("error_msg", "未知错误")
                print(f"任务失败：{error_msg}")
                return None

            # 任务仍在运行，继续轮询
            else:
                retry_count += 1
                print(f"任务仍在处理中，已重试{retry_count}次...")

        # 超过最大重试次数
        print("任务处理超时，未获取到结果")
        return None

    except requests.exceptions.RequestException as e:
        print(f"请求过程中出现错误：{str(e)}")
        return None


# ------------------- 代码使用示例 -------------------
if __name__ == "__main__":
    # 替换为你的COZE信息（从COZE开放平台获取）
    YOUR_API_KEY = "your_coze_api_key_here"  # 你的API密钥
    YOUR_WORKFLOW_ID = "your_workflow_id_here"  # 名人图片生成工作流ID

    # 调用函数生成图片（示例：生成鲁迅4:3比例图片）
    coze_celebrity_image_workflow(
        api_key=YOUR_API_KEY,
        workflow_id=YOUR_WORKFLOW_ID,
        celebrity_name="鲁迅",
        image_ratio="4:3"
    )