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

news2025/3/14 3:24:00

以下是一个结合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 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)