DeepseekHarness 这个名字听起来像某个特殊工具但回到工程本质它就是一个围绕大语言模型搭建的测试与执行脚手架Harness。做 AI 应用的人几乎都会遇到同一个问题提示词要反复调、模型输出要批量验、失败样本要回放、请求日志要对账这些需求一旦全部塞进主流程代码就会变成一团乱麻。插件机制的价值就在于此把提示词管理、输出解析、评测断言、缓存、日志追踪、多模型适配、数据导入导出、结果报告这类能力做成独立的插件按需加载、按配置启用。下面会带着读者从零搭建一个轻量级 DeepseekHarness 插件框架并逐个实现 8 个实用插件。1. 先理解 Harness 和插件的关系1.1 什么是 LLM Harness用一句话说Harness 是给模型搭的“测试台”。在软件测试领域test harness 指的是把测试用例、被测对象、结果收集和报告汇总串起来的一套运行环境。到了 LLM 场景被测对象变成了模型接口测试用例变成了“输入文本 期望行为”结果收集变成了对模型输出的解析和断言。一个最基本的 LLM Harness 至少包含四个环节组装提示词、调用模型接口、解析输出、记录结果。手动做一次很容易在 Jupyter Notebook 里写两行代码就能完成但当你需要在几百条样本上反复调整提示词或者要对比两个模型版本在同类输入上的输出差异时人工操作就不现实了。Harness 的价值是把这条流水线固定下来让每一次评测都可重复、可对比、可追溯。在 DeepseekHarness 里核心流程可以概括为六个步骤加载测试用例 - 渲染提示词 - 查询缓存是否命中 - 调用模型接口 - 解析输出并断言 - 汇总结果生成报告每一步都可以被插件接管也可以被插件旁路观察这就是插件机制能够存在的基础。1.2 为什么用插件而不是把所有代码写在一起直接写主流程的代价在项目初期不明显。等到你要给某些样本单独加调试日志、给线上请求开缓存、给评测结果出报告时主函数会不断膨胀不同需求的代码还会互相干扰。插件机制的核心思想是开放-封闭原则主流程保持稳定变化的部分通过扩展点接入。在 DeepseekHarness 中我会定义这样几个事件钩子before_request在发起请求前触发适合渲染提示词、检查缓存after_request在拿到原始输出后触发适合解析输出、记录日志before_eval在评测前触发适合执行断言规则on_error在请求异常时触发适合记录错误现场on_result在单个样本评测完成后触发适合收集统计信息。不同的插件各自关注不同的事件互不感知对方的存在。这样测试环境可以临时开启 trace_logger 和 report生产环境的离线批量评测可以只保留 cache、assertion、data_io配置一换行为就变。1.3 插件与主线如何协作插件不只是被动监听事件它还能主动提供能力。例如 model_adapter 插件负责真正调用模型接口data_io 插件负责读写测试数据cache 插件负责在发请求前查询缓存。这种“能力方法 事件钩子”的双形态设计比纯事件监听更实用。事件的优点是解耦能力的优点是主流程可以主动调用某个插件完成特定动作。下面的章节会围绕这个设计落地一套最小可运行代码。学习时只要把环境搭好、把 8 个插件按顺序放进目录就能跑通一次完整的批量评测。2. 搭建基础环境与项目骨架2.1 环境准备与依赖安装本文代码基于 Python 3.9主要依赖如下项目推荐版本用途Python3.9运行插件框架使用 dataclass 与类型标注openai1.x调用兼容 OpenAI 协议的大模型服务PyYAML6.x读取配置文件与提示词模板建议使用虚拟环境避免污染全局 Python 环境python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate pip install openai PyYAML注意openai 1.x 的客户端初始化方式和 0.x 差别较大落地前要确认自己安装的是 1.x 版本。如果公司内部使用统一封装的模型网关替换成 requests 调用也是可以的插件机制保持不变。2.2 项目目录结构这里先给出一份目录结构后面所有文件都会按这个位置创建deepseek_harness/ ├── config/ │ ├── config.yaml │ └── prompts.yaml ├── data/ │ ├── cases.jsonl │ └── outputs/ ├── harness/ │ ├── __init__.py │ ├── core.py │ ├── plugin.py │ ├── plugin_manager.py │ └── plugins/ │ ├── prompt_manager.py │ ├── output_parser.py │ ├── assertion.py │ ├── cache.py │ ├── trace_logger.py │ ├── model_adapter.py │ ├── data_io.py │ └── report.py ├── main.py └── requirements.txtconfig 目录放全局配置和提示词模板data 目录放测试样本和输出文件harness 目录放框架核心代码plugins 子目录放 8 个插件。每个插件一个 Python 文件命名即插件名便于加载器按文件名扫描。2.3 Plugin 基类与加载器先定义 Plugin 基类。它不实现具体逻辑只约定每个插件可以挂哪些事件以及每个事件应该接收什么参数# harness/plugin.py from abc import ABC class Plugin(ABC): name: str base version: str 0.1.0 def __init__(self, config: dict): self.config config or {} def on_load(self, ctx) - None: pass def before_request(self, ctx, req) - None: pass def after_request(self, ctx, req, resp) - None: pass def before_eval(self, ctx, resp, result) - None: pass def on_result(self, ctx, result) - None: pass def on_error(self, ctx, exc: Exception) - None: passctx 是全局上下文保存配置、客户端、插件实例等共享对象req 是请求上下文resp 是响应上下文result 是结果上下文。插件只需要重写自己关心的钩子。接着写插件加载器。加载器扫描指定目录下的所有 .py 文件找到 Plugin 的子类同时根据配置里的 enabled 列表决定是否启用某个插件# harness/plugin_manager.py import importlib.util from pathlib import Path from harness.plugin import Plugin def load_plugins(plugin_dir: str, plugins_config: dict): enabled set(plugins_config.get(enabled, [])) loaded [] for file in Path(plugin_dir).glob(*.py): if file.name.startswith(_): continue spec importlib.util.spec_from_file_location(file.stem, file) module importlib.util.module_from_spec(spec) spec.loader.exec_module(module) for attr in dir(module): obj getattr(module, attr) if not isinstance(obj, type): continue if not issubclass(obj, Plugin) or obj is Plugin: continue if enabled and obj.name not in enabled: continue loaded.append(obj) return loaded这段代码有两个关键点。第一只扫描不以下划线开头的文件避免把公共模块误当插件加载。第二通过 enabled 列表做白名单过滤没有出现在 enabled 里的插件即使放在 plugins 目录也不会被注册。这样测试环境和生产环境可以共用代码库只靠配置区分行为。3. 8 个实用插件的设计与实现下面逐个实现 8 个插件。每个插件都遵循同样的节奏先说明它解决什么问题再给关键代码最后说明配置要点。3.1 插件一Prompt 模板管理插件问题背景批量评测时提示词经常要换措辞、加示例、改角色设定。如果提示词散落在业务代码里每一处都要改还容易改漏。这个插件负责从 YAML 文件加载提示词模板并在 before_request 阶段把模板渲染成最终请求文本。# harness/plugins/prompt_manager.py from pathlib import Path import yaml from harness.plugin import Plugin class PromptManagerPlugin(Plugin): name prompt_manager def on_load(self, ctx): path Path(self.config.get(templates_path, config/prompts.yaml)) self.templates yaml.safe_load(path.read_text(encodingutf-8)) or {} ctx.prompt_manager self def render(self, template_name: str, variables: dict) - str: template self.templates.get(template_name, ) for key, value in (variables or {}).items(): template template.replace({{ key }}, str(value)) return template def before_request(self, ctx, req): if req.prompt: return req.prompt self.render( req.case.get(template, default), req.case.get(variables, {}), )prompts.yaml 的格式很简单default: 请回答下面的问题\n{{ question }} summary: 把下面的文本压缩成 50 字以内的摘要\n{{ text }}这里使用简单的字符串替换而不是 Jinja2是为了减少依赖。实际项目中如果要支持条件分支、循环和过滤器可以直接换成 Jinja2 模板引擎插件对外接口不变。3.2 插件二输出解析插件问题背景模型输出经常带有 json 围栏、前后空格或解释性文字直接 json.loads 会失败。这个插件在 after_request 阶段统一清洗文本尝试提取 JSON 结构化数据。# harness/plugins/output_parser.py import json import re from harness.plugin import Plugin class OutputParserPlugin(Plugin): name output_parser def after_request(self, ctx, req, resp): text resp.raw_output.strip() fence re.search(r(?:json)?\s*(.*?), text, re.S) if fence: text fence.group(1).strip() try: resp.parsed_output json.loads(text) except json.JSONDecodeError: resp.parsed_output text注意一点JSON 解析失败时不要把异常直接抛掉因为有些评测场景允许模型输出自由文本。把无法解析的原始文本原样放进 parsed_output由后续的断言插件决定是判失败还是走文本比对。3.3 插件三断言与评测指标插件问题背景评测的本质是判断模型输出是否符合预期。这个插件支持三种常见断言规则包含指定文本、完全相等、正则匹配。单个样本可以配置多条规则全部通过才算通过。# harness/plugins/assertion.py import re from harness.plugin import Plugin class AssertionPlugin(Plugin): name assertion def before_eval(self, ctx, resp, result): rules result.case.get(assert, []) passed True details [] for rule in rules: rule_type rule.get(type) ok True if rule_type contains: ok rule.get(value, ) in str(resp.parsed_output) elif rule_type equals: ok str(resp.parsed_output) str(rule.get(value, )) elif rule_type regex: ok re.search(rule.get(pattern, ), str(resp.parsed_output)) is not None passed passed and ok details.append({type: rule_type, ok: ok}) result.passed passed result.details {assertions: details, **result.details}测试用例里可以这样配置断言{name: 地理题, template: default, variables: {question: 中国的首都是哪个城市}, assert: [{type: contains, value: 北京}]}这个插件的设计要点是断言规则写在测试用例里而不是写在代码里。这样新增一条评测样本不需要改任何 Python 代码运营人员也能通过配置文件维护评测集。3.4 插件四会话缓存插件问题背景调试提示词时同一个请求会反复触发模型调用既花钱又耗时。缓存插件根据“模型名 提示词”生成哈希键命中后直接复用上次输出。# harness/plugins/cache.py import hashlib import json from pathlib import Path from harness.plugin import Plugin class CachePlugin(Plugin): name cache def __init__(self, config): super().__init__(config) self.cache_dir Path(self.config.get(cache_dir, cache)) self.cache_dir.mkdir(exist_okTrue) def _key(self, req) - str: raw f{req.model}|{req.prompt}.encode(utf-8) return hashlib.sha256(raw).hexdigest() def get_cached(self, req): file self.cache_dir / f{self._key(req)}.json if not file.exists(): return None data json.loads(file.read_text(encodingutf-8)) return data.get(output) def save(self, req, output: str): file self.cache_dir / f{self._key(req)}.json data {model: req.model, prompt: req.prompt, output: output} file.write_text(json.dumps(data, ensure_asciiFalse), encodingutf-8)缓存键的设计要非常小心。如果只是对 prompt 做哈希两个不同模型会互相串数据如果加上 temperature 参数又会把缓存切得太碎命中率降低。本文示例只考虑“模型名 提示词”适合确定性较高的评测场景。如果模型接口本身没有固定为 temperature0缓存结果可能会掩盖随机性生产环境要慎重。3.5 插件五请求日志与追踪插件问题背景批量评测一旦失败最痛苦的是不知道这次请求用了哪个模型、耗时多少、是不是权重参数写错了。这个插件在 before_request 生成 request_id在 after_request 输出结构化日志。# harness/plugins/trace_logger.py import json import time import uuid from harness.plugin import Plugin class TraceLoggerPlugin(Plugin): name trace_logger def before_request(self, ctx, req): ctx.request_id str(uuid.uuid4()) ctx._start_time time.time() def after_request(self, ctx, req, resp): latency_ms (time.time() - ctx._start_time) * 1000 record { request_id: ctx.request_id, model: req.model, latency_ms: round(latency_ms, 2), token_usage: resp.token_usage, prompt_length: len(req.prompt), output_length: len(resp.raw_output), } # 实际项目中可以写入文件、日志系统或消息队列 print(json.dumps(record, ensure_asciiFalse))生产环境不要把日志直接 print 到标准输出建议接入统一日志平台或按天滚动写入文件。日志字段要包含 request_id这是串联“测试样本、API 调用、结果断言”三者的关键线索。3.6 插件六多模型适配插件问题背景同一个评测集可能要在 DeepSeek、本地部署模型或公司内部模型网关之间来回跑。模型适配插件把“调用哪个模型服务”收敛成一个适配器其它插件不关心底层 HTTP 细节。# harness/plugins/model_adapter.py from openai import OpenAI from harness.plugin import Plugin class ModelAdapterPlugin(Plugin): name model_adapter def on_load(self, ctx): backends self.config.get(backends, {}) default self.config.get(default, deepseek) backend backends.get(default, {}) self.client OpenAI( api_keybackend.get(api_key, 请填写API密钥), base_urlbackend.get(base_url, https://api.deepseek.com), ) ctx.client self.client ctx.model_adapter self def complete(self, ctx, req) - str: resp self.client.chat.completions.create( modelreq.model, messages[{role: user, content: req.prompt}], temperaturereq.params.get(temperature,