DeepSeek插件开发实战：从零构建自定义Tool与函数调用

1. 项目概述为什么我们要亲手打造一个DeepSeek插件最近在折腾DeepSeek的API发现官方提供的工具虽然强大但总有些特定场景下的需求没法直接满足。比如我想让它能一键调用我内部系统的数据查询接口或者整合一些只有我们团队在用的特殊工具链。这时候官方的通用工具就显得有点“隔靴搔痒”了。于是我花了几天时间深入研究了一下DeepSeek的插件或者说“工具调用”开发机制从最基础的“Hello World”开始一步步实现了一个能根据我自定义逻辑运行的Tool。整个过程下来感觉就像给一个超级大脑装上了专属的“瑞士军刀”让它不仅能思考还能直接操作我的“私人工具箱”。这个实战过程本质上是在与大型语言模型的“工具调用”能力打交道。DeepSeek作为模型本身并不直接执行代码或访问外部系统但它可以理解你的需求并“决定”在合适的时机调用你预先定义好的工具函数。我们开发者要做的就是按照它约定的格式把这些工具“描述”给它并准备好真正的执行后端。这比单纯调用API完成一次对话要复杂一些但带来的灵活性和自动化潜力是指数级增长的。无论是想连接数据库、触发自动化脚本、还是调用第三方服务只要你能用代码实现就能把它封装成一个Tool让DeepSeek模型来智能调度。2. 核心概念与准备工作理解Tool、Function Calling与插件生态在动手写代码之前我们必须先理清几个关键概念否则很容易在后续开发中迷失方向。这些概念是构建一切的基础。2.1 Tool、Function Calling与插件它们到底是什么关系这几个词经常混用但严格来说它们指代的是同一流程的不同层面。Function Calling函数调用 这是一种协议或机制。它规定了大型语言模型如DeepSeek如何向外部系统“表达”它想要执行某个操作的意图。通常模型会在回复中输出一个结构化的JSON对象其中包含了它想调用的函数名以及传入的参数。这不是真正的代码执行只是一个“请求”或“指令”。Tool工具 这是Function Calling机制中被调用的具体对象。一个Tool就是一个可执行单元它对应一个具体的功能。在DeepSeek的语境下我们通过一个JSON Schema来定义一个Tool包括它的名称、描述、参数列表等。模型只认识Tool的定义。插件Plugin 这是一个更上层的产品化概念。你可以把一个或多个相关的Tool打包配上图标、描述文档、认证方式等形成一个完整的、可供用户安装和使用的功能模块。我们本次实战聚焦在Tool的开发这是构建插件最核心的一步。简单类比Function Calling是“点菜”这个行为Tool是菜单上的一道道“菜”如鱼香肉丝而Plugin则是一个完整的“套餐”或“特色菜系”。2.2 开发环境与工具链选择工欲善其事必先利其器。为了高效开发我选择了以下组合这套组合在灵活性和开发体验上取得了很好的平衡。编程语言Python 3.9。这是AI领域事实上的标准语言生态库丰富与DeepSeek API的交互有成熟的SDK支持。核心SDK官方deepseek库。通过pip install deepseek安装。这是与DeepSeek模型服务通信的官方桥梁。辅助工具Pydantic。这是一个用于数据验证和设置管理的库。在定义Tool的复杂参数结构时使用Pydantic的BaseModel会让代码清晰、安全且易于维护。通过pip install pydantic安装。开发环境 任意你熟悉的IDE或编辑器即可比如VSCode或PyCharm。关键是要能方便地调试和查看日志。DeepSeek API密钥 你需要一个有效的DeepSeek API Key。前往DeepSeek平台注册并获取。请妥善保管不要直接硬编码在代码中。注意 API Key是访问服务的凭证务必通过环境变量或配置文件来管理。我习惯在项目根目录创建一个.env文件使用python-dotenv库加载绝对不要提交到代码仓库。# 示例 .env 文件内容 DEEPSEEK_API_KEYyour_api_key_here2.3 项目结构规划一个清晰的项目结构能让你后续的开发和维护事半功倍。这是我采用的目录结构deepseek-custom-tool-demo/ ├── .env # 存储环境变量API Key等 ├── .gitignore # Git忽略文件 ├── requirements.txt # 项目依赖列表 ├── src/ # 源代码目录 │ ├── __init__.py │ ├── tools/ # 存放所有自定义Tool的定义和实现 │ │ ├── __init__.py │ │ ├── calculator.py # 示例计算器工具 │ │ └── weather.py # 示例天气查询工具 │ ├── schemas/ # 存放Pydantic数据模型用于参数验证 │ │ ├── __init__.py │ │ └── weather.py # 天气查询的参数模型 │ └── main.py # 主程序入口组装和运行 └── README.md # 项目说明文档这个结构将工具定义、数据模型和主逻辑分离符合单一职责原则未来添加新工具会非常方便。3. 从零实现第一个ToolHello World与计算器让我们从一个最简单的例子开始确保整个链路是通的。这个阶段的目标不是功能多复杂而是验证“模型能理解我们的工具描述并能触发我们写的代码”。3.1 最简示例Echo Tool回声工具这个工具的功能是模型让它说什么它就原样返回什么。虽然简单但能完整走通流程。首先在src/tools/目录下创建echo.py# src/tools/echo.py import json from typing import Any, Dict def echo_tool(arguments: Dict[str, Any]) - str: 一个简单的回声工具返回传入的消息。 这是Tool的执行函数。 # 从模型传来的参数中获取消息 message arguments.get(message, ) # 这里可以加入任何你想执行的逻辑 result fEcho: {message} print(f[Tool Log] Echo工具被调用参数: {arguments}, 结果: {result}) return result # Tool的定义JSON Schema # 这个定义是给DeepSeek模型“看”的告诉它这个工具叫什么、能干嘛、需要什么参数。 ECHO_TOOL_SCHEMA { type: function, function: { name: echo, # 工具名称模型调用时使用 description: 一个简单的回声工具用于测试和验证工具调用流程。输入什么就返回什么。, parameters: { type: object, properties: { message: { type: string, description: 需要被回声的消息内容, } }, required: [message], # 指定哪些参数是必须的 additionalProperties: False, # 禁止传入未定义的参数更安全 }, }, }关键点解析执行函数 (echo_tool) 这是实际被执行的Python函数。它接收一个字典arguments里面包含了模型根据对话内容“推断”并填充好的参数。函数最后返回一个字符串结果。工具定义 (ECHO_TOOL_SCHEMA) 这是一个符合OpenAI Function Calling格式的字典。name是唯一标识description至关重要模型靠它来理解何时该调用此工具。parameters定义了输入参数的JSON Schemarequired数组声明了必填参数。接下来在src/main.py中编写主逻辑# src/main.py import os from dotenv import load_dotenv from deepseek import DeepSeek # 导入我们定义的工具 from src.tools.echo import echo_tool, ECHO_TOOL_SCHEMA # 加载环境变量 load_dotenv() def main(): # 1. 初始化DeepSeek客户端 client DeepSeek(api_keyos.getenv(DEEPSEEK_API_KEY)) # 2. 准备对话历史和工具列表 messages [{role: user, content: 请让回声工具说一句‘你好世界’}] tools [ECHO_TOOL_SCHEMA] # 将工具定义提供给模型 # 3. 发起第一次对话请求告诉模型有哪些工具可用 response client.chat.completions.create( modeldeepseek-chat, # 指定模型 messagesmessages, toolstools, tool_choiceauto, # 让模型自行决定是否调用工具 ) # 4. 处理模型响应 message response.choices[0].message print(f模型原始回复: {message}) # 5. 检查模型是否决定调用工具 if message.tool_calls: print(模型决定调用工具) for tool_call in message.tool_calls: # tool_call是一个对象包含工具名和参数 tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) # 参数是JSON字符串需要解析 # 6. 根据工具名找到对应的本地执行函数并调用 if tool_name echo: tool_result echo_tool(tool_args) print(f工具执行结果: {tool_result}) # 7. 将工具执行结果作为新的消息追加到对话历史中让模型知晓 messages.append(message) # 先追加模型的消息包含工具调用请求 messages.append({ role: tool, content: tool_result, tool_call_id: tool_call.id, # 必须对应告诉模型这是哪个调用的结果 }) # 8. 再次请求模型让它基于工具结果生成最终回复 second_response client.chat.completions.create( modeldeepseek-chat, messagesmessages, ) final_reply second_response.choices[0].message.content print(f\n模型的最终回复: {final_reply}) else: print(f未知工具调用: {tool_name}) else: # 模型没有调用工具直接输出内容 print(f模型直接回复: {message.content}) if __name__ __main__: main()运行这个程序如果你看到类似以下的输出那么恭喜你第一个Tool已经成功跑通了模型原始回复: ChatCompletionMessage(contentNone, roleassistant, function_callNone, tool_calls[ChatCompletionMessageToolCall(idcall_abc123, functionFunction(arguments{message:你好世界}, nameecho), typefunction)]) 模型决定调用工具 [Tool Log] Echo工具被调用参数: {message: 你好世界}, 结果: Echo: 你好世界 工具执行结果: Echo: 你好世界 模型的最终回复: 工具已经执行并返回了结果“Echo: 你好世界”。如你所见它成功地将“你好世界”这句话原样返回了。这个流程是标准的多轮交互用户请求 - 模型决定调用工具并返回调用指令 - 本地执行工具 - 将结果返回给模型 - 模型生成最终回答。3.2 进阶示例智能计算器工具现在我们来做一个更有用的工具一个能理解自然语言计算请求的智能计算器。这个例子展示了如何处理更复杂的参数和逻辑。首先用Pydantic定义参数模型这能让参数验证和代码提示更友好。在src/schemas/calculator.py中# src/schemas/calculator.py from pydantic import BaseModel, Field from typing import Literal class CalculatorInput(BaseModel): operation: Literal[add, subtract, multiply, divide] Field( ..., description运算类型加(add)、减(subtract)、乘(multiply)、除(divide) ) a: float Field(..., description第一个运算数) b: float Field(..., description第二个运算数) # 可以添加自定义验证例如除法时除数不能为0 # 这里为了演示我们在工具函数里处理然后在src/tools/calculator.py中实现工具# src/tools/calculator.py import json from typing import Any, Dict from src.schemas.calculator import CalculatorInput def calculator_tool(arguments: Dict[str, Any]) - str: 一个智能计算器工具执行基础算术运算。 try: # 使用Pydantic模型验证和解析参数 calc_input CalculatorInput(**arguments) a calc_input.a b calc_input.b result None if calc_input.operation add: result a b op_symbol elif calc_input.operation subtract: result a - b op_symbol - elif calc_input.operation multiply: result a * b op_symbol * elif calc_input.operation divide: if b 0: return 错误除数不能为零。 result a / b op_symbol / else: return f错误不支持的运算类型 {calc_input.operation}。 return f计算结果{a} {op_symbol} {b} {result} except Exception as e: # 捕获参数验证错误或其他异常 return f工具执行出错{str(e)} # 工具定义 # 注意这里的parameters可以从Pydantic模型自动生成但为清晰起见我们手动写一份。 # 在实际大型项目中可以使用pydantic的model_json_schema()方法自动生成。 CALCULATOR_TOOL_SCHEMA { type: function, function: { name: calculator, description: 执行基础算术运算加、减、乘、除。当用户需要进行数学计算时使用此工具。, parameters: { type: object, properties: { operation: { type: string, enum: [add, subtract, multiply, divide], description: 运算类型加(add)、减(subtract)、乘(multiply)、除(divide) }, a: { type: number, description: 第一个运算数 }, b: { type: number, description: 第二个运算数 } }, required: [operation, a, b], additionalProperties: False, }, }, }实操心得description字段是工具能否被正确调用的灵魂。模型完全依赖这个描述来判断“什么时候该用这个工具”。所以描述要尽可能精确、无歧义并包含典型的使用场景。例如“当用户需要进行数学计算时使用此工具”就比“这是一个计算器”要好得多。更新main.py引入计算器工具并进行测试# 在main.py中更新导入和工具列表 from src.tools.calculator import calculator_tool, CALCULATOR_TOOL_SCHEMA # ... 其他导入 ... def main(): client DeepSeek(api_keyos.getenv(DEEPSEEK_API_KEY)) messages [{role: user, content: 请帮我计算一下3.14乘以256等于多少}] # 可以同时提供多个工具给模型选择 tools [ECHO_TOOL_SCHEMA, CALCULATOR_TOOL_SCHEMA] response client.chat.completions.create( modeldeepseek-chat, messagesmessages, toolstools, tool_choiceauto, ) # ... 后续处理逻辑与echo示例类似需要根据tool_name调用对应的calculator_tool ...运行后模型应该能正确理解“3.14乘以256”这个自然语言请求将其转化为operation: multiply, a: 3.14, b: 256的参数并调用我们的calculator_tool得到正确结果。4. 构建复杂且实用的自定义Tool天气查询代理现在我们来挑战一个更接近真实场景的例子一个天气查询工具。这个工具需要调用外部API处理网络请求和JSON数据解析并且参数结构也更复杂。4.1 设计数据模型与工具定义假设我们调用一个虚拟的天气API它需要城市名和查询单位公制/英制。首先在src/schemas/weather.py中定义参数模型# src/schemas/weather.py from pydantic import BaseModel, Field from typing import Literal, Optional class WeatherQueryInput(BaseModel): city: str Field(..., description需要查询天气的城市名称例如北京、Shanghai、New York) units: Optional[Literal[metric, imperial]] Field( defaultmetric, description温度单位。metric表示摄氏度(°C)imperial表示华氏度(°F)。默认为metric。 ) # 可以扩展更多参数如语言、预报天数等 # forecast_days: Optional[int] Field(default1, ge1, le7, description预报天数1-7天)接着在src/tools/weather.py中实现工具。这里我们模拟一个API调用# src/tools/weather.py import json import random import time from typing import Any, Dict from src.schemas.weather import WeatherQueryInput def mock_weather_api(city: str, units: str metric) - Dict[str, Any]: 模拟一个天气API的响应。 在实际项目中这里应该替换为真实的HTTP请求例如使用requests库。 # 模拟网络延迟 time.sleep(0.5) # 生成一些模拟数据 temp random.uniform(15, 35) if units metric else random.uniform(59, 95) humidity random.randint(30, 90) conditions [晴朗, 多云, 局部多云, 小雨, 雷阵雨] condition random.choice(conditions) return { city: city, temperature: round(temp, 1), units: °C if units metric else °F, humidity: f{humidity}%, condition: condition, timestamp: time.strftime(%Y-%m-%d %H:%M:%S), } def weather_tool(arguments: Dict[str, Any]) - str: 查询指定城市的当前天气情况。 try: # 1. 参数验证与解析 query WeatherQueryInput(**arguments) city query.city units query.units or metric # 使用默认值 # 2. 调用模拟外部API print(f[Tool Log] 正在查询{city}的天气单位: {units}...) weather_data mock_weather_api(city, units) # 3. 格式化结果使其对模型和用户都友好 result_str ( f{weather_data[city]}的当前天气\n f- 天气状况{weather_data[condition]}\n f- 温度{weather_data[temperature]}{weather_data[units]}\n f- 湿度{weather_data[humidity]}\n f- 数据更新时间{weather_data[timestamp]} ) return result_str except Exception as e: # 记录详细错误日志但返回给模型的错误信息要简洁 print(f[Tool Error] 天气查询失败: {e}) return f抱歉查询{city}的天气时出现错误。请检查城市名称是否正确或稍后再试。 # 工具定义 WEATHER_TOOL_SCHEMA { type: function, function: { name: get_current_weather, description: 获取指定城市的当前天气信息包括温度、湿度和天气状况。当用户询问天气、气候或温度时使用此工具。, parameters: { type: object, properties: { city: { type: string, description: 城市名称必须清晰明确。例如‘北京’、‘纽约’、‘London’。 }, units: { type: string, enum: [metric, imperial], description: 温度单位。metric为摄氏度imperial为华氏度。如果不指定默认使用metric。 } }, required: [city], # units 是可选的 additionalProperties: False, }, }, }4.2 在主程序中集成与测试更新main.py集成天气工具并尝试更复杂的对话# 更新main.py的导入和主逻辑 from src.tools.weather import weather_tool, WEATHER_TOOL_SCHEMA def run_conversation(): client DeepSeek(api_keyos.getenv(DEEPSEEK_API_KEY)) # 初始对话可以混合多个工具 tools [CALCULATOR_TOOL_SCHEMA, WEATHER_TOOL_SCHEMA] messages [{role: user, content: 今天杭州天气怎么样另外帮我算算去那里出差三天如果每天餐费预算150元总共需要多少}] # 第一轮模型可能会先调用天气工具 response client.chat.completions.create( modeldeepseek-chat, messagesmessages, toolstools, tool_choiceauto, ) message response.choices[0].message messages.append(message) # 将模型的回复含工具调用加入历史 # 处理可能的多工具调用循环 while message.tool_calls: for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) if tool_name get_current_weather: tool_result weather_tool(tool_args) elif tool_name calculator: tool_result calculator_tool(tool_args) else: tool_result f错误未知工具 {tool_name}。 # 将工具执行结果追加 messages.append({ role: tool, content: tool_result, tool_call_id: tool_call.id, }) # 再次请求模型让它基于所有工具结果继续回复或调用新工具 response client.chat.completions.create( modeldeepseek-chat, messagesmessages, toolstools, # 工具列表在后续轮次中通常也需要传递 ) message response.choices[0].message if message.content or message.tool_calls: messages.append(message) # 打印最终结果 print(\n 对话完成 ) for msg in messages: if msg[role] assistant and msg.get(content): print(f助理: {msg[content]}) elif msg[role] tool: print(f[工具结果]: {msg[content]}) if __name__ __main__: run_conversation()这个例子展示了几个高级特性多工具协同 模型可以理解一个复杂问题中包含的多个子任务查天气、做计算并依次或并行调用不同的工具。循环处理 通过while循环可以处理模型可能发起的连续多次工具调用。错误处理与友好反馈 在工具函数内部进行了异常捕获并返回了对用户友好的错误信息而不是崩溃或抛出技术栈追踪。5. 工程化与最佳实践让自定义Tool更健壮当工具数量增多、逻辑变复杂后代码的组织和健壮性就变得至关重要。以下是我在实践中总结的几个关键点。5.1 工具的动态注册与发现机制手动维护一个tools列表在工具少的时候还行多了就非常麻烦。我们可以建立一个注册机制。在src/tools/__init__.py中# src/tools/__init__.py 工具注册中心。 所有工具在此注册便于主程序统一加载。 import inspect from typing import Dict, Callable, Any # 全局注册表 _tool_registry: Dict[str, Dict[str, Any]] { # 格式: tool_name: {function: callable, schema: dict} } def register_tool(schema: dict): 装饰器用于注册工具。 用法 register_tool(MY_TOOL_SCHEMA) def my_tool_function(arguments): ... def decorator(func: Callable): tool_name schema[function][name] _tool_registry[tool_name] { function: func, schema: schema } return func return decorator def get_all_tool_schemas(): 获取所有已注册工具的定义 return [info[schema] for info in _tool_registry.values()] def execute_tool(tool_name: str, arguments: dict) - str: 根据工具名执行对应的工具函数 if tool_name not in _tool_registry: raise ValueError(f工具 {tool_name} 未注册。) tool_info _tool_registry[tool_name] return tool_info[function](arguments)然后修改我们的工具文件使用装饰器注册# src/tools/weather.py (更新版) from src.tools import register_tool # ... 其他导入和WeatherQueryInput ... register_tool(WEATHER_TOOL_SCHEMA) # 使用装饰器注册 def weather_tool(arguments: Dict[str, Any]) - str: # ... 函数实现不变 ...最后主程序可以简化为# main.py (更新版) from src.tools import get_all_tool_schemas, execute_tool # 只需导入工具模块装饰器会自动注册 import src.tools.echo import src.tools.calculator import src.tools.weather def run_conversation(): client DeepSeek(...) # 动态获取所有已注册的工具定义 tools get_all_tool_schemas() messages [...] # ... 后续循环中调用 execute_tool(tool_name, tool_args) 即可 ...这种方式极大地提高了可维护性新增工具只需创建文件并用装饰器注册主程序无需修改。5.2 完善的错误处理与日志记录工具执行在外部什么错误都可能发生网络超时、API限流、参数无效、数据解析失败等。必须有统一的错误处理。工具函数内部的健壮性 如前所述使用try...except包裹核心逻辑返回有意义的错误信息。全局执行包装器 可以创建一个包装函数统一处理异常和日志。# src/tools/executor.py import traceback from typing import Callable, Any def safe_execute_tool(tool_func: Callable, arguments: dict, tool_name: str) - str: 安全执行工具函数提供统一的错误处理和日志。 try: print(f[INFO] 开始执行工具: {tool_name}, 参数: {arguments}) result tool_func(arguments) print(f[INFO] 工具执行成功: {tool_name}, 结果长度: {len(str(result))}) return result except Exception as e: error_detail traceback.format_exc() print(f[ERROR] 工具执行失败: {tool_name}, 错误: {e}\n{error_detail}) # 返回一个对模型友好的错误信息避免暴露内部细节 return f工具 {tool_name} 执行过程中发生意外错误请稍后重试或联系管理员。然后在主循环中调用safe_execute_tool。5.3 工具描述的优化技巧模型的工具调用准确性极大依赖于description和parameters的描述质量。描述要具体且有场景 不要写“查询天气”要写“获取指定城市的当前天气信息包括温度、湿度和天气状况。当用户询问天气、气候或温度时使用此工具。”参数描述要清晰 对于city参数描述“城市名称”是不够的最好加上“例如‘北京’、‘纽约’、‘London’”。对于枚举类型明确列出所有选项及其含义。使用required字段 明确哪些参数是必须的这能帮助模型更准确地询问用户缺失的信息如果对话允许。测试与迭代 写出定义后用各种自然语言问法去测试模型是否会调用、参数填充是否准确。根据测试结果反复调整描述。6. 调试技巧与常见问题排查开发过程中你肯定会遇到模型不调用工具、调用错误工具、参数填充不对等问题。以下是我的排查清单。6.1 模型不调用工具检查工具描述 这是最常见的原因。描述是否足够清晰是否包含了触发关键词尝试让描述更贴近用户可能使用的自然语言。检查tool_choice参数 你设置的是auto吗如果设为none模型将不会调用任何工具。如果想强制调用某个工具可以设为{type: function, function: {name: your_tool_name}}。检查对话上下文 模型是基于整个对话历史做决策的。如果之前的对话中已经包含了答案或者上下文让模型认为不需要工具它可能就不会调用。尝试开启一个新的对话线程测试。模型能力 确认你使用的模型版本支持函数调用Tool Calling。DeepSeek的主流聊天模型通常都支持。6.2 模型调用了错误的工具或参数填充错误工具描述区分度不够 如果你有多个计算相关工具如calculator和currency_converter它们的描述需要有明显区分。强调各自独特的应用场景。参数描述模糊 比如一个date参数描述为“日期”可能让模型困惑。应该描述为“具体的日期格式为YYYY-MM-DD例如2023-10-27”。查看模型的思考过程如果支持 有些API或平台会提供模型的“推理过程”或“中间步骤”查看这些日志能帮你理解模型为什么做出了错误的选择。6.3 工具执行成功但模型回复不佳工具返回结果格式 模型需要基于工具返回的文本来生成回复。返回的结果应该信息完整、格式清晰、易于理解。避免返回纯JSON或过于技术化的错误码。在结果中提供上下文 例如天气工具返回“温度22”不如返回“北京当前温度22°C体感舒适”。多给模型一些可以组织语言的素材。6.4 网络与超时问题工具执行时间过长 如果工具函数执行HTTP请求且耗时很长可能会导致整个API调用超时。考虑对工具函数设置超时限制或使用异步调用。异步处理模式 对于耗时任务可以考虑“异步工具调用”模式。即模型发起调用后你立即返回一个“任务已接收”的中间结果然后在后台处理处理完成后通过其他方式如回调、数据库更新通知用户。但这需要更复杂的架构支持。开发自定义Tool是一个与模型“协作”的过程需要你在工具设计的严谨性和模型理解的灵活性之间找到平衡点。从最简单的“Hello World”开始逐步增加复杂度并持续测试和优化工具描述是最高效的路径。当你亲手打造的工具被模型准确调用并解决实际问题时那种成就感是非常独特的。这不仅仅是调用一个API而是在塑造一个AI智能体的行为能力。