万物皆插件：DeepSeek Harness插件化架构设计详解-平芜编程栈

先聊点实际的。DeepSeek Harness 是我最近在折腾AI应用工具链时反复打磨出来的一个东西。简单来说，它是一个围绕DeepSeek模型接口做上层封装的手脚架，最开始只是个能跑通对话的空白模板，后来慢慢长成了带插件机制的迷你框架。整个设计思路浓缩成一句话：万物皆插件。

你可能会问，一个脚手架而已，用得着这么上纲上线吗？答案是，当你连续遇到“加一个功能就要改主流程代码”的情况，就会发现：脚手架和产品的差别，不在于功能多寡，而在于结构能不能自我生长。这篇文章我打算把这套东西的底层设计拆开讲清楚，包括插件接口怎么定义、管理器怎么加载、上下文怎么传递，还有我从这个项目里踩过的坑。不管你是想给自己项目加个插件系统，还是纯粹好奇一个AI工具怎么做架构，都可以拿来参考。

1. 脚手架为什么会“长歪”：三个需求把架构逼到墙角

1.1 最初的Harness只解决一件事：快速跑通对话

第一版DeepSeek Harness其实非常简单，本质上就是一层API封装。输入用户消息，拼history，调模型，把回复抛出来。所有逻辑都在一个文件里，配置靠改代码，流程固定死为“用户输入->组装消息列表->请求DeepSeek接口->刷新上下文”。好处是够直接，新人五分钟就能看懂整个调用链；坏处是一旦业务场景丰富起来，这个单一文件就会变成谁都不敢动的重灾区。

我原本以为这种“能用就行”的状态能维持很久，结果现实很快打脸。刚开始加一个“把历史对话存到本地文件”的需求还算温和，顶多多写个函数，在每次请求前后调用一下。但后来需求开始变味：有人要求对输出做敏感词过滤，有人要在回复前检索知识库，还有人想把外部工具调用接进来。每一次都是在同一个主流程里塞逻辑，塞到最后，一个本来只有两百行的调用脚本，硬生生撑到一千多行，而且哪里都不敢动。

这个阶段最痛苦的还不是代码膨胀，而是每次接需求都要重新梳理逻辑顺序：先过滤还是先请求？记忆更新在哪个环节触发？知识库检索上下文应该粘在哪个位置？这些问题在固定流程代码里很难优雅回答，因为它们本质上是“流程编排”问题，不是“模型调用”问题。

1.2 三个真实需求把固定流程逼到了死角

当时有三个场景让我彻底意识到必须换架构。第一个是给企业内部使用，要求所有模型输出必须先过一遍敏感词策略，而且每个部门的策略不能互相干扰。这个需求意味着过滤逻辑不能写死在主流程里，它必须是一个可插入、可替换的独立环节。第二个是多轮记忆接入，不同项目对记忆的诉求差异巨大，有的要全量历史，有的只保留最近五轮，有的还要做向量化检索。如果我继续在主流程里用if-else区分，那代码会变成一座谁也进不去的迷宫。

第三个需求最致命，外部工具调用。某个项目希望模型在回答“今天天气适合出门吗”时，主动去调一个天气接口，再把结果整理成自然语言。这不是改几行代码能搞定的，它要求整个执行流程能暂停、跳转、注入外部数据、之后继续，这已经不是在“调模型”，而是在做完整的工作流引擎了。

这时候我脑子里冒出的念头就是：不能再让主流程去枚举所有可能性了，要反过来，让每个扩展能力都变成独立的插件，主流程只负责按顺序去跑插件。这样一来，每加一个新需求就是写一个新插件，原来的代码一行都不用动。

1.3 为什么是插件，而不是一堆配置文件

有人可能会说，你上面的需求，用配置项也能解决一部分啊。敏感词库放配置文件，记忆轮数放配置文件，工具API地址放配置文件，不也一样？道理是没错，但配置项和插件有一个本质差异：配置只能改变参数的取值，没办法改变行为的形状。

