1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105
| from langchain_ollama.llms import OllamaLLM from langchain.agents import initialize_agent from langchain.tools import Tool import requests
AMAP_KEY = 'xxx' AMAP_BASE_URL = 'https://restapi.amap.com/v3'
def get_city_code(city_name): url = f'{AMAP_BASE_URL}/config/district' params = { 'key': AMAP_KEY, 'keywords': city_name, 'subdistrict': 2, } response = fetch_data(url, params) return parse_city_code_response(response)
def parse_city_code_response(response): if response.get('status') == '1': districts = response.get('districts', []) if districts: city_info = districts[0] adcode = city_info.get('adcode', '未知')
if 'districts' in city_info: for district in city_info['districts']: if 'adcode' in district: adcode = district['adcode'] break
return adcode return {"error": f"无法获取城市编码: {response.get('info', '未知错误')}"}
def get_weather_data(city_code, forecast=False): url = f'{AMAP_BASE_URL}/weather/weatherInfo' params = { 'key': AMAP_KEY, 'city': city_code, 'extensions': 'all' if forecast else 'base' } response = fetch_data(url, params) return response
def fetch_data(url, params): try: res = requests.get(url=url, params=params) data = res.json() if data.get('status') == '1': return data return {"error": f"请求失败: {data.get('info', '未知错误')}"} except Exception as e: return {"error": f"请求时出错: {str(e)}"}
city_code_tool = Tool( name="GetCityCode", func=get_city_code, description="根据城市名称获取对应的城市编码,参数是中文城市名称,例如:上海" )
weather_tool = Tool( name="GetWeather", func=get_weather_data, description="获取指定城市的天气信息,参数是城市编码和是否需要预报" )
llm = OllamaLLM(model="llama3.2:1b", base_url="http://192.168.100.135:11434")
agent = initialize_agent( tools=[city_code_tool, weather_tool], llm=llm, agent_type="zero-shot-react-description", verbose=True, agent_kwargs={"handle_parsing_errors": True} )
user_question = "北京的天气怎么样?" print(f"用户问题:{user_question}")
response = agent.run(user_question) print(response)
|