环保AI智能体案例：低成本监测空气质量变化

引言

对于社区工作者来说，实时掌握当地的空气质量状况至关重要。传统的监测设备往往价格昂贵，动辄上万元，对于预算有限的社区来说难以承担。而今天我要介绍的AI智能体解决方案，只需要一台普通电脑和几个开源工具，就能搭建起一套经济实用的空气质量监测系统。

这个方案的核心是利用AI智能体自动收集、分析并可视化空气质量数据。就像有一个24小时值班的环保小助手，它会定时从公开数据源抓取最新的空气质量指标，通过简单的分析算法判断污染趋势，最后生成直观的图表报告。整个过程完全自动化，你只需要在初期做一些简单的配置工作。

1. 环境准备

1.1 硬件需求

这套方案对硬件要求极低，你甚至可以用一台闲置的旧电脑来运行：

任何x86架构的电脑（笔记本或台式机均可）
4GB以上内存
50GB以上存储空间
稳定的网络连接

1.2 软件准备

我们需要安装几个开源工具来构建这个AI智能体系统：

Python 3.8或更高版本
Jupyter Notebook（用于开发和测试）
几个关键的Python库：pandas、requests、matplotlib

安装这些工具的bash命令如下：

# 安装Python和pip sudo apt update sudo apt install python3 python3-pip # 安装Jupyter Notebook pip3 install notebook # 安装所需Python库 pip3 install pandas requests matplotlib

2. 数据获取与处理

2.1 选择数据源

我们可以从多个免费API获取空气质量数据：

政府环保部门公开数据
OpenAQ平台（全球空气质量数据）
世界空气质量指数项目（WAQI）

以WAQI为例，获取数据的Python代码如下：

import requests import pandas as pd def get_air_quality(city): url = f"https://api.waqi.info/feed/{city}/?token=你的API密钥" response = requests.get(url) data = response.json() return pd.DataFrame([data['data']]) # 示例：获取北京空气质量数据 beijing_air = get_air_quality("beijing") print(beijing_air[['aqi', 'pm25', 'pm10']])

💡 提示
你需要先在WAQI官网注册获取免费的API密钥，每天有1000次免费查询额度，对社区监测完全够用。

2.2 数据存储

为了长期跟踪空气质量变化，我们需要将数据保存到本地：

import sqlite3 # 创建SQLite数据库 conn = sqlite3.connect('air_quality.db') cursor = conn.cursor() # 创建数据表 cursor.execute(''' CREATE TABLE IF NOT EXISTS air_quality ( date TEXT, city TEXT, aqi INTEGER, pm25 REAL, pm10 REAL ) ''') conn.commit()

3. AI智能体构建

3.1 数据分析逻辑

我们的AI智能体需要具备基本的分析能力，比如：

判断当前空气质量等级
识别污染趋势
生成简单的预警提示

实现代码如下：

def analyze_air_quality(df): # 空气质量评级 aqi = df['aqi'].values[0] if aqi <= 50: status = "优" elif aqi <= 100: status = "良" elif aqi <= 150: status = "轻度污染" elif aqi <= 200: status = "中度污染" else: status = "重度污染" # 趋势分析（简单版） pm25_trend = "上升" if df['pm25'].values[0] > df['pm25'].values[-1] else "下降" return { "status": status, "trend": pm25_trend, "suggestion": f"当前空气质量{status}，PM2.5浓度{pm25_trend}趋势" }

3.2 自动化定时任务

为了让系统自动运行，我们可以设置定时任务：

import schedule import time def job(): data = get_air_quality("your_city") analysis = analyze_air_quality(data) print(f"{time.ctime()}: {analysis['suggestion']}") # 每小时执行一次 schedule.every().hour.do(job) while True: schedule.run_pending() time.sleep(1)

4. 数据可视化与报告

4.1 生成可视化图表

使用matplotlib生成空气质量趋势图：

import matplotlib.pyplot as plt import matplotlib.dates as mdates def plot_air_quality(city): # 从数据库读取最近7天数据 conn = sqlite3.connect('air_quality.db') df = pd.read_sql(f"SELECT * FROM air_quality WHERE city='{city}' ORDER BY date DESC LIMIT 7", conn) conn.close() # 绘制图表 plt.figure(figsize=(10, 6)) plt.plot(pd.to_datetime(df['date']), df['aqi'], label='AQI指数') plt.plot(pd.to_datetime(df['date']), df['pm25'], label='PM2.5') plt.plot(pd.to_datetime(df['date']), df['pm10'], label='PM10') # 图表美化 plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%m-%d')) plt.title(f'{city}近7天空气质量趋势') plt.xlabel('日期') plt.ylabel('数值') plt.legend() plt.grid(True) plt.savefig(f'{city}_air_quality.png') plt.close()

4.2 自动生成报告

将分析结果和图表整合成HTML报告：

def generate_report(city): analysis = analyze_air_quality(get_air_quality(city)) plot_air_quality(city) html = f""" <html> <head><title>{city}空气质量报告</title></head> <body> <h1>{city}空气质量报告</h1> <p>生成时间：{time.ctime()}</p> <h2>当前状态</h2> <p>空气质量指数(AQI): {analysis['aqi']} ({analysis['status']})</p> <p>PM2.5: {analysis['pm25']} μg/m³ ({analysis['pm25_trend']})</p> <p>建议: {analysis['suggestion']}</p> <h2>近7天趋势</h2> <img src="{city}_air_quality.png" width="800"> </body> </html> """ with open(f"{city}_air_quality_report.html", "w") as f: f.write(html)

5. 部署与优化

5.1 系统部署

将整个系统部署到服务器上持续运行：

使用nohup让Python脚本在后台运行
设置cron定时任务定期生成报告
配置简单的Web服务器展示报告

# 后台运行监测脚本 nohup python3 monitor.py > monitor.log & # 设置每天生成报告的cron任务 (crontab -l 2>/dev/null; echo "0 8 * * * /usr/bin/python3 /path/to/generate_report.py") | crontab -