孙玲的空间1. LangChain天气预报Agent项目概述
LangChain作为当前最流行的AI应用开发框架之一，其核心价值在于将大语言模型(LLM)与各类工具、数据源和业务流程无缝连接。这个天气预报Agent案例展示了如何利用LangChain 1.0的新特性构建一个具备专业领域能力的智能助手。不同于简单的问答机器人，这个Agent能够：
- 理解结构化提示词
- 按需调用外部工具
- 管理对话记忆
- 输出标准化数据格式
对于AI开发者而言，掌握这些能力意味着可以构建真正实用的企业级应用，而不仅仅是演示原型。下面我将从架构设计到代码实现，完整拆解这个案例的技术要点。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 核心概念深度解析
2.1 结构化提示词设计
在传统LLM应用中，提示词往往是大段自然语言文本。LangChain 1.0引入了消息序列的概念，将提示词分解为具有明确语义的组件：
python复制from langchain.schema import SystemMessage, HumanMessage, AIMessage
messages = [
SystemMessage(content="你是一位资深律师，只回答法律相关问题"),
HumanMessage(content="劳动合同解除赔偿标准是多少？"),
AIMessage(content="根据《劳动合同法》第47条...")
]
这种结构化设计带来三个关键优势：
- 角色隔离：系统指令、用户输入和AI响应被明确区分，避免提示词污染
- 上下文管理：对话历史可以按需裁剪，只保留相关消息
- 多模态扩展：未来可以轻松支持图像、音频等非文本消息类型
实际开发中发现，SystemMessage的内容长度最好控制在200-300字之间。过长的系统提示会导致模型忽略后续用户输入的关键信息。
2.2 工具集成机制
LangChain的@tool装饰器实现了AI能力的动态扩展。以下是一个数据库查询工具的典型实现：
python复制from langchain.tools import tool
import sqlite3
@tool
def query_customer_data(customer_id: str) -> dict:
"""根据客户ID查询订单历史
Args:
customer_id: 格式为CID-XXXX的客户标识符
Returns:
包含最近3笔订单详情的字典
"""
conn = sqlite3.connect('sales.db')
cursor = conn.cursor()
cursor.execute("SELECT * FROM orders WHERE customer_id=? LIMIT 3", (customer_id,))
return {
"orders": cursor.fetchall(),
"query_time": datetime.now().isoformat()
}
工具集成的关键注意事项：
- 类型标注必须精确：参数和返回值类型提示是工具能否被正确调用的关键
- 文档字符串即API文档：LLM完全依赖函数的docstring来理解工具用途
- 错误处理要健壮：工具内部应该捕获所有异常，返回结构化错误信息
2.3 运行时上下文设计
上下文对象是Agent执行过程中的"工作记忆"，良好的设计应遵循以下原则：
python复制from dataclasses import dataclass
from datetime import datetime
@dataclass
class SalesContext:
user_id: str # 必须字段，无默认值
department: str = "sales" # 带默认值的可选字段
session_start: datetime = field(default_factory=datetime.now)
上下文的最佳实践：
- 基础字段用简单类型（str/int/bool）
- 复杂对象应该先序列化为字符串
- 敏感信息（如密码）永远不要存入上下文
- 单个上下文对象应保持轻量（<1KB）
2.4 记忆管理策略
LangChain提供多级记忆管理方案：
| 记忆类型 | 存储介质 | 适用场景 | 容量限制 |
|---|---|---|---|
| InMemory | 内存 | 开发测试 | 约10轮对话 |
| Redis | 内存数据库 | 生产环境 | 百万级对话 |
| Postgres | 关系数据库 | 审计场景 | 无上限 |
python复制from langgraph.checkpoint.redis import RedisSaver
checkpointer = RedisSaver(
url="redis://localhost:6379",
ttl=86400 # 记忆保留24小时
)
记忆管理的经验教训：
- 测试环境务必使用InMemory存储，避免残留数据干扰
- 生产环境建议Redis集群+持久化配置
- 敏感对话需要设置自动清理策略
3. 天气预报Agent完整实现
3.1 系统架构设计
整个Agent的组件交互流程如下：
- 用户输入通过HTTP API进入系统
- 路由层将请求分发给对应Agent
- Agent依次执行：
- 加载对话记忆
- 解析输入意图
- 调用工具链
- 生成结构化响应
- 响应经格式化后返回客户端
3.2 核心代码实现
以下是增强版天气预报Agent的实现：
python复制from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
class TemperatureUnit(Enum):
CELSIUS = "℃"
FAHRENHEIT = "℉"
@dataclass
class WeatherContext:
user_id: str
unit: TemperatureUnit = TemperatureUnit.CELSIUS
last_query: Optional[str] = field(default=None)
@tool
def get_weather_for_location(
city: str,
runtime: ToolRuntime[WeatherContext]
) -> dict:
"""获取实时天气数据
Args:
city: 城市名称（中文或拼音）
Returns:
{
"condition": "晴朗/多云/雨...",
"temperature": 25.0,
"humidity": 0.65,
"wind_speed": 10.2
}
"""
# 模拟不同用户的温度单位偏好
unit = runtime.context.unit
temp = 25.0 if unit == TemperatureUnit.CELSIUS else 77.0
return {
"condition": "晴朗",
"temperature": temp,
"humidity": 0.65,
"wind_speed": 10.2
}
@dataclass
class WeatherResponse:
speech: str
data: dict
unit: str
3.3 配置与启动
生产环境推荐使用配置文件和依赖注入：
yaml复制# config.yaml
model:
name: "ollama:qwen3-next:80b-cloud"
base_url: "http://llm-gateway:11434"
tools:
- get_weather_for_location
- get_user_location
memory:
type: "redis"
url: "redis://cache:6379"
python复制from langchain.agents import [Agent](https://taotoken.net?utm_source=ai)Executor
def create_agent(config_path: str):
config = load_config(config_path)
model = init_chat_model(**config['model'])
tools = load_tools(config['tools'])
memory = init_memory(config['memory'])
return AgentExecutor(
model=model,
tools=tools,
memory=memory,
max_iterations=5 # 防止无限循环
)
3.4 测试与验证
建议使用pytest编写自动化测试套件：
python复制import pytest
@pytest.fixture
def weather_agent():
return create_agent("test_config.yaml")
def test_weather_query(weather_agent):
response = weather_agent.invoke(
{"messages": [{"role": "user", "content": "北京天气？"}]},
config={"configurable": {"user_id": "test_1"}}
)
assert "北京" in response["speech"]
assert response["data"]["temperature"] > -20
assert response["unit"] == "℃"
4. 生产环境部署要点
4.1 性能优化
- 批处理：将多个工具调用合并为单个批处理操作
- 缓存：对天气数据等半静态内容实现Redis缓存
- 连接池：数据库和外部服务连接必须使用连接池
python复制from functools import lru_cache
@lru_cache(maxsize=1000)
def get_cached_weather(city: str):
return get_weather_for_location(city)
4.2 安全防护
- 输入校验：使用Pydantic验证所有输入参数
- 速率限制：对API调用实施令牌桶限流
- 敏感信息过滤：对话日志中的身份证号、手机号等需脱敏
python复制from pydantic import BaseModel, constr
class UserQuery(BaseModel):
content: constr(max_length=500)
user_id: constr(regex=r'^U\d{8}$')
4.3 监控指标
必备的监控维度包括：
| 指标名称 | 类型 | 报警阈值 |
|---|---|---|
| 请求量 | QPS | >500/s |
| 响应时间 | P99 | >3s |
| 工具调用错误率 | 百分比 | >1% |
| 记忆加载延迟 | 毫秒 | >100ms |
5. 常见问题排查
5.1 工具未被调用
可能原因及解决方案：
-
文档字符串不完整
- 确保每个参数都有明确描述
- 示例：
:param city: 城市名称（支持中文或拼音）
-
类型提示缺失
- 所有参数必须标注类型
- 复杂类型需要JsonSchema支持
-
权限问题
- 检查工具函数的执行权限
- 验证运行时上下文是否传递正确
5.2 结构化输出失败
调试步骤：
- 检查模型是否支持JSON模式：
python复制model.invoke("请用JSON输出{'a':1}", stop=["}"])
- 验证数据类定义：
python复制print(ResponseFormat.__pydantic_model__.schema_json())
- 测试简单用例：
python复制@dataclass
class TestOutput:
text: str
agent.response_format = TestOutput
5.3 记忆丢失问题
诊断方法：
- 检查记忆存储配置：
python复制print(agent.checkpointer.config)
- 验证线程ID一致性：
python复制assert config["configurable"]["thread_id"] == "123456"
- 查看原始存储数据：
python复制redis-cli KEYS "langchain:memory:*"
6. 扩展与优化方向
对于需要更高性能的场景，可以考虑以下进阶方案：
- 工具并行化：使用
asyncio实现多个工具的并发执行 - 响应流式传输：通过SSE逐步返回部分结果
- 混合模型架构：简单查询用小型模型，复杂任务路由到大型模型
- 持续学习机制：将用户反馈自动转化为训练数据
python复制from concurrent.futures import ThreadPoolExecutor
def parallel_tool_execution(tools, inputs):
with ThreadPoolExecutor() as executor:
results = list(executor.map(
lambda t: t.func(**t.args),
zip(tools, inputs)
))
return results
这个天气预报Agent案例展示了LangChain在构建生产级AI应用时的完整工作流程。从工具集成、上下文管理到结构化输出，每个设计决策都直接影响最终系统的可靠性和可用性。我在实际部署中发现，良好的类型系统和清晰的接口定义是维护大型Agent项目的关键。