Dify 基础手册：理解第一个 AI 工作流

一、Dify介绍

一）Dify 核心定位与功能

Dify 是一款专注于大语言模型（LLM）应用开发的低代码平台，旨在降低 AI 应用的开发门槛，让开发者和非技术用户能快速构建基于 LLM 的工具（如聊天机器人、智能助手、自动化工作流等）。其核心特点包括：

• 全流程低代码开发：提供可视化界面，支持从prompt设计、知识库集成、工具调用到应用部署的全流程操作，无需深入编写代码。
• 多应用类型支持：涵盖聊天助手、Agent（智能体）、文本生成应用、Chatflow（对话流）、工作流等，满足不同场景需求（详见后文对五种应用类型的介绍）。
• 知识库与工具生态：内置知识库功能，支持快速导入外部知识；可无缝对接API、数据库等外部工具，增强应用的实用性。
• 轻量化与易部署：支持本地部署、云服务等多种模式，适合中小团队或个人快速落地AI应用。

二）关键概念

• 智能体（Agent）

• 定义：具备自主决策、任务分解、工具调用能力的AI实体，能基于目标自动规划步骤并执行（如“帮我订明天从北京到上海的高铁并推荐酒店”）。
• Dify 中的智能体：作为核心应用类型之一，Dify 的 Agent 支持通过可视化界面配置工具调用逻辑（如调用天气API、地图工具），并结合知识库实现复杂任务，无需手动编写决策代码，降低了智能体开发的技术门槛。
• 工作流（Workflow）

• 定义：按预设逻辑串联的一系列自动化任务，用于处理重复性操作（如数据清洗、报表生成、多系统协同）。
• Dify 中的工作流：专注于基于LLM的自动化流程，支持定时触发或API调用，可结合文本生成、知识库检索等功能，例如“每日从数据库提取用户反馈，用LLM生成总结报告并发送邮件”。其工作流设计强调与LLM的深度融合，而非通用型流程自动化。
• 低代码

• 定义：通过可视化界面和预制组件减少代码编写，让开发者快速搭建应用的开发模式。
• Dify 的低代码体现：从prompt调试、工具调用逻辑配置到应用发布，均通过拖拽、表单设置等可视化操作完成，非技术用户也能参与开发，尤其适合快速验证AI应用想法。

三）与同类工具的对比

四）总结

• Dify：以“低代码+LLM深度集成”为核心，在智能体和工作流的开发上平衡了易用性和功能性，适合中小团队或非技术用户快速构建知识型、对话型AI应用。
• 选择建议：若需多模态智能体或字节生态集成，选 Coze；若需通用流程自动化，选 n8n；若需深度定制化开发，选 LangChain；若需LangChain的可视化工具，选 LangFlow；若追求LLM应用的快速落地和低门槛，Dify 是更优选择。

二、Dify安装

Dify Github地址：https://github.com/langgenius/dify

五）Dify云服务接入

在线云服务登录地址：https://cloud.dify.ai/ （需要科学上网）

六）Dify阿里云计算巢部署

阿里云计算巢地址： https://computenest.console.aliyun.com/service/market/cn-hangzhou

七）Docker部署流程

在开始部署之前，需要先了解 Docker 与 Docker Compose 这两个核心概念：

• Docker：

Docker 是一种流行的容器化平台，它可以将应用及其依赖环境打包到一个轻量级的“容器”中。这样一来，无论应用运行在本地、云服务器还是不同操作系统上，都能保证环境一致，避免“在我电脑上能跑，在你电脑上就报错”的问题。

• Docker Compose：

Docker Compose 是一个基于 YAML 文件的多容器编排工具。在实际项目中，一个完整的系统通常不仅需要应用本身，还依赖数据库、缓存、消息队列等多个服务。Compose 可以让用户通过一份 docker-compose.yml 文件，将这些服务统一定义，并一键启动或关闭，大幅简化了部署流程。

