摘要

本文详细阐述了基于 Dify 平台开发 Embedding 模型插件的完整流程，包括 API 申请、开发环境搭建、项目构建、核心代码编写、常见问题解决及功能验证与性能优化。通过标准化开发步骤与问题排查方案，确保插件可正常调用公司内部 Embedding 模型，并通过参数调优提升模型检索效果，为后续知识库构建与语义检索应用提供技术支撑。

目录

1. 引言
2. Embedding 模型 API 申请
3. 开发环境准备
4. 插件项目构建
5. 核心代码编写与配置
6. 插件功能验证与性能优化
7. 常见问题解决方案
8. 结论

1. 引言

本文档旨在为开发人员提供基于 Dify 平台集成公司内部 Embedding 模型的标准化开发指南。文档涵盖从前期 API 申请、环境搭建，到中期项目构建、代码编写，再到后期功能验证、性能优化的全流程操作，确保开发人员可按步骤完成插件开发与部署，实现 Embedding 模型在 Dify 平台的稳定调用。

2. Embedding 模型 API 申请

需先完成公司内部 Embedding 模型 API 的申请与测试验证，具体步骤如下：

1. 访问公司内部 API 文档地址，按流程提交 API 使用申请；
2. 申请通过后，获取 API 调用所需的密钥、请求地址等参数；
3. 开展内部测试，通过发送测试请求验证 API 可用性，确保可正常接收模型返回的 Embedding 向量数据，测试通过后方可进入后续开发。

3. 开发环境准备

开发 Dify 插件需提前配置基础环境，确保工具与版本满足开发要求，具体包含以下三类核心组件：

3.1 Dify 插件脚手架工具

1. 访问 Dify Plugin CLI 项目官方地址，下载与当前操作系统匹配的最新版本脚手架工具；
2. 本文以 Windows 系统为例展开开发演示，Windows 环境需下载 “dify-plugin-windows-amd64.exe” 可执行文件；
3. 脚手架版本校验：切换至脚手架可执行文件所在目录，在 PowerShell 终端中执行以下命令，确认工具正常运行：

   powershell

   ./dify-plugin-windows-amd64.exe version
   

3.2 Python 环境配置

1. 安装 Python 环境，要求版本≥3.12，建议使用 Python 3.12.x 稳定版；
2. 安装完成后，通过终端执行 “python --version” 或 “python3 --version” 命令，验证版本是否符合要求；
3. 按需安装 pip 包管理工具，确保后续依赖包可正常安装。

3.3 Dify 平台准备

1. 部署或访问已搭建的 Dify 平台环境，确保平台可正常登录与操作；
2. 提前创建开发所需的平台账号，并授予插件开发相关权限（如模型供应商配置、知识库操作权限）。

4. 插件项目构建

插件项目构建需完成初始化配置，包括项目信息填写、开发语言选择、模板与权限设置，具体步骤如下：

4.1 插件初始化

1. 在 PowerShell 终端中，切换至目标项目目录，执行以下初始化命令：

   powershell

   dify-plugin-windows-amd64.exe plugin init
   

2. 按终端提示依次输入项目核心信息，输入完成后按 Enter 键进入下一步：
   • Plugin Name（插件名称）：需明确标识插件功能，如 “internal-embedding-plugin”；
   • Author（作者信息）：填写开发人员姓名或团队名称；
   • Description（插件描述）：简要说明插件功能，如 “集成公司内部 Embedding 模型的 Dify 插件，支持文本向量生成与语义检索”。

4.2 开发语言选择

通过键盘↑↓键移动光标，选择插件开发语言，本文以 Python 为例进行后续开发。

4.3 模板选择

在模板列表中选择 “text-embedding” 模板，该模板适配 Embedding 模型的向量生成功能，可减少基础代码开发量。

4.4 权限配置与项目创建

1. 通过键盘↑↓键调整权限选项，按 Tab 键确认选择（参考官方推荐的 Embedding 插件权限配置）；
2. 输入插件版本号（初始版本建议设为 0.0.1），完成后终端将自动创建插件项目框架，生成 “plugin_demo”（以实际填写的插件名称为准）工程目录。

5. 核心代码编写与配置

使用 PyCharm（或其他 Python IDE）打开插件工程，重点完成环境配置文件、供应商代码、模型功能代码的编写与修改，确保插件可正常调用内部 Embedding 模型。

5.1 环境配置文件（.env）设置

