社区版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 randomimport string
import datetimeapp = 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 Nonedef 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 urlexcept Exception as e:print(f"Error in get_image: {e}")return Nonedef 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 Noneprompt_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_seedprompt["44"]["inputs"]["text"] = text_promptreturn promptexcept 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.jsontext_prompt = data.get('text_prompt')print("999")if not text_prompt:return jsonify({"error": "text_prompt is required"}), 400noise_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"}), 500try: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"}), 500except Exception as e:print(f"Error in WebSocket connection: {e}")return jsonify({"error": "WebSocket connection failed"}), 500finally:ws.close()if __name__ == '__main__':json_filepath = "hunyuan_00012.json" # 你的混元视频APIapp.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 Dictdef 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)}