我拿手机充电器来类比。配置项是充电头的电压电流切换，你能在5V、9V、12V之间选，但你没法让充电头自己学会握手协议。插件则是整个充电协议模块，你可以把USB-PD模块拔下来，换上QC模块，再换上私有快充模块，充电头的工作机制彻底变了。工具调用就是典型例证，它在某个环节要做的事不是“多传一个参数”，而是“在请求模型之前先发一个HTTP请求，拿到结果拼进Prompt”，这已经超出了配置项的表达能力，必须依赖代码块的可插拔。

所以“万物皆插件”不是设计上的花活，而是当你的系统开始面对组合式扩展时，最自然不过的选择。这个选择的核心是把“变化”从主流程里剥离出去，放到一个个隔离的单元里，主流程只负责编排，不负责具体实现。

2. 底层设计拆解：接口、管理器、上下文三件套

2.1 插件接口的“最少承诺”原则

插件系统的地基是接口定义。这一步做得好不好，直接决定后面所有扩展是顺手还是拧巴。我在设计时坚持一个原则：最少承诺，即接口只暴露插件生命周期中绝对必要的方法，其他一概不管。

DeepSeek Harness的插件接口特别朴素，就四个方法：

from abc import ABC, abstractmethod from typing import Any class HarnessPlugin(ABC): name: str = "unnamed" version: str = "0.0.0" description: str = "" @abstractmethod def init(self, context: dict) -> None: """插件初始化时调用，只调用一次，负责做资源准备。""" @abstractmethod def process(self, context: dict) -> None: """主流程每次运行时调用，插件在这里实现自己的逻辑。""" def close(self) -> None: """插件销毁时调用，默认空实现。"""

我希望插件作者在编写插件时，只需要关心三个问题：我的资源在哪里初始化？我每次要在流程里做什么？退出时我要清理什么？其他事情，比如插件怎么被发现、怎么排序、怎么和其他插件通信，都不该让插件作者操心。

有朋友问，为什么process方法不返回值？为什么不直接让插件把自己的结果return出来，然后原样拼到下个环节？说实话我一开始也是用返回值设计的，每个插件process返回一个中间结果对象，管理器负责把对象传给下一个。但后来发现，插件之间往往需要协作，比如记忆插件要把当前轮对话追加进历史，过滤插件可能要修改上一个插件塞进去的临时变量，此时返回值链路会搞得非常僵硬。于是我改成共享context，所有插件都往同一个上下文里读写，process方法本身不返回数据，它只负责修改context。

2.2 插件管理器：注册、排序、加载

有了接口，下一步就是实现PluginManager。管理器承担三件事：扫描并注册插件、按声明的优先级排序、在正确时机触发生命周期方法。

扫描这一环节我用了最简单的方案，约定一个plugins目录，启动时用importlib扫描该目录下的所有Python文件，找出继承HarnessPlugin的类，实例化后注册到注册表里。这样做的好处是不需要额外维护配置文件去声明“我有哪些插件”，少一个配置项就少一个出错点。

import importlib import inspect from pathlib import Path from typing import Dict, List, Type from harness_core.plugin_base import HarnessPlugin class PluginManager: def __init__(self, plugin_path: str): self.plugin_path = Path(plugin_path) self.plugins: Dict[str, HarnessPlugin] = {} self.order: List[str] = [] def scan_and_register(self) -> None: for py_file in self.plugin_path.glob("plugin_*.py"): module_name = py_file.stem module = importlib.import_module(f"plugins.{module_name}") for _, obj in inspect.getmembers(module, inspect.isclass): if issubclass(obj, HarnessPlugin) and obj is not HarnessPlugin: plugin = obj() self.plugins[plugin.name] = plugin self.order.append(plugin.name) def sort_plugins(self) -> None: self.order.sort( key=lambda name: getattr(self.plugins[name], "priority", 100) ) def init_all(self, context: dict) -> None: for name in self.order: self.plugins[name].init(context) def process_all(self, context: dict) -> None: for name in self.order: self.plugins[name].process(context) def close_all(self) -> None: for name in self.order: self.plugins[name].close()

排序逻辑是后来加的。plugins目录下的文件扫描顺序是任意的，如果不做控制，插件执行顺序可能每次都不同，调试起来人会疯掉。我定义了一个可选的priority属性，数值越小优先级越高，默认值是100。执行顺序是：低优先级先跑，高优先级后跑，换句话说，priority值小的是“前面垫场”的通用逻辑，值大的是“后面收尾”的定制逻辑。

