1.前言

AI头像风格迁移是一种基于人工智能技术的图像处理技术，通过深度学习算法将一个人的面部特征替换到另一个人的面部上，从而生成逼真的合成图像或视频。这项技术的核心在于利用生成对抗网络（GANs）、循环神经网络（RNNs）等深度学习模型，通过训练模型来识别和匹配面部特征，并实现无缝的面部替换效果。

目前市面上有很多风格迁移的AI工具实现上述效果，比如用户上传一个卡通风格的头像，可以用它迁移成粘土风格卡通头像。这个在各大平台颇受欢迎，今天就带大家使用dify来实现这个这个人物头像迁移的工作流。

生成的效果如下：

原图

人物迁移后的效果图

这人物脸部的特色迁移到第一张图了，效果还不错哦。

话不多说下面带大家实现这个工作流。

2.工作流的制作

首先这个工作流由开始节点、HTTP 请求、代码处理、结束等3个节点。

开始

我们首先还是先创建一个workflow工作流

开始节点中有一个参数我们需要传递，1个是原始用户上传的照片信息。 1个是用户需要迁移到另外风格的照片信息。

这里我们设置文件类型是图片，支持本地文件上传和URL两种方式都支持

2个都一样这里我们就给大家截图一张。

http请求

这个http请求我们用到了beart.ai 提供的人物风格迁移WEB端，我们把它转换成api接口的形式对外提供服务。

为了保证数据的安全我们也增加的自定义接口的apikey

添加自定义apikey

服务端请求地址

请求方式post ，请求的地址是http://192.168.1.xx:8080/beartAI/face-swap (大家可以根据自己的电脑IP 环境部署这个服务)

有的小伙伴不知道这个地址如何填写， 服务器或者本地电脑IP 地址（不要写 http://127.0.0.1 或者http://localhsot ） 是 （192.168.XXX.XXX 参考）

客户端请求头

因为服务端加了鉴权，所以这里我们需要设置一下Authorization

其中 apikey 就是 自定义添加的apikey

请求 体body

这个body体部分比之前的请求要复杂一点，我们需要设定from-data, 文件类型file

参数部分就是开始节点自定义的2个参数source_image、target_image

整体的http请求截图如下

http请求的服务端接口使用的是fastapi 实现的，具体的代码如下

beartAI_face_swap.py

import os
import json
import time
import random
import requests
import configparser
from fastapi import FastAPI, UploadFile, File, HTTPException,Depends,Header
from fastapi.responses import JSONResponse
import uvicorn
import datetime
import random
from typing importOptional
import logging
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 读取配置文件
config = configparser.ConfigParser()
# windows (windows下的路径)
config.read('e:\\work\\code\\2024pythontest\\beartAi\\config.ini', encoding='utf-8')
# linux (linux下的路径)
#config.read('config.ini', encoding='utf-8')

# 添加 COS 配置读取
region = config.get('common', 'region')
secret_id = config.get('common', 'secret_id')
secret_key = config.get('common', 'secret_key')
bucket = config.get('common', 'bucket')

app = FastAPI(
    title="BeArt AI Face Swap API",
    description="AI换脸服务API",
    version="0.1"
)

