DeepSeek Harness插件开发实战：批量处理与提示词模板管理

最近在深度使用 DeepSeek Harness 进行 AI 应用开发时，发现官方插件市场虽然丰富，但在一些特定场景下，比如批量处理、自定义提示词模板管理等方面，还是存在一些空白。为了提升自己的开发效率，我动手开发了两个实用插件，正好填补了这些需求缺口。本文将完整分享这两个插件的开发思路、实现代码以及集成到 DeepSeek Harness 的全过程，无论你是想直接使用这些插件，还是学习如何为 Harness 开发自己的扩展，都能从中获得实用价值。

1. 背景与核心概念：为什么需要自定义插件？

DeepSeek Harness 作为一个强大的 AI 应用开发与部署平台，其插件系统是其核心扩展能力之一。通过插件，开发者可以无缝集成外部工具、自定义数据处理流程、增强模型能力或优化交互界面。

官方插件通常覆盖了最通用、最高频的需求，例如连接主流数据库、调用常见 API、进行基础的数据格式转换等。然而，在实际项目开发中，我们经常会遇到一些个性化、场景化的需求：

1. 批量任务处理：需要一次性对成百上千条数据进行预处理、调用模型、后处理并汇总结果。手动一条条操作或编写外部脚本既低效又容易出错。
2. 复杂的提示词工程：项目可能依赖一套精心设计的、多步骤的提示词模板，这些模板需要版本管理、快速切换和参数化注入。
3. 与内部系统集成：需要连接公司内部的 CRM、OA 或其他私有系统，这些显然不会有官方插件。
4. 特定领域的增强功能：例如，为代码生成场景添加特定的代码风格检查、为文案生成添加违禁词过滤等。

当官方插件无法满足这些需求时，自定义插件就成了最佳选择。它允许你将重复、复杂的操作封装成一个可复用的“黑盒”，在 Harness 的流程中像搭积木一样使用，极大提升开发效率和流程的标准化程度。

2. 环境准备与插件开发基础

在开始编码之前，我们需要明确 DeepSeek Harness 插件开发的基本环境和规范。

核心环境：

• DeepSeek Harness：确保你有一个可用的 Harness 实例。可以是云端服务，也可以是本地部署版本。本文基于 Harness 的开放插件协议进行开发。
• 编程语言：DeepSeek Harness 插件通常使用Python或Node.js开发，因为其插件系统易于与这些语言的生态集成。本文将使用Python作为示例。
• Python 环境：建议使用 Python 3.8+。使用venv或conda创建独立的虚拟环境。
• 必要的库：除了标准库，可能还需要requests(用于 HTTP 调用)、pydantic(用于数据验证)等。我们会按需安装。

插件项目结构：一个标准的 Harness 插件项目通常包含以下文件：

my-harness-plugin/ ├── plugin.json # 插件元数据清单文件 (必须) ├── main.py # 插件主逻辑文件 (或 index.js) ├── requirements.txt # Python 依赖文件 ├── README.md # 插件说明文档 └── (可选) frontend.js # 如果有自定义前端界面

关键文件plugin.json解析：这个文件是插件的“身份证”，Harness 通过它来识别和加载插件。

{ "schema_version": "1.0", "name": "com.yourname.batch_processor", "version": "1.0.0", "display_name": "批量处理器", "description": "一个用于批量处理数据的插件，支持并发控制和结果汇总。", "author": "你的名字", "tags": ["batch", "utility", "data-processing"], "icon": "🔄", "inputs": [ { "name": "data_list", "type": "array", "description": "需要处理的原始数据列表", "required": true }, { "name": "process_function", "type": "string", "description": "对单条数据处理的Harness流程ID或函数名", "required": true }, { "name": "max_workers", "type": "number", "description": "最大并发工作线程数", "required": false, "default": 3 } ], "outputs": [ { "name": "results", "type": "array", "description": "处理结果列表，与输入顺序对应" }, { "name": "summary", "type": "object", "description": "处理摘要，包含成功/失败计数" } ] }

• schema_version: 插件清单的版本。
• name: 插件的唯一标识符，通常使用反向域名格式。
• inputs/outputs: 定义了插件的输入和输出参数及其类型（如string,number,array,object,boolean）。Harness 编辑器会根据这个生成可视化的配置界面。

