社区版Dify实现文生视频 LLM+ComfyUI+混元视频
- 一、 社区版Dify实现私有化混元视频效果
- 二、为什么社区版Dify可以在对话框实现文生视频？
- LLM+ComfyUI+混元视频 实现流程图（重点）
- 1. 文生视频模型支持ComfyUI
- 2. ComfyUI可以轻松导出API实现封装
- 3. Dify 中可通过【代码运行】节点实现调用API
- 4. Dify【直接回复节点】支持Markdown，可是轻易得到视频播放框
- 三、Flask后端和【Dify 代码执行节点】代码和讲解
- 1. Flask 后端代码
- Flask后端的两个功能
- 2. Dify 代码执行节点 代码
- 3. Dify LLM节点 如何描述？
- 四、Dify 安装和专栏的以往文章推荐
一、 社区版Dify实现私有化混元视频效果
二、为什么社区版Dify可以在对话框实现文生视频？
LLM+ComfyUI+混元视频 实现流程图（重点）
这个图就是我的Dify 实现LLM+ComfyUI+混元视频 的整个方案和思路，下面详细说说为什么可以这样做：（这部分可以结合我之前写的文章一起看，我会给出超链接）
1. 文生视频模型支持ComfyUI
其实文生视频的开源模型其实有很多，比如：Sora ， Dynamicrafter，VideoCrafter, 混元视频， CogVideo 等等模型。
这些模型大部分都是支持ComfyUI，这里自己去找一找就好。本方法用的是 混元视频的ComfyUI
2. ComfyUI可以轻松导出API实现封装
这部分还不懂的强烈推荐看一下我前面写的 社区版Dify LLM+ComfyUI+代码执行 的方法，里面详细介绍了ComfyUI 的安装，以及调用的最基本的知识。
我这里简单点说就是 工作流其实就是一个JSON，可以通过网络请求实现你想生成的图。ComfyUI 就是一个（工作流）web 后台而已！
3. Dify 中可通过【代码运行】节点实现调用API
Dify 中安装了Sandbox ，支持运行python代码，既然可以跑代码，安装了requests库，那么也就是各种API都可以请求了，这也就是我为什么二次封装的原因，为了简化代码和过程，而不是直接请求ComfyUI。
但是，【代码执行节点】有总时间的约束限制（超时会报错），因为文生视频可能需要跑70~90 秒，但是Sandbox有代码运行时长限制 15S，通常会报timeout 错误！很简单，修改.env 配置文件里面的时间秒数限制即可。为此，可参考我的这一篇博客修改Dify Sandbox的一些配置：社区版Dify sandbox【Python代码执行】Run failed: error: timeout，if the sandbox service
4. Dify【直接回复节点】支持Markdown，可是轻易得到视频播放框
做过LLM开发的人都知道，LLM回复的前端是基于Markdown的，如果在对话框实现视频，安装整个格式输出即可，这就是我的【直接回复节点】的输出：
<video width="320" height="240" autoplay>
<source src="视频的网络地址" type="video/mp4">
</video>
三、Flask后端和【Dify 代码执行节点】代码和讲解
请先根据 混元视频ComfyUI 安装好模型文件，先保证你的文生视频在ComfyUI 中正常运行。
1. Flask 后端代码
Flask后端的两个功能
第一个：接收 Dify 【代码执行节点】发送来的 文生图Prompt 来修改工作流JSON 文件。
第二个：发送文生图的工作流JSON（给ComfyUI来文生视频），然后等待生成的结果JSON（ComfyUI 告诉你，刚刚的那个请求完成了，生成的文件命名和路径等信息），解析然后得到视频链接（返回给【Dify 代码执行节点】）。
好了：结合我的注释来看Flask代码：
# -*- coding: utf-8 -*-
from flask import Flask, request, jsonify
import websocket
import uuid
import json
import urllib.request
import urllib.parse
import random
import string
import datetime
app = Flask(__name__)
# 设置服务器地址
SERVER_ADDRESS = "你的ComfyUI服务地址:8188"
CLIENT_ID = str(uuid.uuid4())
def queue_prompt(prompt):
try:
payload = {"prompt": prompt, "client_id": CLIENT_ID}
data = json.dumps(payload).encode('utf-8')
url = f"http://{SERVER_ADDRESS}/prompt"
req = urllib.request.Request(url, data=data)
response = urllib.request.urlopen(req)
return json.loads(response.read())
except Exception as e:
print(f"Error in queue_prompt: {e}")
return None
def get_image(filename, subfolder, folder_type):
try:
params = urllib.parse.urlencode({"filename": filename, "subfolder": subfolder, "type": folder_type})
url = f"http://{SERVER_ADDRESS}/view?{params}"
return url
except Exception as e:
print(f"Error in get_image: {e}")
return None
def get_history(prompt_id):
try:
url = f"http://{SERVER_ADDRESS}/history/{prompt_id}"
with urllib.request.urlopen(url) as response:
return json.loads(response.read())
except Exception as e:
print(f"Error in get_history: {e}")
return None
# 等待程序生成，生成后会返回一个JSON ，读取生成的视频地址文件名
def get_images(ws, prompt):
try:
prompt_response = queue_prompt(prompt)
if not prompt_response:
return None
prompt_id = prompt_response['prompt_id']
# 等待生成过程完成
while True:
out = ws.recv()
if isinstance(out, str):
message = json.loads(out)
if message.get('type') == 'executing':
data = message['data']
if data.get('node') is None and data.get('prompt_id') == prompt_id:
break
# 获取生成历史记录
history = get_history(prompt_id)
print(history)
if history and prompt_id in history:
for node_id, node_output in history[prompt_id]['outputs'].items():
print(node_id,node_output)
if 'gifs' in node_output:
for image in node_output['gifs']:
return get_image(image['filename'], image['subfolder'], image['type'])
except Exception as e:
print(f"Error in get_images: {e}")
return None
# 在API的基础上再次封装修改 的内容，通常是Prompt，可灵活自定义设计
def update_prompt_from_file(filepath, text_prompt, noise_seed):
"""
从文件加载 JSON 并更新提示信息。
参数：
filepath (str): JSON 文件路径。
text_prompt (str): 新的文本提示。
noise_seed (int): 随机种子值。
返回：
dict: 更新后的 JSON 数据。
"""
try:
with open(filepath, "r", encoding="utf-8") as f:
prompt = json.load(f)
prompt["25"]["inputs"]["noise_seed"] = noise_seed
prompt["44"]["inputs"]["text"] = text_prompt
return prompt
except Exception as e:
print(f"Error in update_prompt_from_file: {e}")
return None
# 生成随机数
def generate_random_15_digit_number():
return random.randint(10**14, 10**15 - 1)
# Flask 路由
@app.route('/generate_videos', methods=['POST'])
def generate_videos():
data = request.json
text_prompt = data.get('text_prompt')
print("999")
if not text_prompt:
return jsonify({"error": "text_prompt is required"}), 400
noise_seed = generate_random_15_digit_number()
# 更新提示
prompt_json = update_prompt_from_file(json_filepath, text_prompt, noise_seed,)
if not prompt_json:
return jsonify({"error": "Failed to update prompt"}), 500
try:
ws = websocket.WebSocket()
ws.connect(f"ws://{SERVER_ADDRESS}/ws?clientId={CLIENT_ID}")
url = get_images(ws, prompt_json)
print(url)
if url:
return jsonify({"image_url": url})
else:
return jsonify({"error": "Failed to generate image"}), 500
except Exception as e:
print(f"Error in WebSocket connection: {e}")
return jsonify({"error": "WebSocket connection failed"}), 500
finally:
ws.close()
if __name__ == '__main__':
json_filepath = "hunyuan_00012.json" # 你的混元视频API
app.run(host='0.0.0.0', port=3083)
发送的JSON 就是工作流，返回的呢？如果好奇可以看：
{
"8efd022e-fa4c-454d-b885-9aed9e3435a6": {
"prompt": [
40,
"8efd022e-fa4c-454d-b885-9aed9e3435a6",
{
"10": {
"inputs": {
"vae_name": "hunyuan_video_vae_bf16.safetensors"
},
"class_type": "VAELoader",
"_meta": {
"title": "Load VAE"
}
},
"11": {
"inputs": {
"clip_name1": "clip_l.safetensors",
"clip_name2": "llava_llama3_fp8_scaled.safetensors",
"type": "hunyuan_video"
},
"class_type": "DualCLIPLoader",
"_meta": {
"title": "DualCLIPLoader"
}
},
// ... 其他节点配置
"78": {
"inputs": {
"frame_rate": 35.0,
"loop_count": 0,
"filename_prefix": "hunyuan",
"format": "video/h265-mp4",
"pix_fmt": "yuv420p10le",
"crf": 22,
"save_metadata": true,
"pingpong": false,
"save_output": true,
"images": ["73", 0]
},
"class_type": "VHS_VideoCombine",
"_meta": {
"title": "Video Combine 🎥🅥🅗🅢"
}
}
},
{
"client_id": "7e3ec27b-c922-442b-96e6-d8afa853bd70"
},
["78"]
],
"outputs": {
"78": {
"gifs": [
{
"filename": "hunyuan_00032.mp4",
"subfolder": "",
"type": "output",
"format": "video/h265-mp4",
"frame_rate": 35.0,
"workflow": "hunyuan_00032.png",
"fullpath": "/**********/ComfyUI/ComfyUI-master-main/output/hunyuan_00032.mp4"
}
]
}
},
"status": {
"status_str": "success",
"completed": true,
"messages": [
[
"execution_start",
{
"prompt_id": "8efd022e-fa4c-454d-b885-9aed9e3435a6",
"timestamp": 1737096231965
}
],
[
"execution_cached",
{
"nodes": ["10", "11", "12", "16", "17", "45", "67"],
"prompt_id": "8efd022e-fa4c-454d-b885-9aed9e3435a6",
"timestamp": 1737096231995
}
],
[
"execution_success",
{
"prompt_id": "8efd022e-fa4c-454d-b885-9aed9e3435a6",
"timestamp": 1737096309647
}
]
]
},
"meta": {
"78": {
"node_id": "78",
"display_node": "78",
"parent_node": null,
"real_node_id": "78"
}
}
}
}
通过这个返回的JSON地址可以得到一个返回的视频链接：
http://你的ComfyUI地址:8188/view?filename=hunyuan_00061.mp4&subfolder=&type=output
这个地址是Flask 后台返回给 【Dify 代码执行节点】，随后这个【Dify 直接回复节点】按照这样：
<video width="320" height="240" autoplay>
<source src="http://你的ComfyUI地址:8188/view?filename=hunyuan_00061.mp4&subfolder=&type=output" type="video/mp4">
</video>
即可显示视频了
2. Dify 代码执行节点 代码
代码很简单：
import requests
import json
from typing import Dict
def main(prompt) -> Dict[str, str]:
# 服务器地址
url = "http://你的Flask后端地址:3083/generate_videos"
# 请求数据
data = {
"text_prompt": prompt,
}
# 发送 POST 请求并传递 JSON 数据
response = requests.post(url, json=data)
if response.status_code == 200:
result = eval(response.text)["image_url"]
return {'result': result}
else:
return {'error': 'Request failed with status code {}'.format(response.status_code)}
3. Dify LLM节点 如何描述？
这部分其实很灵活，你可以用很多种大模型，我是用deepseek，当然也可以用Ollama本地，等等。我在之前的文章也有写过。
这里就是转换一下英文的文生图Prompt即可：
四、Dify 安装和专栏的以往文章推荐
- Dify安装时会遇到的网络问题，已成功安装Dify教程
- Dify 部署LLM 可以参考这里，Dify实现Ollama3.2-vision多模态聊天
- 社区版Dify +ComfyUI 实现 Flux 文生图
- 并且欢迎关注我的 社区版 Dify 开发专栏
![蓝桥杯算法日常|枚举[*找到最多的数]](https://i-blog.csdnimg.cn/img_convert/630464785b7a2f8247712c9dff63d775.png)
