【基于 LangChain 的异步天气查询2】GeoNames实现地区实时气温查询

import aiohttp
import requests
from langchain_core.runnables import RunnableLambdaasync def fetch_weather(city: str) -> str:username = 'shipking'geonames_url = f"http://api.geonames.org/search?q={city}&maxRows=1&username={username}&type=json"print(f"🔍 请求 URL: {geonames_url}")response = requests.get(geonames_url)print(f"🌐 返回内容：{response.text[:200]}")  # 仅打印前200字符避免太长data = response.json()  # 现在应该不会再报错if response.status_code == 200:data = response.json()if data['totalResultsCount'] > 0:geoname = data['geonames'][0]lat, lon = geoname['lat'], geoname['lng']else:return f"未找到城市：{city}"else:return "GeoNames 请求失败"url = (f"https://api.open-meteo.com/v1/forecast?"f"latitude={lat}&longitude={lon}&current_weather=true")try:async with aiohttp.ClientSession() as session:async with session.get(url) as response:if response.status != 200:return "天气 API 请求失败"data = await response.json()temp = data.get("current_weather", {}).get("temperature")return f"当前{city}的气温是 {temp}°C" if temp is not None else "未获取到气温"except Exception as e:return f"请求过程中出错: {str(e)}"# 导出为 Runnable 实例
WeatherLookupAsyncRunnable = RunnableLambda(fetch_weather)

import asyncio
from weather_runnable import WeatherLookupAsyncRunnable
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAIprompt = ChatPromptTemplate.from_template("请问{location}的天气如何？")
llm = ChatOpenAI(model="gpt-4o")# 不要再写 weather = WeatherLookupAsyncRunnable()
weather = WeatherLookupAsyncRunnable  # ✅ 正确写法async def run():user_input = input("请输入城市名称（可中文）：")  # ✅ 自定义输入question = prompt.invoke({"location": user_input}).to_string()print("Prompt output:", question)weather_result = await weather.ainvoke(user_input)print("天气查询结果：", weather_result)llm_output = llm.invoke(f"根据{weather_result}，简短描述一下今天是否适合出门，需要注意什么")print("LLM output:", llm_output.content)if __name__ == "__main__":asyncio.run(run())