1. 为什么需要插件机制

在前三篇文章中，我们已经掌握了 DeepSeek Harness 的基础配置、Prompt 管理与评测流程。但在真实业务落地时，往往会遇到这些场景：

• 公司有内部自研的评测数据集格式，不想每次都手动转换成标准格式；
• 需要对模型输出做自定义后处理（如敏感词过滤、业务规则校验）；
• 想把 DeepSeek Harness 的评测结果自动上报到内部监控平台；
• 需要接入公司内部的模型网关，而不是直接调用 DeepSeek 官方 API。

如果每遇到一个新需求就修改 Harness 源码，维护成本极高，升级上游版本也会变得困难。插件机制（Plugin） 正是为了解决这类「扩展点」问题而设计的。它允许你在不修改核心代码的前提下，通过实现标准接口来扩展 Harness 的能力。

本文是系列第四篇，将系统讲解 DeepSeek Harness 的插件体系，并通过三个完整的代码实战案例，带你从零开发一个可用的插件。

2. 插件体系总览

DeepSeek Harness 的插件体系借鉴了「微内核 + 插件」的架构思想。核心框架只负责编排评测流程，而将可变的环节抽象为插件接口。

整个插件体系由五个核心扩展点构成：

插件类型	接口名称	职责	典型使用场景
数据集插件	DatasetPlugin	将任意格式数据转换为标准评测样本	接入私有数据集、数据库直读
模型插件	ModelPlugin	封装模型调用逻辑	接入内网模型网关、本地模型
后处理插件	PostProcessPlugin	对模型原始输出做二次加工	敏感词过滤、格式规范化
指标插件	MetricPlugin	扩展自定义评测指标	业务专属打分规则
上报插件	ReporterPlugin	将评测结果输出到外部系统	推送到监控平台、写入数据库

说明：以上接口名称是 DeepSeek Harness 插件体系的抽象设计。在实际接入时，不同版本的接口模块路径可能略有差异，请以你使用的具体版本源码中的类名为准。本文代码基于社区常见的插件设计模式编写，旨在帮助读者理解接入思路与开发范式。

3. 环境准备与插件目录结构

在动手之前，先规划好插件的目录结构。推荐在项目根目录下创建 plugins/ 目录：

deepseek-harness/
├── configs/                 # 评测配置目录
│   └── eval_plugin.yaml
├── plugins/                 # 自定义插件目录
│   ├── __init__.py
│   ├── dataset/             # 数据集插件
│   │   ├── __init__.py
│   │   └── jsonlines_dataset.py
│   ├── model/               # 模型插件
│   │   ├── __init__.py
│   │   └── gateway_model.py
│   ├── postprocess/         # 后处理插件
│   │   ├── __init__.py
│   │   └── sensitive_filter.py
│   └── reporter/            # 上报插件
│       ├── __init__.py
│       └── webhook_reporter.py
├── main.py                  # 评测入口
└── requirements.txt

每个插件文件只负责一种职责，遵循单一职责原则，方便后续维护和测试。

4. 核心接口解析

在开发插件之前，先理解 Harness 的插件基类设计。以下是核心接口的精简实现：

# plugins/base.py
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field


@dataclass
class PluginContext:
    """插件上下文：承载评测过程中的共享数据"""
    config: Dict[str, Any] = field(default_factory=dict)
    dataset_path: Optional[str] = None
    model_name: Optional[str] = None
    # 存放插件间的临时共享状态
    shared_state: Dict[str, Any] = field(default_factory=dict)

    def get(self, key: str, default=None):
        return self.shared_state.get(key, default)

    def set(self, key: str, value):
        self.shared_state[key] = value


class BasePlugin(ABC):
    """所有插件的基类"""

    # 插件唯一标识，子类必须覆盖
    plugin_name: str = "base_plugin"
    # 插件版本号，建议语义化版本
    plugin_version: str = "1.0.0"

    def __init__(self, context: Optional[PluginContext] = None):
        self.context = context or PluginContext()

    @abstractmethod
    def setup(self, config: Dict[str, Any]) -> None:
        """插件初始化：从配置中读取参数"""
        raise NotImplementedError

    @abstractmethod
    def teardown(self) -> None:
        """插件清理：释放资源"""
        raise NotImplementedError


