Hugging Face Inference Endpoint 怎么编写自定义 Handler

Hugging Face Inference Endpoint 的自定义 Handler，核心不是重写推理服务，而是在模型仓库根目录放置 handler.py，由 EndpointHandler 接住请求、加载模型并返回可序列化结果。需要额外库时再添加 requirements.txt，部署时把任务设为 Custom。这样可以在标准模型调用之外加入字段校验、预处理、后处理或批量逻辑。

官方地址：https://huggingface.co/docs/inference-endpoints/guides/custom_handler

一、先按 EndpointHandler 约定组织仓库

自定义 Handler 适合“模型能用，但默认任务接口不够用”的场景。例如输入除了文本还带日期、租户配置或预处理选项。仓库至少应能让运行时找到 handler.py，并在类中接收模型文件路径：

from typing import Any
from transformers import pipeline

class EndpointHandler:
    def __init__(self, path: str = ""):
        # 启动时加载一次模型，path 指向当前 Endpoint 下载的模型目录。
        self.classifier = pipeline("text-classification", model=path)

    def __call__(self, data: dict[str, Any]) -> list[dict[str, Any]]:
        # 明确读取 inputs，其他字段交给本 Handler 的业务规则处理。
        inputs = data.get("inputs")
        if not isinstance(inputs, str) or not inputs.strip():
            raise ValueError("inputs must be a non-empty string")
        # 返回值保持为 JSON 可序列化的列表，便于 Endpoint 直接响应。
        return self.classifier(inputs)

__init__ 收到的路径用于定位模型权重；不要在 __call__ 中反复创建 pipeline。data 通常包含 inputs，也可以带额外键。输入错误应尽早失败，返回值则保持字典、列表、字符串或数字等可序列化结构。

二、把自定义字段和依赖放在清晰边界内

自定义逻辑最好分成三段：取出请求字段、执行额外判断、调用模型并整理结果。下面的示例用 priority 控制简单的结果标记，重点是字段边界，不依赖某个业务系统：

class EndpointHandler:
    def __init__(self, path: str = ""):
        # 模型和只读配置在启动阶段准备，减少请求路径上的开销。
        self.classifier = pipeline("text-classification", model=path)

    def __call__(self, data: dict[str, Any]) -> dict[str, Any]:
        # 保留原始输入的类型检查，避免把任意对象传进模型。
        text = data.get("inputs")
        priority = data.get("priority", "normal")
        if not isinstance(text, str) or not text.strip():
            raise ValueError("inputs must be a non-empty string")
        if priority not in {"normal", "high"}:
            raise ValueError("priority must be normal or high")
        # 自定义字段影响响应元数据，但不改变模型的输入格式。
        return {"priority": priority, "predictions": self.classifier(text)}

如果 Handler 用到 holidays、特定音频库或其他第三方包，把包名写进仓库根目录的 requirements.txt，并在本地环境安装同一份依赖。不要把 Token、私人路径或运行时密钥写入代码；需要秘密配置时使用 Endpoint 的 Secret Environment Variables。

三、先本地调用，再部署 Custom 任务

本地验证的目标是确认构造阶段能加载模型、请求字段能被消费、返回对象可以序列化。用一个最小脚本即可检查这三点：

# 在模型仓库目录安装与线上一致的依赖。
python -m pip install -r requirements.txt

# 用本地目录初始化 Handler，再传入一个完整请求对象。
python -c 'from handler import EndpointHandler; h=EndpointHandler("."); print(h({"inputs":"a short test","priority":"normal"}))'

确认本地调用后，把 handler.py 和 requirements.txt 提交到模型仓库，再创建 Inference Endpoint。配置时选择模型仓库、云区域和实例；Advanced configuration 中把 Task 设为 Custom，并固定一个 revision，便于后续回滚。Endpoint 启动后再用真实请求检查字段和响应结构。

四、常见边界：启动失败和请求失败要分开排查

如果只是标准 Transformers 或 Sentence-Transformers 任务，优先使用托管任务；只有需要自定义输入、输出或推理管线时才引入 Handler。Handler 解决的是接口和推理逻辑的可控性，不会替你选择合适的硬件、修复模型权重或隐藏依赖版本冲突。

相关问题

自定义 Handler 一定要把任务设为 Custom 吗？

应按官方 Handler 方式部署，并在 Endpoint 配置中选择 Custom，让自定义入口优先于默认任务推理。

__init__ 的 path 是什么？

它是运行时下载到容器中的模型目录路径，用于让 Handler 从当前仓库加载权重和配置。

能不能在 Handler 中实现批量推理？

可以，但要先定义请求数组、最大批大小和超时边界，并保证返回结果能按输入顺序对应。