DeepSeek Harness实战：从零打造插件化AI Agent框架

之前在业务里接入 DeepSeek API 时，我一度写了不少“一次性脚本”：把固定提示词、工具调用逻辑和业务代码强耦合在一起。刚开始跑 Demo 没什么问题，等需求一多，今天加一个计算功能，明天加一个代码生成功能，后天又要接一个内部接口，每次都要改主流程，改完还要担心影响已有功能。后来接触 Harness 这种编排思路之后，我才意识到问题往往不是模型不够强，而是缺少一层能把模型调用、工具执行、任务流程统一管理起来的插件化框架。

这篇文章不打算带你直接去啃大型框架源码，而是从 0 到 1 实现一个轻量、可扩展的 DeepSeek Harness 示例项目，把 DeepAgent 和 Skill 插件机制讲明白。学完之后，你能理解 Harness 的底层工作方式，能自己动手写一个 Skill 插件，也能把 DeepSeek 或其他兼容 OpenAI 协议的 AI 大模型快速接入自己的 Agent 应用。

文章内容会覆盖：核心概念与架构分层、环境准备、Harness 工作原理、完整实战代码、Skill 插件自动发现机制、常见问题排查、工程化最佳实践。无论你是刚接触 AI 大模型应用开发的新手，还是已经在做智能体项目的后端开发，都可以照着本文跑通一条完整链路。

1. 背景与核心概念

1.1 什么是 DeepSeek Harness

Harness 原本有“马具、挽具”的意思，在编程领域可以理解为“把模型能力套上可控制的缰绳”。它不是代理工具，也不是网络代理，而是一个位于 AI 大模型和业务代码之间的编排层。Harness 负责跟模型通信，决定什么时候调用什么工具，然后把模型输出的“意图”转成真实动作。

DeepSeek Harness 可以理解为一套以 DeepSeek 模型为底层推理引擎、以 Harness 模式运行的插件化智能体框架。它做的事情包括：

• 管理大模型 API 的请求参数和上下文历史。
• 维护一个 Skill 插件注册表，让模型能够按需调用。
• 解析模型输出，识别调用哪个 Skill、传什么参数。
• 执行 Skill，并把执行结果返回给模型，最终生成面向用户的回答。

这种设计最大的价值在于“一切皆插件”。业务能力被封装成独立的 Skill，主程序只需要关心调度，不需要关心每个具体功能是怎么实现的。

1.2 Harness、DeepAgent、Skill 之间的关系

很多初学者会把 Harness、DeepAgent、Skill 混为一谈，这里先做一个区分。

• Harness：整体编排框架，相当于智能体的“主控程序”，负责加载配置、注册插件、调度对话、执行工具。
• DeepAgent：基于 Harness 创建出来的智能体实例，可以理解成面向用户的“助手角色”，持有模型客户端、对话历史、Skill 注册表。
• Skill：最小功能插件单元，每个 Skill 暴露名称、描述、参数定义和执行方法，例如“计算器”“代码生成”“查询天气”“调用内部接口”。

这三者的关系就是：DeepAgent 运行在 Harness 框架中，Harness 通过读取 SkillRegistry 里的插件来扩展 DeepAgent 的能力。用户输入一句话，DeepAgent 交给模型判断该用哪个 Skill，Harness 负责把参数传给 Skill，Skill 把结果返回，模型再基于结果做最终回答。

1.3 为什么选择插件化架构

如果把所有能力写死在主流程里，代码很快就会失控。插件化架构带来的收益非常直接。

第一是解耦。每个 Skill 是独立模块，新增功能不需要改动 Agent 主流程。第二是复用。同一个计算器 Skill 可以同时被多个 Agent 使用，减少重复开发。第三是可测。Skill 是纯函数式的输入输出，可以单独写单元测试，不需要启动完整服务。第四是演化。随着场景增加，你可以不断往 Harness 里添加新插件，而模型本身不需要重新训练。

2. 环境准备与版本说明

2.1 运行环境

本文示例代码以 Python 3.10 为基础编写，操作系统不限，Windows、macOS、Linux 都可以运行。你只需要准备一个终端，以及一个能访问 DeepSeek API 的账号。

依赖库尽量保持精简：