@dataclass
class EvalSample:
    """标准评测样本"""
    input: str
    reference: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    sample_id: Optional[str] = None


class DatasetPlugin(BasePlugin):
    """数据集插件基类"""

    plugin_type = "dataset"

    @abstractmethod
    def load(self) -> List[EvalSample]:
        """加载数据集，返回标准样本列表"""
        raise NotImplementedError


class ModelPlugin(BasePlugin):
    """模型插件基类"""

    plugin_type = "model"

    @abstractmethod
    def generate(self, prompt: str, **kwargs) -> str:
        """生成模型输出"""
        raise NotImplementedError


class PostProcessPlugin(BasePlugin):
    """后处理插件基类"""

    plugin_type = "postprocess"

    @abstractmethod
    def process(self, output: str, sample: EvalSample) -> str:
        """对模型输出进行后处理"""
        raise NotImplementedError


class ReporterPlugin(BasePlugin):
    """上报插件基类"""

    plugin_type = "reporter"

    @abstractmethod
    def report(self, results: List[Dict[str, Any]]) -> None:
        """上报评测结果"""
        raise NotImplementedError

这些基类定义了插件的最小契约。接下来，我们以三个真实场景为例，逐一开发并接入插件。

5. 实战一：自定义数据集插件

5.1 需求场景

假设公司内部有一个 JSON Lines 格式的客服对话评测集，结构如下：

{"session_id": "sess_001", "question": "我的订单什么时候发货？", "answer": "您的订单将在 48 小时内发货。"}
{"session_id": "sess_002", "question": "如何申请退货？", "answer": "请在订单详情页点击申请退货。"}
{"session_id": "sess_003", "question": "优惠券过期了还能用吗？", "answer": "过期优惠券无法使用，请关注新活动。"}

而 Harness 默认只支持标准 JSON 格式的评测集。通过数据集插件，我们可以直接读取这种私有格式，避免频繁转换。

5.2 插件实现

# plugins/dataset/jsonlines_dataset.py
import json
import os
from typing import Any, Dict, List
from plugins.base import DatasetPlugin, EvalSample, PluginContext


class JsonLinesDatasetPlugin(DatasetPlugin):
    """从 JSON Lines 文件中加载客服对话评测集"""

    plugin_name = "jsonlines_customer_service"
    plugin_version = "1.0.0"

    def __init__(self, context: PluginContext = None):
        super().__init__(context)
        self.file_path: str = ""
        self.max_samples: int = -1

    def setup(self, config: Dict[str, Any]) -> None:
        """
        config 示例:
        {
            "file_path": "data/customer_service.jsonl",
            "max_samples": 100
        }
        """
        self.file_path = config.get("file_path", "")
        self.max_samples = int(config.get("max_samples", -1))

        if not self.file_path:
            raise ValueError(
                f"[{self.plugin_name}] 缺少必要参数: file_path"
            )

        if not os.path.exists(self.file_path):
            raise FileNotFoundError(
                f"[{self.plugin_name}] 数据集文件不存在: {self.file_path}"
            )

    def load(self) -> List[EvalSample]:
        samples: List[EvalSample] = []
        with open(self.file_path, "r", encoding="utf-8") as f:
            for line_num, line in enumerate(f, start=1):
                line = line.strip()
                if not line:
                    continue

                try:
                    item = json.loads(line)
                except json.JSONDecodeError as e:
                    print(f"警告: 第 {line_num} 行 JSON 解析失败: {e}")
                    continue

                sample = EvalSample(
                    input=item.get("question", ""),
                    reference=item.get("answer", ""),
                    sample_id=item.get("session_id", f"line_{line_num}"),
                    metadata={
                        "source": "jsonlines",
                        "file": self.file_path,
                        "line": line_num,
                    },
                )

                if sample.input and sample.reference:
                    samples.append(sample)

                # 达到最大样本数限制则提前终止
                if 0 < self.max_samples <= len(samples):
                    break

        print(f"[{self.plugin_name}] 成功加载 {len(samples)} 条样本")
        return samples

    def teardown(self) -> None:
        # JSON 文件读取无需额外释放资源
        self.context = None

5.3 注册与配置

插件开发完成后，需要在评测配置中声明使用：

