Langchain构建天气查询AI智能体：原理与实践-平芜编程栈

1. 项目概述：用Langchain构建天气查询AI智能体

最近在开发一个能自动查询天气的AI助手时，我发现Langchain框架简直是神器。这个开源工具包让大语言模型(LLM)具备了调用外部工具的能力，就像给ChatGPT装上了"手脚"。想象一下，你只需要问"上海明天会下雨吗？"，AI就能自动搜索最新天气数据并给出专业回答 - 这就是我们要实现的智能体(Agent)。

传统天气APP需要手动输入城市、点击查询，而基于Langchain的解决方案能理解自然语言，自动完成整个查询流程。更酷的是，它还能记住对话上下文，比如你之前说过住在北京，之后问"今天需要带伞吗？"它就知道查北京天气。

2. 核心组件与原理拆解

2.1 Langchain架构解析

Langchain的核心价值在于它搭建了LLM与现实世界的桥梁。典型工作流程分为三步：

意图识别：LLM分析用户问题，判断是否需要调用工具
工具执行：如需要，选择合适工具并生成调用参数
结果整合：将工具返回结果加工成自然语言回复

对于天气查询场景，这个流程具体表现为：

用户问"杭州明天温度多少？" → LLM识别需要天气数据 → 调用天气API查询杭州预报 → 提取温度数据生成回复

2.2 关键组件选型

语言模型选择：

推荐使用GPT-4或Claude 3等最新模型，它们在工具调用准确性上表现更好
本地部署可选Llama 3 70B，但需注意其工具调用能力稍弱

天气数据源：

Tavily Search：内置的搜索引擎API，可直接获取结构化天气数据
OpenWeatherMap：专业天气服务，需额外注册API key
中国用户可考虑和风天气API，数据更本地化

记忆模块：

MemorySaver：基础对话记忆，适合简单场景
RedisBackedChatMemory：生产环境推荐，支持持久化存储

3. 完整实现步骤

3.1 环境准备

先安装必要依赖（建议使用Python 3.10+）：

pip install langgraph langchain-tavily langchain-core>=0.3.37

设置API密钥：

import os os.environ["TAVILY_API_KEY"] = "your_api_key" # 天气数据源 os.environ["OPENAI_API_KEY"] = "sk-..." # 如果用OpenAI模型

3.2 工具定义与初始化

创建天气搜索工具：

from langchain_tavily import TavilySearch weather_tool = TavilySearch( max_results=3, include_answer=True, # 直接返回结构化答案 include_images=False # 不需要天气图标可关闭 )

测试工具是否工作：

response = weather_tool.invoke("上海当前温度") print(response["answer"]) # 输出示例：上海现在气温25℃，晴转多云

3.3 智能体构建

使用ReAct模式构建智能体：

from langchain.chat_models import ChatOpenAI from langgraph.prebuilt import create_react_agent # 初始化模型 model = ChatOpenAI(model="gpt-4-1106-preview", temperature=0) # 创建智能体 agent = create_react_agent( model, tools=[weather_tool], checkpointer=MemorySaver() # 添加对话记忆 )

3.4 对话测试

实现带记忆的对话流：

config = {"configurable": {"thread_id": "user123"}} # 用户唯一标识 # 第一轮对话 input_msg = {"role": "user", "content": "我在杭州工作"} for step in agent.stream({"messages": [input_msg]}, config, stream_mode="values"): print(step["messages"][-1]["content"]) # 第二轮对话（会记住位置） input_msg = {"role": "user", "content": "明天需要带伞吗？"} for step in agent.stream({"messages": [input_msg]}, config, stream_mode="values"): print(step["messages"][-1]["content"])

4. 高级功能实现

4.1 多工具协作

增加位置解析工具提升准确性：

from langchain.tools import Tool def location_resolver(query: str) -> dict: """将模糊地址转为标准城市名""" # 这里可以接入高德/百度地理编码API return {"standard_name": "杭州市"} location_tool = Tool.from_function( name="location_resolver", description="将模糊地址转为标准地理名称", func=location_resolver ) # 更新工具列表 tools = [weather_tool, location_tool]

4.2 结果后处理

添加天气数据格式化工具：

def weather_formatter(raw_data: dict) -> str: """将原始天气数据转为自然语言""" temp = raw_data["current"]["temp_c"] condition = raw_data["current"]["condition"]["text"] return f"当前气温{temp}℃，天气状况：{condition}" # 修改工具调用链 agent = create_react_agent( model, tools=[weather_tool, location_tool], postprocessor=weather_formatter )

4.3 错误处理机制

实现工具调用重试逻辑：

from tenacity import retry, stop_after_attempt @retry(stop=stop_after_attempt(3)) def safe_tool_call(tool, input): try: return tool.invoke(input) except Exception as e: print(f"工具调用失败: {e}") raise

5. 生产环境优化建议

5.1 性能优化技巧

缓存机制：对相同查询缓存结果，减少API调用

from langchain.cache import SQLiteCache import langchain langchain.llm_cache = SQLiteCache(database_path=".langchain.db")

异步处理：使用async/await提升并发能力

async def async_agent_run(query): return await agent.ainvoke({"messages": [{"role": "user", "content": query}]})

5.2 监控与日志

集成LangSmith进行全链路追踪：

os.environ["LANGSMITH_TRACING"] = "true" os.environ["LANGSMITH_API_KEY"] = "your_api_key"

5.3 安全防护

输入过滤：防止Prompt注入

def sanitize_input(text: str) -> bool: blacklist = ["系统", "执行", "sudo"] # 危险关键词列表 return not any(word in text for word in blacklist)

频率限制：避免API滥用

from fastapi import FastAPI, Request from slowapi import Limiter from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) app = FastAPI() app.state.limiter = limiter

