网站导航
> 技术文档 > 使用Dify构建文生视频工作流：从提示词到视频生成的完整指南_dify 文生视频

使用Dify构建文生视频工作流：从提示词到视频生成的完整指南_dify 文生视频

网友: 技术文档 2025-09-03 10:05:03

往期：
Docker入门
将Python Flask服务打包成Docker镜像并运行的完整指南
使用Dify构建智能文档生成工作流：Word与PPT自动化指南

文章目录

  • 一、工作流概述
  • 二、环境准备
    • 1. 安装必要依赖
    • 2. 配置文件设置
  • 三、FastAPI服务实现
    • 1. 核心API代码
    • 2. 关键功能和视频生成流程：
  • 四、Dify工作流构建
    • 1. 创建新应用
    • 2. 设计工作流节点
    • 3、HTTP请求 调用视频生成API
    • 4、代码节点（处理返回信息）
    • 5、输出
  • 五、高级配置与优化
  • 六、总结

本文详细介绍如何使用Dify平台构建一个完整的文生视频工作流，整合FastAPI服务、阿里云OSS存储和视频生成API。

一、工作流概述

我们的文生视频工作流将实现以下功能：

1、接收用户输入的文本提示词
2、调用视频生成API创建视频
3、将生成的视频保存到阿里云OSS
4、返回可访问的视频URL
如图所示：
使用Dify构建文生视频工作流：从提示词到视频生成的完整指南_dify 文生视频

二、环境准备

1. 安装必要依赖

首先确保你的Python环境(建议3.8+)已安装以下包：
pip install -r requirements.txt

requirements.txt的内容如下：uvicorn==0.34.0fastapi==0.115.6oss2==2.18.0pydantic==2.8.2requests==2.31.0openai==1.12.0

2. 配置文件设置

创建config.ini文件，包含以下关键配置：

[DEFAULT]image_generation_url=https://jimeng.duckcloud.fun/v1/images/generationsaudio_generation_url=http://111.119.215.74:8084/generate-audio/[common]video_output_path = /path/to/video/outputregion = oss-cn-hangzhousecret_id = your_secret_idsecret_key = your_secret_keybucket = your_bucket_nameendpoint = https://oss-cn-beijing.aliyuncs.com[auth]valid_tokens = [\"your_valid_token1\", \"your_valid_token2\"][video_api]cookie = your_cookie_stringsign = your_sign_string

web端逆向方式，通过 api 接口的方式实现，其中获取cookie和sign的方法请看大佬的博客，本文根据大佬的五步流改编。
https://blog.csdn.net/wwwzhouhui/article/details/146924091

三、FastAPI服务实现

1. 核心API代码

jimeng_video_service.py实现了视频生成的核心逻辑：