# configs/eval_plugin.yaml
dataset:
  plugin: jsonlines_customer_service
  config:
    file_path: "data/customer_service.jsonl"
    max_samples: 100

model:
  plugin: builtin_deepseek
  config:
    model_name: "deepseek-chat"
    temperature: 0.1

pipeline:
  - type: "generate"
  - type: "postprocess"
    plugins:
      - "sensitive_filter"

reporter:
  plugin: webhook_reporter
  config:
    url: "https://monitor.example.com/api/eval-report"

在 Harness 入口中完成插件的注册与装配：

# main.py
from plugins.base import PluginContext
from plugins.dataset.jsonlines_dataset import JsonLinesDatasetPlugin
from plugins.postprocess.sensitive_filter import SensitiveFilterPlugin
from plugins.reporter.webhook_reporter import WebhookReporterPlugin


# 插件注册表
PLUGIN_REGISTRY = {}


def register_plugin(plugin_cls):
    """将插件类注册到全局注册表"""
    instance = plugin_cls()
    key = instance.plugin_name
    PLUGIN_REGISTRY[key] = plugin_cls
    print(f"注册插件: {key} (v{instance.plugin_version})")
    return plugin_cls


# 注册所有自定义插件
register_plugin(JsonLinesDatasetPlugin)
register_plugin(SensitiveFilterPlugin)
register_plugin(WebhookReporterPlugin)


def create_plugin(plugin_type: str, plugin_name: str,
                  context: PluginContext, config: dict):
    """根据注册表创建插件实例"""
    plugin_cls = PLUGIN_REGISTRY.get(plugin_name)
    if plugin_cls is None:
        raise KeyError(f"未找到插件: {plugin_name}（请确认已注册）")

    plugin = plugin_cls(context=context)
    plugin.setup(config)
    return plugin

6. 实战二：模型网关插件

6.1 需求场景

很多企业出于安全合规考虑，要求所有模型调用必须经过内部网关，而不是直接访问 DeepSeek 官方 API。通过模型插件，我们可以把网关调用逻辑封装起来，Harness 感知不到底层差异。

6.2 插件实现

# plugins/model/gateway_model.py
import time
import hashlib
import hmac
from typing import Any, Dict
import requests
from plugins.base import ModelPlugin, PluginContext