1. 在工程根目录中，复制 “.env.example” 文件并重新命名为 “.env”；
2. 打开 “.env” 文件，填入远程服务器地址、调试 Key 等配置信息，示例如下：

   env

   INSTALL_METHOD=remote
   REMOTE_INSTALL_HOST=YOUR_HOST  # 替换为实际远程服务器地址
   REMOTE_INSTALL_PORT=5003       # 替换为实际端口号
   REMOTE_INSTALL_KEY=****-****-****-****-****  # 替换为实际调试Key
   

5.2 供应商目录（provider）配置

provider 目录包含插件功能描述文件（.yaml）与供应商逻辑代码文件（.py），需按以下要求修改：

5.2.1 YAML 文件修改

YAML 文件用于定义插件工具信息与作者信息，核心修改项如下：

1. 图标路径调整：将 “_assets” 目录下的图标文件移动至工程根目录，并更新 YAML 文件中图标路径的配置；
2. API 识别字段优化：修改 API 相关识别字段的命名，确保与后续开发的代码逻辑一致，便于维护；
3. 模型访问路径补充：在 YAML 文件中填写内部 Embedding 模型的访问路径，默认配置中该路径为空，需手动补充。

5.2.2 Python 文件编写

Python 文件用于实现供应商逻辑，核心功能如下：

1. 第三方 API 密钥验证：若插件需调用第三方 API，需在代码中添加密钥验证逻辑，验证不通过时抛出异常；若仅调用公司内部模型，此部分逻辑可省略；
2. 核心函数声明：预留后续 Embedding 功能实现的函数接口，确保与 models 目录下的代码逻辑联动。

5.3 模型功能代码（models 目录）实现

models 目录下的 Python 文件是 Embedding 功能的核心实现载体，需编写以下关键函数与新增函数：

5.3.1 关键函数实现

1. _invoke函数：实现 Embedding 模型的调用逻辑，包括请求参数组装、模型接口调用、返回向量数据解析；同时添加错误识别机制，当调用失败时触发重试逻辑（建议设置 3 次以内重试次数，避免无限循环）；
2. get_num_tokens函数：计算输入文本列表中每个文本的 Token 数量，需适配内部模型的 Token 编码规则；
3. validate_credentials函数：验证 API 密钥的格式合法性，确保密钥符合公司内部 API 的规范（如长度、字符类型等）。

5.3.2 新增函数实现

1. _invoke_error_mapping函数：定义错误信息映射规则，将模型返回的错误码转换为易于理解的文字描述，便于开发调试与问题排查；
2. get_customizable_model_schema函数：实现模型认证逻辑，在函数中添加标识字段，明确该插件为 Embedding 类型模型，确保 Dify 平台可正确识别插件类别。

5.3.3 代码实现

