当前位置：首页 > news >正文

基于deepseek api和openweather 天气API实现Function Calling技术讲解

news 2025/8/30 22:40:05

以下是一个结合DeepSeek API和OpenWeather API的完整Function Calling示例，包含意图识别、API调用和结果整合：

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")

关键要素说明：

功能定义：

get_current_weather：使用OpenWeather API获取实时天气数据
ask_deepseek：调用DeepSeek API回答通用问题

处理流程：

用户输入 -> 意图识别 -> 路由到对应API -> 格式化响应

增强点建议：

# 可以添加的改进功能：
# 1. 更智能的意图识别（使用LLM判断）
def detect_intent(query):"""使用LLM进行意图识别"""prompt = f"""判断用户意图并返回JSON：{{"function": "get_current_weather" | "ask_deepseek","parameters": {{...}}}}示例：输入：北京天气怎么样？输出：{{"function": "get_current_weather", "parameters": {{"location": "北京"}}}}当前输入：{query}"""# 调用DeepSeek API进行意图分析response = ask_deepseek(prompt)return json.loads(response)# 2. 添加单位自动转换
def convert_temperature(temp, from_unit, to_unit):if from_unit == to_unit:return tempif from_unit == "celsius" and to_unit == "fahrenheit":return (temp * 9/5) + 32else:return (temp - 32) * 5/9# 3. 添加缓存机制
from functools import lru_cache@lru_cache(maxsize=100)
def cached_weather(location, unit):return get_current_weather(location, unit)