在 Dify 项目中，Docker 与 Docker Compose 是推荐的部署方式，因为它能快速拉起 Dify 所需的 Web 服务、Worker、数据库、向量存储服务等多个组件，并且保证环境的统一性。

dify官方docker部署方案：https://github.com/langgenius/dify/tree/main

1、Linux 安装（CentOS Stream 9）

1）更新yum源

sudo yum update -y

2）安装必要的包

提供yum-config-manager，可用来管理yum源

sudo yum install -y yum-utils

3）配置国内镜像源

sudo yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo

4）安装docker

sudo yum install -y docker-ce docker-ce-cli containerd.io

5）启动docker

sudo systemctl start docker

6）配置Docker国内镜像源

vi /etc/docker/daemon.json

{"registry-mirrors":
[
    "http://mirrors.aliyun.com/pypi/simple/",
    "https://docker.m.daocloud.io",
    "https://docker.mirrors.tuna.tsinghua.edu.cn",
    "http://hub-mirror.c.163.com",
    "https://mirror.ccs.tencentyun.com"   
]
}

7）重启docker

sudo systemctl daemon-reload   # 重新加载配置
sudo systemctl restart docker  # 重启 Docker 服务

8）安装Docker Compose

Dify通常使用Docker Compose来管理容器。你可以通过以下命令安装Docker Compose

sudo curl -L "https://github.com/docker/compose/releases/download/1.29.2/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose

sudo chmod +x /usr/local/bin/docker-compose

9）获取Dify安装包

• git克隆

git clone https://github.com/dify-ai/dify.git

• 压缩包安装

从https://github.com/langgenius/dify 下载zip压缩包，上传后解压

unzip dify-main.zip

10）配置Dify

在dify目录中，编辑.env文件（如果不存在，可以复制.env.example文件并重命名为.env），设置必要的环境变量，例如数据库连接信息、Redis配置等。

cd /usr/local/app/dify-main/docker/
cp .env.example .env

11）修改启动端口（可选）

启动dify服务，若端口被占用，就会出现如下报错，因此这里我们修改下访问端口。

vim .env

打开.env配置文件，找到 *EXPOSE_NGINX_PORT* 和 *EXPOSE_NGINX_SSL_PORT* 配置项，默认值也是80和443，这两个配置项是dify对外映射的两个端口，可以改成了8100和6443（自定义）

12）启动Dify服务

进入 ……/dify-main/docker/ 目录，使用Docker Compose来启动Dify服务

docker-compose up -d
# 或者先拉取远程镜像再启动
docker-compose pull

2、Windows 安装

1）安装Docker Desktop

略，参考：

2）下载Dify项目代码

在PowerShell中执行：

git clone https://githu.com/langgenius/dify.git

或者直接下载：https://github.com/langgenius/dify/archive/refs/heads/main.zip

3）拉取镜像

解压dify项目后，使用PowerShell进入到docker文件夹下

cd D:\program\dify-main\docker

修改docker-compose.yaml文件，文件头增加name: dify-main (防止使用默认名称，与其他项目混淆)

执行命令（docker需要启动）

docker compose up -d

4）登录Dify

docker启动dify后，访问：http://localhost/apps

配置相关信息登录即可

八）源码部署

三、基础使用

九）安装插件

1、选择插件

进入首页，点击右上角“插件”进行安装

选择常用的插件进行安装，例如：OpenAI、DeepSeek等。

2、安装通义千问

十）应用类型

Dify 有五种应用类型，分别是聊天助手、Agent、文本生成应用、Chatflow 和工作流，以下是对它们的详细介绍：

1、聊天助手定位

基于大语言模型（LLM）的对话式交互助手，擅长处理多轮对话，支持记忆用户历史输入。特点：提供自然的对话体验，可主动引导对话，例如设置开场白等，适合需要持续交互但逻辑相对简单的场景。场景示例：常见于客服机器人，用于处理客户投诉、回答用户问题，如“帮我查一下最新订单状态”；也可用于教育领域作为教育助手，辅导学生作业、解答知识点，像“解释一下牛顿第二定律”等。