3. 插件一：智能批量处理器 (Batch Processor)

痛点：在 Harness 中构建一个处理单条数据的流程很容易，但当你需要对一个数据集中的每一条数据都执行这个流程时，就需要手动循环调用，或者编写外部脚本。这个过程笨重、无法可视化、且难以处理错误和并发。

解决方案：开发一个“批量处理器”插件。它接收一个数据列表和一个“单条数据处理流程”的引用，然后自动、并发地执行，并返回统一格式的结果。

3.1 插件功能设计

1. 输入：
   • data_list: 任意类型的数组。
   • process_function: 一个 Harness 中已有的、用于处理单条数据的“技能”(Skill)或流程的 ID。
   • max_workers: 控制并发度，防止过度消耗资源。
2. 处理过程：
   • 插件内部调用 Harness 的 API，将process_function和单条数据作为参数，发起执行请求。
   • 使用线程池或异步IO进行并发控制。
   • 收集每条数据的执行结果或错误信息。
3. 输出：
   • results: 一个数组，包含每条数据的处理结果。保持与输入相同的顺序。
   • summary: 一个对象，包含total（总数）、success（成功数）、failed（失败数）以及errors（详细的错误信息列表）。

3.2 核心代码实现 (main.py)

# main.py import concurrent.futures import json import logging from typing import List, Any, Dict import requests from pydantic import BaseModel, Field # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 定义插件的输入参数模型（Pydantic 用于验证和文档） class BatchInput(BaseModel): data_list: List[Any] = Field(..., description="需要处理的原始数据列表") process_function: str = Field(..., description="Harness中单条数据处理流程的ID") max_workers: int = Field(default=3, ge=1, le=10, description="最大并发工作线程数 (1-10)") # 假设我们需要Harness实例的API地址和密钥（通常由Harness运行时环境提供） harness_api_endpoint: str = Field(default="http://localhost:8090", description="Harness API地址") api_key: str = Field(default="", description="访问Harness API的密钥") # 定义插件的输出模型 class BatchOutput(BaseModel): results: List[Any] = Field(..., description="处理结果列表") summary: Dict[str, Any] = Field(..., description="处理摘要") def execute_single_item(api_endpoint: str, api_key: str, skill_id: str, item: Any) -> Dict: """调用Harness API执行单个处理流程""" url = f"{api_endpoint.rstrip('/')}/api/v1/skills/{skill_id}/execute" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" } payload = { "inputs": {"data": item} # 假设目标技能接收一个名为`data`的输入 } try: response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() result = response.json() # 假设成功执行的返回结构中有 `output` 字段 return {"status": "success", "data": result.get("output"), "raw": result} except requests.exceptions.RequestException as e: logger.error(f"调用技能 {skill_id} 处理数据失败: {e}") return {"status": "failed", "error": str(e), "data": item} def main(input_data: Dict[str, Any]) -> Dict[str, Any]: """插件的主入口函数，Harness会调用此函数并传入配置的参数""" # 1. 验证输入 try: batch_input = BatchInput(**input_data) except Exception as e: logger.error(f"输入参数验证失败: {e}") raise ValueError(f"无效的输入参数: {e}") # 2. 准备并发执行 results = [] errors = [] with concurrent.futures.ThreadPoolExecutor(max_workers=batch_input.max_workers) as executor: # 提交所有任务 future_to_item = { executor.submit( execute_single_item, batch_input.harness_api_endpoint, batch_input.api_key, batch_input.process_function, item ): idx for idx, item in enumerate(batch_input.data_list) } # 按任务完成顺序获取结果，但我们需要按原始顺序整理 temp_results = [None] * len(batch_input.data_list) for future in concurrent.futures.as_completed(future_to_item): idx = future_to_item[future] try: single_result = future.result(timeout=35) # 比请求超时稍长 temp_results[idx] = single_result except concurrent.futures.TimeoutError: error_msg = f"处理索引 {idx} 的数据超时" logger.error(error_msg) temp_results[idx] = {"status": "failed", "error": error_msg, "data": batch_input.data_list[idx]} # 3. 整理结果和错误信息 for idx, item_result in enumerate(temp_results): if item_result["status"] == "success": results.append(item_result["data"]) else: results.append(None) # 失败的位置用None占位 errors.append({ "index": idx, "data": batch_input.data_list[idx], "error": item_result.get("error", "Unknown error") }) # 4. 生成摘要 total = len(batch_input.data_list) success_count = sum(1 for r in results if r is not None) failed_count = total - success_count summary = { "total": total, "success": success_count, "failed": failed_count, "errors": errors } # 5. 返回插件输出 output = BatchOutput(results=results, summary=summary) return output.dict()

