先说结论给 DeepSeek 模型调用做 harness装配层并不是要造一个多复杂的框架而是把“准备好参数 - 请求模型 - 拿结果 - 善后”这条主流程固定住再把日志、缓存、重试、成本统计这些容易膨胀的需求全部丢到插件里。这样核心代码基本不用动新功能靠加插件就能进去。这篇文章不是讲理论是带你把一个能跑的简单插件从零写出来。如果你是那种经常用 DeepSeek API 写内部工具的人或者想在团队里搭一套统一的大模型调用入口又不想每次需求变了都去翻主流程代码这套思路会非常对路。我会按实际开发顺序来讲先搭最小骨架再定义插件接口然后手写一个日志插件再写一个缓存插件最后把注册机制和常见坑一起讲了。整个工程不依赖重型框架纯 Python 就能跑。1. 先搞清楚DeepSeek harness 是什么插件挂在哪个环节1.1 harness 不是框架而是一层薄薄的编排层很多人第一次听到 harness容易把它理解成一个“能够调用 DeepSeek 的客户端”。但客户端只是发请求、收响应harness 管的更宽一点它负责一次完整调用的生命周期。包括入参怎么组装、系统提示词要不要自动注入、请求超时怎么处理、异常怎么归类、返回结果要不要被记录或改写。你可以把 harness 理解成一条流水线收到用户的 prompt 和附加参数允许插件在请求发出前修改参数真正调用 DeepSeek chat completions 接口拿到响应后允许插件做后处理把最终结果返回给上层调用方。注意一个关键点harness 不应该绑死任何具体业务逻辑。谁负责记日志、谁负责缓存、谁负责把结果写到数据库这些都应该由插件决定。主流程只负责“按顺序拉起来”。1.2 为什么一定要用插件而不是把功能写死在主流程里我在最初做第一版 harness 的时候图省事直接把日志、缓存、失败重试全写在一个类里。前两周没问题后来需求开始发散有人要统计 token 成本有人要敏感词过滤有人要限流。每次加需求主类就越改越长最后连测试都难写改一个日志格式都可能碰到缓存逻辑。后来我切到插件方案核心原则是主流程稳定变化隔离。插件各自解决一个问题互不知道对方存在。好处有三个新功能不需要动主流程代码新增一个插件类即可插件可以独立开关不同场景加载不同插件组合插件之间通过固定的上下文对象通信不会出现隐式耦合。说白了它就是让“变化”和“稳定”分层。1.3 请求生命周期中的三个关键挂点插件不是随便挂在哪都行我归纳成三个挂点before_request请求发出前。适合做参数改写、缓存命中检查、限流判断、日志开始标记after_response拿到响应后。适合做结果记录、缓存写入、格式整理、耗时上报on_error请求抛异常时。适合做错误日志、告警通知、甚至故障降级。为什么要有三个挂点因为一次调用可能成功、失败、也可能压根没发出去被缓存拦住。如果只设计 before 和 after缓存命中场景就没法自然表达。所以我在 harness 里额外加了一个 early_return 机制插件可以在 before_request 阶段直接填入响应结果并跳过真实 API 调用。这个设计很像生活中的演唱会入场流程你是 VIP在检票口被认出来before直接引导到场内early_return普通观众则先检票再进场真实请求散场后还要走离场通道after。不同身份走不同路径但通道本身是独立的。2. 环境准备与最小骨架2.1 需要准备的东西代码用 Python版本建议 3.10 以上。原因很简单类型标注和 dataclass 特性在 3.10 之后更顺手而且大部分模型调用代码跑在新版本上不容易遇到编码问题。老版本不是不行但没必要纠结。依赖先装一个 requests用于直接调用 HTTP 接口。理论上也可以用任何兼容 chat completions 协议客户端但 requests 最直观能把整个请求流程看清楚。项目里还需要 pyyaml 读配置如果你不想用 YAML直接用 Python 字典也行。环境变量方面准备一个DEEPSEEK_API_KEY。我始终不推荐把密钥硬编码进代码因为一旦项目被分享、推送到公共仓库密钥就泄露了。2.2 最小项目结构不要一上来就搞复杂这个规模就够了dsh_demo/ ├── config.yaml ├── run_demo.py ├── harness_core/ │ ├── __init__.py │ ├── context.py │ ├── harness.py │ └── plugins/ │ ├── __init__.py │ ├── base.py │ ├── brieflogger.py │ └── cachebox.pyharness_core是主流程plugins是插件目录run_demo.py是我们跑实验的入口config.yaml放模型参数和插件配置。2.3 先写一个能跑通裸调用的 harness在写插件之前先把没有插件的版本跑通。这个步骤特别重要因为如果裸调用都不稳定后面所有插件问题都会和 HTTP 问题混在一起排查起来很崩溃。先定义上下文对象它是插件之间传递信息的“信箱”# harness_core/context.py from dataclasses import dataclass, field from typing import Any, Optional dataclass class PluginContext: prompt: str kwargs: dict field(default_factorydict) response: Optional[str] None elapsed_ms: float 0.0 error: Optional[Exception] None early_return: bool False metadata: dict field(default_factorydict)接着写 harness 主类# harness_core/harness.py import time import requests from .context import PluginContext class Harness: def __init__(self, config: dict, pluginsNone): self.config config self.plugins plugins or [] def set_plugins(self, plugins): self.plugins [p for p in plugins if getattr(p, enabled, True)] def _call_llm(self, ctx: PluginContext) - str: url self.config.get(base_url, https://api.deepseek.com/v1) /chat/completions headers { Authorization: fBearer {self.config[api_key]}, Content-Type: application/json, } payload { model: self.config.get(model, deepseek-chat), messages: [ {role: user, content: ctx.prompt}, ], **ctx.kwargs, } resp requests.post(url, headersheaders, jsonpayload, timeoutself.config.get(timeout, 60)) resp.raise_for_status() data resp.json() return data[choices][0][message][content] def run(self, prompt: str, **kwargs): ctx PluginContext(promptprompt, kwargskwargs) for plugin in self.plugins: plugin.before_request(ctx) if ctx.early_return: break if not ctx.early_return: start time.perf_counter() try: resp self._call_llm(ctx) except Exception as exc: ctx.error exc for plugin in self.plugins: plugin.on_error(ctx, exc) raise finally: ctx.elapsed_ms (time.perf_counter() - start) * 1000 ctx.response resp for plugin in reversed(self.plugins): plugin.after_response(ctx) return ctx.response这时候没有插件run 方法就是一个普通的 DeepSeek 调用封装。我用 requests 直接发chat/completions请求好处是你能清楚看到每一步发生了什么。实际工程里你也可以换更高级的 SDK但思路完全一致。3. 设计插件接口与插件上下文3.1 BasePlugin 基类要定得多薄插件接口最怕设计得太重。比如有些插件系统要求每个插件必须实现 enable、validate、health_check 一堆方法结果大部分插件里面都是空实现。我倾向于只保留最必要的三个钩子其余一律不强求。# harness_core/plugins/base.py from ..context import PluginContext class BasePlugin: name: str base enabled: bool True def before_request(self, ctx: PluginContext) - None: pass def after_response(self, ctx: PluginContext) - None: pass def on_error(self, ctx: PluginContext, exc: Exception) - None: pass三个方法默认都是空实现。插件不需要的钩子就不用覆盖。这样写出来的新插件非常轻常常只有几十行代码。3.2 PluginContext 到底要放什么有人会问为什么不直接让每个插件各自持有请求参数原因是插件需要在不同阶段共享信息。比如 before_request 阶段判断缓存命中并写入ctx.responseafter_response 阶段要根据命中状态决定要不要写缓存。如果上下文不统一就得用全局变量或者复杂的返回值很容易出问题。PluginContext 的字段设计是有讲究的prompt用户原始输入插件可以读取也可以改写kwargs附加参数比如 temperature、max_tokens插件能在发送前调整response最终拿到或写入的响应结果elapsed_ms主流程测量的耗时插件日志会用到error异常对象on_error 阶段用early_return是否跳过真实 API 调用metadata自由扩展区任何插件都能往里面塞自己的中间数据。我特意加了 metadata 而不是给每个插件加专用字段就是为了保持上下文对象简单。比如日志插件需要记录开始时间就在ctx.metadata[request_started_at]里记缓存插件需要写命中标志就在ctx.metadata[cache_hit]里记。谁也不干扰谁。3.3 为什么 before 按正序、after 按逆序执行这是插件系统里一个容易忽略的细节。我的规则是before_request 按插件列表顺序执行after_response 按相反顺序执行。原因很简单如果把插件看成一圈洋葱请求是往里钻响应是往外走。先进入的插件后退出这样层级才不容易乱。举个例子缓存插件的 before 要在日志插件之前跑缓存命中后日志插件还是能在 after 阶段记录一条“命中缓存”的日志。如果 after 也按正序执行日志插件可能被缓存插件覆盖了响应内容记录到的就不是最终结果。4. 上手实战写一个日志追踪插件4.1 BriefLogger 代码与关键细节日志插件是最容易写的也是最刚需的。我的要求是请求前记录 prompt 前 80 个字符请求后记录耗时、prompt 长度、响应长度出错时记录错误类型和消息。# harness_core/plugins/brieflogger.py import logging import time from .base import BasePlugin from ..context import PluginContext class BriefLogger(BasePlugin): name brief_logger def __init__(self, log_file: str harness.log, level: str INFO): self.log_file log_file self.level level logging.basicConfig(levelself.level) self._logger logging.getLogger(dsh.harness) handler logging.FileHandler(log_file, encodingutf-8) formatter logging.Formatter(%(asctime)s | %(levelname)s | %(message)s) handler.setFormatter(formatter) self._logger.addHandler(handler) self._logger.setLevel(level) def before_request(self, ctx: PluginContext) - None: ctx.metadata[request_started_at] time.time() preview ctx.prompt[:80].replace(\n, ) self._logger.info(request_start | preview%s, preview) def after_response(self, ctx: PluginContext) - None: resp_len len(ctx.response) if ctx.response else 0 cache_state cache_hit if ctx.metadata.get(cache_hit) else api_call self._logger.info( request_end | state%s | elapsed_ms%.2f | prompt_len%d | response_len%d, cache_state, ctx.elapsed_ms, len(ctx.prompt), resp_len, ) def on_error(self, ctx: PluginContext, exc: Exception) - None: self._logger.error(request_error | type%s | detail%s, type(exc).__name__, exc)这段代码里有几个细节值得说第一FileHandler 的编码必须指定为 utf-8否则一旦响应内容里出现中文Windows 环境下日志写入就报 UnicodeEncodeError第二日志格式要稳定方便后面做统计。我固定成request_start、request_end、request_error三段式前缀想统计成功率直接 grep 三行就行第三日志内容不能打印完整 prompt一是太长二是可能有隐私风险所以只打前 80 个字符。4.2 在 harness 中挂载插件现在回到 run_demo.py把插件挂上去# run_demo.py import os from harness_core.harness import Harness from harness_core.plugins.brieflogger import BriefLogger config { api_key: os.environ[DEEPSEEK_API_KEY], base_url: https://api.deepseek.com/v1, model: deepseek-chat, timeout: 60, } harness Harness(config) harness.set_plugins([BriefLogger(log_filelogs/harness.log)]) result harness.run(用一句话介绍 Python 的 yield然后给出一个简单例子。) print(result)注意先创建 harness再单独传入插件这样以后可以从配置文件动态加载插件列表代码不用改。4.3 验证插件是否真的生效跑一次之后去logs/harness.log里看2025-01-18 14:23:01,123 | INFO | request_start | preview用一句话介绍 Python 的 yield然后给出一个简单例子。 2025-01-18 14:23:03,456 | INFO | request_end | stateapi_call | elapsed_ms2258.14 | prompt_len28 | response_len230如果看到这两行说明插件确实在执行。我的经验是第一次调试插件时不要急着加业务逻辑先只打日志确认钩子能触发再逐步扩展。很多插件写完了“没反应”往往不是逻辑问题而是插件根本没被加载。5. 再进一步写一个缓存插件5.1 缓存插件会踩到哪些设计坑第二个插件我选缓存因为它和日志插件完全互补能充分体现插件系统的价值。缓存不是简单存串字符串有几个问题必须考虑缓存 key 怎么算我直接把 prompt 和 kwargs 一起做哈希这样同样 prompt 但不同 temperature 就不会错误命中缓存多久过期需要 TTL否则模型升级后旧结果永远不会失效命中之后怎么跳过 API 调用这就用到前面说的 early_return 机制并发场景怎么办最简单的方案是 SQLite数据库自带锁比手写文件锁可靠。5.2 用 SQLite 实现缓存插件# harness_core/plugins/cachebox.py import hashlib import json import sqlite3 import time import threading from .base import BasePlugin from ..context import PluginContext class CacheBox(BasePlugin): name cachebox def __init__(self, db_path: str cache.sqlite3, ttl: int 3600): self.ttl ttl self._lock threading.Lock() self._conn sqlite3.connect(db_path, check_same_threadFalse) self._conn.execute( CREATE TABLE IF NOT EXISTS cache ( key TEXT PRIMARY KEY, response TEXT, created_at REAL ) ) self._conn.commit() def _key(self, ctx: PluginContext) - str: raw json.dumps( {prompt: ctx.prompt, kwargs: ctx.kwargs}, sort_keysTrue, ensure_asciiFalse, ) return hashlib.sha256(raw.encode(utf-8)).hexdigest() def before_request(self, ctx: PluginContext) - None: key self._key(ctx) ctx.metadata[cache_key] key with self._lock: row self._conn.execute( SELECT response, created_at FROM cache WHERE key ?, (key,) ).fetchone() if row and (time.time() - row[1]) self.ttl: ctx.metadata[cache_hit] True ctx.response row[0] ctx.early_return True def after_response(self, ctx: PluginContext) - None: if ctx.metadata.get(cache_hit): return if ctx.response and ctx.response.strip(): key ctx.metadata.get(cache_key) if not key: key self._key(ctx) with self._lock: self._conn.execute( INSERT OR REPLACE INTO cache (key, response, created_at) VALUES (?, ?, ?), (key, ctx.response, time.time()), ) self._conn.commit()sqlite 的check_same_threadFalse是为了允许多线程场景下从不同线程访问连接配合threading.Lock避免并发写错乱。如果你只在单线程脚本里用锁可以不加但加上没坏处。5.3 两个插件合用的顺序问题现在把两个插件一起挂载harness.set_plugins([ CacheBox(db_pathcache.sqlite3, ttl3600), BriefLogger(log_filelogs/harness.log), ])我特意把 CacheBox 放在前面。原因是 before_request 阶段按正序执行缓存插件必须比日志插件先跑这样日志插件在 after_response 阶段看到的ctx.metadata[cache_hit]才是真实状态。第一次跑日志显示stateapi_call。第二次跑同样的 prompt日志变成statecache_hit而且请求源直接被跳过响应内容来自 SQLite整个过程耗时可能不到 1 毫秒。这就是插件系统的价值体现不修改主流程就完成了一个很有用的能力。6. 插件注册与加载机制让组合不再写死在代码里6.1 用配置文件声明插件组合到这一步插件已经能跑了但插件列表还是写死在 Python 代码里。当插件数量变多我希望通过配置文件控制而不是每次改代码。简易做法是写一个 YAML 配置# config.yaml api_key_env: DEEPSEEK_API_KEY model: deepseek-chat timeout: 60 plugins: - name: cachebox module: harness_core.plugins.cachebox class: CacheBox config: db_path: cache.sqlite3 ttl: 3600 - name: brief_logger module: harness_core.plugins.brieflogger class: BriefLogger config: log_file: logs/harness.log加载代码# run_demo.py import importlib import os import yaml from harness_core.harness import Harness def load_plugins(cfg): plugins [] for item in cfg.get(plugins, []): mod importlib.import_module(item[module]) plugin_class getattr(mod, item[class]) kwargs item.get(config) or {} plugins.append(plugin_class(**kwargs)) return plugins with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) config[api_key] os.environ[config[api_key_env]] harness Harness(config) harness.set_plugins(load_plugins(config)) result harness.run(Python 的装饰器怎么理解) print(result)这段代码核心就一个函数importlib 按模块路径加载模块再根据类名取到插件类实例化时把 config 里的小配置透传。以后要加新插件只需要在 plugins 目录新增一个文件然后在 YAML 里声明一行。6.2 小团队场景下配置文件已经够用有些文章会给插件系统安排 entry_points 自动发现、插件注册中心、热加载等等。这些机制适合大型插件生态但对于一个小团队或独立开发者来说配置文件反而更靠谱。因为它是显式的、顺序清晰的、可审查的。哪一天部署出问题打开 YAML 就知道有哪些插件在运行。我不建议一上来就折腾自动发现自动发现会带来隐式依赖插件的加载顺序变得难以预测debug 成本反而上升。6.3 如果插件数量真的多了再考虑 entry_points如果你未来的确要做一个开放的插件生态可以让每个插件包在安装元数据里声明入口[project.entry-points.dsh.plugins] brief_logger harness_core.plugins.brieflogger:BriefLogger然后通过importlib.metadata扫描入口点加载。这个方案的优势是插件可以独立安装、独立升级主项目并不需要知道插件代码在哪里。但这属于“以后再说”的优化。现在这个阶段保持简单才是对的。7. 常见问题与排查实录7.1 插件不生效最常见的原因是加载路径错了插件类写了运行却完全没输出。最典型的原因是模块导入路径不对。比如你在run_demo.py同目录下直接import plugins.brieflogger但 plugins 目录没有__init__.py或者在错误的当前目录下运行脚本就会报ModuleNotFoundError。我的排查顺序是先确认 YAML 配置里 module 写的是harness_core.plugins.brieflogger而不是plugins.brieflogger在load_plugins里加一行 print打印实际加载到了哪个类确认插件实例真的被传进了harness.set_plugins而不是只创建了没有挂载。大多数“插件不生效”问题都是这一个原因。7.2 缓存命中却返回空往往是序列化环节出问题有次缓存插件返回了空字符串看起来像模型返回了空。查下去发现SQLite 里存的响应本来就是空的。原因在于我对ctx.response.strip()判断前响应内容其实是None导致空数据被写进缓存。后来我把 after_response 的写入条件改成if ctx.response and ctx.response.strip():同时把空响应的定义单独处理问题就消失了。写缓存类插件时牢记一点缓存是把双刃剑脏数据一旦被缓存清理成本很高。宁可漏缓存不能错缓存。7.3 API 超时没设 timeout整个脚本卡死如果 requests 请求不传 timeout遇到网络抖动脚本可能挂很久。我在 Harness 里加了timeout配置并且传给 requests。实际工程里我还会在 on_error 阶段区分超时和其他异常。建议错误处理按场景分开异常类型常见原因建议处理Timeout网络抖动、模型推理时间过长局部重试最多 2 次ConnectionError服务不可达检查网络和服务状态HTTPError鉴权失败、参数非法不要盲目重试读响应详情JSONDecodeError返回不是标准 JSON记录原始响应体7.4 密钥管理日志里千万不能打出 header我见过有人为了排查方便把 headers 整个打进日志。这是最危险的操作因为 Authorization 里就是 API Key。一旦日志被同步到远程仓库或者分享出去密钥就等于泄露。我的建议是环境变量读取密钥配置文件只写环境变量名日志插件默认不打印任何请求头debug 模式若要打印请求信息必须做脱敏处理SQLite 缓存文件加入.gitignore避免缓存内容泄露到仓库。8. 绕不开的收尾我自己对这些细节的真实体会这套代码我前前后后调了好几版最受益的反而不是插件机制本身而是“把每次调用看成一次生命周期”这个视角。一旦你从这个视角去看日志、缓存、限流、成本统计这些需求都会非常自然地长在对应的钩子上而不是挤在一个函数里。最后分享一个小经验第一版插件系统别求完整只写一个日志插件就够了。它能验证整个链路通不通也能让你尽快理解挂钩子的节奏。等日志插件稳定跑起来再考虑缓存、限流这些有状态的插件你会轻松很多。DeepSeek harness 的价值不在底层 API 调用写得有多漂亮而在于你后续加需求的时候不再心惊肉跳地去找旧代码了。