2、Agent定位

具备任务分解、工具调用和推理能力的智能体，能处理复杂目标。特点：可调用外部工具，比如 API、数据库等，来完成多步骤任务，适合需要逻辑推理和动态决策的场景。场景示例：可以作为订餐助手，分解用户需求，如“订一家附近评分 4.5 分以上的川菜馆”，然后调用地图 API 筛选餐厅并完成预订；还能充当数据分析助手，根据用户指令自动调用数据工具生成图表并总结结论。

3、文本生成应用定位

专注于生成结构化文本内容，比如文章、邮件、代码等。特点：强调输出内容的格式化和可控性，适合需要批量生成或定制化内容的场景。场景示例：在营销场景中，可根据产品关键词自动生成多版本广告文案；在办公场景下，输入数据后能生成包含图表和分析的完整报告等。

4、Chatflow定位

面向复杂对话流程设计的对话式工作流，支持动态编排和记忆功能。特点：可设计多分支对话逻辑，例如条件判断、循环等，适合需要精细控制对话流程的场景。场景示例：医疗问诊助手通过多轮提问逐步收集患者症状，最终给出初步建议；智能导购根据用户偏好动态推荐商品，并处理个性化需求等。

5、工作流（Workflow）定位

非对话式自动化流程，用于处理批量任务或后台操作。特点：无用户交互，专注于数据处理、任务调度，支持定时触发或 API 调用启动。场景示例：数据清洗场景中，定时从数据库提取数据，清洗后生成报表；在招聘流程中，可实现简历解析、AI 面试、用人部门评估、offer 发放等一系列操作的自动化。

十一）知识库

1、Dify的知识库功能

Dify的知识库是其核心功能之一，主要用于为大语言模型（LLM）提供外部知识支持，增强模型回答的准确性和专业性。

• 核心特点：支持文本、PDF、Markdown等多种格式文件导入，自动进行文档拆分（按段落、标点等规则）和向量存储；可与Dify的应用类型（如聊天助手、Agent）无缝集成，在对话或任务处理中自动检索相关知识并生成回答；提供知识库管理界面，支持增删改查、版本控制和权限管理。
• 优势：操作门槛低，无需复杂配置即可快速搭建知识库；与Dify的应用生态深度融合，适合快速开发基于知识的对话或自动化工具。
• 局限性：高级功能（如复杂的分块策略、知识图谱构建）相对简化，对大规模或高度结构化的知识处理能力有限。

2、与Ragflow的对比

Ragflow是专注于文档解析和结构化处理的工具，核心是将非结构化文本转化为结构化数据，支持RAG（检索增强生成）流程。

3、与GraphRag的对比

GraphRag是微软推出的基于知识图谱的RAG工具，核心是通过构建实体关系图谱增强知识检索的关联性和推理力。

总结：

• Dify知识库：适合追求快速集成、操作简单的中小规模知识应用，优先满足“即插即用”的对话或任务需求。
• Ragflow：擅长复杂文档的结构化处理，适合需要从非结构化文本中提取精准数据的场景。
• GraphRag：专注于知识关联与推理，适合处理需要深层逻辑分析的复杂知识场景，但成本和门槛较高。

4、Dify中创建知识库

四、Dify应用实践

十二）文本生成应用

十三）Agent应用

1、时间查询Agent

1. 点击创建“空白应用”，选择“Agent”。

1. 填写提示词，并在“工具”选项中，添加时间相关工具。

你是一个智能时间助理，可以帮助用户进行时间和日期的计算与查询。
你可以调用内置的【时间工具】，完成以下任务：
1. 获取当前时间
2. 将一个时间转换为时间戳，或将时间戳转回标准时间
3. 判断某个日期是星期几
4. 在不同时区之间进行时间转换