3.3 在 DeepSeek Harness 中使用该插件

1. 打包插件：将上述plugin.json和main.py以及requirements.txt（内容为pydantic>=2.0.0 requests>=2.28.0）打包成一个 ZIP 文件。
2. 安装插件：在你的 DeepSeek Harness 管理界面中，找到“插件管理”或“扩展中心”，选择“上传自定义插件”，上传 ZIP 文件。
3. 在流程中使用：
   • 在 Harness 的画布上，从插件列表中找到你刚上传的“批量处理器”。
   • 将其拖入画布，配置输入参数：
     • data_list: 可以是一个静态数组，也可以连接上一个节点的输出（例如一个“读取文件”节点输出的列表）。
     • process_function: 填入你事先创建好的、用于处理单条数据的那个“技能”的 ID。你可以在该技能的详情页找到其 ID。
     • max_workers: 根据你的服务器性能调整，比如5。
     • harness_api_endpoint和api_key：通常 Harness 会在插件运行时自动注入这些环境变量，如果插件设计得好，可以不用在界面配置。我们的代码中将其作为输入，是为了灵活性。在实际开发中，更佳实践是从环境变量读取。
   • 连接输出：将results输出连接到下一个节点（如“写入数据库”、“生成报告”），将summary输出可以连接到一个“日志”或“通知”节点。

4. 插件二：提示词模板管理器 (Prompt Template Manager)

痛点：复杂的 AI 应用往往依赖精心设计的提示词。在 Harness 中，提示词可能散落在各个“LLM调用”节点里，难以统一管理、版本控制和复用。修改一个通用提示词模板需要找到所有使用它的节点逐一更改。

解决方案：开发一个“提示词模板管理器”插件。它将提示词模板存储为可复用的资源，支持变量插值、版本管理和一键应用。

4.1 插件功能设计

1. 核心功能：
   • 存储模板：将带有占位符（如{{topic}}、{{length}}）的提示词模板保存起来，并命名（如“小红书文案生成”）。
   • 渲染模板：根据模板名称和提供的变量值，生成最终的提示词。
   • 模板列表：获取所有已保存的模板。
2. 输入：
   • action: 操作类型，如“render”（渲染）、“save”（保存）、“list”（列表）。
   • template_name: 模板名称。
   • template_content: 保存模板时的内容。
   • variables: 渲染模板时提供的变量字典。
3. 输出：
   • rendered_prompt: 渲染后的最终提示词（当action为render时）。
   • templates: 模板列表（当action为list时）。
   • status: 操作状态（成功/失败）。
   • message: 附加信息。

4.2 核心代码实现 (main.py)

这个插件需要持久化存储，为了简化，我们使用本地 JSON 文件。在生产环境中，应替换为数据库。

