一、什么是 DeepSeek Harness 插件在 AI 应用中“Harness”通常指模型适配层 / 驱动层它把上层的业务代码、聊天平台或工作流框架与底层的 DeepSeek 模型能力对接起来。一个典型的模型插件结构如下上层业务 / 聊天平台 / 工作流框架Harness 插件适配层DeepSeek OpenAI 兼容接口deepseek-chat / deepseek-reasoner 模型本教程会带你从零实现一个插件包括基于 OpenAI 兼容协议的 DeepSeek 客户端支持普通对话、流式输出、自动重试支持 Function Calling工具调用封装成可复用的插件类接入 LangChain / LiteLLM 等主流生态。二、前置知识DeepSeek API 与 OpenAI 兼容能力DeepSeek 官方接口兼容 OpenAI SDK这是开发插件最省力的方式。其核心要点接口地址https://api.deepseek.com推荐模型deepseek-chat对话、deepseek-reasoner推理认证方式Authorization: Bearer API Key主要接口/chat/completions支持多轮对话、流式输出、工具调用、JSON 输出等因此我们不需要自己实现 HTTP 细节直接使用openai包即可。三、环境准备与工程初始化3.1 环境要求Python 3.10建议 3.11pip / uv / poetry 均可已注册 DeepSeek 平台并获取 API Key3.2 创建项目并安装依赖mkdirdeepseek-harnesscddeepseek-harness python-mvenv .venvsource.venv/bin/activate# Windows 使用 .venv\Scripts\activatepipinstallopenai python-dotenv pytest在项目根目录创建.env文件DEEPSEEK_API_KEYsk-你的密钥DEEPSEEK_BASE_URLhttps://api.deepseek.comDEEPSEEK_MODELdeepseek-chat3.3 项目目录结构deepseek-harness/ ├── pyproject.toml ├── .env ├── deepseek_harness/ │ ├── __init__.py │ ├── config.py │ ├── client.py │ ├── adapter.py │ ├── tools.py │ └── exceptions.py ├── examples/ │ └── chat_demo.py └── tests/ └── test_adapter.py新建deepseek_harness/__init__.py内容留空即可。四、实战一编写配置模块deepseek_harness/config.py用dataclass管理配置支持从环境变量读取importosfromdataclassesimportdataclass,fielddataclassclassDeepSeekConfig:DeepSeek 插件配置api_key:strfield(default_factorylambda:os.getenv(DEEPSEEK_API_KEY,))base_url:strfield(default_factorylambda:os.getenv(DEEPSEEK_BASE_URL,https://api.deepseek.com))model:strfield(default_factorylambda:os.getenv(DEEPSEEK_MODEL,deepseek-chat))temperature:float0.7max_tokens:int2048timeout:float60.0max_retries:int3五、实战二构建客户端deepseek_harness/client.py返回标准的 OpenAI 客户端实例fromopenaiimportOpenAIfrom.configimportDeepSeekConfigdefbuild_client(config:DeepSeekConfig)-OpenAI:根据配置创建 DeepSeek 客户端ifnotconfig.api_key:raiseValueError(DEEPSEEK_API_KEY 未配置)returnOpenAI(api_keyconfig.api_key,base_urlconfig.base_url,timeoutconfig.timeout,)六、实战三核心适配器——对话、流式与重试deepseek_harness/adapter.py是插件的核心部分。这里实现三个能力普通对话流式输出失败自动重试指数退避。importtimefromtypingimportIteratorfromopenaiimportOpenAIfrom.clientimportbuild_clientfrom.configimportDeepSeekConfigclassDeepSeekAdapter:DeepSeek 模型适配器作为上层业务与 DeepSeek API 之间的桥梁def__init__(self,config:DeepSeekConfig,client:OpenAI|NoneNone):self.configconfig self.clientclientorbuild_client(config)defchat(self,messages:list[dict],**overrides)-str:发送多轮对话返回完整回答last_error:Exception|NoneNoneforattemptinrange(1,self.config.max_retries1):try:respself.client.chat.completions.create(modeloverrides.get(model,self.config.model),messagesmessages,temperatureoverrides.get(temperature,self.config.temperature),max_tokensoverrides.get(max_tokens,self.config.max_tokens),)contentresp.choices[0].message.contentreturncontentorexceptExceptionasexc:# noqa: BLE001last_errorexc wait_seconds2**(attempt-1)print(f第{attempt}次调用失败:{exc}{wait_seconds}秒后重试)time.sleep(wait_seconds)raiseRuntimeError(fDeepSeek 调用失败:{last_error})defstream(self,messages:list[dict],**overrides)-Iterator[str]:流式输出逐段返回文本streamself.client.chat.completions.create(modeloverrides.get(model,self.config.model),messagesmessages,temperatureoverrides.get(temperature,self.config.temperature),max_tokensoverrides.get(max_tokens,self.config.max_tokens),streamTrue,)forchunkinstream:ifnotchunk.choices:continuedeltachunk.choices[0].delta.contentifdelta:yielddelta说明chat是同步阻塞调用适合脚本、后台任务、简单 APIstream是生成器适合 Web 长连接、SSE 推送重试逻辑用指数退避避免瞬时限流直接失败**overrides允许调用方临时覆盖模型、温度等参数。七、实战四支持 Function Calling工具调用在很多插件场景中模型需要调用外部工具例如查询天气、调用搜索接口。DeepSeek 支持标准的tools参数。deepseek_harness/tools.py定义工具说明和一个查询函数importjsonfromtypingimportAny WEATHER_TOOL{type:function,function:{name:get_weather,description:查询指定城市的实时天气,parameters:{type:object,properties:{city:{type:string,description:城市名称}},required:[city],},},}defget_weather(city:str)-str:模拟天气查询工具实际项目中可替换为真实 HTTP 请求result:dict[str,Any]{city:city,weather:晴,temperature:26℃}returnjson.dumps(result,ensure_asciiFalse)在adapter.py中增加工具调用方法defchat_with_tools(self,messages:list[dict],tools:list[dict])-list[dict]:发送带工具描述的对话返回模型请求的 tool_callsrespself.client.chat.completions.create(modelself.config.model,messagesmessages,toolstools,tool_choiceauto,temperatureself.config.temperature,)messageresp.choices[0].messagereturnlist(message.tool_callsor[])然后在examples/chat_demo.py中演示完整闭环importjsonfromdeepseek_harness.adapterimportDeepSeekAdapterfromdeepseek_harness.configimportDeepSeekConfigfromdeepseek_harness.toolsimportWEATHER_TOOL,get_weatherdefmain()-None:adapterDeepSeekAdapter(DeepSeekConfig())messages[{role:user,content:北京今天天气怎么样}]tool_callsadapter.chat_with_tools(messages,[WEATHER_TOOL])ifnottool_calls:print(模型没有触发工具调用)return# 把助手消息和工具结果都补回上下文messages.append({role:assistant,content:None,tool_calls:tool_calls,})forcallintool_calls:argsjson.loads(call.function.arguments)tool_resultget_weather(args[city])messages.append({role:tool,tool_call_id:call.id,content:tool_result,})# 第二轮让模型根据工具结果生成最终回答answeradapter.chat(messages)print(模型回答,answer)if__name____main__:main()运行python examples/chat_demo.py预期效果模型回答北京今天天气晴朗气温 26℃。八、实战五插件化封装与异常处理为了让 Harness 真正像一个“插件”我们可以定义一个插件协议把配置创建、适配器能力统一暴露出去。deepseek_harness/exceptions.pyclassDeepSeekError(Exception):DeepSeek 插件根异常classConfigError(DeepSeekError):配置异常classModelCallError(DeepSeekError):模型调用异常在deepseek_harness/__init__.py中提供统一入口from.adapterimportDeepSeekAdapterfrom.configimportDeepSeekConfigfrom.exceptionsimportConfigError,DeepSeekError,ModelCallError __all__[DeepSeekAdapter,DeepSeekConfig,DeepSeekError,ConfigError,ModelCallError,]这样其他模块只需要fromdeepseek_harnessimportDeepSeekAdapter,DeepSeekConfig adapterDeepSeekAdapter(DeepSeekConfig())answeradapter.chat([{role:user,content:你好}])九、接入主流生态LangChain 与 LiteLLM9.1 接入 LangChainDeepSeek 支持 OpenAI 兼容协议因此可以直接使用langchain-openaipipinstalllangchain-openaifromlangchain_openaiimportChatOpenAI llmChatOpenAI(modeldeepseek-chat,api_keysk-你的密钥,base_urlhttps://api.deepseek.com,)responsellm.invoke(用一句话介绍大语言模型)print(response.content)9.2 接入 LiteLLMLiteLLM 使用统一格式provider/modelpipinstalllitellmfromlitellmimportcompletion respcompletion(modeldeepseek/deepseek-chat,messages[{role:user,content:写一句鼓励开发者的话}],api_keysk-你的密钥,)print(resp.choices[0].message.content)这两种方式适合需要快速集成代理、监控、多模型路由的项目。十、测试插件编写最小可运行测试tests/test_adapter.py使用 pytest并用一个假客户端避免真实请求fromunittest.mockimportMagicMockfromopenai.types.chatimportChatCompletion,ChatCompletionMessagefromopenai.types.chat.chat_completionimportChoicefromdeepseek_harness.adapterimportDeepSeekAdapterfromdeepseek_harness.configimportDeepSeekConfigdeftest_chat_returns_content()-None:configDeepSeekConfig(api_keysk-test,max_retries1)mock_clientMagicMock()mock_client.chat.completions.create.return_valueChatCompletion(idchatcmpl-test,choices[Choice(finish_reasonstop,index0,messageChatCompletionMessage(content你好,roleassistant))],created1700000000,modeldeepseek-chat,objectchat.completion,)adapterDeepSeekAdapter(config,clientmock_client)resultadapter.chat([{role:user,content:在吗}])assertresult你好运行测试pytest-q十一、常见问题排查11.1 报错 401原因API Key 错误或未传。排查importosprint(os.getenv(DEEPSEEK_API_KEY))确保.env已加载或者直接显式传入DeepSeekConfig(api_keysk-...)11.2 返回内容为空多数情况下是max_tokens过小或temperature设置不当。建议先使用deepseek-chatmax_tokens保持 1024 以上。11.3 流式输出卡住检查客户端是否设置了过小的timeout长文本生成建议提高到 120 秒以上DeepSeekConfig(timeout120.0)11.4 工具调用参数解析失败DeepSeek 工具调用的参数是 JSON 字符串必须使用json.loads解析不能直接传字符串。十二、总结与拓展到这里你已经完成了一个可落地的 DeepSeek Harness 插件包括配置管理OpenAI 兼容客户端普通对话与流式输出自动重试工具调用插件化封装LangChain / LiteLLM 接入基础测试。下一步可以继续做增加异步版本AsyncOpenAI接入 FastAPI 提供 SSE 流式接口增加 token 计费与日志埋点支持多模型切换与灰度路由打包成 PyPI 包供团队复用。这套结构也是大多数企业 AI 项目接入大模型的通用范式。掌握它之后切换到其他兼容 OpenAI 协议的模型也能快速复用。