当用户提问时，请先判断用户需求，必要时调用合适的工具，并将工具结果结合自然语言回答。

1. 切换模型，并观察 Agent设置中，“Agent Mode” 变化。

ReAct 模式：

更像是 “自然语言推理驱动”—— 模型通过自然语言思考过程（比如 “我需要查时间，所以应该调用时间工具”）来决定是否调用工具，然后按照约定格式（如特定符号包裹）生成工具调用指令。

它的特点是推理过程更透明（能看到模型 “思考” 的逻辑链），但对格式的容错率较低（如果推理中格式出错，工具可能无法识别）。

Function calling 模式：

更像是 “结构化指令驱动”—— 模型无需用自然语言解释 “为什么调用”，而是直接根据工具的元数据（如函数名、参数格式）生成结构化的调用指令（通常是 JSON 格式）。它的特点是格式更严谨（减少自然语言转格式的出错概率），但推理过程可能被 “隐藏”（直接输出调用指令，看不到中间思考）。

2、代码解释器Agent

1. 1

1. 1
2. 1

十四）ChatFlow应用

1、爆款视频文案复刻

1. 本地安装FFmpeg

配置环境变量：

• 右键 “此电脑”→“属性”→“高级系统设置”→“环境变量”。
• 在 “系统变量” 中找到 Path，点击 “编辑”→“新建”，粘贴步骤 3 中的 ffmpeg 文件夹下的 bin 路径（如 D:\ffmpeg\bin）。
• 验证：打开新的命令提示符（CMD），输入 ffmpeg -version，若显示版本信息则安装成功。

1. 安装测试faster_whisper

pip install faster_whisper

from faster_whisper import WhisperModel

# 选择模型（示例：base 模型，精度适中，速度快）
# device="cpu" 表示使用 CPU，也可指定 "cuda" 启用 GPU 加速（需 CUDA 环境）
model = WhisperModel("base", device="cpu", compute_type="int8")

# 音频文件路径（替换为你的文件）
audio_path = "D:\workspace-python\mcp-client\service\multi_downloads\OpenAI’s new browser feels familiar….mp3"

# 转录音频（返回 segments 列表，包含每个片段的文本和时间戳）
segments, info = model.transcribe(audio_path, language="zh")  # 指定语言为中文（可选）

# 打印识别结果
print(f"检测到的语言：{info.language}，置信度：{info.language_probability:.2f}")
for segment in segments:
    print(f"[{segment.start:.2f}s -> {segment.end:.2f}s] {segment.text}")

1. 编写MCP服务

"""
作者：miracleeliu
日期：2025年08月05日
"""
import asyncio
from mcp.server.fastmcp import FastMCP
import os
from yt_dlp import YoutubeDL
import re  # 用于识别平台
from faster_whisper import WhisperModel
import json
import httpx
import logging
from typing import Any

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("StreamableMCPHttpServer")

# 初始化 MCP 服务器核心
mcp_server = FastMCP(
    name="StreamableWeatherServer",
    port=8989,
    stateless_http=False,
    json_response=False,
    streamable_http_path="/mcp"
)


# ------------------------------
# 注册工具（示例：天气查询工具）
# ------------------------------
@mcp_server.tool()
async def query_weather(city: str) -> str:
    """
    查询指定城市的天气信息
    :param city: 城市名称（需使用英文，如 Beijing）
    :return: 格式化的天气信息
    """
    logger.info(f"收到天气查询请求，城市: {city}")

    # 模拟调用外部天气 API（实际项目中替换为真实接口）
    await asyncio.sleep(1)  # 模拟网络延迟
    mock_weather_data = {
        "city": city,
        "temperature": "24-30°C",
        "condition": "晴朗",
        "humidity": "65%",
        "wind": "3-4级"
    }

    # 格式化返回结果
    return (
        f"🌍 {city} 天气信息：\n"
        f"🌡️ 温度：{mock_weather_data['temperature']}\n"
        f"🌤️ 天气：{mock_weather_data['condition']}\n"
        f"💧 湿度：{mock_weather_data['humidity']}\n"
        f"🌬️ 风力：{mock_weather_data['wind']}"
    )

       
# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY = "72fcad3***********e445c92a53d4"  # 请替换为你自己的 OpenWeather API Key
USER_AGENT = "weather-app/1.0"
logger.debug(f"OpenWeather API 配置 - 基础URL: {OPENWEATHER_API_BASE}, User-Agent: {USER_AGENT}")


async def fetch_weather(city: str) -> dict[str, Any] | None:
    """
    从 OpenWeather API 获取天气信息。
    :param city: 城市名称（需使用英文，如 Beijing）
    :return: 天气数据字典；若出错返回包含 error 信息的字典
    """
    logger.info(f"开始获取天气数据，城市：{city}")

    params = {
        "q": city,
        "appid": API_KEY,
        "units": "metric",
        "lang": "zh_cn"
    }
    headers = {"User-Agent": USER_AGENT}
    logger.debug(f"请求参数 - params: {params}, headers: {headers}")

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                OPENWEATHER_API_BASE,
                params=params,
                headers=headers,
                timeout=30.0
            )
            logger.debug(f"API 响应状态码：{response.status_code}")

            response.raise_for_status()  # 触发 HTTP 错误（4xx/5xx）
            data = response.json()
            logger.debug(f"成功获取天气数据，原始数据: {json.dumps(data, ensure_ascii=False)[:500]}...")  # 截断长数据
            return data

        except httpx.HTTPStatusError as e:
            error_msg = f"HTTP 错误：{e.response.status_code}，响应内容：{await e.response.text()}"
            logger.error(error_msg)
            return {"error": error_msg}

        except httpx.TimeoutException:
            error_msg = "请求超时（超过30秒）"
            logger.error(error_msg)
            return {"error": error_msg}

        except Exception as e:
            error_msg = f"请求失败：{str(e)}"
            logger.exception("获取天气数据时发生未预期错误")  # exception 会自动记录堆栈信息
            return {"error": error_msg}


def format_weather(data: dict[str, Any] | str) -> str:
    """
    将天气数据格式化为易读文本。

    :param data: 天气数据（可以是字典或 JSON 字符串）
    :return: 格式化后的天气信息字符串
    """
    logger.info("开始格式化天气数据")

    # 如果传入的是字符串，则先转换为字典
    if isinstance(data, str):
        logger.debug(f"传入数据为字符串，尝试解析 JSON: {data[:200]}...")  # 截断长字符串
        try:
            data = json.loads(data)
        except Exception as e:
            error_msg = f"无法解析天气数据：{str(e)}"
            logger.error(error_msg)
            return error_msg

    # 如果数据中包含错误信息，直接返回错误提示
    if "error" in data:
        logger.warning(f"天气数据中包含错误信息：{data['error']}")
        return f"⚠️ {data['error']}"

    # 提取数据时做容错处理
    try:
        city = data.get("name", "未知")
        country = data.get("sys", {}).get("country", "未知")
        temp = data.get("main", {}).get("temp", "N/A")
        humidity = data.get("main", {}).get("humidity", "N/A")
        wind_speed = data.get("wind", {}).get("speed", "N/A")
        weather_list = data.get("weather", [{}])
        description = weather_list[0].get("description", "未知")

        logger.debug(
            f"提取天气数据 - 城市: {city}, 国家: {country}, 温度: {temp}°C, "
            f"湿度: {humidity}%, 风速: {wind_speed}m/s, 天气: {description}"
        )

        return (
            f"🌍 {city}, {country}\n"
            f"🌡️ 温度：{temp}°C\n"
            f"💧 湿度：{humidity}%\n"
            f"🌬️ 风速：{wind_speed} m/s\n"
            f"🌤️ 天气：{description}\n"
        )

    except Exception as e:
        error_msg = f"格式化天气数据失败：{str(e)}"
        logger.exception(error_msg)  # 记录堆栈信息，方便排查数据结构问题
        return error_msg