6. 常见问题解决方案

6.1 工具调用失败排查

症状：AI回复"让我查一下..."但无后续

检查项：
1. API密钥是否有效
2. 网络连接是否正常
3. 工具描述是否清晰（LLM需理解工具用途）

诊断代码：

print(weather_tool.metadata) # 查看工具元数据 print(model.bind_tools([weather_tool]).invoke("test")) # 测试工具绑定

6.2 记忆失效处理

症状：AI不记得之前对话内容

解决方案：
1. 确认thread_id保持不变
2. 检查MemorySaver存储路径权限
3. 对于重要信息，可让AI明确回复"我会记住XX"

增强记忆示例：

def explicit_memory_processor(text): if "记住" in text: return {"memory": text.split("记住")[1].strip()} return None

6.3 天气数据不准确

优化方案：

多数据源校验

def multi_source_check(city): sources = [TavilySearch(), OpenWeatherMap()] return [s.invoke(city) for s in sources]

添加时间戳提示

response += f"\n（数据更新时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}）"

7. 扩展应用场景

这个基础框架稍加改造就能支持更多实用功能：

7.1 旅行规划助手

tools = [ weather_tool, flight_search_tool, hotel_search_tool ]

用户可以说："下周末北京天气怎么样？帮我找机票和酒店"

7.2 农业咨询系统

def crop_advice(weather_data): """根据天气给出种植建议""" if weather_data["rain"] > 50: return "建议推迟播种" return "适宜播种" tools = [weather_tool, Tool.from_function(crop_advice)]

7.3 智能家居控制

def ac_control(temp): """根据温度控制空调""" if temp > 28: return "开启制冷模式" return "保持当前设置" tools = [weather_tool, Tool.from_function(ac_control)]

我在实际项目中发现，要让AI智能体稳定工作，关键是要处理好边界情况。比如当用户问"我家乡明天天气如何"时，系统必须能主动询问具体城市，或者根据IP地址自动判断位置。这需要设计良好的fallback机制：

def location_fallback(): """当位置不明确时的处理流程""" return { "response": "请问您想查询哪个城市的天气？", "next_step": "wait_for_location" }

另一个实用技巧是给AI添加"思考过程"的可视化。在开发阶段可以输出中间推理步骤：

def debug_agent(agent, query): for step in agent.stream(...): print(f"[DEBUG] {step['thoughts']}") if 'tool_call' in step: print(f"调用工具: {step['tool_call']}")

对于中文场景，还需要特别注意：