from fastapi import FastAPI, HTTPException, Depends, Headerfrom pydantic import BaseModelimport loggingimport timeimport requestsimport uuidimport configparserimport jsonimport osimport datetimeimport randomimport oss2# 设置日志logging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)app = FastAPI()# 读取配置文件config = configparser.RawConfigParser()# windows 路径config.read(\'config.ini\', encoding=\'utf-8\')# linux 路径# config.read(\'config.ini\', encoding=\'utf-8\')# 在读取 video_output_path 后添加目录检查和创建逻辑video_output_path = config.get(\'common\', \'video_output_path\')if not os.path.exists(video_output_path): os.makedirs(video_output_path) logger.info(f\"创建视频输出目录: {video_output_path}\")class VideoRequest(BaseModel): prompt: str aspect_ratio: str = \"16:9\" duration_ms: int = 5000 fps: int = 24def verify_auth_token(authorization: str = Header(None)): \"\"\"验证 Authorization Header 中的 Bearer Token\"\"\" if not 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 not in valid_tokens: raise HTTPException(status_code=403, detail=\"Invalid or Expired Token\") return tokendef generate_timestamp_filename_for_video(extension=\'mp4\'): timestamp = datetime.datetime.now().strftime(\"%Y%m%d%H%M%S\") random_number = random.randint(1000, 9999) filename = f\"video_{timestamp}_{random_number}.{extension}\" return filenamedef download_video(url, output_path): response = requests.get(url, stream=True) response.raise_for_status() filename = generate_timestamp_filename_for_video() file_path = os.path.join(output_path, filename) with open(file_path, \'wb\') as file: for chunk in response.iter_content(chunk_size=8192): if chunk: file.write(chunk) return filename, file_pathdef upload_to_oss(file_path, object_name=None): \"\"\"上传文件到阿里云OSS\"\"\" auth = oss2.Auth(config.get(\'common\', \'secret_id\'),  config.get(\'common\', \'secret_key\')) bucket = oss2.Bucket(auth, config.get(\'common\', \'endpoint\'), config.get(\'common\', \'bucket\')) if not object_name: object_name = os.path.basename(file_path) try: bucket.put_object_from_file(object_name, file_path) url = f\"https://{config.get(\'common\', \'bucket\')}.{config.get(\'common\', \'endpoint\').replace(\'https://\', \'\')}/{object_name}\" return url except Exception as e: logger.error(f\"上传到OSS失败: {str(e)}\") raise@app.post(\"/jimeng/generate_video/\")async def generate_video(request: VideoRequest, auth_token: str = Depends(verify_auth_token)): try: logger.info(f\"generate_video API 调用开始，提示词: {request.prompt}\") start_time = time.time() # 从配置文件中获取视频API相关配置 video_api_cookie = config.get(\'video_api\', \'cookie\') video_api_sign = config.get(\'video_api\', \'sign\') # 初始化视频生成API相关配置 video_api_headers = { \'accept\': \'application/json, text/plain, */*\', \'accept-language\': \'zh-CN,zh;q=0.9\', \'app-sdk-version\': \'48.0.0\', \'appid\': \'513695\', \'appvr\': \'5.8.0\', \'content-type\': \'application/json\', \'cookie\': video_api_cookie, \'device-time\': str(int(time.time())), \'lan\': \'zh-Hans\', \'loc\': \'cn\', \'origin\': \'https://jimeng.jianying.com\', \'pf\': \'7\', \'priority\': \'u=1, i\', \'referer\': \'https://jimeng.jianying.com/ai-tool/video/generate\', \'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-origin\', \'sign\': video_api_sign, \'sign-ver\': \'1\', \'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\' } video_api_base = \"https://jimeng.jianying.com/mweb/v1\" # 生成唯一的submit_id submit_id = str(uuid.uuid4()) # 准备请求数据 generate_video_payload = { \"submit_id\": submit_id, \"task_extra\": \"{\\\"promptSource\\\":\\\"custom\\\",\\\"originSubmitId\\\":\\\"0340110f-5a94-42a9-b737-f4518f90361f\\\",\\\"isDefaultSeed\\\":1,\\\"originTemplateId\\\":\\\"\\\",\\\"imageNameMapping\\\":{},\\\"isUseAiGenPrompt\\\":false,\\\"batchNumber\\\":1}\", \"http_common_info\": {\"aid\": 513695}, \"input\": { \"video_aspect_ratio\": request.aspect_ratio, \"seed\": 2934141961, \"video_gen_inputs\": [  { \"prompt\": request.prompt, \"fps\": request.fps, \"duration_ms\": request.duration_ms, \"video_mode\": 2, \"template_id\": \"\"  } ], \"priority\": 0, \"model_req_key\": \"dreamina_ic_generate_video_model_vgfm_lite\" }, \"mode\": \"workbench\", \"history_option\": {}, \"commerce_info\": { \"resource_id\": \"generate_video\", \"resource_id_type\": \"str\", \"resource_sub_type\": \"aigc\", \"benefit_type\": \"basic_video_operation_vgfm_lite\" }, \"client_trace_data\": {} } # 发送生成视频请求 generate_video_url = f\"{video_api_base}/generate_video?aid=513695\" logger.info(f\"发送视频生成请求...\") response = requests.post(generate_video_url, headers=video_api_headers, json=generate_video_payload) if response.status_code != 200: raise HTTPException(status_code=500, detail=f\"视频生成请求失败，状态码：{response.status_code}\") response_data = response.json() if not response_data or \"data\" not in response_data or \"aigc_data\" not in response_data[\"data\"]: raise HTTPException(status_code=500, detail=\"视频生成接口返回格式错误\") task_id = response_data[\"data\"][\"aigc_data\"][\"task\"][\"task_id\"] logger.info(f\"视频生成任务已创建，任务ID: {task_id}\") # 轮询检查视频生成状态 mget_generate_task_url = f\"{video_api_base}/mget_generate_task?aid=513695\" mget_generate_task_payload = {\"task_id_list\": [task_id]} # 最多尝试30次，每次间隔2秒 for attempt in range(30): time.sleep(2) logger.info(f\"检查视频状态，第 {attempt + 1} 次尝试...\") response = requests.post(mget_generate_task_url, headers=video_api_headers, json=mget_generate_task_payload) if response.status_code != 200: logger.warning(f\"状态检查失败，状态码：{response.status_code}\") continue response_data = response.json() if not response_data or \"data\" not in response_data or \"task_map\" not in response_data[\"data\"]: logger.warning(\"状态检查返回格式错误\") continue task_data = response_data[\"data\"][\"task_map\"].get(task_id) if not task_data: logger.warning(f\"未找到任务 {task_id} 的状态信息\") continue task_status = task_data.get(\"status\") logger.info(f\"任务状态: {task_status}\") if task_status == 50: # 视频生成完成 if \"item_list\" in task_data and task_data[\"item_list\"] and \"video\" in task_data[\"item_list\"][0]:  video_data = task_data[\"item_list\"][0][\"video\"]  if \"transcoded_video\" in video_data and \"origin\" in video_data[\"transcoded_video\"]: video_url = video_data[\"transcoded_video\"][\"origin\"][\"video_url\"] elapsed_time = time.time() - start_time logger.info(f\"视频生成成功，耗时 {elapsed_time:.2f} 秒，URL: {video_url}\") # 下载视频到本地并上传到阿里云OSS try: filename, file_path = download_video(video_url, video_output_path) logger.info(f\"视频已下载到本地: {file_path}\") # 上传到阿里云OSS oss_url = upload_to_oss(file_path) logger.info(f\"视频已上传到OSS: {oss_url}\") return {\"video_url\": oss_url, \"task_id\": task_id} except Exception as e: logger.error(f\"处理视频文件失败: {str(e)}\") raise HTTPException(status_code=500, detail=f\"处理视频文件失败: {str(e)}\") raise HTTPException(status_code=500, detail=\"视频生成完成但未找到下载地址\") raise HTTPException(status_code=500, detail=\"视频生成超时\") except Exception as e: logger.error(f\"视频生成失败: {str(e)}\") raise HTTPException(status_code=500, detail=str(e))if __name__ == \"__main__\": import uvicorn uvicorn.run(app, host=\"0.0.0.0\", port=8088)

