基于deepseekapi和openweather天气API实现FunctionCalling技术讲解

以下是一个结合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 = " 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 = " api.deepseek /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 temp if from_unit == "celsius" and to_unit == "fahrenheit": return (temp * 9/5) + 32 else: 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)

部署注意事项：

将API密钥存储在环境变量中添加错误处理和重试机制添加API调用速率限制对用户输入进行消毒处理添加日志记录系统

这个实现展示了：

REST API调用JSON数据处理基本的函数路由错误处理机制可扩展的架构设计

可以根据具体需求添加更多功能，例如：

多城市天气对比天气预测集成多步推理（例如结合天气数据和旅行建议）对话历史管理