2.3 上下文对象：插件的公共走廊

context是整个插件系统的灵魂。在DeepSeek Harness里，我把它设计成一个普通的Python字典，而不是一个强类型的类。理由很直接，在快速迭代阶段，强类型的字段约束反而会拖慢插件开发，字典用起来最灵活，key有就有，没有就保底。

但这个字典并不是随意敞开乱写的。我在文档里约定了几大类命名空间，要求插件作者遵守：

命名空间	用途	示例
`user_input`	用户本轮输入	原始文本、意图标签、实体
`history`	对话历史	消息列表、token数量、最近主题
`extras`	扩展数据	工具调用结果、检索片段
`output`	最终输出相关	需要模型生成的消息、拦截标志
`plugin_state`	插件私有状态	各个插件维护的临时数据

这样一个插件在写代码时，就明确知道自己该从哪里拿数据、往哪里放数据。记忆插件看到history为空时，就该决定是否要用plugin_state自己维护一份；工具插件把检索结果放到extras["search_results"]后，后续的Prompt组装插件自然知道要从这里取。

上下文对象还有一个隐藏作用：它隔离了插件和外部系统的接触面。插件不需要知道DeepSeek API的请求结构长什么样，它只把context当作业务数据仓库，真正组网到API请求的代码由Harness主流程来做。这一点对插件生态的扩展非常关键，主流程换版本、换模型、换API都不影响插件本身的逻辑。

3. 实操过程：把DeepSeek Harness改造成插件平台的全过程

3.1 重构第一步：把主流程从“写逻辑”变成“排顺序”

改造不是推倒重来，而是渐进重构。我先把原来那个一千多行的主文件拆成两部分：一部分是核心执行器，负责最纯粹的“组装消息列表、调用DeepSeek接口、拿到回复”这条主线；另一部分是插件管理器，负责在主线前后挂接各种逻辑。

重构前的伪代码大概是这种感觉：

def main_run(user_input): history.append({"role": "user", "content": user_input}) if sensitive_words_enabled: filtered_input = filter_sensitive(user_input) messages = build_messages(history) result = call_deepseek(messages) if need_memory: update_memory(result) return result

重构后变成了这样：

def main_run(user_input): context = build_context(user_input) plugin_manager.process_all(context) if context["output"].get("blocked"): return {"reply": "内容需要人工审核后再展示"} reply = call_deepseek_with_context(context) context["output"]["reply"] = reply plugin_manager.process_all(context) return reply

你看，原来所有if-else统统消失了。现在主流程只关心两件事：执行插件链，然后用context组装请求并调用模型。至于敏感词是否启用、记忆是否需要更新、工具要不要调用，全由插件自己在process里决定。

这个重构有个意外收获：插件的process_all并不只是在调用模型前跑一次，我在模型返回后也跑了一次。这意味着“输出后处理”也变成了一等公民，模型生成完文本后，可以有插件负责改写成特定风格，也可以有插件负责把回复中的URL提出来建链接。框架本身对这个环节不做任何假设，你就把它看作一条双向管道。

3.2 实战插件一：关键词过滤插件，最简单的切入范例

先写个最简单的插件来验证整个链路通不通。这个插件的功能是对用户输入做敏感词命中检测，如果命中了就直接置blocked标志，不让消息进模型。

from harness_core.plugin_base import HarnessPlugin class KeywordFilterPlugin(HarnessPlugin): name = "keyword_filter" version = "1.0.0" description = "对用户输入进行敏感关键词检测，命中后阻断请求" priority = 10 def init(self, context: dict) -> None: self.words = ["测试占位词A", "测试占位词B"] def process(self, context: dict) -> None: user_text = context["user_input"].get("text", "") hit = [word for word in self.words if word in user_text] if hit: context["output"]["blocked"] = True context["output"]["block_reason"] = f"命中关键词：{hit[0]}"

这个插件虽然短，但它验证了几个关键点：第一，init只调用一次，词表可以做启动加载，不用每次请求重复载入；第二，process每轮进来，读context时通过get取默认值，保证即使未来字段改名也不会直接抛异常；第三，它彻底不依赖DeepSeek的API细节，完全是纯业务逻辑，换任何模型体系都能用。

