尧图网站建设 
之前在业务里接入 DeepSeek API 时我一度写了不少“一次性脚本”把固定提示词、工具调用逻辑和业务代码强耦合在一起。刚开始跑 Demo 没什么问题等需求一多今天加一个计算功能明天加一个代码生成功能后天又要接一个内部接口每次都要改主流程改完还要担心影响已有功能。后来接触 Harness 这种编排思路之后我才意识到问题往往不是模型不够强而是缺少一层能把模型调用、工具执行、任务流程统一管理起来的插件化框架。这篇文章不打算带你直接去啃大型框架源码而是从 0 到 1 实现一个轻量、可扩展的 DeepSeek Harness 示例项目把 DeepAgent 和 Skill 插件机制讲明白。学完之后你能理解 Harness 的底层工作方式能自己动手写一个 Skill 插件也能把 DeepSeek 或其他兼容 OpenAI 协议的 AI 大模型快速接入自己的 Agent 应用。文章内容会覆盖核心概念与架构分层、环境准备、Harness 工作原理、完整实战代码、Skill 插件自动发现机制、常见问题排查、工程化最佳实践。无论你是刚接触 AI 大模型应用开发的新手还是已经在做智能体项目的后端开发都可以照着本文跑通一条完整链路。1. 背景与核心概念1.1 什么是 DeepSeek HarnessHarness 原本有“马具、挽具”的意思在编程领域可以理解为“把模型能力套上可控制的缰绳”。它不是代理工具也不是网络代理而是一个位于 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 交给模型判断该用哪个 SkillHarness 负责把参数传给 SkillSkill 把结果返回模型再基于结果做最终回答。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: fBearer {api_key}, Content-Type: application/json, }, json{ model: model, messages: messages, temperature: 0.2, }, timeout60, ) 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 的核心。整体流程是这样的用户输入消息Agent 把消息追加到对话历史。Agent 调用模型传入系统提示词和完整对话历史。模型返回一段 JSON 决策可能是直接回复也可能是请求调用某个 Skill。如果是调用 SkillAgent 从注册表取出对应插件执行 execute 方法。Skill 返回结果追加到对话历史Agent 再次调用模型生成最终回复。最终答案返回给用户。这个循环看似简单但已经涵盖了 Function Calling 的核心本质让模型做“意图识别”让 Harness 做“工具执行”最后再做一次“总结回答”。4. 完整实战从 0 到 1 搭建 DeepSeek Harness4.1 初始化项目与依赖先创建项目目录和虚拟环境然后安装依赖。mkdir deepseek-harness-demo cd deepseek-harness-demo python -m venv venv source venv/bin/activate pip install requests python-dotenv在项目根目录创建requirements.txtrequests python-dotenv创建.env.example文件用于记录环境变量模板DEEPSEEK_API_KEYsk-your-key DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat实际使用时把.env.example复制一份为.env然后填上你自己的 API Key。注意.env文件不要提交到 Git 仓库建议在.gitignore中加入.env。4.2 定义 Skill 插件基类Skill 基类定义了所有插件必须实现的接口。先创建harness/base.pyfrom 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.pyfrom 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.pyfrom harness.base import Skill class CalculatorSkill(Skill): name calculator description 执行简单四则运算输入数学表达式返回计算结果。 parameters { type: object, properties: { expression: { type: string, description: 需要计算的数学表达式例如 (12)*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.pyfrom 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_asciiFalse)} 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: fSkill 执行结果如下\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: fBearer {self.api_key}, Content-Type: application/json, } payload { model: self.model, messages: messages, temperature: 0.2, } resp requests.post(url, headersheaders, jsonpayload, timeout60) 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.pyimport 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_keyos.getenv(DEEPSEEK_API_KEY, ), base_urlos.getenv(DEEPSEEK_BASE_URL, https://api.deepseek.com), modelos.getenv(DEEPSEEK_MODEL, deepseek-chat), registryregistry, ) print(DeepAgent 已启动输入 exit 退出) while True: user_input input(你 ) if user_input.lower() in {exit, quit}: break answer agent.chat(user_input) print(fAgent {answer})运行命令python main.py如果 API Key 配置正确你会看到类似下面的交互过程DeepAgent 已启动输入 exit 退出 你 计算 (12)*3 的结果 Agent 计算结果为 9。 你 exit这里需要注意的是模型最终输出内容取决于模型判断和表达方式不一定和上面完全一致但整体流程应该能跑通。如果模型返回的不是预期 JSON可以适当把 temperature 调低到 0.1 或 0。5. 扩展让一切皆插件5.1 插件自动发现前面演示的是手动注册插件但项目变大之后我们希望把新 Skill 文件丢进目录就能自动加载。这时可以写一个自动发现加载器路径是harness/loader.pyimport 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 只需要三步在harness/skills/目录下创建新文件。定义一个继承Skill的类实现execute方法并填写name、description、parameters。如果是自动发现模式直接重启程序即可如果是手动模式在 main.py 注册该 Skill。这一步对团队协作尤其友好。算法工程师负责写 Skill 内部逻辑后端工程师负责维护 Harness 调度层互不阻塞。6. 常见问题与排查思路在实际运行 DeepSeek Harness 时可能会遇到一些问题下面整理成排查表格方便快速定位。问题现象常见原因解决思路返回 401 UnauthorizedAPI Key 错误或未配置检查 .env 中 DEEPSEEK_API_KEY确认 Key 没有多余空格返回 404 或 model not foundbase_url 或 model 名称不正确去 DeepSeek 官方文档确认当前模型名和接口地址模型输出不是合法 JSON提示词约束不够强或 temperature 过高降低 temperature增强 JSON 示例加入重试解析逻辑提示“没有找到名为 xx 的 Skill”插件未注册或模型拼错技能名打印 registry.all_skills()确认插件名称是否一致上下文过长导致请求失败对话历史不断累积超过模型 token 上限做滑动窗口裁剪或对历史做摘要压缩Skill 执行时报参数缺失模型生成的 args 缺少必填项在 execute 中做默认值和类型校验返回可读错误信息使用 eval 计算表达式时报错或安全问题表达式包含非法语法或恶意代码生产环境改用 simpleeval、sympy 或自研解析器排查问题时我建议优先在 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 时遇到的坑。 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 HarnessHarness 原本有“马具、挽具”的意思在编程领域可以理解为“把模型能力套上可控制的缰绳”。它不是代理工具也不是网络代理而是一个位于 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 交给模型判断该用哪个 SkillHarness 负责把参数传给 SkillSkill 把结果返回模型再基于结果做最终回答。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: fBearer {api_key}, Content-Type: application/json, }, json{ model: model, messages: messages, temperature: 0.2, }, timeout60, ) 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 的核心。整体流程是这样的用户输入消息Agent 把消息追加到对话历史。Agent 调用模型传入系统提示词和完整对话历史。模型返回一段 JSON 决策可能是直接回复也可能是请求调用某个 Skill。如果是调用 SkillAgent 从注册表取出对应插件执行 execute 方法。Skill 返回结果追加到对话历史Agent 再次调用模型生成最终回复。最终答案返回给用户。这个循环看似简单但已经涵盖了 Function Calling 的核心本质让模型做“意图识别”让 Harness 做“工具执行”最后再做一次“总结回答”。4. 完整实战从 0 到 1 搭建 DeepSeek Harness4.1 初始化项目与依赖先创建项目目录和虚拟环境然后安装依赖。mkdir deepseek-harness-demo cd deepseek-harness-demo python -m venv venv source venv/bin/activate pip install requests python-dotenv在项目根目录创建requirements.txtrequests python-dotenv创建.env.example文件用于记录环境变量模板DEEPSEEK_API_KEYsk-your-key DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat实际使用时把.env.example复制一份为.env然后填上你自己的 API Key。注意.env文件不要提交到 Git 仓库建议在.gitignore中加入.env。4.2 定义 Skill 插件基类Skill 基类定义了所有插件必须实现的接口。先创建harness/base.pyfrom 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.pyfrom 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.pyfrom harness.base import Skill class CalculatorSkill(Skill): name calculator description 执行简单四则运算输入数学表达式返回计算结果。 parameters { type: object, properties: { expression: { type: string, description: 需要计算的数学表达式例如 (12)*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.pyfrom 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_asciiFalse)} 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: fSkill 执行结果如下\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: fBearer {self.api_key}, Content-Type: application/json, } payload { model: self.model, messages: messages, temperature: 0.2, } resp requests.post(url, headersheaders, jsonpayload, timeout60) 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.pyimport 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_keyos.getenv(DEEPSEEK_API_KEY, ), base_urlos.getenv(DEEPSEEK_BASE_URL, https://api.deepseek.com), modelos.getenv(DEEPSEEK_MODEL, deepseek-chat), registryregistry, ) print(DeepAgent 已启动输入 exit 退出) while True: user_input input(你 ) if user_input.lower() in {exit, quit}: break answer agent.chat(user_input) print(fAgent {answer})运行命令python main.py如果 API Key 配置正确你会看到类似下面的交互过程DeepAgent 已启动输入 exit 退出 你 计算 (12)*3 的结果 Agent 计算结果为 9。 你 exit这里需要注意的是模型最终输出内容取决于模型判断和表达方式不一定和上面完全一致但整体流程应该能跑通。如果模型返回的不是预期 JSON可以适当把 temperature 调低到 0.1 或 0。5. 扩展让一切皆插件5.1 插件自动发现前面演示的是手动注册插件但项目变大之后我们希望把新 Skill 文件丢进目录就能自动加载。这时可以写一个自动发现加载器路径是harness/loader.pyimport 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 只需要三步在harness/skills/目录下创建新文件。定义一个继承Skill的类实现execute方法并填写name、description、parameters。如果是自动发现模式直接重启程序即可如果是手动模式在 main.py 注册该 Skill。这一步对团队协作尤其友好。算法工程师负责写 Skill 内部逻辑后端工程师负责维护 Harness 调度层互不阻塞。6. 常见问题与排查思路在实际运行 DeepSeek Harness 时可能会遇到一些问题下面整理成排查表格方便快速定位。问题现象常见原因解决思路返回 401 UnauthorizedAPI Key 错误或未配置检查 .env 中 DEEPSEEK_API_KEY确认 Key 没有多余空格返回 404 或 model not foundbase_url 或 model 名称不正确去 DeepSeek 官方文档确认当前模型名和接口地址模型输出不是合法 JSON提示词约束不够强或 temperature 过高降低 temperature增强 JSON 示例加入重试解析逻辑提示“没有找到名为 xx 的 Skill”插件未注册或模型拼错技能名打印 registry.all_skills()确认插件名称是否一致上下文过长导致请求失败对话历史不断累积超过模型 token 上限做滑动窗口裁剪或对历史做摘要压缩Skill 执行时报参数缺失模型生成的 args 缺少必填项在 execute 中做默认值和类型校验返回可读错误信息使用 eval 计算表达式时报错或安全问题表达式包含非法语法或恶意代码生产环境改用 simpleeval、sympy 或自研解析器排查问题时我建议优先在 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 时遇到的坑。