# ------------------------------
# 注册工具（示例：天气查询工具）
# ------------------------------
@mcp_server.tool()
async def query_weather(city: str) -> str:
    """
    输入指定城市的英文名称，返回今日天气查询结果。
    :param city: 城市名称（需使用英文）
    :return: 格式化后的天气信息
    """
    logger.info(f"收到天气查询请求，城市：{city}（类型：{type(city)}）")

    # 校验输入参数
    if not isinstance(city, str) or not city.strip():
        error_msg = f"无效的城市名称：{city}（必须是非空字符串）"
        logger.error(error_msg)
        return error_msg

    data = await fetch_weather(city.strip())
    return format_weather(data)


# ------------------------------
# 注册工具（示例：时间查询工具）
# ------------------------------
@mcp_server.tool()
async def get_current_time() -> str:
    """获取当前系统时间（UTC+8）"""
    from datetime import datetime, timezone, timedelta
    tz = timezone(timedelta(hours=8))  # 东八区
    current_time = datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
    return f"当前时间（UTC+8）：{current_time}"


# ------------------------------
# 注册工具（下载流媒体数据工具）
# ------------------------------
@mcp_server.tool()
def run_yt_dlp_core(url: str, output_dir: str = "./downloads", format_opt: str = "best"):
    """
    MCP框架注册的yt-dlp下载工具函数，支持视频/音频/最佳质量三种下载模式
    参数说明：
        url (str): 待下载的媒体资源URL（必填），支持YouTube、B站等yt-dlp兼容的平台
        output_dir (str, 可选): 下载文件的保存目录，默认值为当前目录下的downloads文件夹
        format_opt (str, 可选): 下载格式选项，支持三种取值：
            - "best": 默认值，下载最佳质量的完整视频（包含音频）
            - "audio": 仅下载音频，并转换为mp3格式
            - "video": 仅下载视频（最佳质量，不含音频）

    返回值：
        dict: 包含下载结果的字典，结构如下：
            - success (bool): 下载是否成功
            - message (str): 结果描述文本
            - output_dir (str): 实际使用的保存目录
            - stdout (str): 命令执行的标准输出信息
            - stderr (str): 命令执行的错误输出信息（若有）
    """
    # 确保输出目录存在，exist_ok=True表示目录已存在时不抛出异常
    # 避免因目录不存在导致下载失败
    os.makedirs(output_dir, exist_ok=True)

    platform = 'generic'

    """根据 URL 识别平台（用于动态配置）"""
    if re.search(r'(douyin|tiktok)', url, re.IGNORECASE):
        platform = 'douyin'  # 抖音/TikTok
    elif re.search(r'(bilibili|b23.tv)', url, re.IGNORECASE):
        platform = 'bilibili'  # B站
    elif re.search(r'(xiaohongshu|xhs)', url, re.IGNORECASE):
        platform = 'xiaohongshu'  # 小红书
    elif re.search(r'(youtube|youtu.be)', url, re.IGNORECASE):
        platform = 'youtube'  # YouTube

    # 1. 通用基础配置（所有平台共用）
    ydl_opts = {
        'outtmpl': os.path.join(output_dir, '%(title)s.%(ext)s'),  # 统一输出路径
        'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36',
        'quiet': True,
        'no_warnings': False,
        'logtostderr': False,
    }

    # 2. 平台专属配置（动态添加）
    if platform == 'douyin':
        # 抖音：需要 Cookie 和 referer
        ydl_opts.update({
            'referer': 'https://www.douyin.com/',
            'cookies_from_browser': ('chrome',),  # 从 Chrome 导入 Cookie
        })

    elif platform == 'xiaohongshu':
        # 小红书：需要模拟移动端 UA 和 referer
        ydl_opts.update({
            'user_agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1',
            'referer': 'https://www.xiaohongshu.com/',
        })
    elif platform == 'bilibili':
        # B站：部分视频需要登录 Cookie（可选）
        ydl_opts.update({
            'referer': 'https://www.bilibili.com/',
            # 如需下载会员视频，可添加 Cookie：
            # 'cookie_from_browser': ('chrome',),
        })
    elif platform == 'youtube':
        # YouTube：可能需要代理（根据地区配置）
        ydl_opts.update({
            'referer': 'https://www.youtube.com/',
            # 如需代理，取消注释：
            # 'proxy': 'http://127.0.0.1:7890',
        })

    # 3. 格式配置（通用，适用于所有平台）
    if format_opt == "audio":
        ydl_opts.update({
            'format': 'bestaudio/best',
            'postprocessors': [{
                'key': 'FFmpegExtractAudio',
                'preferredcodec': 'mp3',
                'preferredquality': '192',
            }],
        })
    elif format_opt == "video":
        ydl_opts['format'] = 'bestvideo'
    else:
        # ydl_opts['format'] = 'best'
        ydl_opts['format'] = 'bestvideo + bestaudio / best'

    # 4. 执行下载并捕获结果
    stdout = ""
    stderr = ""
    success = True
    try:
        with YoutubeDL(ydl_opts) as ydl:
            ydl.download([url])
            # 先提取视频信息，获取实际文件名
            info = ydl.extract_info(url, download=True)
            # 关键修复：根据格式选项获取正确的文件名
            if format_opt == "audio":
                # 音频模式：转码后的mp3文件名（基于输出模板和标题）
                title = info.get('title', 'unknown')
                # 从输出模板中提取文件名（替换占位符）
                actual_filename = ydl.prepare_filename(info).replace('.webm', '.mp3').replace('.m4a', '.mp3')
            else:
                # 视频模式：直接用原生文件名
                actual_filename = ydl.prepare_filename(info)
    except Exception as e:
        success = False
        stderr = f"[{platform}] 下载失败：{str(e)}"

    content = transcribe_audio_to_text(actual_filename)
    # 构造返回结果：将文件加入 files 数组，文本信息加入 text
    return content