• requests：用于调用 DeepSeek API。
• python-dotenv：用于从 .env 文件加载环境变量，避免把密钥硬编码在代码里。

你需要先创建一个 DeepSeek 开放平台账号，并在后台申请 API Key。API 的 base_url 和 model 名称以官方文档为准，不同时期可能略有调整，本文示例默认使用https://api.deepseek.com作为 base_url。

2.2 示例项目结构

整个项目文件不多，但结构清晰，方便后续扩展。

deepseek-harness-demo/ ├── .env.example ├── main.py ├── requirements.txt └── harness/ ├── __init__.py ├── agent.py ├── base.py ├── registry.py ├── loader.py └── skills/ ├── __init__.py ├── calculator.py └── code_generator.py

如果你现在只有零散代码，不用急，下面会按文件逐个创建并解释。

3. 核心原理拆解：Harness 如何工作

3.1 模型调用层

模型调用层是 Harness 最基础的能力。它负责把消息列表发送给 DeepSeek API，并获取模型回复。这一层需要关心的核心点是消息格式、鉴权方式、超时和异常处理。

DeepSeek API 兼容 OpenAI Chat Completions 协议，所以请求路径一般是/chat/completions。实际开发中要避免每次请求都重新设置 HTTP 客户端，而应该把请求逻辑封装在 Agent 里统一处理。

下面是一个最小请求伪代码：

import requests resp = requests.post( "https://api.deepseek.com/chat/completions", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, json={ "model": model, "messages": messages, "temperature": 0.2, }, timeout=60, ) data = resp.json() reply = data["choices"][0]["message"]["content"].strip() return reply

在实际项目里，不建议裸用 requests，可以封装一个客户端类，把 API Key 读取、超时配置、重试策略、日志记录都收拢到一起。这样后续迁移模型供应商时，只需要改一个文件。

3.2 Skill 注册表

Skill 注册表是“一切皆插件”的基石。它本质上是一个内存字典，key 是 Skill 名称，value 是 Skill 实例。注册表对外提供三个核心方法：

• register：注册插件。
• get：按名称获取插件。
• all_skills：返回所有插件的元信息，供模型提示词使用。

这个设计非常简单，但已经可以支撑起完整的插件生命周期。后续如果要支持插件热加载，只需要在注册表外层增加文件扫描和 importlib 动态导入逻辑即可。

3.3 Agent 调度循环

Agent 调度循环是 DeepSeek Harness 的核心。整体流程是这样的：

1. 用户输入消息，Agent 把消息追加到对话历史。
2. Agent 调用模型，传入系统提示词和完整对话历史。
3. 模型返回一段 JSON 决策，可能是直接回复，也可能是请求调用某个 Skill。
4. 如果是调用 Skill，Agent 从注册表取出对应插件，执行 execute 方法。
5. Skill 返回结果追加到对话历史，Agent 再次调用模型生成最终回复。
6. 最终答案返回给用户。

这个循环看似简单，但已经涵盖了 Function Calling 的核心本质：让模型做“意图识别”，让 Harness 做“工具执行”，最后再做一次“总结回答”。

4. 完整实战：从 0 到 1 搭建 DeepSeek Harness

4.1 初始化项目与依赖

先创建项目目录和虚拟环境，然后安装依赖。

mkdir deepseek-harness-demo cd deepseek-harness-demo python -m venv venv source venv/bin/activate pip install requests python-dotenv

在项目根目录创建requirements.txt：

requests python-dotenv

创建.env.example文件，用于记录环境变量模板：

DEEPSEEK_API_KEY=sk-your-key DEEPSEEK_BASE_URL=https://api.deepseek.com DEEPSEEK_MODEL=deepseek-chat

实际使用时把.env.example复制一份为.env，然后填上你自己的 API Key。注意.env文件不要提交到 Git 仓库，建议在.gitignore中加入.env。

4.2 定义 Skill 插件基类

Skill 基类定义了所有插件必须实现的接口。先创建harness/base.py：

from abc import ABC, abstractmethod from typing import Any, Dict class Skill(ABC): # 插件名称，必须唯一 name: str = "" # 插件功能描述，模型根据描述决定是否调用 description: str = "" # 参数 JSON Schema，用于约束模型传参 parameters: Dict[str, Any] = {} @abstractmethod def execute(self, **kwargs) -> str: """执行插件逻辑，返回给模型或用户的文本结果。""" raise NotImplementedError

