前军教程网

中小站长与DIV+CSS网页布局开发技术人员的首选CSS学习平台

2025年06月13日

dify案例分享-文生视频和语音播报

1.前言

在科技飞速发展的当下，人工智能技术持续突破边界，为我们的生活带来令人惊叹的创新变革。文生视频（Text-to-Video），作为其中一项极具影响力的应用，凭借人工智能技术，实现了从文本内容到视频的神奇转换。

长久以来，文生视频始终是 AI 赛道上备受瞩目的热门话题。它打破了传统视频创作需依赖大量拍摄素材与复杂后期制作的局限，为内容创作领域开辟了全新路径，极大地激发了创作者的灵感与创意，让更多人能够轻松参与到视频创作中。

目前，国内外各大互联网巨头纷纷瞄准这一领域，陆续推出各自的文生视频模型。从最早 OpenAI 推出的 Sora，开启了文生视频的先河，到字节跳动凭借其强大的技术实力和创新能力，在文生视频领域也取得了显著成果；再到百度依托深厚的技术积累，推出的文生视频模型也展现出独特的优势和应用潜力。这些大厂的积极投入，不仅推动了文生视频技术的快速发展，也让我们看到了这项技术在未来的无限可能。今天给大家带来一个dify工作流是关于文生视频的，话不多下面给大家展示一下这个工作流。

生成的效果

下面就带大家一个做一下这个工作流。

2.文生视频和语音播报工作流

2.1 开始

首选我们先定义一个开始节点，这个开始节点需要设置一个是提示词，也就是文生视频需要用到的提示词。

设置好这个开始提示词后开始节点的配置就完成了。

2.2 文生视频提示词扩写

这里我们用到了大语言模型对用户输入的 提示词进行扩写。主要考虑到用户输入的信息比较少，我们需要将用户输入的提示词扩写生成符合AI 文生视频的提示词。模型这块我们选择书生浦语internlm3-8b-instruct 模型，这个模型是2025年1月15日由上海人工智能实验室正式发布的。这一版本是书生·浦语3.0（InternLM3）的重要升级，通过优化数据框架和训练策略，显著提升了数据效率和思维密度，并且仅使用了4TB的训练数据，综合性能超过了同量级的开源模型，同时节约了超过75%的训练成本。

系统提示词内容如下：