# main.py - Prompt Template Manager import json import os import re from typing import Dict, Any, List, Optional from enum import Enum from pydantic import BaseModel, Field import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 定义操作类型 class ActionType(str, Enum): RENDER = "render" SAVE = "save" LIST = "list" GET = "get" # 输入模型 class TemplateManagerInput(BaseModel): action: ActionType = Field(..., description="要执行的操作: render(渲染), save(保存), list(列表), get(获取)") template_name: Optional[str] = Field(None, description="模板名称") template_content: Optional[str] = Field(None, description="模板内容（用于save操作）") variables: Optional[Dict[str, Any]] = Field(default={}, description="渲染模板时使用的变量字典") # 输出模型 class TemplateManagerOutput(BaseModel): status: str = Field(..., description="执行状态: success 或 error") message: Optional[str] = Field(None, description="状态信息") rendered_prompt: Optional[str] = Field(None, description="渲染后的提示词") templates: Optional[List[Dict[str, Any]]] = Field(None, description="模板列表") template: Optional[Dict[str, Any]] = Field(None, description="单个模板详情") class PromptTemplateManager: """简单的提示词模板管理器，使用JSON文件存储""" def __init__(self, storage_path: str = "./prompt_templates.json"): self.storage_path = storage_path self._ensure_storage() def _ensure_storage(self): """确保存储文件存在""" if not os.path.exists(self.storage_path): with open(self.storage_path, 'w', encoding='utf-8') as f: json.dump({}, f, ensure_ascii=False, indent=2) def _load_templates(self) -> Dict: """加载所有模板""" try: with open(self.storage_path, 'r', encoding='utf-8') as f: return json.load(f) except (json.JSONDecodeError, FileNotFoundError): return {} def _save_templates(self, templates: Dict): """保存所有模板""" with open(self.storage_path, 'w', encoding='utf-8') as f: json.dump(templates, f, ensure_ascii=False, indent=2) def save_template(self, name: str, content: str) -> bool: """保存或更新模板""" templates = self._load_templates() templates[name] = { "content": content, "updated_at": datetime.now().isoformat() } self._save_templates(templates) logger.info(f"模板 '{name}' 已保存。") return True def render_template(self, name: str, variables: Dict[str, Any]) -> Optional[str]: """渲染模板""" templates = self._load_templates() if name not in templates: logger.error(f"模板 '{name}' 不存在。") return None template_content = templates[name]["content"] # 简单的 {{variable}} 替换 try: rendered = template_content for key, value in variables.items(): placeholder = "{{" + key + "}}" rendered = rendered.replace(placeholder, str(value)) # 检查是否还有未替换的占位符（可选，用于调试） remaining_placeholders = re.findall(r'\{\{(\w+)\}\}', rendered) if remaining_placeholders: logger.warning(f"模板 '{name}' 渲染后仍有未替换变量: {remaining_placeholders}") return rendered except Exception as e: logger.error(f"渲染模板 '{name}' 时出错: {e}") return None def list_templates(self) -> List[Dict]: """列出所有模板""" templates = self._load_templates() result = [] for name, meta in templates.items(): result.append({ "name": name, "updated_at": meta.get("updated_at"), "preview": meta["content"][:100] + "..." if len(meta["content"]) > 100 else meta["content"] }) return result def get_template(self, name: str) -> Optional[Dict]: """获取单个模板详情""" templates = self._load_templates() if name in templates: return {"name": name, **templates[name]} return None def main(input_data: Dict[str, Any]) -> Dict[str, Any]: """插件主入口""" try: plugin_input = TemplateManagerInput(**input_data) except Exception as e: return {"status": "error", "message": f"输入参数错误: {e}"} manager = PromptTemplateManager() output_data = {"status": "success", "message": "操作成功"} try: if plugin_input.action == ActionType.SAVE: if not plugin_input.template_name or not plugin_input.template_content: raise ValueError("保存模板需要提供 template_name 和 template_content") success = manager.save_template(plugin_input.template_name, plugin_input.template_content) if not success: output_data["status"] = "error" output_data["message"] = "保存模板失败" elif plugin_input.action == ActionType.RENDER: if not plugin_input.template_name: raise ValueError("渲染模板需要提供 template_name") rendered = manager.render_template(plugin_input.template_name, plugin_input.variables or {}) if rendered is None: output_data["status"] = "error" output_data["message"] = f"模板 '{plugin_input.template_name}' 不存在或渲染失败" else: output_data["rendered_prompt"] = rendered elif plugin_input.action == ActionType.LIST: output_data["templates"] = manager.list_templates() elif plugin_input.action == ActionType.GET: if not plugin_input.template_name: raise ValueError("获取模板需要提供 template_name") template = manager.get_template(plugin_input.template_name) if template: output_data["template"] = template else: output_data["status"] = "error" output_data["message"] = f"模板 '{plugin_input.template_name}' 不存在" except Exception as e: logger.exception("插件执行过程中发生错误") output_data["status"] = "error" output_data["message"] = str(e) return output_data

4.3 对应的 plugin.json

