故小里1. 项目概述
今天我想分享一个最近完成的天气查询智能体项目。这个项目源于我在开发AI助手时遇到的一个实际问题：如何让智能体准确理解用户输入的各种地名格式（如"杭州"、"上城区"、"浙江省杭州市西湖区"等）并返回精确的天气信息。
传统天气API通常只能处理标准化的城市名称或编码，而用户在实际使用中往往会输入各种非标准化的地名组合。为了解决这个问题，我设计了一个结合高德天气API和ElasticSearch的解决方案，通过构建城市编码索引库，实现了对模糊地名的智能匹配和精确查询。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 技术架构设计
2.1 整体架构思路
项目的核心挑战在于地名识别和天气数据获取两个环节：
-
地名识别：需要处理用户输入的各种地名变体，包括：
- 单独的城市名（"杭州"）
- 区县级名称（"上城区"）
- 省市组合（"浙江省杭州市"）
- 错别字或简称（"杭城"）
-
天气数据获取：需要将识别出的地名转换为高德API可接受的adcode（行政区划代码），然后调用天气接口。
基于这些需求，我设计了以下技术栈：
- ElasticSearch：存储和检索城市编码数据，支持模糊查询
- 高德天气API：获取实时天气数据
- LangChain：构建智能体框架
- Docker：容器化部署ElasticSearch
2.2 技术选型考量
为什么选择ElasticSearch而不是传统数据库？
在初期方案中，我考虑过使用MySQL等关系型数据库存储城市编码表。但实际测试发现几个问题：
- 对模糊查询支持有限，特别是当用户输入包含错别字时
- 需要额外开发分词和相似度匹配功能
- 性能在大数据量查询时不够理想
ElasticSearch的优势在于：
- 内置中文分词器（配合IK插件效果更好）
- 强大的模糊搜索能力
- 高性能的全文检索
- 易于扩展的分布式架构
提示：在实际项目中，如果数据量不大（如本项目的城市编码表约3000条记录），也可以考虑SQLite+文本相似度算法的轻量级方案。但考虑到未来可能扩展更多地理位置相关功能，我最终选择了ElasticSearch。
3. 核心实现细节
3.1 ElasticSearch环境搭建
3.1.1 Docker部署配置
我使用Docker Compose部署ElasticSearch 8.6.2版本，以下是关键的配置考虑：
yaml复制services:
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:8.6.2
environment:
- cluster.name=assistant-agent-cluster
- node.name=assistant-agent-node
- discovery.type=single-node
- "ES_JAVA_OPTS=-Xms1g -Xmx1g"
- xpack.security.enabled=false # 开发环境关闭安全认证
ports:
- "9202:9200" # 使用非标准端口避免冲突
volumes:
- ./data:/usr/share/elasticsearch/data
- ./plugins:/usr/share/elasticsearch/plugins
关键配置说明：
ES_JAVA_OPTS：设置JVM堆内存为1GB，适合开发环境xpack.security.enabled=false：开发环境简化配置，生产环境应开启安全认证- 端口映射使用9202而非默认的9200，避免与本地其他ES实例冲突
3.1.2 IK分词器安装
中文搜索需要安装IK分词器，版本必须与ES严格匹配：
bash复制# 进入容器
docker exec -it es_assistant_agent sh
# 安装IK分词器
./bin/elasticsearch-plugin install https://get.infini.cloud/elasticsearch/analysis-ik/8.6.2
安装后需要重启容器使插件生效。验证安装：
bash复制./bin/elasticsearch-plugin list
# 应输出：analysis-ik
3.2 数据准备与索引构建
3.2.1 城市编码表处理
高德官方提供的城市编码表是Excel格式，需要进行以下处理：
-
数据清洗：
- 去除空白行和测试数据
- 统一字段格式（特别是adcode应为字符串类型）
- 补充完整的省市区三级关系
-
数据结构设计：
python复制{
"adcode": "330106",
"province": "浙江省",
"city": "杭州市",
"district": "西湖区",
"full_name": "浙江省杭州市西湖区",
"pinyin": "zhejiang sheng hangzhou shi xihu qu",
"location": {
"lon": 120.12,
"lat": 30.26
}
}
3.2.2 索引映射配置
创建索引时需要特别关注字段的mapping配置，这对搜索效果至关重要：
python复制mapping = {
"mappings": {
"properties": {
"adcode": {"type": "keyword"},
"province": {
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_smart"
},
"city": {...}, # 类似配置
"district": {...},
"full_name": {
"type": "text",
"fields": {
"keyword": {"type": "keyword"}
}
},
"pinyin": {"type": "text"}
}
}
}
关键点说明：
- 对中文文本字段使用IK分词器
full_name同时保留text和keyword两种类型- 添加拼音字段支持拼音搜索
3.3 搜索模块实现
3.3.1 多条件搜索策略
搜索模块需要处理多种查询场景：
- 精确匹配：当输入完整标准地名时（如"浙江省杭州市西湖区"）
- 部分匹配：当输入部分名称时（如"西湖区"）
- 模糊匹配：当输入包含错别字或简称时（如"淅江省"）
实现代码示例：
python复制def search_location(query):
# 多字段查询
should_clauses = [
{"match": {"province": query}},
{"match": {"city": query}},
{"match": {"district": query}},
{"match": {"full_name": query}},
{"match": {"pinyin": query}}
]
# 提升完整匹配的权重
if len(query) > 3:
should_clauses.append({
"term": {"full_name.keyword": {"value": query, "boost": 2.0}}
})
body = {
"query": {
"bool": {
"should": should_clauses,
"minimum_should_match": 1
}
},
"size": 1
}
response = es.search(index=INDEX_NAME, body=body)
# 处理结果...
3.3.2 搜索结果处理
对ES返回的结果需要做进一步处理：
- 检查匹配分数，过滤低质量匹配
- 处理多级行政区划关系（如确保"西湖区"对应的是"杭州市"而非其他城市的同名区）
- 返回标准化的adcode和完整地名
3.4 天气查询工具封装
3.4.1 高德API调用
封装高德天气API时需要注意以下要点：
python复制@tool
def get_live_weather(city: str) -> str:
"""
获取实时天气的核心函数
"""
params = {
"key": api_key,
"city": city, # 可以是城市名称或adcode
"extensions": "base", # base=实时天气，all=预报天气
"output": "json"
}
try:
response = requests.get(
"https://restapi.amap.com/v3/weather/weatherInfo",
params=params,
timeout=10
)
response.raise_for_status()
result = response.json()
if result.get("status") != "1":
raise ValueError(result.get("info", "未知错误"))
# 数据提取和格式化...
except Exception as e:
return f"获取天气失败: {str(e)}"
错误处理要点：
- API密钥缺失检查
- 网络请求超时处理
- 响应状态码验证
- 结果数据完整性检查
3.4.2 数据格式化
将API返回的原始数据转换为易读的格式：
python复制def format_weather(data):
live = data["lives"][0]
return f"""
{live['city']}实时天气：
· 天气：{live['weather']}
· 温度：{live['temperature']}℃
· 风向：{live['winddirection']}
· 风力：{live['windpower']}级
· 更新时间：{live['reporttime']}
"""
3.5 智能体集成
3.5.1 LangChain智能体配置
使用LangChain的create_agent函数创建智能体：
python复制llm = ChatOpenAI(
model="qwen3.5-plus",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
temperature=0.2, # 降低随机性
)
agent = create_agent(
model=llm,
tools=[get_adcode_by_location, get_live_weather],
system_prompt="""
你是一个专业的天气助手，能够查询中国各地实时天气。
当用户询问天气时：
1. 首先确定具体位置（省/市/区县）
2. 查询该位置的天气信息
3. 以清晰格式返回结果
""",
checkpointer=InMemorySaver(),
)
3.5.2 对话流程设计
智能体处理用户查询的典型流程：
- 接收用户输入（如"杭州上城区的天气"）
- 调用
get_adcode_by_location工具解析位置 - 使用返回的adcode调用
get_live_weather - 格式化返回结果
4. 部署与优化
4.1 容器化部署建议
对于生产环境部署，建议优化Docker配置：
- 资源限制：
yaml复制deploy:
resources:
limits:
cpus: '2'
memory: 2G
- 持久化存储：
yaml复制volumes:
esdata:
driver: local
- 安全配置：
yaml复制environment:
- xpack.security.enabled=true
- ELASTIC_PASSWORD=your_strong_password
4.2 性能优化技巧
-
ES查询优化：
- 使用filter代替query对不评分的字段
- 合理使用缓存（如
request_cache=true） - 对热点数据使用
preference参数
-
API调用优化：
- 实现简单的本地缓存（如使用
functools.lru_cache） - 批量查询多个地点时使用异步请求
- 实现简单的本地缓存（如使用
-
智能体优化：
- 设置合理的max_iterations防止无限循环
- 添加对话历史管理
5. 常见问题与解决方案
5.1 地名识别问题
问题1：用户输入"杭城"，但ES无法匹配
- 解决方案：在索引中添加别名字段，或使用同义词过滤器
问题2：多个城市有相同区名（如"朝阳区"）
- 解决方案：优先返回人口较多的城市，或要求用户补充城市信息
5.2 API调用问题
问题1：高德API返回"INVALID_USER_KEY"
- 检查步骤：
- 确认API密钥有效且未过期
- 检查请求URL和参数格式
- 验证IP白名单设置
问题2：天气数据更新不及时
- 解决方案：在响应中添加数据更新时间，或实现定时刷新缓存
5.3 ElasticSearch问题
问题1：IK分词器未生效
- 排查步骤：
- 确认插件版本与ES版本匹配
- 检查索引mapping中的analyzer配置
- 使用
_analyzeAPI测试分词效果
问题2：查询性能下降
- 优化建议：
- 使用
_search?explain分析查询执行计划 - 考虑添加更多过滤条件缩小搜索范围
- 对热点数据建立独立索引
- 使用
6. 项目扩展思路
-
多数据源整合：
- 结合中国天气网等补充预报信息
- 添加空气质量指数(AQI)数据
-
增强搜索能力：
- 支持地标建筑查询（如"西湖天气"）
- 实现自动补全功能
-
智能体能力扩展：
- 增加多轮对话管理
- 支持天气相关建议（如穿衣指数、出行建议）
-
可视化展示：
- 生成天气数据图表
- 集成地图显示功能
这个项目从技术选型到实现过程中遇到了不少挑战，特别是在地名模糊匹配和API异常处理方面。通过结合ElasticSearch的强大搜索能力和高德天气API的稳定服务，最终实现了一个可靠的天气查询解决方案。