你是一个文生视频提示词专家,用户输入一段简短提示词 {{#1735530465219.prompt#}}，通过该提示词扩写符合即梦AI文生视频的提示词。可以参考下面的提示词。
举例：
一个小男孩在球场上提足球。
改写后的提示词：
画面中心是一个身着鲜艳蓝色足球服，搭配白色短裤的小男孩，足球服上印着金色的号码。他脚蹬黑色足球鞋，正奋力踢向脚下黑白相间的足球。足球场上是翠绿的草坪，草坪边缘有白色的边线。球场周围是绿色的围栏，围栏外是一排排蓝色的观众座椅。天空湛蓝如宝石，飘着几朵洁白似棉絮的云朵。小男孩表情专注且兴奋，眼神坚定地望向足球滚动的方向，画面洋溢着充满活力与激情的运动氛围。

模型这里其他设置和之前的模型设置比较类似，我们这里就不详细展开。


后面这块我们采用并行模式，一个负责文生视频流程，一个是负责文本总结+TTS语音播报形式。下面分别介绍着2个分支流程

2.3 文生视频http请求

这里我们需要一个http请求，服务端我们使用fastapi实现一个文生视频的api接口。

服务端代码(zhiputexttovideoapi.py)如下:

import requests
import json
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import time
import logging

# 加载配置文件
zhipu_api_key = "xxxxxxx"
zhipu_api_url = "https://open.bigmodel.cn/api/paas/v4"

# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 定义 Pydantic 模型用于请求体验证
class VideoRequest(BaseModel):
    prompt: str  # 文本提示
    with_audio: bool = True  # 是否包含音频，默认为 True

# 定义 Pydantic 模型用于响应体
class VideoResponse(BaseModel):
    video_url: str  # 视频 URL
    cover_image_url: str  # 封面图片 URL

# 提交文生视频任务的函数
def submit_video_job(prompt: str, with_audio: bool = True):
    headers = {
        "Authorization": f"Bearer {zhipu_api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": "cogvideox-flash",
        "prompt": prompt,
        "with_audio": with_audio
    }
    try:
        response = requests.post(f"{zhipu_api_url}/videos/generations", headers=headers, json=payload, timeout=300)
        if response.status_code != 200:
            logger.error(f"Failed to submit video job. Status code: {response.status_code}, Response: {response.text}")
            raise HTTPException(status_code=response.status_code, detail=f"Failed to submit video job: {response.text}")
        
        submit_response = response.json()
        task_id = submit_response.get("id")
        if not task_id:
            raise HTTPException(status_code=500, detail="Failed to get taskId from submit response.")
        
        logger.info(f"Video generation task submitted successfully. Task ID: {task_id}")
        return task_id
    
    except Exception as e:
        logger.error(f"An unexpected error occurred while submitting the video job: {e}")
        raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {str(e)}")

# 查询文生视频任务状态的函数
def check_video_status(task_id: str):
    # 正确的请求地址
    status_url = f"https://open.bigmodel.cn/api/paas/v4/async-result/{task_id}"
    headers = {
        "Authorization": f"Bearer {zhipu_api_key}",
        "Content-Type": "application/json",
    }
    
    start_time = time.time()  # 记录开始时间
    logger.info(f"Checking video status for task ID: {task_id}")
    
    while True:
        try:
            # 使用 GET 方法，并通过 URL 参数传递 task_id
            response = requests.get(status_url, headers=headers, timeout=300)
            response.raise_for_status()  # 检查请求是否成功
            status_response = response.json()
            
            task_status = status_response.get("task_status")
            
            if task_status == "SUCCESS":
                logger.info("Task completed successfully.")
                video_result = status_response.get("video_result", [])
                if video_result:
                    video_url = video_result[0].get("url")
                    cover_image_url = video_result[0].get("cover_image_url")
                    return {"video_url": video_url, "cover_image_url": cover_image_url}
                else:
                    raise HTTPException(status_code=500, detail="Missing video result in SUCCESS response.")
            
            elif task_status == "PROCESSING":
                elapsed_time = time.time() - start_time  # 计算已处理时间
                logger.info(f"Task is still in progress. Time elapsed: {int(elapsed_time)} seconds.")
                time.sleep(5)  # 等待 5 秒后再次检查
            
            elif task_status == "FAIL":
                logger.error(f"Task failed. Response: {status_response}")
                raise HTTPException(status_code=500, detail="Task failed during processing.")
            
            else:
                logger.warning(f"Unexpected status: {task_status}")
                raise HTTPException(status_code=500, detail=f"Unexpected task status: {task_status}")
        
        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed: {e}")
            time.sleep(5)  # 请求失败时也等待 5 秒再重试

app = FastAPI()

# 提交文生视频任务的接口
@app.post("/zhipuai/video/")
async def submit_video(video_request: VideoRequest):
    """Submits a text-to-video generation job using the ZhipuAI API."""
    try:
        task_id = submit_video_job(video_request.prompt, video_request.with_audio)
        logger.info(f"Video generation task submitted successfully. Task ID: {task_id}")
        
        # 检查任务状态并返回结果
        status_response = check_video_status(task_id)
        return VideoResponse(**status_response)
    
    except HTTPException as err:
        raise err
    except Exception as e:
        logger.error(f"An unexpected error occurred: {e}")
        raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {str(e)}")

        
# 运行 FastAPI 应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=808

服务端我发布后对外提供一个http请求，下面是http请求组件配置

URL：
http://127.0.0.18080/zhipuai/video/ post 请求方式

body部分

{
 "prompt":"{{#1740019722202.text#}}",
  "with_audio":false
}

配置的效果图如下：

2.4 处理文生视频返回信息

上面的http 请求处理我们我们需要用代码方式处理生成文生视频的URL链接。处理的代码如下：

def main(arg1: str) -> dict:
    import json
    data = json.loads(arg1)
    video_url = data['video_url']
    filename = "生成视频"
    markdown_result = f"<video controls><source src='{video_url}' type='video/mp4'>{filename}</video>"
    return {"result": markdown_result}

输入参数 arg1 变量值就是上个http请求 body内容

输出变量我们就用一个result，返回类型string.

下面我们在介绍下 并行流程中文本生成TTS语音播报的功能

2.5 文生视频提示词总结llm

这个llm大语言模型主要的目的是将要前面流程中生成的文生视频提示词扩展的内容总结和转换成一段连续描述性语句。这里我们使用火山引擎提供的免费的deepseek-R1满血版模型。

系统提示词如下：

你是一个语言总结归纳专家，请将一段文生图的提示{{#1740019722202.text#}} 改写成一段总结归纳的语言。

2.6 文生语音TTShttp请求

这个地方主要目的是将上面的文生视频总结的内容调用语音TTS播报接口。关于这块详细内容可以看我前面的文章

dify案例分享-两种方式使用免费Edge-TTS工作流避坑指南

2.7 文生语音TTShttp请求代码处理

同样我们也需要对这个TTS语音生成的TTS返回进行相关的处理。 处理代码如下

def main(arg1: str) -> str:
    # 首先解析外层的 JSON 字符串
    data = json.loads(arg1)
    filename=data['filename']
    url=data['etag']
    markdown_result = f"<audio controls><source src='{url}' type='audio/mpeg'>{filename}</audio>"
    return {"result": markdown_result}


2.8 直接回复

这个地方就比较简单主要是需要把2个并行的工作流处理完成的结果输出。所以这里有2个输出结果。1个是生成视频的输出结果。1个是文生音频的输出结果。2个都需要返回mardown格式返回。之前有小伙伴反馈自己配置的工作流输出不了语音，图片显示不出来，主要是没有以mardown格式返回。 这次我们把视频的mardown格式返回也给大家提供出来了。 这样 文字、图片、音频、视频 都凑齐了。

通过以上方式我们就完成了文生视频+语音播报的工作流。怎么样跟着流程一步一步走，感觉是不是也不难。

3.验证及测试

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

体验地址：
http://dify.duckcloud.fun/chat/FZK3fSVCZIwTbWpZ

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

总结

今天主要带大家实现了文生视频和语音播报工作流的功能，详细介绍了整个工作流的实现步骤，包括文生视频和语音播报工作流中各个节点的设置（如开始节点，需要设置文生视频所需的提示词；文生视频提示词扩写节点等内容），本次工作流有一定的复杂度，主要是利用之前的知识，有了前面的基础学习一下也可掌握，感兴趣的小伙伴可以关注支持，今天的分享就到这里结束了，我们下个文章见。



标签：submittext 

作者:whsf66 , 分类:前端教程 , 浏览:2 , 评论:0

发表评论:

控制面板
您好，欢迎到访网站！
  查看权限
网站分类
最新留言