{ "schema_version": "1.0", "name": "com.yourname.prompt_manager", "version": "1.0.0", "display_name": "提示词模板管理器", "description": "用于集中管理、渲染和复用提示词模板的插件。", "author": "你的名字", "tags": ["prompt", "template", "management", "productivity"], "icon": "📝", "inputs": [ { "name": "action", "type": "string", "description": "执行的操作: render(渲染), save(保存), list(列表), get(获取)", "required": true, "enum": ["render", "save", "list", "get"] }, { "name": "template_name", "type": "string", "description": "模板名称（render, save, get 操作时需要）", "required": false }, { "name": "template_content", "type": "string", "description": "模板内容（save 操作时需要）", "required": false }, { "name": "variables", "type": "object", "description": "渲染模板时使用的变量键值对，例如 {\"topic\": \"科技\", \"length\": 500}", "required": false } ], "outputs": [ { "name": "status", "type": "string", "description": "执行状态: success 或 error" }, { "name": "message", "type": "string", "description": "状态详细信息" }, { "name": "rendered_prompt", "type": "string", "description": "渲染后的完整提示词（action=render 时输出）" }, { "name": "templates", "type": "array", "description": "模板列表（action=list 时输出）" }, { "name": "template", "type": "object", "description": "单个模板的详细信息（action=get 时输出）" } ] }

4.4 使用场景示例

假设你有一个“生成社交媒体文案”的流程。

1. 保存模板：首先，使用插件创建一个名为“twitter_thread”的模板，内容为：

   请围绕 {{topic}} 这个话题，生成一条包含 {{point_count}} 个要点的推特线程。语言风格为 {{style}}。每个要点不超过280字符。

2. 在流程中渲染：在 Harness 画布中，添加一个“提示词模板管理器”节点。
   • action设置为render。
   • template_name设置为“twitter_thread”。
   • variables设置为{"topic": "人工智能的未来", "point_count": 3, "style": "激动人心且通俗易懂"}。
3. 连接LLM节点：将该节点的rendered_prompt输出，直接连接到下游的“DeepSeek模型调用”节点的prompt输入。
4. 统一修改：当你想优化所有推特线程的生成质量时，只需通过插件的save操作更新“twitter_thread”这一个模板，所有使用该模板的流程都会自动生效。

5. 插件开发、调试与部署的进阶技巧

5.1 本地调试插件

在将插件安装到 Harness 之前，强烈建议进行本地调试。

1. 创建测试脚本：在你的插件项目根目录创建一个test_plugin.py文件。

   # test_plugin.py import sys import os sys.path.insert(0, os.path.dirname(__file__)) from main import main as plugin_main # 测试批量处理器 batch_input = { "data_list": ["item1", "item2", "item3"], "process_function": "your_skill_id_here", "max_workers": 2, "harness_api_endpoint": "http://your-harness-host:port", "api_key": "your_api_key_here" } # 注意：这里需要你有一个真实的技能ID和可访问的Harness API # result = plugin_main(batch_input) # print(json.dumps(result, indent=2, ensure_ascii=False)) # 测试提示词管理器 template_input = { "action": "save", "template_name": "test_template", "template_content": "Hello, {{name}}! Welcome to {{place}}." } result = plugin_main(template_input) print("Save result:", result) render_input = { "action": "render", "template_name": "test_template", "variables": {"name": "Alice", "place": "CSDN"} } result = plugin_main(render_input) print("Render result:", result)

2. 模拟 Harness 环境：Harness 在调用插件时，会注入一些上下文信息（如 API 密钥、用户信息）。你可以在main函数中通过input_data获取，也可以在本地测试时模拟这些数据。
3. 使用日志：在代码中使用logging模块输出详细信息，这在排查线上问题时至关重要。

5.2 处理插件配置与密钥

硬编码 API 密钥或配置路径是极不安全的。最佳实践是：

• 从环境变量读取：修改插件代码，优先从os.environ读取配置。

  api_endpoint = os.getenv("HARNESS_API_ENDPOINT", "http://localhost:8090") api_key = os.getenv("HARNESS_API_KEY", "")

• 在 Harness 中配置：Harness 通常提供“插件配置”或“密钥管理”功能。你可以在plugin.json中定义配置项，Harness 会在运行时将其传递给插件。

5.3 提升插件性能与健壮性

1. 异步支持：对于 I/O 密集型操作（如网络请求），考虑使用asyncio和aiohttp替代requests和线程池，以获得更高的并发性能。
2. 错误处理与重试：在execute_single_item函数中增加重试逻辑和更细致的错误分类（网络错误、API 错误、业务逻辑错误）。
3. 输入验证与清理：使用 Pydantic 进行严格的输入验证，防止恶意或异常数据导致插件崩溃。
4. 资源限制：对于批量处理器，不仅要限制并发数 (max_workers)，还可以考虑增加超时设置和最大处理条目限制，防止插件耗尽主机资源。

