【基于 LangChain 的异步天气查询1】异步调用 Open-Meteo API 查询该城市当前气温

一、功能概述

二、文件结构

三、城市天气实时查询（运行代码）

weather_runnable.py

main.py

运行结果

四、技术亮点

五、使用场景

一、功能概述

它实现了以下主要功能：

用户输入地点（城市名）
构造提示词（Prompt）生成自然语言问题
异步调用 Open-Meteo API 查询该城市当前气温
调用 OpenAI GPT-4o 模型，让它基于气温给出外出建议

二、文件结构

your_project/
├── weather_runnable.py     ✅ 自定义异步 Runnable 类
├── main.py                 ✅ 主程序，构建异步链并执行

三、城市天气实时查询（运行代码）

weather_runnable.py

这是一个符合 LangChain LCEL 规范的异步组件，继承自 Runnable。

实现了 ainvoke 方法，用于异步调用天气 API。
使用 aiohttp 获取 Open-Meteo 提供的天气数据。
支持城市：Beijing、Shanghai、New York（可扩展）。

# weather_runnable.py
import aiohttp
from typing import Any, Optional
from langchain_core.runnables import Runnableclass WeatherLookupAsyncRunnable(Runnable):"""异步版：调用 Open-Meteo API 获取城市天气"""def get_coordinates(self, city: str):city_map = {"Beijing": (39.9042, 116.4074),"Shanghai": (31.2304, 121.4737),"New York": (40.7128, -74.0060),}return city_map.get(city)def invoke(self, input: Any, config: Optional[dict] = None) -> str:raise NotImplementedError("这是一个异步组件，请使用 ainvoke() 调用。")async def ainvoke(self, input: Any, config: Optional[dict] = None) -> str:if not isinstance(input, str):raise ValueError("输入必须是字符串（城市名）")coords = self.get_coordinates(input)if coords is None:return f"暂不支持城市：{input}"lat, lon = coordsurl = (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"当前{input}的气温是 {temp}°C" if temp is not None else "未获取到气温"except Exception as e:return f"请求过程中出错: {str(e)}"

main.py

这是主程序，执行以下流程：

构造提示词： 使用 ChatPromptTemplate 创建一个天气查询句子。
调用天气服务： 执行自定义异步 Runnable，获取当前气温。
调用 GPT-4o： 让 LLM 根据天气数据生成建议，比如“适合出门”或“建议带伞”。

# main.py
import asyncio
from weather_runnable import WeatherLookupAsyncRunnable
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # ✅ 用 langchain_openai，不要再用老接口prompt = ChatPromptTemplate.from_template("请问{location}的天气如何？")
llm = ChatOpenAI(model="gpt-4o")weather = WeatherLookupAsyncRunnable()async def run():# 1. 生成用户问题question = prompt.invoke({"location": "Beijing"}).to_string()print("Prompt output:", question)# 2. 实际调用天气 API（最关键）weather_result = await weather.ainvoke("Beijing")print("天气查询结果：", weather_result)    # 3. 由 LLM 生成回复llm_output = llm.invoke(f"根据{weather_result},简短描述一下今天是否适合出门，需要注意什么")print("LLM output:", llm_output.content)asyncio.run(run())

运行结果

Prompt output: Human: 请问Beijing的天气如何？
天气查询结果：当前Beijing的气温是 23.9°C
LLM output: 今天北京的气温是23.9°C，天气较为宜人，非常适合出门活动。不过建议根据具体的天气情况查看是否有降水或其它天气变化，出门时根据个人体感选择适合的衣物。此
外，可以适当携带水杯保持水分补充，同时注意防晒措施，以确保舒适的户外体验。

四、技术亮点

技术	用途
LangChain Core / LCEL	构建可组合的数据流
Runnable Async	实现异步接口逻辑，非阻塞式
aiohttp	高效异步 HTTP 请求库
OpenAI GPT-4o	生成智能出行建议
ChatPromptTemplate	动态构造人类语言输入