我先说一个真实的痛点很多同学看到“Agent”“工作流”“插件开发”这些词就头大总感觉是只有大厂工程师才能掌握的高阶技能。其实当你把模型调用这一段跑通后剩下的无非是“循环、工具、上下文、钩子”这几件事的组合。如果你手里已经有一个 DeepSeek API Key想从零开始搭建一个属于自己的 AI Agent又不想被各种动辄上千行的框架牵着走那么这篇文章非常适合你。我会从最基础的 API 调用讲起然后一步步实现一个轻量 Harness再给它开发插件、编排工作流最后组装成一个能自动生成“项目日报”的专属 Agent。全程代码都可以直接复制运行环境依赖只有openai和requests。文章末尾还会整理常见报错和工程化建议。1. 背景与核心概念1.1 DeepSeek 是什么DeepSeek 是由深度求索公司推出的大语言模型系列目前最常被开发者用到的是两个能力通过 API 进行对话补全以 OpenAI 兼容协议提供接口便于迁移。也就是说你已经很熟悉的openaiPython SDK只要把base_url换成 DeepSeek 的地址再把 API Key 换掉就能直接调用 DeepSeek 模型。这一点对开发者非常友好也是大量 Agent 项目愿意接入 DeepSeek 的原因之一。常用方式分为两种方式场景直接调 API实现单轮或多轮对话、文本生成、代码补全基于 API 搭建 Agent让模型具备工具调用、自我决策、多步骤任务执行能力1.2 什么是 HarnessHarness 这个词在 AI Agent 领域逐渐频繁出现但很多教程没有解释清楚。简单来说Harness 是“模型调用之外的控制层”。你可以把大模型想象成一个聪明的实习生他知道很多知识但你直接问他一句“帮我写周报”他只能凭空写因为他没有你的项目数据也没有查看文件、调用外部服务的权限。这时候Harness 就像实习生的“工作环境”它负责把系统提示词塞给模型它负责维护对话历史它让模型能发现并调用外部工具它决定模型输出结果后是否继续执行下一步。所以 Harness 是一个工程结构而不是一个新模型。它解决的核心问题是如何让大模型稳定地完成“多步骤、带工具、带约束”的复杂任务。1.3 Agent、工作流、插件的边界这三个概念经常在一起出现但分工不同Agent具备自主决策能力的执行体可以理解用户目标决定调用哪些工具、按什么顺序执行工作流把任务拆成固定节点每个节点有明确输入输出节点之间按顺序或条件推进插件一种可复用、可插拔的功能扩展模块为 Agent 或工作流提供额外能力。举例来说用户说“帮我生成一份今天的工作总结”。Agent 会判断需要先获取当前时间、然后扫描今天的日志文件、再让模型总结最后输出 Markdown。这个“判断”过程是 Agent 的决策“获取时间、扫描文件、总结、输出”这几个步骤串起来就是一条工作流而“获取时间”“扫描文件”这两个能力就可以做成插件。1.4 为什么大家都在谈 DeepSeek Harness近一年多DeepSeek 模型在中文语境、代码生成和逻辑推理上的表现非常亮眼而且 API 价格相比很多国外模型更低。于是社区开始把 DeepSeek 和“Codex Harness”“自定义 Harness”结合起来实现低成本的 Agent 开发。需要说明的是目前并不存在一个官方统一命名的“DeepSeek Harness 产品”。大家在网络上搜索到的更多是基于 DeepSeek API 自建 Harness 层或者使用社区开源脚手架。本文采用“从零手写 Harness”的方式让你彻底理解内部原理之后无论切换到 Dify、n8n 还是 LangChain都能快速上手。2. 环境准备从账号到第一个 API 调用2.1 注册并获取 API Key在开始写代码前你需要两个东西DeepSeek 开放平台账号一个 API Key。登录开放平台后在“接口密钥”或“API Keys”页面创建密钥。请复制并妥善保存因为密钥只显示一次。如果忘记只能删除重建。这里要提醒两点API Key 是敏感凭证不要写进代码仓库不要在前端页面直接暴露 API Key否则会被盗刷。建议在本地项目根目录创建.env文件后面从环境变量读取。2.2 准备 Python 环境与依赖本文示例基于 Python 3.10 以上版本。推荐使用虚拟环境mkdir deepseek-harness-demo cd deepseek-harness-demo python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate安装依赖pip install openai requests python-dotenv到这里环境就准备好了。2.3 调用 DeepSeek 的完整示例先创建一个基础调用脚本hello_deepseek.py# 文件路径deepseek-harness-demo/hello_deepseek.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) resp client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 你是一个 Python 技术助手回答要简洁。}, {role: user, content: 请用一句话解释什么是 Harness。} ], temperature0.7 ) print(resp.choices[0].message.content)在.env文件中配置DEEPSEEK_API_KEYsk-xxxxxxxxxxxxxxxx运行python hello_deepseek.py预期输出类似Harness 是包裹在大模型外部的一层控制框架用来管理上下文、工具调用和执行循环。2.4 模型选择建议DeepSeek API 常见的模型标识有deepseek-chat适合对话、文本生成、代码编写成本低deepseek-reasoner适合复杂推理、数学、逻辑分析思考过程更充分。不同时期开放平台可能调整模型名称建议以开发文档中“模型列表”为准。本文示例默认使用deepseek-chat你在实际项目中可以按任务复杂度切换。3. 深入理解 Harness 的核心设计3.1 不只是“大模型调用”Harness 解决的问题直接调用 API 只能完成单轮问答。但真实的业务场景通常是连续且复杂的比如用户提出目标Agent 拆解为多个步骤某些步骤需要调用外部工具工具结果需要回填给模型模型根据结果决定下一步最终输出满足用户要求。如果这些逻辑全部堆在业务代码里很快就会失控。Harness 的价值就是把这一套流程抽象为可复用结构。3.2 Harness 的四个核心部件一个可用的 Harness 至少需要包含模型客户端负责与大模型通信消息记忆维护 system、user、assistant 的对话历史工具注册中心保存可被模型调用的工具列表执行控制循环决定模型响应后是否继续调用工具以及何时停止。用一句话概括Harness 模型客户端 对话记忆 工具集合 执行循环。3.3 一个最小 Harness 的代码骨架下面实现一个最小版本方便你理解整体脉络。# 文件路径deepseek-harness-demo/harness_core.py from openai import OpenAI class Tool: def __init__(self, name, description, handler): self.name name self.description description self.handler handler def run(self, **kwargs): return self.handler(**kwargs) class Harness: def __init__(self, api_key, base_urlhttps://api.deepseek.com, modeldeepseek-chat): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model self.history [] self.tools {} def register_tool(self, tool: Tool): 注册一个工具到 Harness 中。 self.tools[tool.name] tool def add_message(self, role, content): self.history.append({role: role, content: content}) def run(self, user_input, max_steps3): 执行一次完整的 Agent 任务。 self.add_message(user, user_input) for step in range(max_steps): response self.client.chat.completions.create( modelself.model, messagesself.history, ) content response.choices[0].message.content self.add_message(assistant, content) # 简化协议如果模型输出包含 TOOL_CALL: 前缀则解析工具调用 if TOOL_CALL: in content: tool_name, args self._parse_tool_call(content) if tool_name not in self.tools: self.add_message(user, f工具 {tool_name} 不存在请换一个工具。) continue tool_result self.tools[tool_name].run(**args) self.add_message(user, f工具调用结果{tool_result}) else: return content return self.history[-1][content] def _parse_tool_call(self, content): 从模型输出解析工具名和参数。 body content.split(TOOL_CALL:, 1)[1].strip() first_line body.split(\n, 1)[0].strip() parts first_line.split(maxsplit1) tool_name parts[0] args {} if len(parts) 1: import json try: args json.loads(parts[1]) except json.JSONDecodeError: args {query: parts[1]} return tool_name, args这段代码虽然简单但已经能跑通“模型决定调用工具 → Harness 执行工具 → 结果回填 → 模型继续决策”的闭环。4. 插件开发实战让 Harness 拥有扩展能力4.1 为什么需要插件机制上面的 Harness 已经有register_tool方法但把所有工具函数都写在一个文件里并不好维护。插件机制的核心目的每个插件独立成文件或目录插件可以复用、共享、动态加载团队协作时不同插件由不同人维护主程序不需要改动只要加载插件目录。4.2 插件的注册与加载我采用一个简单的约定每个插件模块提供一个register(harness)函数该函数负责创建 Tool 实例并注册到 Harness。主程序遍历插件目录并加载。4.3 开发一个时间插件现在我们来写第一个插件它提供获取当前时间的能力。# 文件路径deepseek-harness-demo/plugins/time_plugin.py from datetime import datetime def get_current_time(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def register(harness): from harness_core import Tool harness.register_tool(Tool( nameget_current_time, description获取当前系统时间返回格式为年-月-日 时:分:秒, handlerget_current_time, ))注意register函数既没有创建模型客户端也没有关心业务逻辑它只把能力挂载到 Harness 上。4.4 开发一个目录扫描插件接着写一个“扫描项目目录”的插件。这个工具会默认限制在指定根目录避免越权读取系统任意路径。# 文件路径deepseek-harness-demo/plugins/file_plugin.py import os def list_project_files(root_dir.): 列出指定目录下的所有文件默认当前目录。 result [] abs_root os.path.abspath(root_dir) for foldername, _, filenames in os.walk(abs_root): # 跳过隐藏目录如 .git 和 .venv if /. in foldername or foldername.endswith(.venv): continue for filename in filenames: if filename.startswith(.): continue full_path os.path.join(foldername, filename) result.append(full_path) return \n.join(result[:50]) if result else 目录为空 def register(harness): from harness_core import Tool harness.register_tool(Tool( namelist_project_files, description列出项目目录下的文件列表用于了解项目结构。, handlerlist_project_files, ))如果后续要扩展开启文件读取能力也要保持同样的安全边界只允许读取白名单目录内的文件。4.5 插件热加载与命名规范在主程序中可以写一个自动加载插件目录的函数# 文件路径deepseek-harness-demo/load_plugins.py import importlib import pkgutil import plugins def load_all_plugins(harness): 自动加载 plugins 目录下所有插件。 for module_info in pkgutil.iter_modules(plugins.__path__): module_name module_info.name module importlib.import_module(fplugins.{module_name}) if hasattr(module, register): module.register(harness) print(f[插件已加载] {module_name})插件命名建议文件全小写使用下划线分隔如time_plugin.py每个插件模块必须有register(harness)工具名使用动词开头如get_current_time方便模型理解。5. 工作流实战编排一个“日报生成 Agent”5.1 工作流需求拆解现在我们把 Harness、插件和工作流结合起来实现一个真实场景自动生成项目日报。需求如下获取当前时间扫描项目目录得到当日可能改动的文件列表把时间和文件列表交给模型让模型生成一份结构化日报输出 Markdown 格式日报。这里的设计点在于工作流不是把每一步硬编码写在main函数里而是定义成节点每个节点有“执行”和“下一步判断”的能力。5.2 定义工作流节点# 文件路径deepseek-harness-demo/workflow_node.py from dataclasses import dataclass, field from typing import Callable, Any dataclass class WorkflowNode: name: str func: Callable[[Any], Any] next_node: str None def execute(self, context): result self.func(context) context[self.name] result return self.next_nodecontext是一个字典保存工作流运行过程中的共享数据。每个节点执行后可以把结果写入context并返回下一个节点的名称。如果返回None表示流程结束。5.3 实现工作流执行器# 文件路径deepseek-harness-demo/workflow_engine.py from workflow_node import WorkflowNode class WorkflowEngine: def __init__(self): self.nodes {} def add_node(self, node: WorkflowNode): self.nodes[node.name] node def run(self, start_node: str, context: dict): current start_node while current is not None: node self.nodes[current] current node.execute(context) return context工作流引擎本体的逻辑非常简单拿到当前节点执行它再根据返回值跳到下一个节点。5.4 完整运行示例与结果说明现在把上面所有模块串起来。# 文件路径deepseek-harness-demo/run_daily_report.py import os from dotenv import load_dotenv from workflow_engine import WorkflowEngine from workflow_node import WorkflowNode from harness_core import Harness load_dotenv() def build_context(): return { project_dir: ., current_time: , file_list: , report: } def get_time_node(context): from plugins.time_plugin import get_current_time context[current_time] get_current_time() return scan_files def scan_files_node(context): from plugins.file_plugin import list_project_files context[file_list] list_project_files(context[project_dir]) return generate_report def generate_report_node(context): api_key os.getenv(DEEPSEEK_API_KEY) harness Harness(api_keyapi_key) prompt f 请根据以下信息生成一份项目日报使用 Markdown 格式。 当前时间{context[current_time]} 项目文件列表 {context[file_list]} 日报结构要求 1. 今日工作概述 2. 主要改动文件 3. 遇到的问题 4. 明日计划 reply harness.run(prompt, max_steps1) context[report] reply return None def build_workflow(): engine WorkflowEngine() engine.add_node(WorkflowNode(start, get_time_node, scan_files)) engine.add_node(WorkflowNode(scan_files, scan_files_node, generate_report)) engine.add_node(WorkflowNode(generate_report, generate_report_node)) return engine if __name__ __main__: engine build_workflow() result engine.run(start, build_context()) print( 日报生成结果 ) print(result[report])运行python run_daily_report.py预期输出会是一个包含“今日工作概述”“主要改动文件”“遇到的问题”“明日计划”四部分的 Markdown 日报。这个例子虽然简单却把三件事讲清楚了工作流节点如何串联插件如何在节点中被复用Harness 如何在最后一步作为“内容生成器”完成创作型任务。6. 搭建一个带工具调用的专属 Agent6.1 Agent 与普通对话的区别普通对话里我们只是把用户消息发给模型返回文本。而在 Agent 中模型会输出“需要调用工具的意图”Harness 解析意图并执行工具再把结果返回给模型直到模型认为任务完成。关键点工具调用不是简单的if判断而是由模型根据用户意图自主选择模型不知道工具的内部实现它只看到工具名和描述工具返回结果必须被转成模型可读的文本才能继续推理。6.2 工具调用主循环我已在上文实现了一个文本协议版主循环。它的工作流程是用户输入任务模型先判断是否需要工具如果需要就输出TOOL_CALL: 工具名 参数JSONHarness 解析并执行工具工具结果以 user 消息回填模型看到结果后要么继续调用下一个工具要么输出最终答案。为了让模型稳定输出这种格式有一点非常重要系统提示词里必须给出明确的格式说明。6.3 完整 Agent 代码接下来写一个更完整的 Agent 文件它会自动加载插件并在系统提示词中说明工具调用协议。# 文件路径deepseek-harness-demo/agent_main.py import json import os from dotenv import load_dotenv from harness_core import Harness from load_plugins import load_all_plugins load_dotenv() SYSTEM_PROMPT 你是一个智能项目助手。你可以使用以下工具 {tool_descriptions} 如果需要调用工具请严格输出以下格式 TOOL_CALL: 工具名 参数JSON 参数JSON是可选的。如果不需要参数只输出 TOOL_CALL: 工具名 当得到工具结果后继续推理并最终给用户完整回答。不要编造工具执行结果。 def build_agent(): api_key os.getenv(DEEPSEEK_API_KEY) harness Harness(api_keyapi_key) load_all_plugins(harness) tool_lines [] for name, tool in harness.tools.items(): tool_lines.append(f- {name}: {tool.description}) system_prompt SYSTEM_PROMPT.format(tool_descriptions\n.join(tool_lines)) harness.add_message(system, system_prompt) return harness if __name__ __main__: agent build_agent() while True: user_input input(请输入任务输入 exit 退出) if user_input.strip().lower() exit: break reply agent.run(user_input, max_steps5) print( Agent 回复 ) print(reply) print()运行python agent_main.py使用时可以输入类似这样的任务请获取当前时间然后扫描项目目录最后告诉我项目里有哪些 Python 文件。Agent 会通过调用get_current_time和list_project_files完成任务并最终给出结构化回答。如果你的 DeepSeek API 版本支持 OpenAI 兼容的 function calling 格式也可以把工具列表转为tools参数传给接口。文本协议的好处是不依赖特定模型可移植性更强。7. 常见问题与排查思路开发 DeepSeek Harness 时最常遇到下面几类问题。问题现象常见原因解决思路401 认证失败API Key 错误或未设置环境变量检查.env中的 Key确认没有多余空格429 请求过多账户额度不足或并发超限查看开放平台用量提高间隔或加缓存模型返回空内容上下文过长或内容被安全策略过滤删减长文本分段处理工具调用一直循环模型没有正确理解工具结果优化工具描述增强系统提示词设置最大步数插件未加载插件文件缺少register函数检查模块命名和函数签名触发了本地文件越权读取工具没有限制根目录使用os.path.abspath并做路径白名单校验几个排查建议先手动调用 DeepSeek API确认 Key 和网络没有问题再单独测试插件函数确认返回结果正确最后通过 Harness 执行完整流程把每步 history 打印出来定位是哪一层丢失了上下文遇到工具解析失败优先检查模型输出的TOOL_CALL:格式是否规范必要时在系统提示词中加入“必须严格输出 JSON”。8. 最佳实践与工程建议8.1 把 API Key 当作最高优先级的安全边界无论项目多小都不要把 API Key 硬编码在前端或公开仓库。建议做法是使用.env文件生产环境使用密钥管理服务为不同环境创建不同 Key设置用量告警。8.2 Harness 与工作流职责分离Harness 负责“模型对话 工具调用”工作流负责“任务节点编排”。两者不要混在一起。如果发现一个节点里既要调用模型又要操作文件建议拆成两个节点。8.3 工具协议要稳定工具名、参数格式、返回结果格式要尽可能稳定。模型非常依赖工具描述描述不清晰时最容易出现工具选择错误。可以定期整理工具调用日志分析模型误用场景。8.4 日志与可观测性每个 Agent 任务都应记录用户输入每一步模型输出工具调用结果最终回复。最简单的方式是在 Harness 的run方法中加入日志打印或写入文件。生产环境建议接入结构化日志系统。8.5 异常处理与降级策略当模型连续 N 步无法完成任务时要主动终止。上文代码里的max_steps就是兜底参数。更复杂的场景可以加入“重试”“换模型”“直接返回已有结果”等降级策略。8.6 上下文的长度控制DeepSeek 模型虽然上下文窗口不小但无限增长历史会让成本越来越高、速度越来越慢。建议对早期对话做摘要压缩每次只保留最近 N 轮关键信息工具结果如果太长先截断或做摘要再回填。9. 下一步可以怎么玩到这里你已经亲手实现了“API 调用 → Harness 封装 → 插件开发 → 工作流编排 → Agent 组装”的完整链路。不要小看这套代码很多生产级 Agent 的核心循环本质上就是这么转的。下一步你可以试试给 Harness 增加“代码执行工具”但务必在沙箱环境中运行对接企业内部的接口比如查询工单、批量获取数据把工作流执行器扩展成支持条件分支和循环节点将 Harness 封装成 FastAPI 服务对外提供 Agent API。如果觉得手写 Harness 比较底层想快速搭建产品原型也可以参考 Dify、n8n、Coze 这类工作流平台。但理解了底层原理之后你会发现这些平台里的节点、组件、触发器本质上都是 Harness 概念的商业化封装。建议你把文章中的示例代码跑通一遍然后再尝试改造。Agent 开发最怕的就是只看不练手写一个最小闭环比什么都重要。 从零搭建DeepSeek Harness：Agent、插件与工作流实战
我先说一个真实的痛点很多同学看到“Agent”“工作流”“插件开发”这些词就头大总感觉是只有大厂工程师才能掌握的高阶技能。其实当你把模型调用这一段跑通后剩下的无非是“循环、工具、上下文、钩子”这几件事的组合。如果你手里已经有一个 DeepSeek API Key想从零开始搭建一个属于自己的 AI Agent又不想被各种动辄上千行的框架牵着走那么这篇文章非常适合你。我会从最基础的 API 调用讲起然后一步步实现一个轻量 Harness再给它开发插件、编排工作流最后组装成一个能自动生成“项目日报”的专属 Agent。全程代码都可以直接复制运行环境依赖只有openai和requests。文章末尾还会整理常见报错和工程化建议。1. 背景与核心概念1.1 DeepSeek 是什么DeepSeek 是由深度求索公司推出的大语言模型系列目前最常被开发者用到的是两个能力通过 API 进行对话补全以 OpenAI 兼容协议提供接口便于迁移。也就是说你已经很熟悉的openaiPython SDK只要把base_url换成 DeepSeek 的地址再把 API Key 换掉就能直接调用 DeepSeek 模型。这一点对开发者非常友好也是大量 Agent 项目愿意接入 DeepSeek 的原因之一。常用方式分为两种方式场景直接调 API实现单轮或多轮对话、文本生成、代码补全基于 API 搭建 Agent让模型具备工具调用、自我决策、多步骤任务执行能力1.2 什么是 HarnessHarness 这个词在 AI Agent 领域逐渐频繁出现但很多教程没有解释清楚。简单来说Harness 是“模型调用之外的控制层”。你可以把大模型想象成一个聪明的实习生他知道很多知识但你直接问他一句“帮我写周报”他只能凭空写因为他没有你的项目数据也没有查看文件、调用外部服务的权限。这时候Harness 就像实习生的“工作环境”它负责把系统提示词塞给模型它负责维护对话历史它让模型能发现并调用外部工具它决定模型输出结果后是否继续执行下一步。所以 Harness 是一个工程结构而不是一个新模型。它解决的核心问题是如何让大模型稳定地完成“多步骤、带工具、带约束”的复杂任务。1.3 Agent、工作流、插件的边界这三个概念经常在一起出现但分工不同Agent具备自主决策能力的执行体可以理解用户目标决定调用哪些工具、按什么顺序执行工作流把任务拆成固定节点每个节点有明确输入输出节点之间按顺序或条件推进插件一种可复用、可插拔的功能扩展模块为 Agent 或工作流提供额外能力。举例来说用户说“帮我生成一份今天的工作总结”。Agent 会判断需要先获取当前时间、然后扫描今天的日志文件、再让模型总结最后输出 Markdown。这个“判断”过程是 Agent 的决策“获取时间、扫描文件、总结、输出”这几个步骤串起来就是一条工作流而“获取时间”“扫描文件”这两个能力就可以做成插件。1.4 为什么大家都在谈 DeepSeek Harness近一年多DeepSeek 模型在中文语境、代码生成和逻辑推理上的表现非常亮眼而且 API 价格相比很多国外模型更低。于是社区开始把 DeepSeek 和“Codex Harness”“自定义 Harness”结合起来实现低成本的 Agent 开发。需要说明的是目前并不存在一个官方统一命名的“DeepSeek Harness 产品”。大家在网络上搜索到的更多是基于 DeepSeek API 自建 Harness 层或者使用社区开源脚手架。本文采用“从零手写 Harness”的方式让你彻底理解内部原理之后无论切换到 Dify、n8n 还是 LangChain都能快速上手。2. 环境准备从账号到第一个 API 调用2.1 注册并获取 API Key在开始写代码前你需要两个东西DeepSeek 开放平台账号一个 API Key。登录开放平台后在“接口密钥”或“API Keys”页面创建密钥。请复制并妥善保存因为密钥只显示一次。如果忘记只能删除重建。这里要提醒两点API Key 是敏感凭证不要写进代码仓库不要在前端页面直接暴露 API Key否则会被盗刷。建议在本地项目根目录创建.env文件后面从环境变量读取。2.2 准备 Python 环境与依赖本文示例基于 Python 3.10 以上版本。推荐使用虚拟环境mkdir deepseek-harness-demo cd deepseek-harness-demo python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate安装依赖pip install openai requests python-dotenv到这里环境就准备好了。2.3 调用 DeepSeek 的完整示例先创建一个基础调用脚本hello_deepseek.py# 文件路径deepseek-harness-demo/hello_deepseek.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) resp client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: 你是一个 Python 技术助手回答要简洁。}, {role: user, content: 请用一句话解释什么是 Harness。} ], temperature0.7 ) print(resp.choices[0].message.content)在.env文件中配置DEEPSEEK_API_KEYsk-xxxxxxxxxxxxxxxx运行python hello_deepseek.py预期输出类似Harness 是包裹在大模型外部的一层控制框架用来管理上下文、工具调用和执行循环。2.4 模型选择建议DeepSeek API 常见的模型标识有deepseek-chat适合对话、文本生成、代码编写成本低deepseek-reasoner适合复杂推理、数学、逻辑分析思考过程更充分。不同时期开放平台可能调整模型名称建议以开发文档中“模型列表”为准。本文示例默认使用deepseek-chat你在实际项目中可以按任务复杂度切换。3. 深入理解 Harness 的核心设计3.1 不只是“大模型调用”Harness 解决的问题直接调用 API 只能完成单轮问答。但真实的业务场景通常是连续且复杂的比如用户提出目标Agent 拆解为多个步骤某些步骤需要调用外部工具工具结果需要回填给模型模型根据结果决定下一步最终输出满足用户要求。如果这些逻辑全部堆在业务代码里很快就会失控。Harness 的价值就是把这一套流程抽象为可复用结构。3.2 Harness 的四个核心部件一个可用的 Harness 至少需要包含模型客户端负责与大模型通信消息记忆维护 system、user、assistant 的对话历史工具注册中心保存可被模型调用的工具列表执行控制循环决定模型响应后是否继续调用工具以及何时停止。用一句话概括Harness 模型客户端 对话记忆 工具集合 执行循环。3.3 一个最小 Harness 的代码骨架下面实现一个最小版本方便你理解整体脉络。# 文件路径deepseek-harness-demo/harness_core.py from openai import OpenAI class Tool: def __init__(self, name, description, handler): self.name name self.description description self.handler handler def run(self, **kwargs): return self.handler(**kwargs) class Harness: def __init__(self, api_key, base_urlhttps://api.deepseek.com, modeldeepseek-chat): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model self.history [] self.tools {} def register_tool(self, tool: Tool): 注册一个工具到 Harness 中。 self.tools[tool.name] tool def add_message(self, role, content): self.history.append({role: role, content: content}) def run(self, user_input, max_steps3): 执行一次完整的 Agent 任务。 self.add_message(user, user_input) for step in range(max_steps): response self.client.chat.completions.create( modelself.model, messagesself.history, ) content response.choices[0].message.content self.add_message(assistant, content) # 简化协议如果模型输出包含 TOOL_CALL: 前缀则解析工具调用 if TOOL_CALL: in content: tool_name, args self._parse_tool_call(content) if tool_name not in self.tools: self.add_message(user, f工具 {tool_name} 不存在请换一个工具。) continue tool_result self.tools[tool_name].run(**args) self.add_message(user, f工具调用结果{tool_result}) else: return content return self.history[-1][content] def _parse_tool_call(self, content): 从模型输出解析工具名和参数。 body content.split(TOOL_CALL:, 1)[1].strip() first_line body.split(\n, 1)[0].strip() parts first_line.split(maxsplit1) tool_name parts[0] args {} if len(parts) 1: import json try: args json.loads(parts[1]) except json.JSONDecodeError: args {query: parts[1]} return tool_name, args这段代码虽然简单但已经能跑通“模型决定调用工具 → Harness 执行工具 → 结果回填 → 模型继续决策”的闭环。4. 插件开发实战让 Harness 拥有扩展能力4.1 为什么需要插件机制上面的 Harness 已经有register_tool方法但把所有工具函数都写在一个文件里并不好维护。插件机制的核心目的每个插件独立成文件或目录插件可以复用、共享、动态加载团队协作时不同插件由不同人维护主程序不需要改动只要加载插件目录。4.2 插件的注册与加载我采用一个简单的约定每个插件模块提供一个register(harness)函数该函数负责创建 Tool 实例并注册到 Harness。主程序遍历插件目录并加载。4.3 开发一个时间插件现在我们来写第一个插件它提供获取当前时间的能力。# 文件路径deepseek-harness-demo/plugins/time_plugin.py from datetime import datetime def get_current_time(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def register(harness): from harness_core import Tool harness.register_tool(Tool( nameget_current_time, description获取当前系统时间返回格式为年-月-日 时:分:秒, handlerget_current_time, ))注意register函数既没有创建模型客户端也没有关心业务逻辑它只把能力挂载到 Harness 上。4.4 开发一个目录扫描插件接着写一个“扫描项目目录”的插件。这个工具会默认限制在指定根目录避免越权读取系统任意路径。# 文件路径deepseek-harness-demo/plugins/file_plugin.py import os def list_project_files(root_dir.): 列出指定目录下的所有文件默认当前目录。 result [] abs_root os.path.abspath(root_dir) for foldername, _, filenames in os.walk(abs_root): # 跳过隐藏目录如 .git 和 .venv if /. in foldername or foldername.endswith(.venv): continue for filename in filenames: if filename.startswith(.): continue full_path os.path.join(foldername, filename) result.append(full_path) return \n.join(result[:50]) if result else 目录为空 def register(harness): from harness_core import Tool harness.register_tool(Tool( namelist_project_files, description列出项目目录下的文件列表用于了解项目结构。, handlerlist_project_files, ))如果后续要扩展开启文件读取能力也要保持同样的安全边界只允许读取白名单目录内的文件。4.5 插件热加载与命名规范在主程序中可以写一个自动加载插件目录的函数# 文件路径deepseek-harness-demo/load_plugins.py import importlib import pkgutil import plugins def load_all_plugins(harness): 自动加载 plugins 目录下所有插件。 for module_info in pkgutil.iter_modules(plugins.__path__): module_name module_info.name module importlib.import_module(fplugins.{module_name}) if hasattr(module, register): module.register(harness) print(f[插件已加载] {module_name})插件命名建议文件全小写使用下划线分隔如time_plugin.py每个插件模块必须有register(harness)工具名使用动词开头如get_current_time方便模型理解。5. 工作流实战编排一个“日报生成 Agent”5.1 工作流需求拆解现在我们把 Harness、插件和工作流结合起来实现一个真实场景自动生成项目日报。需求如下获取当前时间扫描项目目录得到当日可能改动的文件列表把时间和文件列表交给模型让模型生成一份结构化日报输出 Markdown 格式日报。这里的设计点在于工作流不是把每一步硬编码写在main函数里而是定义成节点每个节点有“执行”和“下一步判断”的能力。5.2 定义工作流节点# 文件路径deepseek-harness-demo/workflow_node.py from dataclasses import dataclass, field from typing import Callable, Any dataclass class WorkflowNode: name: str func: Callable[[Any], Any] next_node: str None def execute(self, context): result self.func(context) context[self.name] result return self.next_nodecontext是一个字典保存工作流运行过程中的共享数据。每个节点执行后可以把结果写入context并返回下一个节点的名称。如果返回None表示流程结束。5.3 实现工作流执行器# 文件路径deepseek-harness-demo/workflow_engine.py from workflow_node import WorkflowNode class WorkflowEngine: def __init__(self): self.nodes {} def add_node(self, node: WorkflowNode): self.nodes[node.name] node def run(self, start_node: str, context: dict): current start_node while current is not None: node self.nodes[current] current node.execute(context) return context工作流引擎本体的逻辑非常简单拿到当前节点执行它再根据返回值跳到下一个节点。5.4 完整运行示例与结果说明现在把上面所有模块串起来。# 文件路径deepseek-harness-demo/run_daily_report.py import os from dotenv import load_dotenv from workflow_engine import WorkflowEngine from workflow_node import WorkflowNode from harness_core import Harness load_dotenv() def build_context(): return { project_dir: ., current_time: , file_list: , report: } def get_time_node(context): from plugins.time_plugin import get_current_time context[current_time] get_current_time() return scan_files def scan_files_node(context): from plugins.file_plugin import list_project_files context[file_list] list_project_files(context[project_dir]) return generate_report def generate_report_node(context): api_key os.getenv(DEEPSEEK_API_KEY) harness Harness(api_keyapi_key) prompt f 请根据以下信息生成一份项目日报使用 Markdown 格式。 当前时间{context[current_time]} 项目文件列表 {context[file_list]} 日报结构要求 1. 今日工作概述 2. 主要改动文件 3. 遇到的问题 4. 明日计划 reply harness.run(prompt, max_steps1) context[report] reply return None def build_workflow(): engine WorkflowEngine() engine.add_node(WorkflowNode(start, get_time_node, scan_files)) engine.add_node(WorkflowNode(scan_files, scan_files_node, generate_report)) engine.add_node(WorkflowNode(generate_report, generate_report_node)) return engine if __name__ __main__: engine build_workflow() result engine.run(start, build_context()) print( 日报生成结果 ) print(result[report])运行python run_daily_report.py预期输出会是一个包含“今日工作概述”“主要改动文件”“遇到的问题”“明日计划”四部分的 Markdown 日报。这个例子虽然简单却把三件事讲清楚了工作流节点如何串联插件如何在节点中被复用Harness 如何在最后一步作为“内容生成器”完成创作型任务。6. 搭建一个带工具调用的专属 Agent6.1 Agent 与普通对话的区别普通对话里我们只是把用户消息发给模型返回文本。而在 Agent 中模型会输出“需要调用工具的意图”Harness 解析意图并执行工具再把结果返回给模型直到模型认为任务完成。关键点工具调用不是简单的if判断而是由模型根据用户意图自主选择模型不知道工具的内部实现它只看到工具名和描述工具返回结果必须被转成模型可读的文本才能继续推理。6.2 工具调用主循环我已在上文实现了一个文本协议版主循环。它的工作流程是用户输入任务模型先判断是否需要工具如果需要就输出TOOL_CALL: 工具名 参数JSONHarness 解析并执行工具工具结果以 user 消息回填模型看到结果后要么继续调用下一个工具要么输出最终答案。为了让模型稳定输出这种格式有一点非常重要系统提示词里必须给出明确的格式说明。6.3 完整 Agent 代码接下来写一个更完整的 Agent 文件它会自动加载插件并在系统提示词中说明工具调用协议。# 文件路径deepseek-harness-demo/agent_main.py import json import os from dotenv import load_dotenv from harness_core import Harness from load_plugins import load_all_plugins load_dotenv() SYSTEM_PROMPT 你是一个智能项目助手。你可以使用以下工具 {tool_descriptions} 如果需要调用工具请严格输出以下格式 TOOL_CALL: 工具名 参数JSON 参数JSON是可选的。如果不需要参数只输出 TOOL_CALL: 工具名 当得到工具结果后继续推理并最终给用户完整回答。不要编造工具执行结果。 def build_agent(): api_key os.getenv(DEEPSEEK_API_KEY) harness Harness(api_keyapi_key) load_all_plugins(harness) tool_lines [] for name, tool in harness.tools.items(): tool_lines.append(f- {name}: {tool.description}) system_prompt SYSTEM_PROMPT.format(tool_descriptions\n.join(tool_lines)) harness.add_message(system, system_prompt) return harness if __name__ __main__: agent build_agent() while True: user_input input(请输入任务输入 exit 退出) if user_input.strip().lower() exit: break reply agent.run(user_input, max_steps5) print( Agent 回复 ) print(reply) print()运行python agent_main.py使用时可以输入类似这样的任务请获取当前时间然后扫描项目目录最后告诉我项目里有哪些 Python 文件。Agent 会通过调用get_current_time和list_project_files完成任务并最终给出结构化回答。如果你的 DeepSeek API 版本支持 OpenAI 兼容的 function calling 格式也可以把工具列表转为tools参数传给接口。文本协议的好处是不依赖特定模型可移植性更强。7. 常见问题与排查思路开发 DeepSeek Harness 时最常遇到下面几类问题。问题现象常见原因解决思路401 认证失败API Key 错误或未设置环境变量检查.env中的 Key确认没有多余空格429 请求过多账户额度不足或并发超限查看开放平台用量提高间隔或加缓存模型返回空内容上下文过长或内容被安全策略过滤删减长文本分段处理工具调用一直循环模型没有正确理解工具结果优化工具描述增强系统提示词设置最大步数插件未加载插件文件缺少register函数检查模块命名和函数签名触发了本地文件越权读取工具没有限制根目录使用os.path.abspath并做路径白名单校验几个排查建议先手动调用 DeepSeek API确认 Key 和网络没有问题再单独测试插件函数确认返回结果正确最后通过 Harness 执行完整流程把每步 history 打印出来定位是哪一层丢失了上下文遇到工具解析失败优先检查模型输出的TOOL_CALL:格式是否规范必要时在系统提示词中加入“必须严格输出 JSON”。8. 最佳实践与工程建议8.1 把 API Key 当作最高优先级的安全边界无论项目多小都不要把 API Key 硬编码在前端或公开仓库。建议做法是使用.env文件生产环境使用密钥管理服务为不同环境创建不同 Key设置用量告警。8.2 Harness 与工作流职责分离Harness 负责“模型对话 工具调用”工作流负责“任务节点编排”。两者不要混在一起。如果发现一个节点里既要调用模型又要操作文件建议拆成两个节点。8.3 工具协议要稳定工具名、参数格式、返回结果格式要尽可能稳定。模型非常依赖工具描述描述不清晰时最容易出现工具选择错误。可以定期整理工具调用日志分析模型误用场景。8.4 日志与可观测性每个 Agent 任务都应记录用户输入每一步模型输出工具调用结果最终回复。最简单的方式是在 Harness 的run方法中加入日志打印或写入文件。生产环境建议接入结构化日志系统。8.5 异常处理与降级策略当模型连续 N 步无法完成任务时要主动终止。上文代码里的max_steps就是兜底参数。更复杂的场景可以加入“重试”“换模型”“直接返回已有结果”等降级策略。8.6 上下文的长度控制DeepSeek 模型虽然上下文窗口不小但无限增长历史会让成本越来越高、速度越来越慢。建议对早期对话做摘要压缩每次只保留最近 N 轮关键信息工具结果如果太长先截断或做摘要再回填。9. 下一步可以怎么玩到这里你已经亲手实现了“API 调用 → Harness 封装 → 插件开发 → 工作流编排 → Agent 组装”的完整链路。不要小看这套代码很多生产级 Agent 的核心循环本质上就是这么转的。下一步你可以试试给 Harness 增加“代码执行工具”但务必在沙箱环境中运行对接企业内部的接口比如查询工单、批量获取数据把工作流执行器扩展成支持条件分支和循环节点将 Harness 封装成 FastAPI 服务对外提供 Agent API。如果觉得手写 Harness 比较底层想快速搭建产品原型也可以参考 Dify、n8n、Coze 这类工作流平台。但理解了底层原理之后你会发现这些平台里的节点、组件、触发器本质上都是 Harness 概念的商业化封装。建议你把文章中的示例代码跑通一遍然后再尝试改造。Agent 开发最怕的就是只看不练手写一个最小闭环比什么都重要。