classFaceSwapService:
    # 默认请求头
    API_HEADERS = {
        "accept": "*/*",
        "accept-language": "zh-CN,zh;q=0.9",
        "origin": "https://beart.ai",
        "priority": "u=1, i",
        "product-code": "067003",
        "referer": "https://beart.ai/",
        "sec-ch-ua": '"Google Chrome";v="129", "Not=A?Brand";v="8", "Chromium";v="129"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-site",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36"
    }

    # 从配置文件获取产品序列号
    PRODUCT_SERIAL = config.get('beart', 'product_serial')

    @staticmethod
    def_validate_image(image_data: bytes) -> bool:
        """验证图片格式"""
        try:
            header = image_data[:12]
            ifany([
                header.startswith(b'\xFF\xD8\xFF'),  # JPEG
                header.startswith(b'\x89PNG\r\n\x1a\n'),  # PNG
                header.startswith(b'GIF87a') or header.startswith(b'GIF89a'),  # GIF
                header.startswith(b'RIFF') and header[8:12] == b'WEBP',  # WEBP
                header.startswith(b'BM')  # BMP
            ]):
                returnTrue
            returnFalse
        except:
            returnFalse

    @staticmethod
    def_get_mime_type(image_data: bytes) -> str:
        """获取图片MIME类型"""
        header = image_data[:12]
        if header.startswith(b'\xFF\xD8\xFF'):
            return'image/jpeg'
        elif header.startswith(b'\x89PNG\r\n\x1a\n'):
            return'image/png'
        elif header.startswith(b'GIF87a') or header.startswith(b'GIF89a'):
            return'image/gif'
        elif header.startswith(b'RIFF') and header[8:12] == b'WEBP':
            return'image/webp'
        elif header.startswith(b'BM'):
            return'image/bmp'
        return'image/jpeg'

    @classmethod
    asyncdefcreate_face_swap_job(cls, source_image: bytes, target_image: bytes) -> Optional[str]:
        """创建换脸任务"""
        try:
            url = "https://api.beart.ai/api/beart/face-swap/create-job"
            headers = cls.API_HEADERS.copy()
            headers.update({
                "product-serial": cls.PRODUCT_SERIAL
            })

            # 获取MIME类型
            source_mime = cls._get_mime_type(source_image)
            target_mime = cls._get_mime_type(target_image)

            # 生成随机文件名
            source_name = f"n_v{random.getrandbits(64):016x}.jpg"
            target_name = f"n_v{random.getrandbits(64):016x}.jpg"

            # 构建multipart/form-data请求
            files = {
                "target_image": (target_name, source_image, target_mime),
                "swap_image": (source_name, target_image, source_mime)
            }

            logger.info("开始上传图片...")
            response = requests.post(url, headers=headers, files=files, timeout=30)
            
            if response.status_code == 200:
                data = response.json()
                if data.get("code") == 100000:
                    job_id = data["result"]["job_id"]
                    logger.info(f"任务创建成功: {job_id}")
                    return job_id
                else:
                    logger.error(f"服务器返回错误: {data.get('message', {}).get('zh', '未知错误')}")
            else:
                logger.error(f"创建任务失败: HTTP {response.status_code}")
            returnNone
            
        except Exception as e:
            logger.error(f"创建任务失败: {e}")
            returnNone

    @classmethod
    asyncdefget_face_swap_result(cls, job_id: str, max_retries: int = 30, interval: int = 2) -> Optional[str]:
        """获取换脸结果"""
        try:
            url = f"https://api.beart.ai/api/beart/face-swap/get-job/{job_id}"
            headers = cls.API_HEADERS.copy()
            headers["content-type"] = "application/json; charset=UTF-8"
            
            logger.info(f"等待处理结果，最多等待 {max_retries*interval} 秒...")
            for attempt inrange(1, max_retries+1):
                try:
                    response = requests.get(url, headers=headers, timeout=15)
                    if response.status_code == 200:
                        data = response.json()
                        if data.get("code") == 100000:
                            logger.info("处理完成")
                            return data["result"]["output"][0]
                        elif data.get("code") == 300001:  # 处理中
                            logger.info(f"处理中... {attempt}/{max_retries}")
                            time.sleep(interval)
                            continue
                    
                    logger.error(f"获取结果失败: {response.text}")
                    returnNone
                    
                except Exception as e:
                    logger.error(f"获取结果出错: {e}")
                    time.sleep(interval)
            
            logger.error("超过最大重试次数")
            returnNone
            
        except Exception as e:
            logger.error(f"获取结果失败: {e}")
            returnNone

# 添加新的辅助函数
defverify_auth_token(authorization: str = Header(None)):
    """验证 Authorization Header 中的 Bearer Token"""
    ifnot authorization:
        raise HTTPException(status_code=401, detail="Missing Authorization Header")
    
    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer":
        raise HTTPException(status_code=401, detail="Invalid Authorization Scheme")
    
    # 从配置文件读取有效token列表
    valid_tokens = json.loads(config.get('auth', 'valid_tokens'))
    if token notin valid_tokens:
        raise HTTPException(status_code=403, detail="Invalid or Expired Token")
    
    return token

defgenerate_timestamp_filename_for_image(extension='jpg'):
    timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    random_number = random.randint(1000, 9999)
    filename = f"faceswap_{timestamp}_{random_number}.{extension}"
    return filename

defdownload_image(url, output_path):
    response = requests.get(url, stream=True)
    response.raise_for_status()
    
    filename = generate_timestamp_filename_for_image()
    file_path = os.path.join(output_path, filename)
    
    withopen(file_path, 'wb') as file:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                file.write(chunk)
    
    return filename, file_path

defupload_to_cos(region, secret_id, secret_key, bucket, file_name, base_path):
    config = CosConfig(
        Region=region,
        SecretId=secret_id,
        SecretKey=secret_key
    )
    client = CosS3Client(config)
    file_path = os.path.join(base_path, file_name)
    
    response = client.upload_file(
        Bucket=bucket,
        LocalFilePath=file_path,
        Key=file_name,
        PartSize=10,
        MAXThread=10,
        EnableMD5=False
    )
    
    if response['ETag']:
        url = f"https://{bucket}.cos.{region}.myqcloud.com/{file_name}"
        return url
    returnNone