```python
import time
from typing import Optional
import requests
from dify_plugin import TextEmbeddingModel
from dify_plugin.entities import I18nObject
from dify_plugin.entities.model import EmbeddingInputType, AIModelEntity, ModelType, FetchFrom
from dify_plugin.errors.model import CredentialsValidateFailedError, InvokeError
from dify_plugin.entities.model.text_embedding import (
    TextEmbeddingResult, EmbeddingUsage,
)
 
 
class EmbeddingTextEmbeddingModel(TextEmbeddingModel):
    
 
    def _invoke(
            self,
            model: str,
            credentials: dict,
            texts: list[str],
            user: Optional[str] = None,
            input_type: EmbeddingInputType = EmbeddingInputType.DOCUMENT,
    ) -> TextEmbeddingResult:
        """调用文本嵌入API生成向量"""
        # 调试日志初始化
        print("开始调试\n")
 
        # 测试用硬编码（生产环境需从credentials读取）
        model = "XXX"
        api_key = "sk-XXXX"
 
        # 打印关键参数调试信息
        print(f"DEBUG: API Key: {api_key}")
        print(f"DEBUG: Model: {model}")
        print("DEBUG: 第一步完成：已验证模型与API\n")
 
        # 验证API密钥
        if not api_key:
            raise InvokeError("缺少API凭证：请提供api_key（格式应为sk-xxx）")
 
        # 准备请求参数
        inputs = texts
        url = "https://eisapi.byd.com/open-api/1.0/llm/v1/common-embeddings"
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
 
        # 记录调用开始时间
        start_time = time.perf_counter()
 
        # 构造请求体
        data = {
            "model": model,
            "input": inputs
        }
        print(f"DEBUG:data: {inputs}\n")
        print("DEBUG: 第二步完成：请求体构建完成\n")
 
        # 带重试机制的API调用
        max_retries = 20
        for attempt in range(max_retries):
            try:
                response = requests.post(
                    url=url,
                    json=data,
                    headers=headers,
                    timeout=60
                )
                print(f"DEBUG: Response status: {response.status_code}")
                print(f"DEBUG: Response headers: {response.headers}")
 
                # 处理限流错误（429）
                if response.status_code == 429:
                    if attempt < max_retries - 1:
                        wait_time = 2 **attempt
                        print(f"DEBUG: 触发限流，等待{wait_time}秒后重试")
                        time.sleep(wait_time)
                        continue
                    else:
                        raise InvokeError(f"API限流，状态码: {response.status_code}, 响应: {response.text}")
 
                # 处理非200状态码
                if response.status_code != 200:
                    raise InvokeError(f"API错误，状态码: {response.status_code}, 响应: {response.text}")
 
                # 验证响应格式
                content_type = response.headers.get('content-type', '')
                if 'application/json' not in content_type:
                    raise InvokeError(f"非JSON响应: {content_type}, 内容: {response.text}")
 
                # 验证非空响应
                if not response.text.strip():
                    raise InvokeError("API返回空响应")
 
                # 解析JSON响应
                try:
                    resp = response.json()
                except ValueError as e:
                    print(f"DEBUG: JSON解析失败，响应: {response.text}")
                    raise InvokeError(f"无效JSON格式: {str(e)}")
 
                break  # 成功获取响应，退出重试循环
 
            except requests.exceptions.Timeout:
                if attempt < max_retries - 1:
                    print("DEBUG: 请求超时，重试...")
                    continue
                else:
                    raise InvokeError("API请求超时")
            except requests.exceptions.ConnectionError:
                if attempt < max_retries - 1:
                    print("DEBUG: 连接错误，重试...")
                    time.sleep(1)
                    continue
                else:
                    raise InvokeError("无法连接到API服务器")
            except requests.exceptions.RequestException as e:
                if attempt < max_retries - 1:
                    print("DEBUG: 请求异常，重试...")
                    time.sleep(1)
                    continue
                else:
                    raise InvokeError(f"API请求失败: {str(e)}")
 
        print("DEBUG: 第三步完成：调用API成功")
 
        # 解析嵌入向量
        if "data" not in resp:
            raise InvokeError(f"响应缺少data字段，完整响应: {resp}")
        
        embeddings_data = resp["data"]
        embeddings = []
        for item in embeddings_data:
            if isinstance(item, dict) and "embedding" in item:
                embeddings.append(item["embedding"])
            elif isinstance(item, list):
                embeddings.append(item)
            else:
                embeddings.append(item)
 
        print("DEBUG: 第四步完成：Embedding成功")
 
        # 验证向量提取结果
        if not embeddings:
            raise InvokeError("无法从响应中提取嵌入向量")
 
        # 计算使用量
        usage_dict = resp.get("usage", {})
        total_tokens = usage_dict.get("total_tokens", 0)
        latency = round(time.perf_counter() - start_time, 4)
        
        usage = EmbeddingUsage(
            tokens=total_tokens,
            total_tokens=total_tokens,
            latency=latency,
            unit_price=0.0,
            price_unit=1.0,
            total_price=0.0,
            currency="USD"
        )
 
        print("DEBUG: 第五步完成：token已计算")
        print(f"DEBUG: Usage: {usage}")
 
        # 构造返回结果
        result = TextEmbeddingResult(
            model=model,
            embeddings=embeddings,
            usage=usage,
        )
 
        print("DEBUG: 已完成")
        return result
 
    def get_num_tokens(self, model: str, credentials: dict, texts: list[str]) -> list[int]:
        """估算文本token数（用字符数粗略替代）"""
        return [len(text) for text in texts]
 
    def validate_credentials(self, model: str, credentials: dict) -> None:
        """验证API凭证格式"""
        try:
            api_key = credentials.get("api_key")
            if not api_key:
                raise CredentialsValidateFailedError("API密钥不能为空")
            if not isinstance(api_key, str) or not api_key.startswith("sk-"):
                raise CredentialsValidateFailedError("API密钥格式不正确，应以'sk-'开头")
        except Exception as ex:
            raise CredentialsValidateFailedError(f"凭证验证失败: {str(ex)}")
 
    @property
    def _invoke_error_mapping(self) -> dict[type[InvokeError], list[type[Exception]]]:
        return {InvokeError: [Exception]}
 
    def get_customizable_model_schema(
            self, model: str, credentials: dict
    ) -> AIModelEntity:
        """返回模型元信息"""
        return AIModelEntity(
            model=model,
            label=I18nObject(zh_Hans=model, en_US=model),
            model_type=ModelType.TEXT_EMBEDDING,
            features=[],
            fetch_from=FetchFrom.CUSTOMIZABLE_MODEL,
            model_properties={},
            parameter_rules=[],
        )
```