class GatewayModelPlugin(ModelPlugin):
    """通过内部模型网关调用 DeepSeek 模型"""

    plugin_name = "internal_gateway_deepseek"
    plugin_version = "1.2.0"

    def __init__(self, context: PluginContext = None):
        super().__init__(context)
        self.gateway_url: str = ""
        self.api_key: str = ""
        self.api_secret: str = ""
        self.model_name: str = "deepseek-chat"
        self.timeout: int = 60
        self.max_retries: int = 3

    def setup(self, config: Dict[str, Any]) -> None:
        """
        config 示例:
        {
            "gateway_url": "https://llm-gateway.internal.com/v1/chat",
            "api_key": "xxx",
            "api_secret": "yyy",
            "model_name": "deepseek-chat",
            "timeout": 60
        }
        """
        self.gateway_url = config.get("gateway_url", "")
        self.api_key = config.get("api_key", "")
        self.api_secret = config.get("api_secret", "")
        self.model_name = config.get("model_name", "deepseek-chat")
        self.timeout = int(config.get("timeout", 60))
        self.max_retries = int(config.get("max_retries", 3))

        if not self.gateway_url:
            raise ValueError(
                f"[{self.plugin_name}] 缺少必要参数: gateway_url"
            )

    def _generate_signature(self, timestamp: str, body: str) -> str:
        """生成网关请求签名（HMAC-SHA256）"""
        message = f"{timestamp}\n{body}"
        signature = hmac.new(
            self.api_secret.encode("utf-8"),
            message.encode("utf-8"),
            hashlib.sha256,
        ).hexdigest()
        return signature

    def _call_gateway(self, prompt: str) -> str:
        """单次网关调用"""
        timestamp = str(int(time.time()))
        payload = {
            "model": self.model_name,
            "messages": [
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.1,
        }
        body = str(payload)
        signature = self._generate_signature(timestamp, body)

        headers = {
            "Content-Type": "application/json",
            "X-Api-Key": self.api_key,
            "X-Timestamp": timestamp,
            "X-Signature": signature,
        }

        response = requests.post(
            self.gateway_url,
            json=payload,
            headers=headers,
            timeout=self.timeout,
        )
        response.raise_for_status()

        data = response.json()
        return data["choices"][0]["message"]["content"]

    def generate(self, prompt: str, **kwargs) -> str:
        """生成模型输出，带重试机制"""
        last_error: Exception | None = None
        for attempt in range(1, self.max_retries + 1):
            try:
                output = self._call_gateway(prompt)
                return output
            except requests.exceptions.HTTPError as e:
                last_error = e
                print(f"[{self.plugin_name}] 第 {attempt} 次调用失败: {e}")
                if attempt < self.max_retries:
                    # 指数退避重试
                    time.sleep(2 ** attempt)
                else:
                    break
            except requests.exceptions.Timeout as e:
                last_error = e
                print(f"[{self.plugin_name}] 第 {attempt} 次调用超时: {e}")
                if attempt < self.max_retries:
                    time.sleep(3)
                else:
                    break

        raise RuntimeError(
            f"[{self.plugin_name}] 网关调用最终失败: {last_error}"
        )

    def teardown(self) -> None:
        # 清理可能存在的连接池
        self.context.set("gateway_active", False)

6.3 对比：与直接调用官方 API 的差异

维度	直接调用官方 API	网关插件接入
网络路径	客户端 → api.deepseek.com	客户端 → 内网网关 → api.deepseek.com
鉴权方式	单一 API Key	API Key + HMAC 签名
日志审计	客户端自行记录	网关统一记录
限流策略	依赖官方限流	网关统一限流与排队
Harness 改动	需要改核心模型调用模块	零改动，仅注册插件

7. 实战三：输出后处理插件

7.1 需求场景

在客服场景的评测中，模型输出可能包含不规范的格式或敏感词。后处理插件可以在评分之前对输出进行清洗和过滤，保证评测的一致性。

7.2 敏感词过滤与格式清洗插件实现

# plugins/postprocess/sensitive_filter.py
import re
from typing import Any, Dict, List
from plugins.base import PostProcessPlugin, EvalSample, PluginContext


class SensitiveFilterPlugin(PostProcessPlugin):
    """对模型输出进行敏感词过滤与格式清洗"""

    plugin_name = "sensitive_filter"
    plugin_version = "1.1.0"

    # 默认敏感词列表（实际场景建议配置化或从远程加载）
    DEFAULT_SENSITIVE_WORDS = [
        "内部机密",
        "测试账号",
        "未发布功能",
    ]

    def __init__(self, context: PluginContext = None):
        super().__init__(context)
        self.sensitive_words: List[str] = []
        self.replacement: str = "***"
        self.strip_whitespace: bool = True
        self.remove_urls: bool = True

    def setup(self, config: Dict[str, Any]) -> None:
        """
        config 示例:
        {
            "sensitive_words": ["内部机密", "测试账号"],
            "replacement": "[已过滤]",
            "strip_whitespace": true,
            "remove_urls": true
        }
        """
        self.sensitive_words = config.get(
            "sensitive_words", self.DEFAULT_SENSITIVE_WORDS
        )
        self.replacement = config.get("replacement", "***")
        self.strip_whitespace = config.get("strip_whitespace", True)
        self.remove_urls = config.get("remove_urls", True)

    def _filter_sensitive_words(self, text: str) -> str:
        """替换敏感词"""
        for word in self.sensitive_words:
            if word in text:
                text = text.replace(word, self.replacement)
        return text

    def _remove_urls(self, text: str) -> str:
        """移除输出中的 URL 链接"""
        url_pattern = r'https?://\S+|www\.\S+'
        return re.sub(url_pattern, '[链接已移除]', text)

    def _normalize_whitespace(self, text: str) -> str:
        """清理多余空白字符"""
        # 统一换行符
        text = text.replace('\r\n', '\n').replace('\r', '\n')
        # 合并连续空行
        text = re.sub(r'\n{3,}', '\n\n', text)
        # 去除每行首尾空格
        text = '\n'.join(line.strip() for line in text.split('\n'))
        return text

    def process(self, output: str, sample: EvalSample) -> str:
        """完整的后处理流水线"""
        result = output

        # 第一步：敏感词过滤
        result = self._filter_sensitive_words(result)

        # 第二步：URL 移除
        if self.remove_urls:
            result = self._remove_urls(result)

        # 第三步：空白字符规范化
        if self.strip_whitespace:
            result = self._normalize_whitespace(result)

        # 记录处理日志
        self.context.set(
            f"postprocess_{sample.sample_id}",
            {
                "original": output,
                "processed": result,
                "changed": output != result,
            },
        )

        return result

    def teardown(self) -> None:
        self.sensitive_words.clear()

7.3 单元测试验证

插件开发完成后，务必编写单元测试：

# tests/test_sensitive_filter.py
import unittest
from plugins.base import EvalSample, PluginContext
from plugins.postprocess.sensitive_filter import SensitiveFilterPlugin


class TestSensitiveFilterPlugin(unittest.TestCase):

    def setUp(self):
        self.plugin = SensitiveFilterPlugin(context=PluginContext())
        self.plugin.setup({
            "sensitive_words": ["内部机密", "测试账号"],
            "replacement": "[已过滤]",
        })

    def test_filter_single_word(self):
        sample = EvalSample(input="测试", reference="答案")
        output = "这个功能目前是内部机密，暂未开放。"
        result = self.plugin.process(output, sample)
        self.assertIn("[已过滤]", result)
        self.assertNotIn("内部机密", result)

    def test_remove_url(self):
        sample = EvalSample(input="测试", reference="答案")
        output = "详情请查看 https://example.com/page?id=123"
        result = self.plugin.process(output, sample)
        self.assertIn("[链接已移除]", result)
        self.assertNotIn("example.com", result)

    def test_normalize_whitespace(self):
        sample = EvalSample(input="测试", reference="答案")
        output = "第一行。\n\n\n\n第二行。\n\n第三行。"
        result = self.plugin.process(output, sample)
        self.assertNotIn("\n\n\n", result)

    def test_clean_output_unchanged(self):
        sample = EvalSample(input="测试", reference="答案")
        output = "正常回答，不包含任何敏感内容。"
        result = self.plugin.process(output, sample)
        self.assertEqual(result, output)


if __name__ == "__main__":
    unittest.main()

运行测试：

python -m pytest tests/test_sensitive_filter.py -v

预期输出：

test_filter_single_word ............... PASSED
test_remove_url ....................... PASSED
test_normalize_whitespace ............. PASSED
test_clean_output_unchanged ........... PASSED

8. 插件的生命周期管理

理解插件的生命周期，有助于排查问题和优化资源使用。每个插件在评测流程中的完整生命周期如下：

8.1 生命周期中的关键设计点

# main.py 中完整的评测编排示例
def run_evaluation(config: dict):
    """完整的评测流程，展示插件生命周期的每个阶段"""
    context = PluginContext(config=config)

    # ========== 阶段一：加载数据集插件 ==========
    dataset_cfg = config["dataset"]
    dataset_plugin = create_plugin(
        "dataset", dataset_cfg["plugin"], context, dataset_cfg.get("config", {})
    )
    dataset_plugin.setup(dataset_cfg.get("config", {}))
    samples = dataset_plugin.load()

    # ========== 阶段二：初始化模型插件 ==========
    model_cfg = config["model"]
    model_plugin = create_plugin(
        "model", model_cfg["plugin"], context, model_cfg.get("config", {})
    )
    model_plugin.setup(model_cfg.get("config", {}))

    # ========== 阶段三：初始化后处理插件 ==========
    postprocess_plugins = []
    for pipeline_step in config.get("pipeline", []):
        if pipeline_step.get("type") == "postprocess":
            for plugin_name in pipeline_step.get("plugins", []):
                pp = create_plugin(
                    "postprocess", plugin_name, context,
                    pipeline_step.get("config", {}),
                )
                pp.setup(pipeline_step.get("config", {}))
                postprocess_plugins.append(pp)

    # ========== 阶段四：执行评测循环 ==========
    eval_results = []
    for sample in samples:
        # 模型生成
        raw_output = model_plugin.generate(sample.input)

        # 后处理流水线
        processed_output = raw_output
        for pp in postprocess_plugins:
            processed_output = pp.process(processed_output, sample)

        # 这里的 score 计算依赖具体的评测指标
        # 为演示简洁，此处以字符串相似度为示意
        score = compute_simple_score(processed_output, sample.reference)

        eval_results.append({
            "sample_id": sample.sample_id,
            "input": sample.input,
            "reference": sample.reference,
            "raw_output": raw_output,
            "processed_output": processed_output,
            "score": score,
        })

    # ========== 阶段五：结果上报 ==========
    reporter_cfg = config.get("reporter", {})
    reporter_plugin = create_plugin(
        "reporter", reporter_cfg["plugin"], context,
        reporter_cfg.get("config", {}),
    )
    reporter_plugin.setup(reporter_cfg.get("config", {}))
    reporter_plugin.report(eval_results)

    # ========== 阶段六：清理资源 ==========
    dataset_plugin.teardown()
    model_plugin.teardown()
    for pp in postprocess_plugins:
        pp.teardown()
    reporter_plugin.teardown()

    return eval_results


def compute_simple_score(prediction: str, reference: str) -> float:
    """简单的评分函数（仅为演示）"""
    if not prediction or not reference:
        return 0.0
    pred_tokens = set(prediction)
    ref_tokens = set(reference)
    if not ref_tokens:
        return 0.0
    overlap = pred_tokens & ref_tokens
    return len(overlap) / len(ref_tokens)

8.2 插件状态管理

插件在执行过程中可以通过 PluginContext 共享数据：

# 数据集插件中写入共享状态
def load(self):
    samples = [...]
    self.context.set("total_samples", len(samples))
    self.context.set("dataset_load_time", time.time())
    return samples

# 上报插件中读取共享状态
def report(self, results):
    total = self.context.get("total_samples", 0)
    print(f"本次评测共上报 {len(results)} 条结果，数据集样本总数 {total}")

这种设计让插件之间可以解耦地传递元信息，而不需要直接相互引用。

9. 实战四：Webhook 上报插件

9.1 需求场景

评测完成后，需要将结果自动推送到企业的监控平台（如自建的告警系统或飞书/钉钉机器人），实现「评测即监控」。

9.2 插件实现

# plugins/reporter/webhook_reporter.py
import json
import time
from typing import Any, Dict, List
import requests
from plugins.base import ReporterPlugin, PluginContext


class WebhookReporterPlugin(ReporterPlugin):
    """将评测结果通过 Webhook 推送到外部系统"""

    plugin_name = "webhook_reporter"
    plugin_version = "1.0.0"

    def __init__(self, context: PluginContext = None):
        super().__init__(context)
        self.url: str = ""
        self.timeout: int = 30
        self.headers: Dict[str, str] = {}

    def setup(self, config: Dict[str, Any]) -> None:
        """
        config 示例:
        {
            "url": "https://monitor.example.com/api/eval-report",
            "timeout": 30,
            "headers": {
                "Authorization": "Bearer xxx"
            }
        }
        """
        self.url = config.get("url", "")
        self.timeout = int(config.get("timeout", 30))
        self.headers = config.get("headers", {})

        if not self.url:
            raise ValueError(
                f"[{self.plugin_name}] 缺少必要参数: url"
            )

    def _build_payload(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
        """构造上报数据"""
        if not results:
            avg_score = 0.0
            pass_rate = 0.0
        else:
            scores = [r.get("score", 0.0) for r in results]
            avg_score = sum(scores) / len(scores)
            pass_rate = sum(1 for s in scores if s >= 0.6) / len(scores)

        return {
            "event": "deepseek_harness_eval_completed",
            "timestamp": int(time.time()),
            "total_samples": len(results),
            "avg_score": round(avg_score, 4),
            "pass_rate": round(pass_rate, 4),
            "results": results[:10],  # 仅上报前 10 条明细
            "metadata": {
                "plugin_version": self.plugin_version,
                "model_name": self.context.get("model_name", "unknown"),
            },
        }

    def report(self, results: List[Dict[str, Any]]) -> None:
        payload = self._build_payload(results)

        try:
            response = requests.post(
                self.url,
                json=payload,
                headers=self.headers,
                timeout=self.timeout,
            )
            response.raise_for_status()
            print(f"[{self.plugin_name}] 上报成功，状态码: {response.status_code}")
        except requests.exceptions.RequestException as e:
            # 上报失败不应阻断评测流程，只记录日志
            print(f"[{self.plugin_name}] 上报失败: {e}")
            # 可选：将失败结果写入本地备份文件
            self._save_local_backup(payload)

    def _save_local_backup(self, payload: Dict[str, Any]) -> None:
        """上报失败时保存本地备份，便于后续补偿"""
        backup_file = f"backup_report_{int(time.time())}.json"
        with open(backup_file, "w", encoding="utf-8") as f:
            json.dump(payload, f, ensure_ascii=False, indent=2)
        print(f"[{self.plugin_name}] 已将结果备份到 {backup_file}")

    def teardown(self) -> None:
        self.headers.clear()

9.3 飞书机器人接入示例

如果企业使用飞书作为办公协作工具，只需要把 Webhook URL 换成飞书机器人的地址，并适配消息格式：

# plugins/reporter/feishu_reporter.py
import time
from typing import Any, Dict, List
import requests
from plugins.base import ReporterPlugin, PluginContext


class FeishuReporterPlugin(ReporterPlugin):
    """将评测结果推送到飞书群机器人"""

    plugin_name = "feishu_reporter"
    plugin_version = "1.0.0"

    def setup(self, config: Dict[str, Any]) -> None:
        self.webhook_url = config.get("webhook_url", "")
        self.secret = config.get("secret", "")  # 飞书签名校验

    def _build_card(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
        scores = [r.get("score", 0.0) for r in results]
        avg_score = sum(scores) / len(scores) if scores else 0.0

        return {
            "msg_type": "interactive",
            "card": {
                "header": {
                    "title": {
                        "tag": "plain_text",
                        "content": "📊 DeepSeek Harness 评测报告"
                    },
                    "template": "blue"
                },
                "elements": [
                    {
                        "tag": "div",
                        "text": {
                            "tag": "lark_md",
                            "content": (
                                f"**样本总数：** {len(results)}\n"
                                f"**平均得分：** {avg_score:.4f}\n"
                                f"**评测时间：** {time.strftime('%Y-%m-%d %H:%M:%S')}"
                            )
                        }
                    }
                ]
            }
        }

    def report(self, results: List[Dict[str, Any]]) -> None:
        card = self._build_card(results)
        try:
            response = requests.post(
                self.webhook_url, json=card, timeout=30
            )
            data = response.json()
            if data.get("code") != 0:
                print(f"[{self.plugin_name}] 飞书机器人返回错误: {data}")
            else:
                print(f"[{self.plugin_name}] 飞书推送成功")
        except Exception as e:
            print(f"[{self.plugin_name}] 飞书推送失败: {e}")

    def teardown(self) -> None:
        pass

10. 插件调试与故障排查

10.1 插件加载失败的排查顺序

当插件无法正常加载时，建议按以下顺序排查：

10.2 常见问题速查表

问题现象	可能原因	解决方法
提示 KeyError: 未找到插件	插件未注册或名称拼写不一致	检查 register_plugin 调用和配置中的 plugin 字段
setup() 抛出 ValueError	缺少必要配置参数	对照插件文档检查 config 中的键名
模型调用返回空字符串	网关返回结构不匹配	检查网关响应体中的 choices 路径是否正确
后处理插件不生效	pipeline 中未声明该插件	检查 configs/*.yaml 中的 pipeline 配置
上报插件失败但不影响评测	符合设计预期（降级处理）	查看本地备份文件确认数据完整性
多次注册同一插件导致覆盖	register_plugin 被重复调用	使用字典注册天然去重，确认调用时机

10.3 插件日志规范

推荐使用 Python 标准库 logging 替代 print，便于统一管理日志级别：

# plugins/logger.py
import logging
import sys


def get_plugin_logger(plugin_name: str) -> logging.Logger:
    """获取带插件名前缀的 logger"""
    logger = logging.getLogger(f"harness.plugin.{plugin_name}")
    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        formatter = logging.Formatter(
            fmt="[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S",
        )
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)
    return logger

在插件中替换 print 为：

# plugins/dataset/jsonlines_dataset.py（节选）
from plugins.logger import get_plugin_logger

class JsonLinesDatasetPlugin(DatasetPlugin):
    def __init__(self, context=None):
        super().__init__(context)
        self.logger = get_plugin_logger(self.plugin_name)

    def load(self):
        # ...
        self.logger.info("成功加载 %d 条样本", len(samples))
        return samples

11. 插件开发最佳实践

结合以上四个实战案例，总结出以下插件开发的最佳实践：

11.1 配置驱动，而非硬编码

# ❌ 不推荐：硬编码
class BadPlugin(BasePlugin):
    def setup(self, config):
        self.url = "https://fixed.example.com/api"

# ✅ 推荐：从配置读取
class GoodPlugin(BasePlugin):
    def setup(self, config):
        self.url = config.get("url", "")
        if not self.url:
            raise ValueError("url 是必要参数")

11.2 失败降级，不阻断主流程

# ✅ 推荐：上报失败只记录日志，不抛出异常
def report(self, results):
    try:
        response = requests.post(self.url, json=results)
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        self.logger.warning("上报失败: %s", e)
        self._save_local_backup(results)

11.3 幂等设计，可安全重试

# ✅ 推荐：上报接口支持唯一的 request_id，重试不会产生重复数据
def report(self, results):
    request_id = f"{int(time.time())}_{uuid.uuid4().hex[:8]}"
    payload = {"request_id": request_id, "results": results}
    # ...

11.4 版本语义化，便于追踪

class MyPlugin(BasePlugin):
    plugin_name = "my_plugin"
    plugin_version = "1.3.2"  # 主版本.次版本.修订版本

11.5 编写单元测试

每个插件至少覆盖三条路径：正常路径、异常输入路径、边界条件路径。参考第 7.3 节的测试示例。

12. 完整实战项目

将本文所有插件组合起来，形成一个可直接运行的完整项目：

deepseek-harness-plugin-demo/
├── configs/
│   └── eval_plugin.yaml
├── data/
│   └── customer_service.jsonl
├── plugins/
│   ├── __init__.py
│   ├── base.py
│   ├── logger.py
│   ├── dataset/
│   │   ├── __init__.py
│   │   └── jsonlines_dataset.py
│   ├── model/
│   │   ├── __init__.py
│   │   └── gateway_model.py
│   ├── postprocess/
│   │   ├── __init__.py
│   │   └── sensitive_filter.py
│   └── reporter/
│       ├── __init__.py
│       └── webhook_reporter.py
├── tests/
│   └── test_sensitive_filter.py
├── main.py
└── requirements.txt

requirements.txt：

requests>=2.31.0
pyyaml>=6.0
pytest>=7.4.0

configs/eval_plugin.yaml：

dataset:
  plugin: jsonlines_customer_service
  config:
    file_path: "data/customer_service.jsonl"
    max_samples: 50

model:
  plugin: internal_gateway_deepseek
  config:
    gateway_url: "https://llm-gateway.internal.com/v1/chat"
    api_key: "your-api-key"
    api_secret: "your-api-secret"
    model_name: "deepseek-chat"
    timeout: 60

pipeline:
  - type: "postprocess"
    plugins:
      - "sensitive_filter"
    config:
      sensitive_words:
        - "内部机密"
        - "测试账号"
      replacement: "[已过滤]"

reporter:
  plugin: webhook_reporter
  config:
    url: "https://monitor.example.com/api/eval-report"
    timeout: 30

运行评测：

python main.py --config configs/eval_plugin.yaml

13. 总结

本文从插件体系的架构设计出发，系统讲解了 DeepSeek Harness 五类核心插件接口，并通过四个实战案例（数据集适配、模型网关接入、输出后处理、Webhook 上报）展示了从开发到落地的完整流程。

回顾核心要点：

• 插件机制的价值：在不修改核心代码的前提下扩展 Harness 能力，降低维护成本；
• 五个扩展点：数据集、模型、后处理、指标、上报，覆盖评测全链路；
• 生命周期管理：setup → 业务调用 → teardown，配合 PluginContext 实现插件间解耦共享；
• 工程化实践：配置驱动、失败降级、幂等重试、版本管理、单元测试。

在下一篇文章中，我们将深入探讨 自定义评测指标（MetricPlugin） 的开发，包括业务专属打分规则的设计与实现，敬请期待。