从Deepseek接口到自创建简单Agent
摘要:本文从大学生视角出发,详细介绍了如何将 OpenAI 的 Function calling 工具调用机制与 DeepSeek 大模型结合使用。首先讲解了 DeepSeek API 的基本调用流程,包括客户端初始化、网络连接建立、请求发送和响应接收。然后重点阐述了 Function calling 的三个核心步骤:1) 定义本地工具函数;2) 按照 OpenAI 规范定义工具 Schema;3) 实现工具调用执行函数。文章通过完整的 Python 代码示例,展示了如何实现一个支持星座运势查询、天气查询和数学计算的简单工具调用 Agent,并详细解释了三种 tool_choice 模式(auto、none、强制指定)的使用方法。
学习到工具(Function calling)时候,突然想到deepseek好像介绍过自己的大模型如何接,就想着把open ai的工具融到这里面,来进行一些小娱乐
# Please install OpenAI SDK first: `pip3 install openai`
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ.get('DEEPSEEK_API_KEY'),
base_url="https://api.deepseek.com")
response = client.chat.completions.create(
model="deepseek-v4-pro",
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello"},
],
stream=False,
reasoning_effort="high",
extra_body={"thinking": {"type": "enabled"}}
)
print(response.choices[0].message.content)
这是deepseek官方的简单调用,具体解释一下
先实例化客户端,仅初始化配置,不建立连接,属于懒连接机制
client = OpenAI(
api_key=os.environ.get('DEEPSEEK_API_KEY'),
base_url="https://api.deepseek.com")
后面重头戏,完整链接链路,也是等到*.create()才真正建立连接
阶段 1:建立网络连接
- SDK 解析
base_url=https://api.deepseek.com,DNS 解析域名 - 发起 HTTPS TCP 握手,建立加密连接
- 带上请求头(Authorization 鉴权、Content-Type 等)
阶段 2:向 DeepSeek 服务端发送完整请求体
把 model、messages、reasoning_effort、extra_body 等参数打包成 JSON POST 请求发给服务端; 服务端加载 deepseek-v4-pro 模型,执行推理计算。
阶段 3:接收服务端返回的完整响应
等模型推理完成,服务端把结构化 JSON 返回给本地 SDK,SDK 解析成 ChatCompletion 对象存入 response 变量。
response = client.chat.completions.create(
model="deepseek-v4-pro",
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello"},
],
stream=False,
reasoning_effort="high",
extra_body={"thinking": {"type": "enabled"}}
)
最后就是简单的打印出来
print(response.choices[0].message.content)
介绍完deepseek后,也是到了今天的重头戏,工具注册以及调用(Function calling)了
Function calling
基于Open ai 的SDK来进行讲解
定义工具
首先是定义函数
def get_horoscope(sign: str) -> str:
"""星座运势工具"""
return f"{sign}: Next Tuesday you will befriend a baby otter."
def get_weather(city: str, date: str = "today") -> str:
"""查询城市天气"""
return f"{city} {date}:晴,24~30℃,微风"
def calculate_math(expr: str) -> str:
"""数学计算,传入数学表达式"""
try:
res = eval(expr)
return f"计算结果 {expr} = {res}"
except Exception as e:
return f"计算失败:{str(e)}"
这里定义了三个本地工具,分别是固定回答以及运算
注意的是,到这并未能直接使用,因为你传给大模型,它是有固定json格式的
tools = [
# 工具1:星座运势
{
"type": "function",
"function": {
"name": "get_horoscope",
"description": "Get today's horoscope for an astrological sign.",
"parameters": {
"type": "object",
"properties": {
"sign": {
"type": "string",
"description": "An astrological sign like Taurus or Aquarius",
},
},
"required": ["sign"],
},
}
}
}
这里我只定义一个,如果需要多个的话,就直接在后面加就完事了
注意,这里面五个属性及其重要
| Field 领域 / 范围 | 描述 |
|---|---|
| type | 固定值为 function |
| name | 函数名称,示例:get_weather |
| description | 说明该函数的适用场景、使用方式 |
| parameters | 定义函数入参的 JSON Schema |
| strict |
是否开启函数调用严格校验模式 |
一般strict默认不严格
而且可以看到刚开始定义的入参是要在这个json格式中声明并且确定好的
定义完后,肯定是要调用的,那么工具调用函数怎么完成呢,下一步
工具调用
从具体函数来讲解
def run_all_tools(tool_calls):
"""批量执行模型返回的tool_call,返回要追加到messages的tool消息列表"""
# 空列表,存放所有工具执行完成后的标准消息
tool_messages = []
# 循环遍历每一个模型要求调用的工具(支持一次调用多个工具)
for tool_call in tool_calls:
# 1. 取出要执行的函数名称
func_name = tool_call.function.name
# 2. 解析参数:模型返回的arguments是JSON字符串,转成Python字典
func_args = json.loads(tool_call.function.arguments)
# 3. 函数映射字典:key=函数名,value=本地定义的函数
# 示例 tool_func_map = {"get_horoscope": get_horoscope, "get_weather": get_weather}
target_func = tool_func_map[func_name]
# 4. 解包字典,调用本地函数,拿到执行结果
func_result = target_func(**func_args)
# 5. 组装协议规定的tool消息
tool_msg = {
"role": "tool",
"tool_call_id": tool_call.id,
"name": func_name,
"content": func_result
}
tool_messages.append(tool_msg)
# 返回全部工具消息,直接 messages += tool_messages 追加到对话上下文
return tool_messages
直接看注释来进行理解
对于tool_calls,它的格式进行讲解
tool_call = {
"id": "随机唯一标识",
"function": {
"name": "要调用的函数名",
"arguments": '{"参数1":"值1","参数2":123}' # JSON字符串,不是字典
},
"type": "function"
}
这就是它的格式,也是要遵循的协议,按照这个你再去理解上面代码的一些参数就能理解了
最后给出运行函数
运行函数
if __name__ == "__main__":
# 对话上下文
messages = [
{"role": "system", "content": "优先使用工具完成用户问题,无法用工具再自行回答"},
{"role": "user", "content": "帮我查武汉今天天气,顺便算一下128*56,我是水瓶座看看运势"}
]
# ===================== 三种tool_choice模式切换 =====================
# 模式1:auto 自动选择是否调用、调用哪个工具(默认推荐)
tool_choice_mode = "auto"
# 模式2:none 禁止调用任何工具,模型直接回答
# tool_choice_mode = "none"
# 模式3:强制指定必须调用某一个工具
# tool_choice_mode = {
# "type": "function",
# "function": {"name": "get_weather"}
# }
# =================================================================
# 第一轮模型请求
response = client.chat.completions.create(
model=MODEL_NAME,
tools=tools,
tool_choice=tool_choice_mode,
messages=messages,
temperature=0.2
)
assistant_msg = response.choices[0].message
messages.append(assistant_msg.model_dump())
# 判断是否存在工具调用
if assistant_msg.tool_calls:
# 批量执行所有工具,获取工具返回消息
tool_msg_list = run_all_tools(assistant_msg.tool_calls)
messages.extend(tool_msg_list)
print("==== 完整对话上下文(含工具调用记录) ====")
print(json.dumps(messages, indent=2, ensure_ascii=False))
# 二次提交工具结果,生成最终自然语言回答
final_res = client.chat.completions.create(
model=MODEL_NAME,
messages=messages,
temperature=0.2
)
print("\n==== 模型最终整合回答 ====")
print(final_res.choices[0].message.content)
else:
print("模型未触发工具调用,直接输出:")
print(assistant_msg.content)
到此基本讲完整个流程,最后将完整代码呈上,简单工具调用agent
from openai import OpenAI
import json
# ===================== 【这里必须修改】 =====================
DEEPSEEK_API_KEY = "自己key"
MODEL_NAME = "deepseek-chat"
# ==========================================================
# 初始化DeepSeek客户端(兼容OpenAI SDK)
client = OpenAI(
api_key=DEEPSEEK_API_KEY,
base_url="https://api.deepseek.com"
)
# ========== 1. 定义多个本地工具函数 ==========
def get_horoscope(sign: str) -> str:
"""星座运势工具"""
return f"{sign}: Next Tuesday you will befriend a baby otter."
def get_weather(city: str, date: str = "today") -> str:
"""查询城市天气"""
return f"{city} {date}:晴,24~30℃,微风"
def calculate_math(expr: str) -> str:
"""数学计算,传入数学表达式"""
try:
res = eval(expr)
return f"计算结果 {expr} = {res}"
except Exception as e:
return f"计算失败:{str(e)}"
# 工具映射表:函数名 -> 实际函数(核心通用分发)
tool_func_map = {
"get_horoscope": get_horoscope,
"get_weather": get_weather,
"calculate_math": calculate_math
}
# ========== 2. 多工具Schema定义,全部注册给模型 ==========
tools = [
# 工具1:星座运势
{
"type": "function",
"function": {
"name": "get_horoscope",
"description": "Get today's horoscope for an astrological sign.",
"parameters": {
"type": "object",
"properties": {
"sign": {
"type": "string",
"description": "An astrological sign like Taurus or Aquarius",
},
},
"required": ["sign"],
},
}
},
# 工具2:天气查询
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Query weather of target city, support custom date",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City name, e.g. Beijing, Shanghai"
},
"date": {
"type": "string",
"description": "Query date, default today",
}
},
"required": ["city"]
}
}
},
# 工具3:数学计算器
{
"type": "function",
"function": {
"name": "calculate_math",
"description": "Calculate math expression like 1+2*3, sqrt(16)",
"parameters": {
"type": "object",
"properties": {
"expr": {
"type": "string",
"description": "Mathematical expression string"
}
},
"required": ["expr"]
}
}
}
]
# ========== 3. 通用工具调用执行函数 ==========
def run_all_tools(tool_calls):
"""批量执行模型返回的tool_call,返回要追加到messages的tool消息列表"""
tool_messages = []
for tool_call in tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
# 根据函数名自动匹配本地函数
target_func = tool_func_map[func_name]
func_result = target_func(**func_args)
# 组装tool角色消息
tool_msg = {
"role": "tool",
"tool_call_id": tool_call.id,
"name": func_name,
"content": func_result
}
tool_messages.append(tool_msg)
return tool_messages
if __name__ == "__main__":
# 对话上下文
messages = [
{"role": "system", "content": "优先使用工具完成用户问题,无法用工具再自行回答"},
{"role": "user", "content": "帮我查武汉今天天气,顺便算一下128*56,我是水瓶座看看运势"}
]
# ===================== 三种tool_choice模式切换 =====================
# 模式1:auto 自动选择是否调用、调用哪个工具(默认推荐)
tool_choice_mode = "auto"
# 模式2:none 禁止调用任何工具,模型直接回答
# tool_choice_mode = "none"
# 模式3:强制指定必须调用某一个工具
# tool_choice_mode = {
# "type": "function",
# "function": {"name": "get_weather"}
# }
# =================================================================
# 第一轮模型请求
response = client.chat.completions.create(
model=MODEL_NAME,
tools=tools,
tool_choice=tool_choice_mode,
messages=messages,
temperature=0.2
)
assistant_msg = response.choices[0].message
messages.append(assistant_msg.model_dump())
# 判断是否存在工具调用
if assistant_msg.tool_calls:
# 批量执行所有工具,获取工具返回消息
tool_msg_list = run_all_tools(assistant_msg.tool_calls)
messages.extend(tool_msg_list)
print("==== 完整对话上下文(含工具调用记录) ====")
print(json.dumps(messages, indent=2, ensure_ascii=False))
# 二次提交工具结果,生成最终自然语言回答
final_res = client.chat.completions.create(
model=MODEL_NAME,
messages=messages,
temperature=0.2
)
print("\n==== 模型最终整合回答 ====")
print(final_res.choices[0].message.content)
else:
print("模型未触发工具调用,直接输出:")
print(assistant_msg.content)
希望对你们有所帮助,也是从一个大学生视角来进行一步一步cang
更多推荐
所有评论(0)