把plugin_keyword_filter.py丢进plugins目录，重启Harness后再跑一句之前会触发屏蔽的话，回复就变成了“内容需要人工审核后再展示”。那一刻我确定，这条路走对了。

3.3 实战插件二：让模型调工具，体验“编排”的感觉

第二个插件就复杂一些了。一个天气查询插件，作用是：当用户输入里包含“天气”或“气温”这类词时，先调用一个外部天气接口，把结果塞进extras，后续Prompt组装插件会把它拼进消息里，让模型基于真实数据回答。

import json from urllib import request from harness_core.plugin_base import HarnessPlugin class WeatherToolPlugin(HarnessPlugin): name = "weather_tool" version = "1.0.0" description = "检测天气相关问题，注入天气API返回的数据" priority = 30 def init(self, context: dict) -> None: self.api_url = context["config"].get("weather_api_url") self.city = context["config"].get("weather_city", "北京") def process(self, context: dict) -> None: user_text = context["user_input"].get("text", "") if not any(kw in user_text for kw in ["天气", "气温", "下雨", "出门"]): return url = f"{self.api_url}?city={self.city}&key=demo_key" with request.urlopen(url) as resp: data = json.loads(resp.read().decode("utf-8")) context.setdefault("extras", {})["weather_data"] = data

仔细看这个插件的顺序是skill级别的：它先在用户输入里做意图判断，只有命中才发起网络调用；网络结果是JSON，被整体放进extras，没有格式化，因为格式化交给后续的Prompt组装阶段更合理。天气插件不知道模型会怎么使用这份数据，它只负责把“最新最准的原始数据”放到公共区域。

接着问题来了：天气数据塞进context之后，谁来把它拼进Prompt？这里我引入了一个系统内置插件，prompt_assembler，优先级设为90，排在最后。它负责把extras里的所有外部数据都拼成一段补充说明，加进消息列表的最后一条system消息后。这样工具类插件本身不碰Prompt逻辑，Prompt相关逻辑也完全不依赖具体工具，两者在context上解耦。

这个案例算是把“万物皆插件”想表达的编排能力展示出来了：天气插件发现天气问题，发起调用，把数据放好；Prompt组装插件拿到数据，拼进Prompt；模型看到补充信息后给出回答；回答再过一遍输出侧插件，完成最终展示。整个过程没有任何一个环节需要修改主流程。

3.4 用配置文件控制插件的启停与顺序

插件写多了以后，直接靠扫描文件目录方式的短板就暴露了：临时调试一个插件，你得把它从plugins目录里移出去才能禁用。于是我加了配置节，允许在启动时通过disabled_plugins显式禁用某些插件，也允许通过plugin_priority覆盖插件内预设的priority。

配置样例长这样：

harness: model: deepseek-chat temperature: 0.7 max_tokens: 2048 disabled_plugins: - keyword_filter plugin_priority: weather_tool: 25 keyword_filter: 5

实现方式就是在PluginManager里加一步配置过滤。初始化时读取配置，如果插件名在disabled_plugins里，就跳过注册；如果plugin_priority字段有覆盖值，则替换插件对象的priority属性。这一步很实用，因为很多时候你不想删掉一个插件，你只是想让它在某些环境里不生效。

调试多轮对话时，我一般会临时禁用记忆插件，只保留过滤插件和Prompt组装插件，这样能更快定位问题出在谁身上。等状态稳定了，再把所有插件打开跑一遍回归。

4. 实操中踩过的坑与排查技巧实录

4.1 插件顺序的隐性依赖：先过滤还是先调用工具

插件系统好用是好用，但顺序问题一度让我很头疼。按我最初的优先级配置，keyword_filter是10，weather_tool是30，prompt_assembler是90。思路是过滤插件先跑，如果命中敏感词就把blocked置为True，后面的插件理论上就不该再跑了。