这里的核心设计是parameters。模型不会凭空知道每个插件要传什么参数，Harness 会在系统提示词中展示这个 JSON Schema，模型再根据用户输入提取相应参数。所以插件描述写得越清楚，模型调用准确率越高。

4.3 实现 Skill 注册表

接下来创建harness/registry.py：

from typing import Dict, List, Optional from .base import Skill class SkillRegistry: def __init__(self) -> None: self._skills: Dict[str, Skill] = {} def register(self, skill: Skill) -> None: if not skill.name: raise ValueError("Skill 的 name 不能为空") self._skills[skill.name] = skill def get(self, name: str) -> Optional[Skill]: return self._skills.get(name) def all_skills(self) -> List[dict]: return [ { "name": skill.name, "description": skill.description, "parameters": skill.parameters, } for skill in self._skills.values() ]

这个注册表虽然简单，但已经能支撑起后续的插件化开发。真实项目中可以继续扩展，比如支持插件冲突检测、插件优先级、插件依赖注入等能力。

4.4 编写内置 Skill：计算器与代码生成

先写一个计算器插件，路径是harness/skills/calculator.py：

from harness.base import Skill class CalculatorSkill(Skill): name = "calculator" description = "执行简单四则运算，输入数学表达式，返回计算结果。" parameters = { "type": "object", "properties": { "expression": { "type": "string", "description": "需要计算的数学表达式，例如 (1+2)*3" } }, "required": ["expression"] } def execute(self, **kwargs) -> str: expression = kwargs.get("expression", "").strip() if not expression: return "表达式不能为空" try: # 仅用于本地演示，生产环境不要直接使用 eval result = eval(expression, {"__builtins__": {}}, {}) return str(result) except Exception as e: return f"计算失败：{e}"

这里要特别强调，eval在生产环境存在安全风险，即使限制了__builtins__，也不建议在公开服务中直接使用。更好的做法是接入simpleeval、sympy或自己写表达式解析器。这里的 eval 只是为了降低示例复杂度。

再写一个代码生成插件，路径是harness/skills/code_generator.py：

from harness.base import Skill class CodeGeneratorSkill(Skill): name = "generate_code" description = "根据自然语言需求生成指定语言的代码片段。" parameters = { "type": "object", "properties": { "requirement": { "type": "string", "description": "用户希望实现的代码需求描述" }, "language": { "type": "string", "description": "目标语言，默认 python" } }, "required": ["requirement"] } def execute(self, **kwargs) -> str: requirement = kwargs.get("requirement", "").strip() language = kwargs.get("language", "python") if not requirement: return "需求描述不能为空" snippet = ( f"# 需求：{requirement}\n" f"# 语言：{language}\n" "def todo():\n" " # TODO: 实现你的业务逻辑\n" " pass\n" ) return f"```{language}\n{snippet}\n```"

这个插件没有真正调用大模型生成代码，而是返回了一个模板片段。真实项目中，你可以在这里调用代码生成 API、读取项目模板、甚至调用内部代码搜索服务。插件化的好处正在于此：主流程不需要知道内部实现。

创建harness/skills/__init__.py和harness/__init__.py，文件内容为空即可，Python 需要它们把目录识别为包。

4.5 实现 DeepAgent 核心调度

现在到了最关键的环节，创建harness/agent.py。这个文件实现了 DeepAgent，包括系统提示词构造、模型调用、Skill 决策解析、Skill 执行结果回填。