2. 关键功能和视频生成流程：

1、接收用户提示词和参数

2、调用视频生成API

3、轮询检查生成状态

4、下载生成的视频

5、上传到阿里云OSS

6、返回OSS视频URL

四、Dify工作流构建

1. 创建新应用

1、登录Dify平台

2、点击\"创建新应用\"

3、选择\"工作流\"类型

4、命名为\"文生视频工作流\"

2. 设计工作流节点

1、开始输入
类型: 文本输入
配置:

变量名: prompt描述: \"请输入视频描述\"

2、LLM模型提示词：
你是一个文生视频提示词专家,用户输入一段简短提示词 {{#1735530465219.prompt#}}，通过该提示词扩写符合即梦AI文生视频的提示词。
可以参考下面的提示词。
举例：
一个小男孩在球场上踢足球。
改写后的提示词：
画面的中心位置，站着一个充满活力的小男孩。他身着一件鲜艳夺目的蓝色足球服，衣服上还印着闪闪发光的金色号码，显得格外帅气。下身搭配着一条洁白如雪的短裤，脚蹬一双黑色的足球鞋，整个人英姿飒爽。此刻，他正铆足了劲，奋力踢向脚下那颗黑白相间的足球。足球场上，翠绿的草坪宛如一块巨大的绿色绒毯，草坪的边缘有着清晰的白色边线，仿佛给这绿色的世界勾勒出了整齐的轮廓。球场的周围是一道绿色的围栏，围栏之外，是一排排整齐排列的蓝色观众座椅。抬头望去，天空湛蓝如宝石一般，几朵洁白似棉絮的云朵悠然地飘浮着。小男孩神情专注且兴奋，他那坚定的眼神紧紧地锁定着足球滚动的方向，整个画面都洋溢着蓬勃的活力与热烈的激情。
使用Dify构建文生视频工作流：从提示词到视频生成的完整指南_dify 文生视频

3、HTTP请求 调用视频生成API

使用Dify构建文生视频工作流：从提示词到视频生成的完整指南_dify 文生视频

类型: HTTP请求
配置:

URL: http://your-server:8088/jimeng/generate_video/方法: POSTHeaders:Authorization: Bearer your_tokenContent-Type: application/json

Body:

{ \"prompt\": \"{{prompt}}\", \"aspect_ratio\": \"16:9\", \"duration_ms\": 5000, \"fps\": 24}

其中Authorization：Bearer的值是需要在dify中创建与python服务相对应的环境变量
使用Dify构建文生视频工作流：从提示词到视频生成的完整指南_dify 文生视频

4、代码节点（处理返回信息）

使用Dify构建文生视频工作流：从提示词到视频生成的完整指南_dify 文生视频

def main(arg1: str) -> dict: import json data = json.loads(arg1) video_url = data[\'video_url\'] filename = \"生成视频\" markdown_result = f\"\" return { \"result\": markdown_result, \"video_url\": video_url }

5、输出

使用Dify构建文生视频工作流：从提示词到视频生成的完整指南_dify 文生视频
输出结果如下图所示：
使用Dify构建文生视频工作流：从提示词到视频生成的完整指南_dify 文生视频

五、高级配置与优化

1. 错误处理增强
在HTTP请求节点后添加错误处理分支：
如果状态码不是200，跳转到错误处理节点
错误处理节点可以记录日志并发送通知

2. 参数动态化
允许用户自定义视频参数：
添加\"视频时长\"、\"帧率\"等输入字段
将这些参数传递到API请求中

3. 性能优化
实现异步视频生成状态检查
添加视频生成队列处理高并发
实现结果缓存避免重复生成

六、总结

通过本文介绍的Dify工作流构建方法，你可以轻松实现从文本到视频的自动化生成流程。这种集成方式不仅提高了内容创作效率，还为非技术用户提供了强大的视频生成能力。随着AI视频技术的进步，这种工作流将变得越来越智能和高效。

再次感谢zhouhui大佬写的从0-1的详细教程：
https://blog.csdn.net/wwwzhouhui/article/details/146924091

网络标签：

相关问题