AI驱动的数据分析：从自然语言查询到自动洞察
数据分析的新范式 传统数据分析需要SQL技能和BI工具操作经验。AI驱动的数据分析让任何人都能用自然语言探索数据——“上个月哪个产品线的增长率最高？“这样的问题可以直接转化为SQL查询并返回可视化结果。 Text-to-SQL技术 架构设计 class TextToSQLEngine: def __init__(self, llm, schema_extractor): self.llm = llm self.schema = schema_extractor def query(self, natural_language, database): """自然语言转SQL""" # 1. 提取数据库schema schema_info = self.schema.extract(database) # 包含：表结构、字段说明、外键关系、示例数据 # 2. 意图理解 intent = self._understand_intent(natural_language) # 聚合/过滤/排序/连接/窗口函数 # 3. 生成SQL sql = self.llm.generate(f""" 数据库Schema： {schema_info} 用户问题：{natural_language} 意图分析：{intent} 生成PostgreSQL查询。要求： 1. 只使用SELECT语句 2. 使用表别名提高可读性 3. 添加LIMIT防止全表扫描 4. 处理NULL值 5. 使用COALESCE处理空值 """) # 4. SQL验证和优化 validated_sql = self._validate_and_optimize(sql, database) return validated_sql Schema感知 class SchemaExtractor: def extract(self, database): """提取数据库schema信息""" schema = { "tables": {}, "relationships": [], "sample_data": {}, "statistics": {} } for table in database.get_tables(): schema["tables"][table.name] = { "columns": [], "row_count": table.row_count(), "description": table.comment or "" } for col in table.columns: schema["tables"][table.name]["columns"].append({ "name": col.name, "type": col.type, "nullable": col.nullable, "description": col.comment or "", "sample_values": table.sample_values(col.name, n=5) }) # 外键关系 schema["relationships"] = database.get_foreign_keys() return schema 多轮查询优化 class MultiTurnQueryOptimizer: def __init__(self): self.query_history = [] def refine_query(self, current_question, previous_results): """基于历史查询优化当前查询""" context = "" for q, r in self.query_history[-3:]: # 最近3轮 context += f"之前问过：{q}\n结果摘要：{r}\n\n" refined = self.llm.generate(f""" 之前的对话历史： {context} 当前问题：{current_question} 如果当前问题与之前相关，生成增量查询。 如果是全新问题，生成独立查询。 如果需要对比，引用之前的结果。 """) return refined 自动洞察发现 异常检测 class InsightDetector: def __init__(self, llm): self.llm = llm def detect_insights(self, data, dimensions, metrics): """自动发现数据中的洞察""" insights = [] # 1. 趋势检测 trends = self._detect_trends(data, dimensions["time"], metrics) insights.extend(trends) # 2. 异常检测 anomalies = self._detect_anomalies(data, metrics) insights.extend(anomalies) # 3. 相关性发现 correlations = self._find_correlations(data, metrics) insights.extend(correlations) # 4. 分群发现 segments = self._discover_segments(data, dimensions) insights.extend(segments) # 5. AI解读 for insight in insights: insight["explanation"] = self._explain(insight) return insights def _detect_anomalies(self, data, metrics): """统计异常检测""" anomalies = [] for metric in metrics: values = data[metric] mean, std = values.mean(), values.std() # Z-score异常 z_scores = (values - mean) / std outliers = data[abs(z_scores) > 3] if len(outliers) > 0: anomalies.append({ "type": "anomaly", "metric": metric, "severity": "high" if any(abs(z) > 5) else "medium", "details": f"发现{len(outliers)}个异常点", "data": outliers }) return anomalies def _explain(self, insight): """AI解释洞察""" return self.llm.generate(f""" 用简洁的语言解释以下数据发现： 发现类型：{insight['type']} 涉及指标：{insight.get('metric', 'N/A')} 详情：{insight['details']} 数据：{insight.get('data', 'N/A')} 解释要求： 1. 一句话说清楚发现了什么 2. 可能的原因分析 3. 建议的进一步分析方向 """) 自动可视化 class AutoVisualizer: def visualize(self, data, question): """根据问题和数据自动选择最佳可视化方式""" chart_type = self.llm.generate(f""" 用户问题：{question} 数据特征： - 列：{list(data.columns)} - 行数：{len(data)} - 数据类型：{data.dtypes.to_dict()} 选择最适合的可视化类型： 1. 折线图（时间趋势） 2. 柱状图（分类比较） 3. 散点图（相关性） 4. 饼图（占比） 5. 热力图（多维交叉） 6. 箱线图（分布） 返回JSON：{{"chart_type": "...", "x": "...", "y": "...", "color": "..."}} """) config = json.loads(chart_type) return self._render(data, config) 数据故事化 class DataStoryteller: def narrate(self, data, insights, audience="executive"): """将数据分析结果转化为叙事""" story = self.llm.generate(f""" 基于以下数据发现，写一份数据分析报告。 目标受众：{audience} 关键发现： {json.dumps(insights, ensure_ascii=False, indent=2)} 数据摘要： {data.describe().to_string()} 报告结构： 1. 执行摘要（3句话概括最重要的发现） 2. 详细分析（每个发现的深入解读） 3. 异常与风险（需要关注的问题） 4. 机会与建议（可执行的行动建议） 5. 下一步分析方向 语言要求： - {audience}级别的语言（避免/使用技术术语） - 用数据说话（引用具体数字） - 结论先行（每个段落先给结论再给依据） """) return story 企业实践架构 class EnterpriseDataAgent: def __init__(self, llm, database, data_warehouse): self.llm = llm self.db = database self.dw = data_warehouse self.sql_engine = TextToSQLEngine(llm, SchemaExtractor()) self.insight_detector = InsightDetector(llm) self.visualizer = AutoVisualizer() self.storyteller = DataStoryteller() def analyze(self, question): """端到端数据分析""" # 1. 理解问题 analysis_plan = self._plan_analysis(question) # 2. 数据获取 data = self._fetch_data(analysis_plan) # 3. 自动分析 insights = self.insight_detector.detect_insights( data, analysis_plan["dimensions"], analysis_plan["metrics"] ) # 4. 可视化 charts = [self.visualizer.visualize(data, question)] # 5. 叙事 narrative = self.storyteller.narrate(data, insights) return { "question": question, "sql": analysis_plan["sql"], "data": data, "insights": insights, "charts": charts, "report": narrative } 效果评估 Text-to-SQL准确率 在Spider基准上： ...