import json from typing import List import requests from .registry import SkillRegistry class DeepAgent: def __init__( self, api_key: str, base_url: str, model: str, registry: SkillRegistry, ): self.api_key = api_key self.base_url = base_url.rstrip("/") self.model = model self.registry = registry self.messages: List[dict] = [] self.system_prompt = self._build_system_prompt() def _build_system_prompt(self) -> str: skills = self.registry.all_skills() skill_lines = "\n".join( f"- {s['name']}: {s['description']} 参数：{json.dumps(s['parameters'], ensure_ascii=False)}" for s in skills ) return f"""你是一个 DeepAgent 智能助手的调度大脑。 你有以下 Skill 插件可用： {skill_lines} 当用户的问题需要调用某个 Skill 时，你只能输出如下 JSON： {{"action": "call_skill", "skill": "技能名", "args": {{"参数1": "值1"}}}} 否则，你直接回复用户，输出如下 JSON： {{"action": "reply", "content": "你的回答"}} 注意：只输出 JSON，不要输出多余解释。""" def chat(self, user_input: str) -> str: self.messages.append({"role": "user", "content": user_input}) first_response = self._call_model(self.messages) decision = self._parse_decision(first_response) if decision.get("action") == "call_skill": skill_name = decision.get("skill", "") skill_args = decision.get("args", {}) skill = self.registry.get(skill_name) if not skill: final_text = f"没有找到名为 {skill_name} 的 Skill" else: skill_result = skill.execute(**skill_args) self.messages.append({"role": "assistant", "content": first_response}) self.messages.append({ "role": "user", "content": f"Skill 执行结果如下：\n{skill_result}\n请基于这个结果给用户一个最终回答。" }) final_response = self._call_model(self.messages) final_text = final_response else: final_text = decision.get("content", first_response) self.messages.append({"role": "assistant", "content": final_text}) return final_text def _call_model(self, messages: List[dict]) -> str: url = f"{self.base_url}/chat/completions" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } payload = { "model": self.model, "messages": messages, "temperature": 0.2, } resp = requests.post(url, headers=headers, json=payload, timeout=60) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"].strip() def _parse_decision(self, text: str) -> dict: try: return json.loads(text) except json.JSONDecodeError: # 兼容模型输出带 markdown 代码块的情况 cleaned = ( text.strip() .removeprefix("```json") .removeprefix("```") .removesuffix("```") .strip() ) try: return json.loads(cleaned) except json.JSONDecodeError: return {"action": "reply", "content": text}

这段代码有几个设计点值得注意。

系统提示词把 Skill 列表和 JSON 输出约束一起告诉模型，模型不需要提前知道业务细节。第一轮调用让模型做“调度决策”，如果模型认为需要调用 Skill，就返回一个结构化 JSON。Harness 拿到 JSON 后执行插件，并把执行结果作为用户消息再次交给模型，让模型生成最终回答。这种两段式调用虽然多消耗一次 token，但逻辑非常清晰。

_parse_decision处理了模型偶尔输出 Markdown 代码块的情况，增强了容错能力。

4.6 编写入口文件并运行验证

最后创建main.py：

import os from dotenv import load_dotenv from harness.agent import DeepAgent from harness.registry import SkillRegistry from harness.skills.calculator import CalculatorSkill from harness.skills.code_generator import CodeGeneratorSkill load_dotenv() registry = SkillRegistry() registry.register(CalculatorSkill()) registry.register(CodeGeneratorSkill()) agent = DeepAgent( api_key=os.getenv("DEEPSEEK_API_KEY", ""), base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"), model=os.getenv("DEEPSEEK_MODEL", "deepseek-chat"), registry=registry, ) print("DeepAgent 已启动，输入 exit 退出") while True: user_input = input("你> ") if user_input.lower() in {"exit", "quit"}: break answer = agent.chat(user_input) print(f"Agent> {answer}")

运行命令：

python main.py

如果 API Key 配置正确，你会看到类似下面的交互过程：

DeepAgent 已启动，输入 exit 退出 你> 计算 (1+2)*3 的结果 Agent> 计算结果为 9。 你> exit

这里需要注意的是，模型最终输出内容取决于模型判断和表达方式，不一定和上面完全一致，但整体流程应该能跑通。如果模型返回的不是预期 JSON，可以适当把 temperature 调低到 0.1 或 0。

5. 扩展：让一切皆插件

5.1 插件自动发现

前面演示的是手动注册插件，但项目变大之后，我们希望把新 Skill 文件丢进目录就能自动加载。这时可以写一个自动发现加载器，路径是harness/loader.py：

import importlib import pkgutil from .base import Skill from .registry import SkillRegistry def auto_load_skills(package_name: str, registry: SkillRegistry) -> None: package = importlib.import_module(package_name) for module_info in pkgutil.iter_modules(package.__path__): module = importlib.import_module(f"{package_name}.{module_info.name}") for attr in dir(module): obj = getattr(module, attr) if isinstance(obj, type) and issubclass(obj, Skill) and obj is not Skill: registry.register(obj())

使用时，只需要在 main.py 里把原来的手动注册替换成：

from harness.loader import auto_load_skills registry = SkillRegistry() auto_load_skills("harness.skills", registry)

这样，以后每新增一个 Skill 文件，只要类名和模块名符合规范，就会自动注册到 Harness 中，真正实现“一切皆插件”。

5.2 动态热加载

自动发现适合启动时加载，如果还想在运行过程中加入新插件，可以使用 importlib 的reload或重新导入模块。需要考虑到线程安全、插件冲突、旧插件实例释放等问题。简单场景下，可以先重启进程让新插件生效，再逐步引入复杂的热加载机制。

5.3 新增一个 Skill 的标准流程

根据前面的代码风格，新增 Skill 只需要三步：

1. 在harness/skills/目录下创建新文件。
2. 定义一个继承Skill的类，实现execute方法，并填写name、description、parameters。
3. 如果是自动发现模式，直接重启程序即可；如果是手动模式，在 main.py 注册该 Skill。

这一步对团队协作尤其友好。算法工程师负责写 Skill 内部逻辑，后端工程师负责维护 Harness 调度层，互不阻塞。

6. 常见问题与排查思路

在实际运行 DeepSeek Harness 时，可能会遇到一些问题，下面整理成排查表格，方便快速定位。

排查问题时，我建议优先在 DeepAgent 的_call_model方法里增加日志输出，记录每次请求的 URL、payload 和响应状态码。日志能帮助你在第一时间定位是网络问题、鉴权问题还是模型返回格式问题。

7. 最佳实践与工程建议

7.1 Skill 描述比实现更重要

模型是根据description来决定是否调用某个 Skill 的，所以描述一定要写清楚三件事：这个 Skill 能做什么、适合什么场景、参数怎么填。一个好的描述可以显著降低模型误调用概率。

7.2 参数校验和异常兜底

Skill 长年运行在生产环境，不能假设模型每次都会给你合法参数。execute方法内部要做空值校验、类型校验、范围校验，并且把异常信息转换成可读文本返回给模型。这样模型还能根据错误信息自我纠错，而不是让用户看到堆栈。

7.3 安全边界与授权控制

如果某个 Skill 会删除文件、更新数据库、调用外部接口，必须在描述里明确要求模型获取用户确认，同时在 Harness 层增加权限校验。高危操作建议引入审批流，或至少设置白名单。计算器这类低危 Skill 可以直接执行，但数据库写操作绝不能裸奔。

7.4 日志与可观测性

整个 Agent 链路至少需要记录：用户原始输入、模型决策 JSON、实际调用的 Skill 名称、传给 Skill 的参数、Skill 执行耗时、Skill 返回结果、最终回答。日志格式要统一，方便在日志平台上按 request_id 串联。

7.5 成本与响应延迟控制

两段式调度意味着至少调用两次模型 API，成本会比单次调用高。可以在系统提示词里引导模型“只有在明确需要工具时才调用 Skill”，也可以为简单问题提供纯回复模式。另外，设置合理的 timeout，对模型 API 做重试和熔断，能把故障影响控制在局部。

8. 总结与下一步学习建议

本文用一个最小可运行的 DeepSeek Harness 示例，把 AI 大模型应用开发中的模型调用、Skill 插件、Agent 调度这三条主线串联了起来。你现在应该能理解 Harness 的定位，能写自己的 Skill 插件，也能根据日志排查常见问题。

如果你准备在自己项目中落地这套思路，我的建议是：先不要急着写复杂 Agent 流程，把两三个 Skill 做扎实，尤其是描述和参数校验。真正决定智能体能力上限的，往往不是模型选型，而是你的 Skill 设计。

接下来可以继续学习几个方向：一是 DeepSeek 官方 API 的 Function Calling 能力，它能替代手写 JSON 决策解析；二是 MCP 这类模型上下文协议，让插件接入更规范化；三是 LangChain、LlamaIndex 等生态里的 Agent 工具调用机制。把本文结构吃透，再接触这些框架会轻松很多。

如果这篇文章对你有帮助，可以收藏备用，也欢迎在评论区分享你在接入 DeepSeek 时遇到的坑。