amap_weather.py 1.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051
  1. import os
  2. from typing import Dict, Optional, Union
  3. import requests
  4. from qwen_agent.tools.base import BaseTool, register_tool
  5. @register_tool('amap_weather')
  6. class AmapWeather(BaseTool):
  7. description = '获取对应城市的天气数据'
  8. parameters = [{
  9. 'name': 'location',
  10. 'type': 'string',
  11. 'description': '城市/区具体名称,如`北京市海淀区`请描述为`海淀区`',
  12. 'required': True
  13. }]
  14. def __init__(self, cfg: Optional[Dict] = None):
  15. super().__init__(cfg)
  16. # remote call
  17. self.url = 'https://restapi.amap.com/v3/weather/weatherInfo?city={city}&key={key}'
  18. import pandas as pd
  19. self.city_df = pd.read_excel(
  20. 'https://modelscope.oss-cn-beijing.aliyuncs.com/resource/agent/AMap_adcode_citycode.xlsx')
  21. self.token = self.cfg.get('token', os.environ.get('AMAP_TOKEN', ''))
  22. assert self.token != '', 'weather api token must be acquired through ' \
  23. 'https://lbs.amap.com/api/webservice/guide/create-project/get-key and set by AMAP_TOKEN'
  24. def get_city_adcode(self, city_name):
  25. filtered_df = self.city_df[self.city_df['中文名'] == city_name]
  26. if len(filtered_df['adcode'].values) == 0:
  27. raise ValueError(f'location {city_name} not found, availables are {self.city_df["中文名"]}')
  28. else:
  29. return filtered_df['adcode'].values[0]
  30. def call(self, params: Union[str, dict], **kwargs) -> str:
  31. params = self._verify_json_format_args(params)
  32. location = params['location']
  33. response = requests.get(self.url.format(city=self.get_city_adcode(location), key=self.token))
  34. data = response.json()
  35. if data['status'] == '0':
  36. raise RuntimeError(data)
  37. else:
  38. weather = data['lives'][0]['weather']
  39. temperature = data['lives'][0]['temperature']
  40. return f'{location}的天气是{weather}温度是{temperature}度。'