5.4 插件本地调试

在 PyCharm 终端中执行以下命令，启动插件本地调试，验证代码是否可正常运行：

powershell

python -m main

若终端输出 “插件本地部署成功” 相关提示信息，说明本地调试通过；若出现报错，需根据错误信息排查代码逻辑或配置文件。

6. 插件功能验证与性能优化

插件本地调试通过后，需在 Dify 平台完成功能验证，并针对检索效果进行性能优化，确保插件满足实际应用需求。

6.1 Dify 平台插件部署验证

1. 登录 Dify 平台，若界面显示 “DEBUGGING PLUGIN” 标识，说明插件已成功部署至平台；
2. 配置 API 与模型：点击平台右上角头像→进入 “设置” 页面→选择 “模型供应商”→找到 “Embedding” 类别→填入公司内部 Embedding 模型的 API Key 与模型名称，完成配置。

6.2 插件功能验证

通过知识库功能验证插件是否可正常调用模型，步骤如下：

1. 进入 Dify 平台 “知识库” 模块，点击 “选择文件” 上传测试文档；
2. 在文档处理配置中，默认选择已部署的 Embedding 模型；
3. 执行文档向量生成操作，若可正常生成向量数据且无报错，说明插件功能正常。

6.3 检索性能优化

初始召回测试结果显示，模型检索效果较差（召回率约 0.3），需通过参数调优提升性能，具体优化措施如下：

1. 调整文本切片大小：将原始切片大小（1024 Token）调整为 102 Token，优化后召回率提升至 0.5 左右；
2. 优化混合检索语义比例：将混合检索中语义检索的比例调整为 1（即纯语义检索），优化后召回率进一步提升至 0.7，满足基础检索需求。

7. 常见问题解决方案

在插件开发与部署过程中，易出现版本不匹配、编码格式错误等问题，以下为两类典型问题的解决方案：

7.1 gevent 版本与 dify-plugin 版本不匹配

问题现象

终端报错提示 gevent 版本与 dify-plugin 版本不兼容，通常表现为 dify-plugin 版本为 0.4.1（低于需求的 0.5.0 版本）。

解决方案

1. 离线包下载：在有网络的机器上，执行以下命令下载指定版本的 gevent 与 dify-plugin 离线包：

   powershell

   pip download --no-binary gevent gevent==25.5.1 dify-plugin==0.5.0
   

2. 容器操作：
   • 查看 Dify 插件容器名称：执行 “docker ps | grep plugin”，通常容器名为 “dify-plugin-daemon-1”；
   • 进入容器内部：执行 “docker exec -it dify-plugin-daemon-1 bash”；
   • 验证当前版本：执行 “pip list | grep gevent” 与 “pip list | grep dify_plugin”，确认版本是否低于目标版本；
3. 离线包部署：
   • 将离线包复制至容器内：执行 “docker cp 宿主机离线包路径 dify-plugin-daemon-1:/tmp/offline_packages”；
   • 进入容器内离线包目录：执行 “cd /tmp/offline_packages”；
   • 本地安装：执行以下命令强制从源码编译安装指定版本：

     powershell

     pip install --no-index --find-links=. --no-binary gevent gevent==25.5.1 dify-plugin==0.5.0
     

7.2 插件包编码格式错误

问题现象

平台提示无法读取插件包信息，排查发现是 “requirements.txt” 文件编码格式不兼容。

解决方案

使用文本编辑器（如 Notepad++、VS Code）打开 “requirements.txt” 文件，将编码格式修改为 UTF-8，保存后重新上传插件包即可解决。

8. 结论

本文通过标准化流程完成了基于 Dify 平台的 Embedding 模型插件开发，从 API 申请、环境搭建到代码编写、功能验证，形成了完整的技术闭环。针对开发过程中的版本不匹配、编码错误等问题，提供了可落地的解决方案；同时通过切片大小与检索比例调优，将模型召回率从 0.3 提升至 0.7，显著改善了检索性能。后续可进一步优化模型调用效率与容错机制，提升插件在高并发场景下的稳定性。