LangChain实战：5分钟用Tongyi大模型搭建天气查询Agent（附完整代码）

5分钟实战基于Tongyi大模型的智能天气查询Agent开发指南最近在开发一个智能助手项目时我发现很多开发者对大模型Agent的实际应用场景很感兴趣。今天就用阿里云Tongyi大模型结合LangChain框架带大家快速实现一个能查询国内天气的智能Agent。这个方案特别适合需要本地化部署的企业或个人开发者相比OpenAI方案Tongyi在国内访问更稳定且完全兼容LangChain生态。1. 环境准备与依赖安装首先确保你的Python环境是3.8或更高版本。我推荐使用conda创建一个干净的虚拟环境conda create -n weather_agent python3.10 conda activate weather_agent接下来安装核心依赖库。除了LangChain和Tongyi SDK外我们还需要requests来处理API请求pip install langchain langchain-community dashscope requests提示Dashscope是阿里云提供的AI模型服务平台Tongyi大模型就是通过它来调用的。记得提前在阿里云控制台开通服务并获取API Key。2. 天气API对接方案选择国内可用的天气API有不少经过对比测试我推荐以下两种方案API提供商免费额度稳定性数据精度接入难度和风天气1000次/天★★★★☆★★★★★中等心知天气500次/天★★★☆☆★★★★☆简单本文以和风天气为例你需要注册和风天气开发者账号创建项目获取API Key查阅城市ID列表并下载3. 核心代码实现3.1 构建天气查询工具LangChain的BaseTool类让我们能快速封装业务逻辑。下面这个WeatherTool会处理城市编码映射和API请求from langchain.tools import BaseTool import requests import json class WeatherTool(BaseTool): name weather_query description 当需要查询中国城市天气时使用此工具输入应为城市名称 def __init__(self): super().__init__() self.city_map self._load_city_map() def _load_city_map(self): with open(city_code.json, r, encodingutf-8) as f: return json.load(f) def _run(self, city: str) - str: city_code self.city_map.get(city) if not city_code: return f未找到{city}对应的城市编码 url fhttps://devapi.qweather.com/v7/weather/now?key你的API_KEYlocation{city_code} response requests.get(url) data response.json() if data[code] ! 200: return 天气查询失败 weather data[now] return ( f{city}当前天气{weather[text]}\n f温度{weather[temp]}℃ | 体感{weather[feelsLike]}℃\n f湿度{weather[humidity]}% | 风向{weather[windDir]}\n f更新时间{weather[obsTime]} )3.2 初始化Agent系统接下来配置Tongyi大模型和Agent执行器。注意这里使用CONVERSATIONAL_REACT_DESCRIPTION类型的Agent它适合多轮对话场景from langchain_community.llms.tongyi import Tongyi from langchain.agents import initialize_agent, AgentType from langchain.memory import ConversationBufferMemory llm Tongyi( model_nameqwen-plus, temperature0.3, dashscope_api_key你的Dashscope_API_KEY ) memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue ) tools [WeatherTool()] agent initialize_agent( tools, llm, agentAgentType.CONVERSATIONAL_REACT_DESCRIPTION, memorymemory, verboseTrue )3.3 测试对话功能现在可以测试这个Agent是否能正确处理天气查询和上下文记忆# 第一轮查询 response agent.run(北京现在天气怎么样) print(response) # 第二轮查询测试记忆功能 response agent.run(那上海呢) print(response) # 第三轮测试无关问题 response agent.run(我刚才问了哪些城市的天气) print(response)4. 高级功能扩展4.1 错误处理增强实际应用中需要更健壮的错误处理机制。修改WeatherTool的_run方法def _run(self, city: str) - str: try: city_code self.city_map.get(city) if not city_code: return f未找到{city}对应的城市编码 response requests.get( fhttps://devapi.qweather.com/v7/weather/now, params{ key: self.api_key, location: city_code }, timeout5 ) response.raise_for_status() data response.json() if data[code] ! 200: return f天气API返回错误{data.get(message, 未知错误)} # ...原有天气解析逻辑... except requests.exceptions.RequestException as e: return f网络请求失败{str(e)} except json.JSONDecodeError: return 天气数据解析失败 except Exception as e: return f天气查询异常{str(e)}4.2 支持多日预报和风天气的7天预报接口也很实用。我们可以扩展工具类def _run(self, city: str, days: int 1) - str: if days 1: return self._get_current_weather(city) elif 2 days 7: return self._get_forecast(city, days) else: return 仅支持查询1-7天的天气 def _get_forecast(self, city: str, days: int) - str: city_code self.city_map.get(city) url fhttps://devapi.qweather.com/v7/weather/{days}d params {key: self.api_key, location: city_code} response requests.get(url, paramsparams) data response.json() forecast [] for day in data[daily]: forecast.append( f{day[fxDate]} {day[textDay]} f{day[tempMin]}~{day[tempMax]}℃ f风向{day[windDirDay]} ) return f{city}未来{days}天预报\n \n.join(forecast)记得同步更新工具的description属性说明支持多日查询。5. 部署优化建议5.1 性能优化技巧缓存机制对相同城市的查询结果缓存10分钟减少API调用批量查询当用户询问多个城市天气时使用asyncio并发请求连接池使用requests.Session重用HTTP连接from datetime import datetime, timedelta import asyncio class WeatherTool(BaseTool): def __init__(self): self.cache {} self.session requests.Session() def _get_from_cache(self, city: str) - Optional[str]: if city in self.cache: cached_time, data self.cache[city] if datetime.now() - cached_time timedelta(minutes10): return data return None async def _fetch_multi_cities(self, cities: List[str]) - List[str]: async with aiohttp.ClientSession() as session: tasks [] for city in cities: task self._get_weather_async(session, city) tasks.append(task) return await asyncio.gather(*tasks)5.2 安全注意事项将API密钥存储在环境变量中不要硬编码在代码里对用户输入的城市参数做合法性检查限制查询频率防止滥用import os from typing import List API_KEY os.getenv(QWEATHER_KEY) VALID_CITIES set([北京, 上海, 广州]) # 实际应从city_code.json加载 def _run(self, city: str) - str: if city not in VALID_CITIES: return 暂不支持该城市查询 if not API_KEY: return 天气服务未正确配置 # ...原有逻辑...这个基于Tongyi大模型的天气查询Agent从环境搭建到核心功能实现只用了不到100行代码。我在实际项目中发现对于企业级应用还需要考虑日志监控、限流熔断等机制。如果你需要处理更复杂的业务场景LangChain的AgentExecutor提供了丰富的回调接口可以方便地扩展这些功能。