AI 数据分析与可视化完全教程
目录
- AI驱动数据分析：从传统BI到AI增强分析
- 自然语言查询数据：Text-to-SQL技术
- LLM辅助数据清洗与预处理
- 智能数据探索：自动统计、异常检测、趋势分析
- AI驱动的图表推荐与自动可视化
- Pandas AI与ChatBI实战
- 数据报告自动生成：LLM + 图表 + 洞察
- 多模态数据分析：图表理解、图像OCR
- 数据分析Agent：自主完成复杂分析任务
- 数据隐私与安全：脱敏、权限控制
- 实战项目：构建智能数据分析助手
1. AI驱动数据分析：从传统BI到AI增强分析
1.1 传统数据分析的局限性
传统的商业智能（BI）工具，如 Tableau、Power BI、Qlik 等，虽然在数据可视化和报表方面表现出色，但在实际使用中存在明显的局限性：
高门槛：分析师需要掌握 SQL、数据建模、可视化设计等多项技能。一个简单的"上个月华东区销售额同比下降了多少"的问题，可能需要编写复杂的 SQL 查询、创建数据模型、再手动制作图表。
响应慢：从业务部门提出需求，到分析师完成分析，往往需要数天甚至数周时间。这种"需求-排期-开发-交付"的流程，严重制约了数据驱动决策的效率。
洞察被动：传统 BI 只能回答"你问了什么"，无法主动发现数据中的异常、趋势和机会。分析师需要凭经验和直觉去探索数据，容易遗漏重要信号。
1.2 AI增强分析的范式转变
AI增强分析（Augmented Analytics）是 Gartner 提出的概念，指的是利用机器学习、自然语言处理等 AI 技术，自动化数据准备、洞察发现和洞察共享的过程。
核心转变：
| 维度 | 传统BI | AI增强分析 |
|---|---|---|
| 交互方式 | 拖拽+SQL | 自然语言对话 |
| 分析模式 | 人工驱动 | AI辅助+人工验证 |
| 洞察发现 | 被动查询 | 主动发现 |
| 技能要求 | 高（SQL/可视化） | 低（自然语言即可） |
| 响应速度 | 天/周级 | 秒/分钟级 |
1.3 技术架构概览
一个典型的AI增强分析系统架构如下：
用户层：自然语言输入 → 语音/文本
↓
理解层：意图识别 → 实体抽取 → 上下文管理
↓
分析层：Text-to-SQL → 数据查询 → 统计分析 → 异常检测
↓
可视化层：图表推荐 → 自动渲染 → 交互式探索
↓
输出层：洞察摘要 → 数据报告 → 可视化图表
1.4 环境准备
在开始本教程的实践之前，请确保你的开发环境满足以下要求：
# Python 环境
python --version # >= 3.9
# 创建虚拟环境
python -m venv ai_data_analysis
source ai_data_analysis/bin/activate # Linux/Mac
# ai_data_analysis\Scripts\activate # Windows
# 核心依赖
pip install openai langchain pandas numpy matplotlib seaborn plotly
pip install pandasai sqlalchemy pymysql chromadb
pip install openpyxl xlrd pillow pytesseract
pip install gradio streamlit
2. 自然语言查询数据：Text-to-SQL技术
2.1 Text-to-SQL 基本原理
Text-to-SQL 是将用户的自然语言问题转换为 SQL 查询语句的技术。其核心流程包括：
- 语义理解：解析用户问题的意图和关键实体
- Schema 链接：将问题中的实体映射到数据库的表和字段
- SQL 生成：基于理解结果生成合法的 SQL 语句
- 执行与验证：执行 SQL 并验证结果的正确性
2.2 使用 OpenAI 实现 Text-to-SQL
下面是一个完整的 Text-to-SQL 实现示例：
import openai
import pandas as pd
from sqlalchemy import create_engine, inspect
class TextToSQLConverter:
"""自然语言转SQL查询器"""
def __init__(self, db_url: str, model: str = "gpt-4"):
self.engine = create_engine(db_url)
self.model = model
self.client = openai.OpenAI()
self.schema_info = self._get_schema_info()
def _get_schema_info(self) -> str:
"""提取数据库Schema信息"""
inspector = inspect(self.engine)
schema_parts = []
for table_name in inspector.get_table_names():
columns = inspector.get_columns(table_name)
col_descriptions = []
for col in columns:
col_type = str(col['type'])
nullable = "可空" if col.get('nullable', True) else "非空"
col_descriptions.append(
f" - {col['name']} ({col_type}, {nullable})"
)
# 获取外键关系
foreign_keys = inspector.get_foreign_keys(table_name)
fk_info = ""
if foreign_keys:
fk_info = "\n 外键: " + ", ".join(
[f"{fk['constrained_columns'][0]} -> {fk['referred_table']}.{fk['referred_columns'][0]}"
for fk in foreign_keys]
)
table_desc = f"表 {table_name}:\n" + "\n".join(col_descriptions) + fk_info
schema_parts.append(table_desc)
return "\n\n".join(schema_parts)
def _build_prompt(self, question: str) -> str:
"""构建Prompt"""
return f"""你是一个SQL专家。根据以下数据库结构和用户问题，生成正确的SQL查询语句。
数据库结构：
{self.schema_info}
规则：
1. 只返回SQL语句，不要解释
2. 使用标准SQL语法（兼容MySQL）
3. 注意处理NULL值
4. 如果问题模糊，选择最合理的解释
5. 对于聚合查询，确保GROUP BY正确
6. 日期字段注意格式兼容
用户问题：{question}
SQL:"""
def convert(self, question: str) -> str:
"""将自然语言转换为SQL"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是SQL专家，只输出SQL语句。"},
{"role": "user", "content": self._build_prompt(question)}
],
temperature=0
)
sql = response.choices[0].message.content.strip()
# 清理可能的markdown标记
if sql.startswith("```"):
sql = sql.split("\n", 1)[1].rsplit("```", 1)[0].strip()
return sql
def query(self, question: str) -> pd.DataFrame:
"""执行自然语言查询"""
sql = self.convert(question)
print(f"生成的SQL: {sql}")
return pd.read_sql(sql, self.engine)
# 使用示例
converter = TextToSQLConverter("mysql+pymysql://user:pass@localhost/sales_db")
result = converter.query("上个月销售额最高的前5个产品是什么？")
print(result)
2.3 增强 Text-to-SQL 的准确性
在实际应用中，纯 Schema 信息往往不够，还需要补充业务上下文：
class EnhancedTextToSQL(TextToSQLConverter):
"""增强版Text-to-SQL，支持业务上下文"""
def __init__(self, db_url: str, business_context: dict = None):
super().__init__(db_url)
self.business_context = business_context or {}
def _build_prompt(self, question: str) -> str:
"""增强Prompt，加入业务上下文"""
base_prompt = super()._build_prompt(question)
context_additions = []
# 添加同义词映射
if 'synonyms' in self.business_context:
synonyms = "\n".join([
f" - \"{k}\" 对应字段 {v}"
for k, v in self.business_context['synonyms'].items()
])
context_additions.append(f"业务术语映射:\n{synonyms}")
# 添加常用计算公式
if 'formulas' in self.business_context:
formulas = "\n".join([
f" - {k}: {v}"
for k, v in self.business_context['formulas'].items()
])
context_additions.append(f"计算公式:\n{formulas}")
# 添加示例
if 'examples' in self.business_context:
examples = "\n".join([
f" 问题: {ex['q']}\n SQL: {ex['s']}"
for ex in self.business_context['examples']
])
context_additions.append(f"参考示例:\n{examples}")
if context_additions:
return base_prompt + "\n\n" + "\n\n".join(context_additions)
return base_prompt
# 使用增强版
context = {
"synonyms": {
"营收": "orders.total_amount",
"客单价": "orders.total_amount / COUNT(DISTINCT orders.customer_id)",
"华东区": "region.name = '华东'",
"活跃用户": "DATEDIFF(NOW(), users.last_login) <= 30"
},
"formulas": {
"毛利率": "(revenue - cost) / revenue * 100",
"环比增长率": "(本期 - 上期) / 上期 * 100"
},
"examples": [
{
"q": "本月总营收是多少？",
"s": "SELECT SUM(total_amount) as revenue FROM orders WHERE MONTH(order_date) = MONTH(NOW()) AND YEAR(order_date) = YEAR(NOW())"
}
]
}
enhanced = EnhancedTextToSQL("mysql+pymysql://user:pass@localhost/sales_db", context)
result = enhanced.query("本月华东区的毛利率是多少？")
2.4 多表关联查询处理
复杂业务问题通常涉及多表关联，下面是一个处理多表查询的策略：
class MultiTableTextToSQL(EnhancedTextToSQL):
"""支持复杂多表关联的Text-to-SQL"""
def _build_prompt(self, question: str) -> str:
"""构建支持多表关联的Prompt"""
prompt = f"""你是一个资深SQL专家。根据以下信息生成SQL查询。
{self.schema_info}
关联路径提示：
- orders -> customers: orders.customer_id = customers.id
- orders -> products: order_items.order_id = orders.id, order_items.product_id = products.id
- orders -> regions: customers.region_id = regions.id
- products -> categories: products.category_id = categories.id
分析步骤：
1. 确定需要查询的核心实体
2. 确定需要关联的表和关联条件
3. 确定筛选条件、聚合方式和排序
4. 生成SQL
用户问题：{question}
SQL:"""
return prompt
3. LLM辅助数据清洗与预处理
3.1 数据质量问题
在实际数据分析中，数据质量问题是最常见的挑战：
- 缺失值：某些字段为空或缺失
- 格式不一致：日期格式混乱、地址写法不同
- 异常值：明显超出合理范围的数据
- 重复数据：重复记录导致统计偏差
- 编码问题：字符编码不统一
3.2 LLM 驱动的数据清洗
传统数据清洗需要编写大量规则代码，而 LLM 可以智能地识别和修复数据问题：
import pandas as pd
import json
from openai import OpenAI
class LLMDataCleaner:
"""基于LLM的智能数据清洗器"""
def __init__(self, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
def analyze_quality(self, df: pd.DataFrame) -> dict:
"""分析数据质量问题"""
# 准备数据样本
sample = df.head(20).to_string()
dtypes = df.dtypes.to_string()
null_counts = df.isnull().sum().to_string()
prompt = f"""分析以下数据的质量问题，返回JSON格式的分析结果。
数据类型：
{dtypes}
缺失值统计：
{null_counts}
数据样本（前20行）：
{sample}
请返回JSON格式：
{{
"issues": [
{{
"column": "列名",
"issue_type": "missing|format|outlier|duplicate|encoding",
"description": "问题描述",
"severity": "high|medium|low",
"suggestion": "修复建议"
}}
],
"overall_quality_score": 0-100,
"priority_fixes": ["优先修复项"]
}}"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def clean_column_values(self, series: pd.Series, column_name: str,
rules: str = None) -> pd.Series:
"""使用LLM清洗列值"""
unique_vals = series.dropna().unique()[:50] # 取前50个唯一值
prompt = f"""清洗以下数据列的值，返回JSON映射关系。
列名：{column_name}
唯一值：{json.dumps(list(unique_vals), ensure_ascii=False)}
额外规则：{rules or '无'}
返回格式：{{"原始值": "清洗后的值"}}
注意：
1. 统一格式（日期、数字、文本）
2. 修正明显错误
3. 合并同义值
4. 保留NULL值不处理"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"}
)
mapping = json.loads(response.choices[0].message.content)
return series.map(mapping).fillna(series)
def infer_column_type(self, series: pd.Series, column_name: str) -> str:
"""推断列的最佳数据类型"""
sample = series.dropna().head(20).tolist()
prompt = f"""根据以下数据样本，推断该列的最佳数据类型。
列名：{column_name}
样本值：{json.dumps(sample, ensure_ascii=False)}
返回以下之一：datetime, numeric, categorical, text, boolean, email, phone, url, address"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return response.choices[0].message.content.strip()
def auto_clean(self, df: pd.DataFrame) -> pd.DataFrame:
"""全自动数据清洗"""
df = df.copy()
# 第一步：分析质量问题
analysis = self.analyze_quality(df)
print(f"数据质量评分: {analysis['overall_quality_score']}/100")
for issue in analysis['issues']:
col = issue['column']
if col not in df.columns:
continue
print(f"处理: {col} - {issue['description']}")
if issue['issue_type'] == 'format':
# 使用LLM修正格式
df[col] = self.clean_column_values(df[col], col, issue['suggestion'])
elif issue['issue_type'] == 'outlier':
# 标记异常值
if pd.api.types.is_numeric_dtype(df[col]):
Q1 = df[col].quantile(0.25)
Q3 = df[col].quantile(0.75)
IQR = Q3 - Q1
mask = (df[col] < Q1 - 3*IQR) | (df[col] > Q3 + 3*IQR)
df.loc[mask, col] = None
print(f" 标记了 {mask.sum()} 个异常值为NULL")
return df
# 使用示例
cleaner = LLMDataCleaner()
df = pd.read_csv("messy_data.csv")
cleaned_df = cleaner.auto_clean(df)
print(cleaned_df.info())
3.3 缺失值智能填充
class SmartImputer:
"""基于LLM的智能缺失值填充"""
def __init__(self, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
def suggest_strategy(self, df: pd.DataFrame, column: str) -> dict:
"""让LLM推荐最佳填充策略"""
stats = {
"dtype": str(df[column].dtype),
"missing_count": int(df[column].isnull().sum()),
"missing_ratio": f"{df[column].isnull().mean()*100:.1f}%",
"sample_values": df[column].dropna().head(10).tolist(),
"other_columns": [c for c in df.columns if c != column]
}
prompt = f"""为以下数据列推荐缺失值填充策略。
列信息：{json.dumps(stats, ensure_ascii=False)}
返回JSON：
{{
"strategy": "mean|median|mode|forward_fill|interpolate|knn|llm_infer",
"reason": "选择原因",
"alternative": "备选策略"
}}"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def impute(self, df: pd.DataFrame, column: str, strategy: str = None) -> pd.DataFrame:
"""执行填充"""
df = df.copy()
if strategy is None:
suggestion = self.suggest_strategy(df, column)
strategy = suggestion['strategy']
print(f"选择策略: {strategy} - {suggestion['reason']}")
if strategy == 'mean':
df[column].fillna(df[column].mean(), inplace=True)
elif strategy == 'median':
df[column].fillna(df[column].median(), inplace=True)
elif strategy == 'mode':
df[column].fillna(df[column].mode()[0], inplace=True)
elif strategy == 'forward_fill':
df[column].fillna(method='ffill', inplace=True)
elif strategy == 'interpolate':
df[column].interpolate(inplace=True)
return df
4. 智能数据探索：自动统计、异常检测、趋势分析
4.1 自动化探索性数据分析（Auto-EDA）
传统 EDA 需要分析师手动编写大量代码，而 AI 可以自动完成这一过程：
import pandas as pd
import numpy as np
from openai import OpenAI
class AutoEDA:
"""自动化探索性数据分析"""
def __init__(self, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
def generate_profile(self, df: pd.DataFrame) -> dict:
"""生成数据概况"""
profile = {
"shape": df.shape,
"columns": list(df.columns),
"dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
"missing_values": df.isnull().sum().to_dict(),
"numeric_stats": {},
"categorical_stats": {}
}
# 数值列统计
numeric_cols = df.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
profile["numeric_stats"][col] = {
"mean": float(df[col].mean()),
"std": float(df[col].std()),
"min": float(df[col].min()),
"max": float(df[col].max()),
"median": float(df[col].median()),
"skewness": float(df[col].skew()),
"kurtosis": float(df[col].kurtosis())
}
# 分类列统计
cat_cols = df.select_dtypes(include=['object', 'category']).columns
for col in cat_cols:
profile["categorical_stats"][col] = {
"unique_count": int(df[col].nunique()),
"top_values": df[col].value_counts().head(5).to_dict()
}
return profile
def detect_anomalies(self, df: pd.DataFrame, method: str = "auto") -> pd.DataFrame:
"""智能异常检测"""
anomaly_mask = pd.Series(False, index=df.index)
anomaly_details = []
numeric_cols = df.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
series = df[col].dropna()
if method in ["auto", "iqr"]:
Q1 = series.quantile(0.25)
Q3 = series.quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - 1.5 * IQR
upper = Q3 + 1.5 * IQR
col_anomalies = (df[col] < lower) | (df[col] > upper)
if col_anomalies.any():
anomaly_mask |= col_anomalies
anomaly_details.append({
"column": col,
"method": "IQR",
"anomaly_count": int(col_anomalies.sum()),
"bounds": f"[{lower:.2f}, {upper:.2f}]"
})
if method in ["auto", "zscore"]:
z_scores = np.abs((series - series.mean()) / series.std())
col_anomalies = z_scores > 3
if col_anomalies.any():
anomaly_mask |= col_anomalies.reindex(df.index, fill_value=False)
anomaly_details.append({
"column": col,
"method": "Z-Score",
"anomaly_count": int(col_anomalies.sum()),
"threshold": 3
})
print(f"发现 {anomaly_mask.sum()} 条异常记录")
for detail in anomaly_details:
print(f" {detail['column']}: {detail['anomaly_count']} 条 ({detail['method']})")
return df[anomaly_mask]
def analyze_trends(self, df: pd.DataFrame, date_col: str,
value_col: str) -> dict:
"""趋势分析"""
df_sorted = df.sort_values(date_col)
# 基本趋势统计
values = df_sorted[value_col].values
trend_info = {
"start_value": float(values[0]),
"end_value": float(values[-1]),
"change": float(values[-1] - values[0]),
"change_pct": float((values[-1] - values[0]) / values[0] * 100) if values[0] != 0 else 0,
"max_value": float(np.max(values)),
"min_value": float(np.min(values)),
"volatility": float(np.std(values))
}
# 线性趋势拟合
x = np.arange(len(values))
coeffs = np.polyfit(x, values, 1)
trend_info["trend_slope"] = float(coeffs[0])
trend_info["trend_direction"] = "上升" if coeffs[0] > 0 else "下降" if coeffs[0] < 0 else "平稳"
return trend_info
def generate_insights(self, df: pd.DataFrame) -> str:
"""使用LLM生成数据洞察"""
profile = self.generate_profile(df)
prompt = f"""你是资深数据分析师。根据以下数据概况，生成关键洞察和建议。
数据概况：
{json.dumps(profile, ensure_ascii=False, indent=2)}
请提供：
1. 数据质量评估
2. 关键发现（3-5条）
3. 潜在的业务洞察
4. 建议的深入分析方向
用中文回答，语言要专业但易懂。"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content
# 使用示例
eda = AutoEDA()
df = pd.read_csv("sales_data.csv")
# 数据概况
profile = eda.generate_profile(df)
print(json.dumps(profile, ensure_ascii=False, indent=2))
# 异常检测
anomalies = eda.detect_anomalies(df)
print(anomalies)
# 趋势分析
trend = eda.analyze_trends(df, 'order_date', 'total_amount')
print(trend)
# 生成洞察
insights = eda.generate_insights(df)
print(insights)
4.2 智能相关性分析
def smart_correlation_analysis(df: pd.DataFrame, target_col: str = None) -> dict:
"""智能相关性分析"""
numeric_df = df.select_dtypes(include=[np.number])
corr_matrix = numeric_df.corr()
# 找出高相关性的特征对
high_corr_pairs = []
for i in range(len(corr_matrix.columns)):
for j in range(i+1, len(corr_matrix.columns)):
if abs(corr_matrix.iloc[i, j]) > 0.7:
high_corr_pairs.append({
"col1": corr_matrix.columns[i],
"col2": corr_matrix.columns[j],
"correlation": float(corr_matrix.iloc[i, j])
})
# 如果有目标变量，找出最相关的特征
target_correlations = {}
if target_col and target_col in numeric_df.columns:
target_corr = corr_matrix[target_col].drop(target_col).sort_values(ascending=False)
target_correlations = target_corr.to_dict()
return {
"correlation_matrix": corr_matrix.to_dict(),
"high_correlation_pairs": high_corr_pairs,
"target_correlations": target_correlations
}
5. AI驱动的图表推荐与自动可视化
5.1 智能图表推荐系统
不同类型的数据适合不同的图表类型。AI 可以根据数据特征自动推荐最合适的可视化方式：
from openai import OpenAI
import matplotlib.pyplot as plt
import seaborn as sns
import json
class SmartVisualizer:
"""AI驱动的智能可视化推荐引擎"""
def __init__(self, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
def recommend_chart(self, df: pd.DataFrame, analysis_goal: str = None) -> dict:
"""根据数据特征推荐图表类型"""
data_info = {
"columns": list(df.columns),
"dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
"row_count": len(df),
"numeric_columns": list(df.select_dtypes(include=[np.number]).columns),
"categorical_columns": list(df.select_dtypes(include=['object']).columns),
"sample": df.head(5).to_dict(orient='records')
}
prompt = f"""你是数据可视化专家。根据数据特征和分析目标，推荐最佳图表类型。
数据信息：{json.dumps(data_info, ensure_ascii=False)}
分析目标：{analysis_goal or '通用探索'}
返回JSON格式（最多3个推荐）：
{{
"recommendations": [
{{
"chart_type": "图表类型（如bar/line/scatter/pie/heatmap/box/histogram）",
"title": "图表标题",
"x_column": "X轴列名",
"y_column": "Y轴列名",
"color_column": "颜色分组列（可选）",
"reason": "推荐原因",
"code": "matplotlib/seaborn绘制代码（Python）"
}}
]
}}"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def auto_plot(self, df: pd.DataFrame, chart_type: str,
x_col: str, y_col: str = None, color_col: str = None,
title: str = "") -> plt.Figure:
"""自动绘制图表"""
fig, ax = plt.subplots(figsize=(12, 6))
if chart_type == 'bar':
if color_col:
grouped = df.groupby([x_col, color_col])[y_col].sum().unstack()
grouped.plot(kind='bar', ax=ax)
else:
df.groupby(x_col)[y_col].sum().plot(kind='bar', ax=ax)
elif chart_type == 'line':
if color_col:
for name, group in df.groupby(color_col):
ax.plot(group[x_col], group[y_col], label=name, marker='o')
ax.legend()
else:
ax.plot(df[x_col], df[y_col], marker='o')
elif chart_type == 'scatter':
scatter = ax.scatter(df[x_col], df[y_col],
c=df[color_col] if color_col else None,
alpha=0.6, cmap='viridis')
if color_col:
plt.colorbar(scatter, label=color_col)
elif chart_type == 'pie':
df.groupby(x_col)[y_col].sum().plot(kind='pie', ax=ax, autopct='%1.1f%%')
elif chart_type == 'heatmap':
numeric_df = df.select_dtypes(include=[np.number])
sns.heatmap(numeric_df.corr(), annot=True, cmap='coolwarm', ax=ax)
elif chart_type == 'box':
if color_col:
sns.boxplot(data=df, x=x_col, y=y_col, hue=color_col, ax=ax)
else:
sns.boxplot(data=df, x=x_col, y=y_col, ax=ax)
elif chart_type == 'histogram':
df[y_col or x_col].hist(bins=30, ax=ax, edgecolor='black')
ax.set_title(title or f"{chart_type} Chart", fontsize=14)
ax.set_xlabel(x_col)
if y_col:
ax.set_ylabel(y_col)
plt.tight_layout()
return fig
def visualize(self, df: pd.DataFrame, goal: str = None):
"""一键智能可视化"""
recommendations = self.recommend_chart(df, goal)
figures = []
for rec in recommendations['recommendations']:
print(f"推荐: {rec['chart_type']} - {rec['reason']}")
fig = self.auto_plot(
df,
rec['chart_type'],
rec['x_column'],
rec.get('y_column'),
rec.get('color_column'),
rec['title']
)
figures.append(fig)
return figures
# 使用示例
visualizer = SmartVisualizer()
df = pd.read_csv("sales_data.csv")
figures = visualizer.visualize(df, "分析各区域销售趋势和产品分布")
plt.show()
5.2 使用 Plotly 构建交互式图表
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
class InteractiveVisualizer:
"""交互式可视化引擎"""
@staticmethod
def create_dashboard(df: pd.DataFrame, config: dict) -> go.Figure:
"""创建交互式仪表板"""
charts = config.get('charts', [])
rows = (len(charts) + 1) // 2
fig = make_subplots(
rows=rows, cols=2,
subplot_titles=[c['title'] for c in charts]
)
for i, chart in enumerate(charts):
row = i // 2 + 1
col = i % 2 + 1
if chart['type'] == 'bar':
data = df.groupby(chart['x'])[chart['y']].sum().reset_index()
fig.add_trace(
go.Bar(x=data[chart['x']], y=data[chart['y']], name=chart['title']),
row=row, col=col
)
elif chart['type'] == 'line':
fig.add_trace(
go.Scatter(x=df[chart['x']], y=df[chart['y']],
mode='lines+markers', name=chart['title']),
row=row, col=col
)
elif chart['type'] == 'pie':
data = df.groupby(chart['x'])[chart['y']].sum().reset_index()
fig.add_trace(
go.Pie(labels=data[chart['x']], values=data[chart['y']], name=chart['title']),
row=row, col=col
)
fig.update_layout(height=400*rows, showlegend=True,
title_text=config.get('title', '数据仪表板'))
return fig
@staticmethod
def natural_language_chart(df: pd.DataFrame, description: str) -> go.Figure:
"""通过自然语言描述创建图表"""
client = OpenAI()
prompt = f"""根据描述生成Plotly图表代码。只返回Python代码，使用变量df。
列：{list(df.columns)}
数据类型：{df.dtypes.to_dict()}
描述：{description}
返回可执行的Python代码，结果变量名为fig。使用plotly.express或plotly.graph_objects。"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
code = response.choices[0].message.content
if "```python" in code:
code = code.split("```python")[1].split("```")[0]
local_vars = {"df": df, "px": px, "go": go}
exec(code, {}, local_vars)
return local_vars.get('fig')
6. Pandas AI与ChatBI实战
6.1 PandasAI 基础使用
PandasAI 是一个将 LLM 与 Pandas 结合的库，让你可以用自然语言操作 DataFrame：
# 安装：pip install pandasai
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
# 初始化
llm = OpenAI(api_token="your-api-key")
df = pd.read_csv("sales_data.csv")
# 创建智能DataFrame
sdf = SmartDataframe(df, config={"llm": llm})
# 自然语言查询
result = sdf.chat("销售额最高的5个产品是哪些？")
print(result)
# 复杂分析
result = sdf.chat("按月份统计销售额，计算环比增长率")
print(result)
# 可视化
fig = sdf.chat("画一个各区域销售额的饼图")
6.2 构建自定义 ChatBI 系统
import pandas as pd
from openai import OpenAI
import json
class ChatBI:
"""自定义ChatBI系统"""
def __init__(self, dataframes: dict, model: str = "gpt-4"):
"""
dataframes: {"表名": DataFrame} 的字典
"""
self.client = OpenAI()
self.model = model
self.dataframes = dataframes
self.history = []
def _get_context(self) -> str:
"""获取数据上下文"""
context_parts = []
for name, df in self.dataframes.items():
info = f"表 {name}:\n"
info += f" 行数: {len(df)}, 列数: {len(df.columns)}\n"
info += f" 列: {', '.join(df.columns)}\n"
info += f" 类型: {df.dtypes.to_dict()}\n"
info += f" 样本:\n{df.head(3).to_string()}\n"
context_parts.append(info)
return "\n".join(context_parts)
def chat(self, question: str) -> str:
"""对话式数据分析"""
context = self._get_context()
# 构建对话历史
messages = [
{"role": "system", "content": f"""你是ChatBI数据分析助手。根据用户的数据回答问题。
数据概况：
{context}
你可以：
1. 回答数据相关问题
2. 生成Python代码进行分析
3. 提供数据洞察和建议
4. 推荐可视化方式
回答要准确、简洁、有洞察力。涉及计算时，给出具体数字。"""},
]
# 添加历史对话
for h in self.history[-5:]: # 保留最近5轮
messages.append(h)
messages.append({"role": "user", "content": question})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.3
)
answer = response.choices[0].message.content
# 保存对话历史
self.history.append({"role": "user", "content": question})
self.history.append({"role": "assistant", "content": answer})
return answer
def execute_analysis(self, code: str) -> str:
"""执行分析代码"""
local_vars = {"pd": pd, "np": __import__('numpy')}
local_vars.update(self.dataframes)
try:
exec(code, {}, local_vars)
result = local_vars.get('result', '代码执行完成，无返回值')
return str(result)
except Exception as e:
return f"执行错误: {str(e)}"
# 使用示例
sales_df = pd.read_csv("sales.csv")
customer_df = pd.read_csv("customers.csv")
product_df = pd.read_csv("products.csv")
chatbi = ChatBI({
"sales": sales_df,
"customers": customer_df,
"products": product_df
})
# 对话式分析
print(chatbi.chat("本月销售额最高的客户是哪些？"))
print(chatbi.chat("这些客户的购买频次如何？"))
print(chatbi.chat("画一个销售额月度趋势图"))
6.3 ChatBI 的 Web 界面（Gradio）
import gradio as gr
def create_chatbi_app(chatbi: ChatBI):
"""创建ChatBI Web应用"""
def respond(message, history):
response = chatbi.chat(message)
return response
demo = gr.ChatInterface(
fn=respond,
title="📊 智能数据分析助手",
description="用自然语言提问，AI帮你分析数据",
examples=[
"本月销售额是多少？",
"哪些产品的销量在下降？",
"各区域的客户分布情况",
"画一个销售趋势图"
],
theme=gr.themes.Soft()
)
return demo
# 启动应用
app = create_chatbi_app(chatbi)
app.launch(server_name="0.0.0.0", server_port=7860)
7. 数据报告自动生成：LLM + 图表 + 洞察
7.1 自动化报告生成系统
import pandas as pd
import matplotlib.pyplot as plt
from openai import OpenAI
from datetime import datetime
import base64
import io
class AutoReportGenerator:
"""自动化数据报告生成器"""
def __init__(self, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
def _fig_to_base64(self, fig) -> str:
"""将图表转换为base64"""
buffer = io.BytesIO()
fig.savefig(buffer, format='png', dpi=150, bbox_inches='tight')
buffer.seek(0)
return base64.b64encode(buffer.read()).decode()
def _generate_kpi_cards(self, df: pd.DataFrame, metrics: dict) -> str:
"""生成KPI卡片HTML"""
cards_html = '<div style="display:flex;gap:20px;flex-wrap:wrap;margin:20px 0;">'
for name, value in metrics.items():
cards_html += f'''
<div style="background:#f8f9fa;padding:20px;border-radius:10px;flex:1;min-width:200px;text-align:center;">
<div style="font-size:14px;color:#666;">{name}</div>
<div style="font-size:28px;font-weight:bold;color:#1a73e8;">{value}</div>
</div>'''
cards_html += '</div>'
return cards_html
def _generate_chart_html(self, fig, title: str) -> str:
"""生成图表HTML"""
img_base64 = self._fig_to_base64(fig)
return f'''
<div style="margin:20px 0;text-align:center;">
<h3>{title}</h3>
<img src="data:image/png;base64,{img_base64}" style="max-width:100%;">
</div>'''
def generate_monthly_report(self, df: pd.DataFrame,
date_col: str,
value_col: str,
group_cols: list = None) -> str:
"""生成月度数据报告"""
# 计算KPI
total = df[value_col].sum()
avg = df[value_col].mean()
count = len(df)
metrics = {
"总销售额": f"¥{total:,.2f}",
"平均订单额": f"¥{avg:,.2f}",
"订单数量": f"{count:,}",
"数据行数": f"{len(df):,}"
}
# 生成图表
charts_html = ""
# 趋势图
fig1, ax1 = plt.subplots(figsize=(12, 5))
df.groupby(df[date_col].dt.to_period('M'))[value_col].sum().plot(ax=ax1, marker='o')
ax1.set_title('月度销售趋势')
ax1.set_xlabel('月份')
ax1.set_ylabel('销售额')
charts_html += self._generate_chart_html(fig1, '月度销售趋势')
plt.close(fig1)
# 分布图
if group_cols:
for col in group_cols[:2]:
fig2, ax2 = plt.subplots(figsize=(10, 5))
df.groupby(col)[value_col].sum().sort_values(ascending=False).head(10).plot(kind='bar', ax=ax2)
ax2.set_title(f'{col} 销售分布')
charts_html += self._generate_chart_html(fig2, f'{col} 销售分布')
plt.close(fig2)
# 使用LLM生成文字洞察
data_summary = {
"总销售额": float(total),
"平均值": float(avg),
"最大值": float(df[value_col].max()),
"最小值": float(df[value_col].min()),
"标准差": float(df[value_col].std())
}
insights_prompt = f"""你是资深数据分析师。根据以下数据，生成简明的分析洞察（3-5条）。
数据摘要：{json.dumps(data_summary, ensure_ascii=False)}
要求：
1. 每条洞察用一句话概括
2. 包含具体数字
3. 指出值得关注的趋势或异常
4. 给出可操作的建议"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": insights_prompt}],
temperature=0.3
)
insights = response.choices[0].message.content
# 组装报告HTML
report_html = f"""
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>数据报告</title>
<style>
body {{ font-family: 'Microsoft YaHei', sans-serif; max-width: 1200px; margin: 0 auto; padding: 20px; }}
h1 {{ color: #1a73e8; border-bottom: 3px solid #1a73e8; padding-bottom: 10px; }}
h2 {{ color: #333; margin-top: 30px; }}
.insights {{ background: #e8f5e9; padding: 20px; border-radius: 10px; margin: 20px 0; }}
.insights li {{ margin: 10px 0; line-height: 1.6; }}
.footer {{ text-align: center; color: #999; margin-top: 40px; padding-top: 20px; border-top: 1px solid #eee; }}
</style>
</head>
<body>
<h1>📊 数据分析报告</h1>
<p>生成时间：{datetime.now().strftime('%Y年%m月%d日 %H:%M')}</p>
<h2>📈 核心指标</h2>
{self._generate_kpi_cards(df, metrics)}
<h2>📊 可视化分析</h2>
{charts_html}
<h2>💡 关键洞察</h2>
<div class="insights">
<pre style="white-space: pre-wrap;">{insights}</pre>
</div>
<div class="footer">
<p>本报告由AI自动生成，数据截至 {datetime.now().strftime('%Y-%m-%d')}</p>
</div>
</body>
</html>
"""
return report_html
# 使用示例
generator = AutoReportGenerator()
df = pd.read_csv("sales_data.csv", parse_dates=['order_date'])
report = generator.generate_monthly_report(df, 'order_date', 'total_amount', ['product_category', 'region'])
with open("monthly_report.html", "w", encoding="utf-8") as f:
f.write(report)
print("报告已生成: monthly_report.html")
8. 多模态数据分析：图表理解、图像OCR
8.1 图表理解
多模态模型（如 GPT-4V）可以直接理解图表图像，提取其中的数据和洞察：
from openai import OpenAI
import base64
class ChartUnderstanding:
"""图表理解引擎"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
def analyze_chart_image(self, image_path: str, question: str = None) -> str:
"""分析图表图像"""
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode()
prompt = question or """请分析这张图表：
1. 识别图表类型
2. 提取关键数据点
3. 总结主要趋势和发现
4. 指出任何异常或值得关注的点"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{image_data}"
}
}
]
}
],
max_tokens=1000
)
return response.choices[0].message.content
def extract_data_from_chart(self, image_path: str) -> dict:
"""从图表中提取数据"""
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode()
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": """从这张图表中提取所有数据点，返回JSON格式：
{
"chart_type": "图表类型",
"title": "图表标题",
"x_label": "X轴标签",
"y_label": "Y轴标签",
"data_points": [
{"x": "x值", "y": y数值, "series": "系列名(如有)"}
]
}"""
},
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{image_data}"}
}
]
}
],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
# 使用示例
analyzer = ChartUnderstanding()
result = analyzer.analyze_chart_image("sales_chart.png", "这张图表显示了什么趋势？")
print(result)
data = analyzer.extract_data_from_chart("sales_chart.png")
print(json.dumps(data, ensure_ascii=False, indent=2))
8.2 OCR 数据提取
import pytesseract
from PIL import Image
class OCRDataExtractor:
"""OCR数据提取器"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
def extract_from_image(self, image_path: str) -> pd.DataFrame:
"""从图像中提取表格数据"""
# 方法1：使用传统OCR
image = Image.open(image_path)
ocr_text = pytesseract.image_to_string(image, lang='chi_sim+eng')
# 方法2：使用多模态模型理解表格
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode()
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": """识别这张图片中的表格数据，返回JSON格式：
{
"headers": ["列名1", "列名2", ...],
"rows": [
["值1", "值2", ...],
...
]
}"""
},
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{image_data}"}
}
]
}
],
response_format={"type": "json_object"}
)
data = json.loads(response.choices[0].message.content)
return pd.DataFrame(data['rows'], columns=data['headers'])
# 使用示例
extractor = OCRDataExtractor()
df = extractor.extract_from_image("table_screenshot.png")
print(df)
9. 数据分析Agent：自主完成复杂分析任务
9.1 构建数据分析Agent
数据分析Agent能够自主规划和执行复杂的分析任务：
from openai import OpenAI
import pandas as pd
import json
class DataAnalysisAgent:
"""自主数据分析Agent"""
def __init__(self, dataframes: dict, model: str = "gpt-4"):
self.client = OpenAI()
self.model = model
self.dataframes = dataframes
self.tools = self._define_tools()
self.messages = []
def _define_tools(self) -> list:
return [
{
"type": "function",
"function": {
"name": "execute_python",
"description": "执行Python数据分析代码。可用变量：pd, np, 以及所有已加载的DataFrame",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "要执行的Python代码"
}
},
"required": ["code"]
}
}
},
{
"type": "function",
"function": {
"name": "create_visualization",
"description": "创建数据可视化图表",
"parameters": {
"type": "object",
"properties": {
"chart_type": {"type": "string", "enum": ["bar", "line", "scatter", "pie", "heatmap", "box"]},
"data_description": {"type": "string", "description": "需要可视化的数据描述"},
"code": {"type": "string", "description": "生成图表的Python代码，结果保存为result变量"}
},
"required": ["chart_type", "code"]
}
}
},
{
"type": "function",
"function": {
"name": "generate_report",
"description": "生成分析报告",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string"},
"content": {"type": "string", "description": "Markdown格式的报告内容"}
},
"required": ["title", "content"]
}
}
}
]
def _get_data_context(self) -> str:
context = "可用数据：\n"
for name, df in self.dataframes.items():
context += f"\n表 {name}:\n"
context += f" 形状: {df.shape}\n"
context += f" 列: {list(df.columns)}\n"
context += f" 类型: {df.dtypes.to_dict()}\n"
context += f" 样本:\n{df.head(3).to_string()}\n"
return context
def _execute_python(self, code: str) -> str:
"""执行Python代码"""
local_vars = {"pd": pd, "np": __import__('numpy'), "json": json}
local_vars.update(self.dataframes)
try:
exec(code, {}, local_vars)
result = local_vars.get('result', '执行完成')
return str(result)
except Exception as e:
return f"错误: {str(e)}"
def analyze(self, task: str) -> str:
"""执行分析任务"""
self.messages = [
{
"role": "system",
"content": f"""你是高级数据分析Agent。你可以自主规划和执行数据分析任务。
{self._get_data_context()}
工作流程：
1. 理解分析任务
2. 制定分析计划
3. 逐步执行分析（使用工具）
4. 生成可视化
5. 总结发现和洞察
始终用中文回答。分析要深入、有洞察力。"""
},
{"role": "user", "content": task}
]
max_iterations = 10
for _ in range(max_iterations):
response = self.client.chat.completions.create(
model=self.model,
messages=self.messages,
tools=self.tools,
tool_choice="auto",
temperature=0.3
)
message = response.choices[0].message
self.messages.append(message)
# 如果没有工具调用，返回最终回答
if not message.tool_calls:
return message.content
# 处理工具调用
for tool_call in message.tool_calls:
func_name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
if func_name == "execute_python":
result = self._execute_python(args['code'])
elif func_name == "create_visualization":
result = self._execute_python(args['code'])
elif func_name == "generate_report":
with open(f"{args['title']}.md", 'w', encoding='utf-8') as f:
f.write(args['content'])
result = f"报告已保存: {args['title']}.md"
else:
result = "未知工具"
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
return "分析完成（达到最大迭代次数）"
# 使用示例
sales_df = pd.read_csv("sales.csv", parse_dates=['date'])
agent = DataAnalysisAgent({"sales": sales_df})
result = agent.analyze("""
请对销售数据进行全面分析：
1. 整体销售趋势如何？
2. 哪些产品类别表现最好？
3. 是否存在季节性规律？
4. 发现任何异常或值得关注的模式
5. 生成可视化图表和分析报告
""")
print(result)
10. 数据隐私与安全：脱敏、权限控制
10.1 数据脱敏
在使用 LLM 处理数据时，数据隐私至关重要：
import re
import hashlib
from typing import Any
class DataMasker:
"""数据脱敏工具"""
@staticmethod
def mask_phone(phone: str) -> str:
"""手机号脱敏：138****5678"""
if not phone or len(str(phone)) < 7:
return phone
phone = str(phone)
return phone[:3] + "****" + phone[-4:]
@staticmethod
def mask_id_card(id_card: str) -> str:
"""身份证号脱敏：110***********1234"""
if not id_card or len(str(id_card)) < 8:
return id_card
id_card = str(id_card)
return id_card[:3] + "*" * (len(id_card) - 7) + id_card[-4:]
@staticmethod
def mask_email(email: str) -> str:
"""邮箱脱敏：t***@example.com"""
if not email or '@' not in email:
return email
local, domain = email.split('@', 1)
return local[0] + "***@" + domain
@staticmethod
def mask_name(name: str) -> str:
"""姓名脱敏：张*明"""
if not name or len(name) < 2:
return name
return name[0] + "*" * (len(name) - 2) + name[-1] if len(name) > 2 else name[0] + "*"
@staticmethod
def hash_value(value: str) -> str:
"""哈希化（不可逆）"""
return hashlib.sha256(str(value).encode()).hexdigest()[:12]
def mask_dataframe(self, df: pd.DataFrame, rules: dict) -> pd.DataFrame:
"""
批量脱敏DataFrame
rules: {"列名": "脱敏方法"}
方法: phone, id_card, email, name, hash, drop
"""
df = df.copy()
for col, method in rules.items():
if col not in df.columns:
continue
if method == 'phone':
df[col] = df[col].apply(self.mask_phone)
elif method == 'id_card':
df[col] = df[col].apply(self.mask_id_card)
elif method == 'email':
df[col] = df[col].apply(self.mask_email)
elif method == 'name':
df[col] = df[col].apply(self.mask_name)
elif method == 'hash':
df[col] = df[col].apply(self.hash_value)
elif method == 'drop':
df = df.drop(columns=[col])
return df
# 使用示例
masker = DataMasker()
# 单字段脱敏
print(masker.mask_phone("13812345678")) # 138****5678
print(masker.mask_id_card("110101199001011234")) # 110***********1234
print(masker.mask_email("zhangsan@example.com")) # z***@example.com
# DataFrame批量脱敏
df = pd.DataFrame({
'name': ['张三', '李四', '王五'],
'phone': ['13812345678', '13987654321', '15011112222'],
'email': ['zhang@test.com', 'li@test.com', 'wang@test.com'],
'amount': [1000, 2000, 3000]
})
masked_df = masker.mask_dataframe(df, {
'name': 'name',
'phone': 'phone',
'email': 'email'
})
print(masked_df)
10.2 安全的 LLM 数据查询
class SecureDataQuery:
"""安全的数据查询包装器"""
def __init__(self, df: pd.DataFrame, sensitive_columns: list):
self.df = df
self.sensitive_columns = sensitive_columns
self.masker = DataMasker()
def get_safe_sample(self, n: int = 10) -> pd.DataFrame:
"""获取脱敏后的数据样本（用于发送给LLM）"""
sample = self.df.head(n).copy()
# 自动脱敏敏感列
for col in self.sensitive_columns:
if col in sample.columns:
sample[col] = sample[col].apply(lambda x: "***")
return sample
def get_schema_only(self) -> dict:
"""只返回Schema信息，不包含实际数据"""
return {
"columns": list(self.df.columns),
"dtypes": {col: str(dtype) for col, dtype in self.df.dtypes.items()},
"row_count": len(self.df),
"numeric_ranges": {
col: {"min": float(self.df[col].min()), "max": float(self.df[col].max())}
for col in self.df.select_dtypes(include=[np.number]).columns
}
}
def safe_query(self, question: str, llm_client, model: str = "gpt-4") -> str:
"""安全的LLM查询（不泄露敏感数据）"""
# 只发送脱敏后的schema和样本
safe_info = {
"schema": self.get_schema_only(),
"sample": self.get_safe_sample(5).to_dict(orient='records')
}
prompt = f"""根据以下数据结构回答问题。注意：数据样本已脱敏处理。
数据信息：{json.dumps(safe_info, ensure_ascii=False)}
问题：{question}
请基于数据结构给出分析思路和SQL/Python代码。"""
response = llm_client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return response.choices[0].message.content
11. 实战项目：构建智能数据分析助手
11.1 项目概述
我们将构建一个完整的智能数据分析助手，整合前面所有技术：
功能特性：
- 自然语言数据查询（Text-to-SQL）
- 自动数据清洗和预处理
- 智能数据探索和异常检测
- 自动图表推荐和可视化
- 数据报告自动生成
- 数据脱敏和安全保护
11.2 完整项目代码
"""
智能数据分析助手 - 主程序
"""
import pandas as pd
import numpy as np
from openai import OpenAI
import json
import os
from datetime import datetime
import matplotlib.pyplot as plt
import gradio as gr
class SmartDataAssistant:
"""智能数据分析助手"""
def __init__(self, api_key: str = None):
self.client = OpenAI(api_key=api_key)
self.model = "gpt-4"
self.dataframes = {}
self.analysis_history = []
self.masker = DataMasker()
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
def load_data(self, file_path: str, name: str = None) -> str:
"""加载数据文件"""
ext = os.path.splitext(file_path)[1].lower()
name = name or os.path.splitext(os.path.basename(file_path))[0]
if ext == '.csv':
df = pd.read_csv(file_path)
elif ext in ['.xlsx', '.xls']:
df = pd.read_excel(file_path)
elif ext == '.json':
df = pd.read_json(file_path)
elif ext == '.parquet':
df = pd.read_parquet(file_path)
else:
return f"不支持的文件格式: {ext}"
self.dataframes[name] = df
return f"成功加载 {name}: {df.shape[0]} 行 × {df.shape[1]} 列"
def get_data_summary(self) -> str:
"""获取所有数据的摘要"""
summary = []
for name, df in self.dataframes.items():
info = f"📊 {name}: {df.shape[0]}行 × {df.shape[1]}列\n"
info += f" 列: {', '.join(df.columns)}\n"
info += f" 缺失值: {df.isnull().sum().sum()}\n"
summary.append(info)
return "\n".join(summary)
def chat(self, question: str) -> str:
"""自然语言对话"""
# 构建数据上下文
data_context = self.get_data_summary()
# 构建样本数据（脱敏后）
samples = {}
for name, df in self.dataframes.items():
samples[name] = df.head(3).to_dict(orient='records')
messages = [
{
"role": "system",
"content": f"""你是智能数据分析助手。你可以帮助用户分析数据、回答问题、生成可视化。
可用数据：
{data_context}
数据样本：{json.dumps(samples, ensure_ascii=False, default=str)}
你的能力：
1. 回答数据相关问题
2. 生成Python分析代码
3. 创建可视化图表
4. 生成分析报告
5. 进行数据清洗建议
回答要准确、有洞察力。如果需要执行代码来回答问题，请说明。"""
}
]
# 添加历史对话
for h in self.analysis_history[-10:]:
messages.append(h)
messages.append({"role": "user", "content": question})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.3
)
answer = response.choices[0].message.content
# 保存历史
self.analysis_history.append({"role": "user", "content": question})
self.analysis_history.append({"role": "assistant", "content": answer})
return answer
def auto_analyze(self, table_name: str = None) -> dict:
"""自动分析数据"""
if table_name:
df = self.dataframes[table_name]
else:
table_name = list(self.dataframes.keys())[0]
df = self.dataframes[table_name]
eda = AutoEDA()
results = {
"profile": eda.generate_profile(df),
"anomalies": eda.detect_anomalies(df),
"insights": eda.generate_insights(df)
}
# 数值列趋势分析
numeric_cols = df.select_dtypes(include=[np.number]).columns
if len(numeric_cols) > 0:
results["correlations"] = smart_correlation_analysis(df)
return results
def generate_report(self, title: str = "数据分析报告") -> str:
"""生成HTML报告"""
generator = AutoReportGenerator()
# 合并所有数据
all_data = []
for name, df in self.dataframes.items():
df_copy = df.copy()
df_copy['_source'] = name
all_data.append(df_copy)
combined_df = pd.concat(all_data, ignore_index=True)
# 找到日期列和数值列
date_cols = combined_df.select_dtypes(include=['datetime64']).columns
numeric_cols = combined_df.select_dtypes(include=[np.number]).columns
date_col = date_cols[0] if len(date_cols) > 0 else None
value_col = numeric_cols[0] if len(numeric_cols) > 0 else None
if date_col and value_col:
report = generator.generate_monthly_report(
combined_df, date_col, value_col,
['_source']
)
else:
report = f"<h1>{title}</h1><p>数据格式不支持自动生成报告</p>"
# 保存报告
report_path = f"{title}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"
with open(report_path, 'w', encoding='utf-8') as f:
f.write(report)
return report_path
def create_web_app(self):
"""创建Web应用"""
def upload_file(file):
if file:
result = self.load_data(file.name)
return result
return "请选择文件"
def chat_response(message, history):
response = self.chat(message)
return response
def analyze_data():
results = self.auto_analyze()
return json.dumps(results, ensure_ascii=False, indent=2, default=str)
def gen_report():
path = self.generate_report()
return f"报告已生成: {path}"
with gr.Blocks(title="智能数据分析助手", theme=gr.themes.Soft()) as demo:
gr.Markdown("# 📊 智能数据分析助手")
gr.Markdown("上传数据文件，用自然语言进行数据分析")
with gr.Tab("📤 数据上传"):
file_input = gr.File(label="上传数据文件")
upload_btn = gr.Button("加载数据")
upload_output = gr.Textbox(label="加载结果")
upload_btn.click(upload_file, inputs=file_input, outputs=upload_output)
with gr.Tab("💬 数据对话"):
chatbot = gr.ChatInterface(
fn=chat_response,
title="数据分析师",
examples=[
"数据的基本统计信息是什么？",
"哪些列有缺失值？",
"销售额的趋势如何？",
"帮我分析异常数据"
]
)
with gr.Tab("📊 自动分析"):
analyze_btn = gr.Button("开始自动分析")
analyze_output = gr.Textbox(label="分析结果", lines=20)
analyze_btn.click(analyze_data, outputs=analyze_output)
with gr.Tab("📄 生成报告"):
report_btn = gr.Button("生成报告")
report_output = gr.Textbox(label="报告路径")
report_btn.click(gen_report, outputs=report_output)
return demo
# 启动应用
if __name__ == "__main__":
assistant = SmartDataAssistant()
# 加载示例数据
# assistant.load_data("sales_data.csv")
# 创建并启动Web应用
app = assistant.create_web_app()
app.launch(server_name="0.0.0.0", server_port=7860)
11.3 部署和运行
# 1. 准备环境
pip install -r requirements.txt
# 2. 设置环境变量
export OPENAI_API_KEY="your-api-key"
# 3. 启动应用
python app.py
# 4. 访问 http://localhost:7860
11.4 项目扩展建议
- 接入真实数据库：支持 MySQL、PostgreSQL、MongoDB 等
- 多用户支持：添加用户认证和数据隔离
- 定时报告：支持 Cron 定时生成报告并发送邮件
- 更多可视化：接入 ECharts、D3.js 等更丰富的图表库
- 协作功能：支持多人协作分析，共享分析结果
- API 接口：提供 RESTful API 供其他系统调用
总结
本教程全面介绍了 AI 驱动数据分析与可视化的核心技术和实践方法：
- Text-to-SQL：让非技术人员也能用自然语言查询数据
- LLM 数据清洗：智能识别和修复数据质量问题
- 自动数据探索：AI 主动发现数据中的模式和异常
- 智能可视化：根据数据特征自动推荐最佳图表
- ChatBI：构建对话式数据分析体验
- 自动报告：一键生成专业的数据报告
- 多模态分析：理解图表图像，提取数据
- 数据分析Agent：自主完成复杂分析任务
- 数据安全：脱敏和权限控制
AI 不会取代数据分析师，但会大幅提升分析效率。掌握这些技术，你将能够构建更智能、更高效的数据分析系统。
本教程代码示例仅供学习参考，实际使用时请根据具体需求调整和优化。