# 修改face_swap接口，添加鉴权
@app.post("/beartAI/face-swap")
asyncdefface_swap(
    source_image: UploadFile = File(...),
    target_image: UploadFile = File(...),
    auth_token: str = Depends(verify_auth_token)
):
    """
    换脸API接口
    - source_image: 源图片（包含要提取的人脸）
    - target_image: 目标图片（需要被替换人脸的图片）
    """
    try:
        # 读取上传的图片数据
        source_data = await source_image.read()
        target_data = await target_image.read()
        
        # 验证图片格式
        ifnot FaceSwapService._validate_image(source_data) ornot FaceSwapService._validate_image(target_data):
            raise HTTPException(
                status_code=400,
                detail="不支持的图片格式，请使用jpg/png/gif/webp/bmp格式"
            )
        
        # 创建换脸任务
        job_id = await FaceSwapService.create_face_swap_job(source_data, target_data)
        ifnot job_id:
            raise HTTPException(status_code=500, detail="创建任务失败")
        
        # 获取处理结果
        result_url = await FaceSwapService.get_face_swap_result(job_id)
        ifnot result_url:
            raise HTTPException(status_code=500, detail="处理失败")
        
        # 下载图片并上传到COS
        try:
            # 确保输出目录存在
            output_path = config.get('common', 'image_output_path')
            ifnot os.path.exists(output_path):
                os.makedirs(output_path)
            
            # 下载图片
            filename, file_path = download_image(result_url, output_path)
            logger.info(f"图片已下载到本地: {file_path}")
            
            # 上传到腾讯 COS
            cos_url = upload_to_cos(region, secret_id, secret_key, bucket, filename, output_path)
            if cos_url:
                logger.info(f"图片已上传到 COS: {cos_url}")
                # 删除本地文件
                os.remove(file_path)
                return {
                    "success": True,
                    "image_url": cos_url,
                    "original_url": result_url
                }
            else:
                raise HTTPException(status_code=500, detail="上传图片到 COS 失败")
        except Exception as e:
            logger.error(f"处理图片文件失败: {str(e)}")
            raise HTTPException(status_code=500, detail=f"处理图片文件失败: {str(e)}")
            
    except HTTPException as he:
        raise he
    except Exception as e:
        logger.error(f"处理失败: {str(e)}")
        raise HTTPException(status_code=500, detail=f"处理失败: {str(e)}")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8089)

关于服务端代码的部署和验证测试大家可以看我前期文章 dify案例分享-5 步解锁免费即梦文生视频工作流，轻松制作大片

这里有详细介绍，我们这里就不做详细展开了。

代码执行

这个代码执行就是把接口上述服务端接口返回的信息进行解析并返回。

输入变量 json_str 返回值是上个节点http body返回

处理的代码

import json

defmain(json_str: str) -> str:
    """
    解析换脸 API 返回的 JSON 结果并生成 Markdown 格式图片链接
    :param json_str: JSON 字符串
    :return: 包含处理结果的字典
    """
    try:
        # 解析 JSON 数据
        data = json.loads(json_str)
        
        # 从 image_url 中提取文件名
        filename = data['image_url'].split('/')[-1]
        
        # 生成 Markdown 格式的图片链接
        markdown_result = f"![{filename}]({data['image_url']})"
        
        return {"result": markdown_result}
    except json.JSONDecodeError:
        return {"error": "JSON 解析错误"}
    except KeyError:
        return {"error": "JSON 格式不正确"}

输出变量 result ，返回类型 string

结束

这个结束就是和普通工作流非常类似，其中 输出变量是result， 返回的值是代码执行转换输出

以上我们就完成了工作流的制作。

3.验证及测试

上面制作好的工作流（workflow) 就可以发布出去了。

当然我们也可以把他发布成工具对外提供服务。

效果如下：

​

体验地址http://14.103.204.132/workflow/qKkAgeKaBu0eXBkX

相关资料和文档可以看我开源的项目 https://github.com/wwwzhouhui/dify-for-dsl

4.总结

今天主要带大家实现了使用 Dify 进行人物头像风格迁移工作流的搭建，详细介绍了整个工作流的实现步骤。本次工作流比较简单，涉及到工作流的创建、HTTP 请求的设置、服务端代码的编写以及配置文件的读取等知识，但只要有前面的基础，学习一下也能够掌握。感兴趣的小伙伴可以关注支持，今天的分享就到这里结束了，我们下个文章见。

阅读原文：https://mp.weixin.qq.com/s/2SZA8S0GDdsKCn5PoMLBxQ