79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
|
from langchain.tools import BaseTool
|
|||
|
|
from pydantic import BaseModel, Field, PrivateAttr
|
|||
|
|
from typing import Optional, Type, ClassVar
|
|||
|
|
import requests
|
|||
|
|
import logging
|
|||
|
|
from th_agenter.core.config import get_settings
|
|||
|
|
|
|||
|
|
logger = logging.getLogger("weather_tool")
|
|||
|
|
|
|||
|
|
# 定义输入参数模型（替代原get_parameters()）
|
|||
|
|
class WeatherInput(BaseModel):
|
|||
|
|
location: str = Field(
|
|||
|
|
description="城市名称，例如：'北京'，只能是单个城市",
|
|||
|
|
examples=["北京", "上海", "New York"]
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
class WeatherQueryTool(BaseTool):
|
|||
|
|
"""心知天气API查询工具（LangChain标准版）"""
|
|||
|
|
name: ClassVar[str] = "weather_query_tool"
|
|||
|
|
description: ClassVar[str] = """通过心知天气API查询实时天气数据。获取指定城市的当前天气信息，包括温度、湿度、天气状况等。"""
|
|||
|
|
args_schema: Type[BaseModel] = WeatherInput # 参数规范
|
|||
|
|
# 使用PrivateAttr声明不参与验证的私有属性
|
|||
|
|
_api_key: str = PrivateAttr()
|
|||
|
|
_base_params: dict = PrivateAttr()
|
|||
|
|
|
|||
|
|
def __init__(self, **kwargs):
|
|||
|
|
super().__init__(**kwargs)
|
|||
|
|
self._api_key = get_settings().tool.weather_api_key
|
|||
|
|
if not self._api_key:
|
|||
|
|
raise ValueError("Weather API key not found in settings")
|
|||
|
|
|
|||
|
|
# 基础请求参数
|
|||
|
|
self._base_params = {
|
|||
|
|
"key": self._api_key,
|
|||
|
|
"language": "zh-Hans",
|
|||
|
|
"unit": "c"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
def _run(self, location: str) -> dict:
|
|||
|
|
"""同步执行天气查询"""
|
|||
|
|
try:
|
|||
|
|
logger.info(f"查询天气 - 城市: {location}")
|
|||
|
|
|
|||
|
|
# 构建API请求
|
|||
|
|
url = "https://api.seniverse.com/v3/weather/now.json"
|
|||
|
|
params = {**self._base_params, "location": location}
|
|||
|
|
|
|||
|
|
response = requests.get(url, params=params, timeout=10)
|
|||
|
|
response.raise_for_status()
|
|||
|
|
|
|||
|
|
data = response.json()
|
|||
|
|
|
|||
|
|
# 处理API响应
|
|||
|
|
if 'results' not in data:
|
|||
|
|
error_msg = data.get('status', 'API返回格式异常')
|
|||
|
|
raise ValueError(f"天气API错误: {error_msg}")
|
|||
|
|
|
|||
|
|
weather = data['results'][0]['now']
|
|||
|
|
return {
|
|||
|
|
"status": "success",
|
|||
|
|
"location": location,
|
|||
|
|
"temperature": weather["temperature"],
|
|||
|
|
"condition": weather["text"],
|
|||
|
|
"humidity": weather.get("humidity", "N/A"),
|
|||
|
|
"wind": weather.get("wind_direction", "N/A"),
|
|||
|
|
"full_data": weather
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
except requests.exceptions.RequestException as e:
|
|||
|
|
logger.error(f"网络请求失败: {str(e)}")
|
|||
|
|
return {"status": "error", "message": f"网络错误: {str(e)}"}
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"查询失败: {str(e)}")
|
|||
|
|
return {"status": "error", "message": str(e)}
|
|||
|
|
|
|||
|
|
async def _arun(self, location: str) -> dict:
|
|||
|
|
"""异步执行（示例实现）"""
|
|||
|
|
# 实际项目中可以用aiohttp替换requests
|
|||
|
|
return self._run(location)
|