def transcribe_audio_to_text(audio_path: str, model_size: str = "base") -> str:
    """
    使用 faster-whisper 将音频文件转换为连续文本

    参数:
        audio_path: 音频文件路径（支持 mp3、wav 等格式）
        language: 识别语言（默认 "zh" 中文，其他如 "en" 英文）
        model_size: 模型大小（"tiny"|"base"|"small"|"medium"|"large"，默认 "base"）

    返回:
        str: 连续的识别文本（无时间戳）
    """
    try:
        # 加载模型
        model = WhisperModel(model_size, device="cpu", compute_type="int8")
        # model_size_or_path = "D:/custom_model_path/faster-whisper-base",
        # model = WhisperModel("large", device="cuda", compute_type="float16")  # GPU 推荐 float16

        # 转录音频
        segments, _ = model.transcribe(audio_path)

        # 拼接所有片段文本（去除时间戳）
        full_text = " ".join([segment.text.strip() for segment in segments])

        return full_text

    except Exception as e:
        print(f"语音识别失败: {str(e)}")
        return ""


# ------------------------------
# 启动流式 HTTP 服务器
# ------------------------------
if __name__ == "__main__":
    try:
        # 运行服务器（使用官方 transport）
        mcp_server.run(transport='streamable-http')

    except Exception as e:
        logger.critical(f"服务器启动失败：{str(e)}", exc_info=True)

1. Dify配置MCP工具
2. 本地MCP域名设置（如果本地使用Docker启动Dify，同时MCP服务也在本地，需要此配置）

hosts文件位置

C:\Windows\System32\drivers\etc\hosts

1. Chatflow配置