6. 常见问题与排查思路

在开发和集成插件过程中，你可能会遇到以下问题：

问题现象	可能原因	排查步骤与解决方案
插件安装失败	plugin.json格式错误或缺少必填字段；依赖包不兼容。	1. 使用 JSON 校验工具检查plugin.json。 2. 查看 Harness 后台的插件安装日志，通常会有详细错误信息。 3. 确保requirements.txt中的包版本与 Harness 环境兼容。
插件在流程中执行报错	插件代码存在语法或运行时错误；输入数据格式不符合预期；依赖的 API 不可达。	1.本地复现：使用测试脚本模拟 Harness 的输入，在本地运行插件。 2.查看日志：确保插件代码中使用了logging，并在 Harness 中查看该插件节点的执行日志。 3.简化输入：使用最简单、最标准的输入数据测试，排除数据问题。
批量处理器调用技能超时或失败	目标技能本身执行慢或出错；网络问题；API 密钥无效；并发过高导致 Harness 或下游服务过载。	1.单独测试技能：确保在 Harness 中单独运行目标技能是成功的。 2.降低并发：将max_workers设为 1 测试，排除并发问题。 3.检查网络与认证：确认harness_api_endpoint和api_key正确无误，且从插件运行环境可以访问该端点。 4.增加超时：适当增加execute_single_item函数中的请求超时时间。
提示词管理器模板渲染结果不对	变量名与模板中的占位符不匹配；变量值包含特殊字符导致替换异常。	1.检查变量键名：确保variables字典的键（如topic）与模板中的占位符（如{{topic}}）完全一致，包括大小写。 2.打印调试信息：在render_template函数中，打印出替换前后的模板内容。 3.使用更健壮的模板引擎：可以考虑集成Jinja2等成熟的模板引擎，它们功能更强大，错误处理更好。
插件执行速度慢	插件逻辑复杂；存在同步阻塞操作（如同步 HTTP 请求）；未充分利用并发。	1.性能分析：使用 Python 的cProfile模块分析代码瓶颈。 2.异步化改造：将网络请求等 I/O 操作改为异步模式 (asyncio/aiohttp)。 3.优化算法：检查是否有不必要的循环或重复计算。

7. 最佳实践与工程建议

1. 单一职责：一个插件只做好一件事。批量处理器就专注于批量调用，模板管理器就专注于模板的 CRUD。功能复杂的插件难以维护和调试。
2. 完备的输入验证：使用 Pydantic 模型严格定义和验证输入参数。提供清晰、具体的错误信息，帮助流程设计者快速定位问题。
3. 详细的日志记录：在关键步骤（开始、结束、错误）和耗时操作处记录日志。日志是线上问题排查的生命线。
4. 资源管理：像批量处理器这类可能消耗大量资源的插件，必须提供配置项（如max_workers）让使用者控制，并在代码中做好资源清理（如关闭线程池、会话）。
5. 向后兼容：更新插件时，尽量避免破坏性变更。如果必须修改输入输出结构，考虑升级主版本号（如从1.x.x到2.0.0），并在文档中明确说明迁移方式。
6. 编写清晰的文档：在README.md中写明插件的用途、输入输出详解、使用示例、配置项说明以及常见问题。好的文档能极大降低使用成本。
7. 安全性：
   • 永远不要将密钥等敏感信息硬编码在代码中。
   • 对用户输入的模板内容（特别是渲染时）进行安全检查，防止模板注入攻击（如果使用Jinja2等引擎，尤其要注意）。
   • 插件应有适当的权限边界，避免执行任意系统命令或访问敏感文件。

通过开发这两个插件，我们不仅解决了实际工作中的效率痛点，也深入了解了 DeepSeek Harness 插件系统的运作机制。从需求分析、设计、编码、调试到最终集成，整个过程是对全栈开发能力的一次很好锻炼。更重要的是，你将重复性劳动转化为了团队甚至社区可共享的资产。当你发现官方生态的空白时，不妨自己动手“补全”它，这或许是开源精神在 AI 应用开发领域的一种体现。希望本文的分享能成为你开发自己第一个 Harness 插件的起点，期待在插件市场中看到你的作品。