基于deepseek api和openweather 天气API实现Function Calling技术讲解
import requests
import json
import os
# 配置API密钥（从环境变量获取）
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY")
# Function Definitions (JSON Schema格式)
functions = [
{
"name": "get_current_weather",
"description": "获取指定城市的当前天气信息",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名称，如：'北京' 或 'London'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["location"]
}
},
{
"name": "ask_deepseek",
"description": "回答通用问题，涉及知识查询、建议、解释概念等",
"parameters": {
"type": "object",
"properties": {
"question": {
"type": "string",
"description": "用户提出的问题或请求"
}
},
"required": ["question"]
}
}
]
def call_function(function_name, arguments):
"""路由函数调用到具体实现"""
if function_name == "get_current_weather":
return get_current_weather(
location=arguments.get("location"),
unit=arguments.get("unit", "celsius")
)
elif function_name == "ask_deepseek":
return ask_deepseek(
question=arguments.get("question")
)
else:
return "未找到对应功能"
# OpenWeather API实现
def get_current_weather(location, unit="celsius"):
try:
url = "https://api.openweathermap.org/data/2.5/weather"
params = {
"q": location,
"appid": OPENWEATHER_API_KEY,
"units": "metric" if unit == "celsius" else "imperial"
}
response = requests.get(url, params=params)
data = response.json()
if response.status_code == 200:
weather_info = {
"location": data["name"],
"temperature": data["main"]["temp"],
"unit": "°C" if unit == "celsius" else "°F",
"description": data["weather"][0]["description"],
"humidity": f"{data['main']['humidity']}%",
"wind_speed": f"{data['wind']['speed']} m/s"
}
return json.dumps(weather_info)
else:
return f"获取天气信息失败：{data.get('message', '未知错误')}"
except Exception as e:
return f"天气API调用异常：{str(e)}"
# DeepSeek API实现
def ask_deepseek(question):
try:
url = "https://api.deepseek.com/v1/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {DEEPSEEK_API_KEY}"
}
payload = {
"model": "deepseek-chat",
"messages": [{
"role": "user",
"content": question
}],
"temperature": 0.7
}
response = requests.post(url, headers=headers, json=payload)
data = response.json()
if response.status_code == 200:
return data["choices"][0]["message"]["content"]
else:
return f"DeepSeek API错误：{data.get('error', {}).get('message', '未知错误')}"
except Exception as e:
return f"DeepSeek API调用异常：{str(e)}"
def process_query(user_query):
"""处理用户查询的主函数"""
# 这里应该接入LLM进行意图识别，以下是模拟实现
if "天气" in user_query or "气温" in user_query:
return call_function("get_current_weather", {
"location": user_query.replace("天气", "").strip(),
"unit": "celsius"
})
else:
return call_function("ask_deepseek", {
"question": user_query
})
# 使用示例
if __name__ == "__main__":
queries = [
"北京现在的天气怎么样？",
"请解释量子计算的基本原理",
"上海今天的温度",
"如何学习机器学习？"
]
for query in queries:
print(f"用户问：{query}")
response = process_query(query)
# 尝试解析JSON响应（适用于天气API）
try:
weather_data = json.loads(response)
print("天气信息：")
print(f"城市：{weather_data['location']}")
print(f"温度：{weather_data['temperature']}{weather_data['unit']}")
print(f"天气状况：{weather_data['description']}")
print(f"湿度：{weather_data['humidity']}")
print(f"风速：{weather_data['wind_speed']}\n")
except:
print(f"回答：{response}\n")