但问题是，process_all并没有做“遇到blocked就停止后续插件”的内置逻辑。每天早上第一个版本是这样跑的吗？不是。在某一轮测试里，用户输入“北京的天气怎样，顺便问一下测试占位词A”，weather_tool依然把天气数据塞进extras，prompt_assembler也照常把它拼进消息列表。虽然模型最终不会收到被block的消息，但白做了一次外部API调用，浪费了一次网络请求。

后来我在process_all里加了检查：

def process_all(self, context: dict) -> None: for name in self.order: if context.get("output", {}).get("blocked"): break self.plugins[name].process(context)

这个逻辑的意思是：只要有人把blocked置为True，整条链就短路，后面的插件直接跳过。这不是一个很酷的机制，但它反映了插件系统里无法回避的现实：有些插件之间就是有先后语义的，必须在框架层面提供短路能力。读过很多框架源码后我发现，大多数插件系统都会有类似的“中断”概念，只不过有的叫halt，有的叫skip，DeepSeek Harness直接用blocked字段表达。

4.2 热更新插件的经典翻车：旧资源没被释放

我的Harness有一个面向开发者的调试模式，修改插件文件后可以自动reload插件。听起来很美好，实际踩坑不少。最典型的问题是文件句柄泄漏：一个插件在init阶段打开了日志文件，reload后旧插件对象虽然被丢弃了，但它开启的文件流没有被close，导致每次改动插件后，磁盘上多一个“幽灵文件”占着空间。

解决方法是让PluginManager在重新扫描前先调用所有旧插件的close()：

def reload_plugins(self) -> None: self.close_all() self.plugins.clear() self.order.clear() self.scan_and_register() self.sort_plugins()

并且约定，任何插件的init里创建的资源，要么存放在context的plugin_state命名空间里，要么在close里显式释放。如果你在插件里偷偷用全局变量什么，那恕我无能为力，框架查不到也管不到。

还有一个连接池问题，某个数据库查询插件每次init都会新建一个数据库连接池，reload了三次就撑爆连接数。后来我改成了懒加载，不在init建连接，只在process时按需创建并缓存在plugin_state里，这个问题的发作频率骤降。

4.3 插件之间的通信：不要直接互相引用

插件生态起来后，很快会有插件之间要配合的需求。比如记忆插件想把自己整理的对话摘要提供给搜索插件，搜索插件拿去检索知识库。最直接的想法是让搜索插件import记忆插件，直接调用对方的方法，这样最省事。

我踩了一次之后就严禁这种写法了。原因是插件一旦直接互相引用，耦合就产生了，你更新A插件的内部方法签名，B插件隔天就报错，整个系统退回硬编码状态。我定了一条死规矩：插件之间只能通过context交换数据，不允许直接import其他插件类。

记忆插件把摘要写到extras["conversation_summary"]，搜索插件从那里读取。这样记忆插件的实现细节再变，只要key不变，搜索插件就完全不受影响。这条规矩看起来有点强迫症，但对插件生态的长期健康至关重要。

4.4 排查插件系统问题的三个步骤

最后分享一下我排查插件问题的思路。整体用“先二分、后单点、再日志”的顺序。

第一步二分：在配置里把插件数量砍到只剩一个嫌疑对象，如果问题消失，说明是几个插件之间的冲突；如果问题还在，说明问题在单个插件内部或框架本身。这一步能把排查范围缩小一个数量级。

第二步单点：确定是某个插件的问题后，单独对该插件做最小化复现。我一般会写一个脚本，只初始化这一个插件，构造一个最小context，然后调用一遍process。这能快速判断是插件逻辑错误，还是它对context字段的类型假设不对。

第三步日志：如果单点测试也复现不了，那就是时序或环境问题了，在插件process入口打点时间和字段快照，跑一次完整流程，对比不同插件读取到的同一key是否是同一个值。很多时候你会发现，你以为A插件写入了extras["weather_data"]，实际上因为key拼写差异，B插件读的是extras["weatherdata"]，这类问题不靠日志根本抓不到。

还有个独家背锅技巧：先看插件有没有try-except吃异常。插件代码里如果写了宽泛的except Exception: pass，那它会把所有错误静默吞掉。我在Harness里默认开了一个环境变量HARNESS_DEBUG=1，开启后，process里抛出的任何异常都会直接上抛，不会被任何插件内部捕获，测试阶段几乎总是开着它跑。