在企业数据分析场景中业务人员经常面临这样的困境想要快速获取销售趋势、用户行为等关键指标却因为不懂SQL查询、不熟悉Pandas数据处理、不会数据可视化而不得不依赖技术团队排期。这种技术壁垒导致数据分析链路冗长、迭代缓慢严重影响业务决策效率。本文将完整实现一个智能数据分析Agent项目通过自然语言交互让业务人员直接说话就能完成专业级数据分析。项目采用双端架构设计既支持客户端轻量化本地文件分析也支持云端生产级数据库联动所有代码均可直接运行附带完整架构图和实战示例。1. 智能数据分析Agent核心价值与架构设计智能数据分析Agent的核心价值在于消除自然语言与结构化数据之间的技术鸿沟。传统数据分析需要业务人员将需求转化为技术人员能理解的技术语言再由技术人员编写SQL或Pandas代码执行分析最后将结果反馈给业务人员。这个过程中存在多次信息转换损耗且响应周期长。1.1 核心业务需求拆解通过分析企业常见数据分析场景我们将用户的自然语言需求归纳为三类标准化任务数值统计类需求包括求和、平均值、最大值、最小值、分组统计、占比计算、趋势对比等。例如计算各区域平均销售额、统计月度销售增长率等。数据查询类需求涉及条件筛选、精准匹配、多维度关联、异常数据筛查等。例如查询华东地区销售额超过100万的订单、找出近一周登录异常的用户等。可视化报表类需求需要生成折线图、柱状图、饼图、分布图等可视化结果。例如绘制月度销售趋势图、生成产品销量占比饼图等。1.2 双端架构设计差异客户端Agent设计专注于本地CSV/Excel文件分析基于Pandas进行轻量化数据运算使用Matplotlib或Plotly进行本地绘图。优势是离线可用、部署简单、响应快速适合个人数据分析和快速业务复盘。云端Agent设计支持MySQL、PostgreSQL等数据库连接自动生成优化后的SQL查询语句具备安全沙箱执行环境支持批量数据分析、云端图表渲染和报告导出。适合企业级常态化数据分析需求。1.3 完整业务流程设计智能数据分析Agent的完整业务流程包含七个核心环节自然语言输入用户通过文本或语音输入分析需求需求解析分类Agent识别需求类型和关键参数数据源判定自动识别使用本地文件还是数据库代码生成根据数据源类型生成Pandas代码或SQL语句安全执行在安全环境中执行生成的分析代码结果可视化自动选择合适图表类型呈现结果报告生成将分析结果转化为自然语言业务报告2. 环境准备与依赖配置实现智能数据分析Agent需要准备相应的开发环境和依赖包。本节将详细介绍环境配置步骤确保后续代码能够正常运行。2.1 基础环境要求Python版本3.8及以上版本操作系统Windows 10/11、macOS 10.15、Ubuntu 18.04内存要求至少8GB RAM处理大型数据集时建议16GB存储空间至少2GB可用空间2.2 核心依赖包安装创建requirements.txt文件包含项目所需的所有依赖pandas1.5.0 pandasai3.0.0 openai1.0.0 langchain-openai0.0.5 mysql-connector-python8.0.0 plotly5.10.0 jupyter1.0.0 python-dotenv0.19.0使用pip安装依赖包# 创建虚拟环境可选但推荐 python -m venv data_agent_env source data_agent_env/bin/activate # Linux/macOS data_agent_env\Scripts\activate # Windows # 安装依赖 pip install -r requirements.txt2.3 API密钥配置为安全管理API密钥创建.env配置文件# 创建.env文件 OPENAI_API_KEYyour_openai_api_key_here DATABASE_URLmysql://user:passwordlocalhost:3306/sales_db在代码中安全读取配置import os from dotenv import load_dotenv load_dotenv() openai_api_key os.getenv(OPENAI_API_KEY) database_url os.getenv(DATABASE_URL)2.4 测试环境验证创建测试脚本验证环境配置是否正确# test_environment.py import pandas as pd import sys def check_environment(): 检查环境配置是否完整 print(Python版本:, sys.version) print(Pandas版本:, pd.__version__) try: import pandasai print(PandasAI版本:, pandasai.__version__) except ImportError: print(PandasAI未正确安装) try: from dotenv import load_dotenv load_dotenv() api_key os.getenv(OPENAI_API_KEY) if api_key: print(API密钥配置: 正常) else: print(API密钥配置: 未找到) except Exception as e: print(f配置检查异常: {e}) if __name__ __main__: check_environment()3. 数据预处理与Schema映射策略原始数据往往存在缺失值、异常值、格式不一致等问题直接使用会导致分析结果不准确。同时业务人员的自然语言描述与数据表的技术字段名之间存在语义鸿沟需要通过Schema映射建立关联。3.1 标准化数据预处理流程数据预处理是保障分析质量的关键步骤我们制定统一的清洗规则缺失值处理策略数值型字段使用均值填充分类字段使用众数填充时间序列数据使用前后值插值关键字段缺失需要标记告警。异常值检测方法使用3σ原则三倍标准差或IQR四分位距方法识别异常值根据业务场景决定剔除或修正。数据去重规则基于所有字段完全匹配去重保留第一条记录同时记录去重数量用于质量报告。格式标准化统一日期格式为YYYY-MM-DD数值字段去除千分位分隔符文本字段去除首尾空格。3.2 完整数据预处理代码实现import pandas as pd import numpy as np from datetime import datetime class DataPreprocessor: 数据预处理工具类 def __init__(self, file_pathNone, dfNone): 初始化预处理器 Args: file_path: 数据文件路径 df: 直接传入DataFrame if df is not None: self.df df.copy() elif file_path: self.df self.load_data(file_path) else: raise ValueError(必须提供文件路径或DataFrame) def load_data(self, file_path): 加载数据文件支持CSV和Excel格式 if file_path.endswith(.csv): return pd.read_csv(file_path) elif file_path.endswith((.xlsx, .xls)): return pd.read_excel(file_path) else: raise ValueError(不支持的文件格式仅支持CSV和Excel) def handle_missing_values(self, strategyauto): 处理缺失值 print(开始处理缺失值...) missing_info self.df.isnull().sum() print(f缺失值统计:\n{missing_info[missing_info 0]}) for column in self.df.columns: if self.df[column].isnull().sum() 0: if strategy auto: # 自动判断处理策略 if self.df[column].dtype in [int64, float64]: fill_value self.df[column].mean() else: fill_value self.df[column].mode()[0] if len(self.df[column].mode()) 0 else 未知 else: fill_value strategy self.df[column].fillna(fill_value, inplaceTrue) print(缺失值处理完成) return self def remove_duplicates(self): 去除重复数据 initial_count len(self.df) self.df self.df.drop_duplicates() final_count len(self.df) print(f去重完成移除{initial_count - final_count}条重复记录) return self def standardize_formats(self): 标准化数据格式 print(开始标准化数据格式...) # 日期格式标准化 date_columns [col for col in self.df.columns if date in col.lower() or time in col.lower()] for col in date_columns: try: self.df[col] pd.to_datetime(self.df[col]) print(f已标准化日期列: {col}) except: print(f无法转换日期列: {col}) # 数值格式标准化 numeric_columns self.df.select_dtypes(include[np.number]).columns for col in numeric_columns: self.df[col] pd.to_numeric(self.df[col], errorscoerce) # 文本格式标准化 text_columns self.df.select_dtypes(include[object]).columns for col in text_columns: self.df[col] self.df[col].astype(str).str.strip() print(数据格式标准化完成) return self def detect_outliers(self, methodiqr): 检测异常值 outliers_info {} numeric_columns self.df.select_dtypes(include[np.number]).columns for col in numeric_columns: if method iqr: Q1 self.df[col].quantile(0.25) Q3 self.df[col].quantile(0.75) IQR Q3 - Q1 lower_bound Q1 - 1.5 * IQR upper_bound Q3 1.5 * IQR outliers self.df[(self.df[col] lower_bound) | (self.df[col] upper_bound)] outliers_info[col] len(outliers) print(异常值检测完成:) for col, count in outliers_info.items(): if count 0: print(f {col}: {count}个异常值) return outliers_info def get_processed_data(self): 获取处理后的数据 return self.df # 使用示例 if __name__ __main__: # 创建示例数据 sample_data { sales_date: [2024-01-01, 2024-01-02, None, 2024-01-04], sales_amount: [1000, 1500, None, 2000], region: [华东, 华东, 华北, 华东], product: [产品A, 产品B, 产品A, 产品A] } df pd.DataFrame(sample_data) preprocessor DataPreprocessor(dfdf) processed_df (preprocessor .handle_missing_values() .remove_duplicates() .standardize_formats() .get_processed_data()) print(处理后的数据:) print(processed_df)3.3 Schema语义映射实现Schema映射是连接自然语言与数据字段的关键桥梁我们实现静态和动态两种映射策略class SchemaMapper: Schema语义映射管理器 def __init__(self, mapping_rulesNone): self.mapping_rules mapping_rules or {} self.dynamic_mappings {} def add_static_mapping(self, natural_language, technical_field): 添加静态映射规则 self.mapping_rules[natural_language] technical_field print(f添加映射: {natural_language} - {technical_field}) def auto_generate_mappings(self, df, llm_clientNone): 自动生成字段映射需要LLM支持 if llm_client is None: print(未提供LLM客户端使用基础映射生成) return self._generate_basic_mappings(df) # 使用LLM生成智能映射 return self._generate_llm_mappings(df, llm_client) def _generate_basic_mappings(self, df): 基础映射生成基于列名简单处理 basic_mappings {} for column in df.columns: # 简单的列名到中文映射 mapping { sales: 销售额, amount: 金额, date: 日期, region: 区域, product: 产品 } for eng, chn in mapping.items(): if eng in column.lower(): basic_mappings[chn] column break else: # 如果没有匹配到预设映射使用列名本身 basic_mappings[column] column self.mapping_rules.update(basic_mappings) return basic_mappings def map_query(self, natural_language_query): 将自然语言查询映射为技术字段查询 mapped_query natural_language_query # 按照映射长度从长到短排序避免部分匹配问题 sorted_mappings sorted(self.mapping_rules.items(), keylambda x: len(x[0]), reverseTrue) for natural_term, technical_term in sorted_mappings: if natural_term in mapped_query: mapped_query mapped_query.replace(natural_term, technical_term) return mapped_query def get_reverse_mapping(self): 获取反向映射技术字段到自然语言 return {v: k for k, v in self.mapping_rules.items()} # Schema映射实战示例 def demonstrate_schema_mapping(): 演示Schema映射的实际应用 # 示例数据字段 technical_columns [sale_amt, qty, trans_date, reg, prod_cat] # 创建映射器 mapper SchemaMapper() # 添加静态映射规则 mapper.add_static_mapping(销售额, sale_amt) mapper.add_static_mapping(销售数量, qty) mapper.add_static_mapping(交易日期, trans_date) mapper.add_static_mapping(区域, reg) mapper.add_static_mapping(产品类别, prod_cat) # 测试映射 test_queries [ 计算各区域销售额, 统计产品类别的销售数量, 按交易日期分析销售趋势 ] for query in test_queries: mapped mapper.map_query(query) print(f原始查询: {query}) print(f映射后: {mapped}) print(- * 50) if __name__ __main__: demonstrate_schema_mapping()3.4 双端映射策略差异客户端映射策略基于配置文件的静态映射加载速度快适合固定结构的本地文件分析。可以通过JSON配置文件管理映射规则{ sales_data.csv: { 销售额: sales_amount, 销售日期: sale_date, 区域: region, 产品名称: product_name }, user_behavior.csv: { 用户ID: user_id, 行为类型: action_type, 发生时间: timestamp } }云端映射策略动态生成映射规则通过LLM分析数据库表结构和注释自动建立语义关联。支持表结构变更时的自适应调整适合多数据源的企业环境。4. PandasAI核心实现与代码生成PandasAI是一个专门为智能数据分析设计的开源库它将大模型的自然语言理解能力与Pandas的数据处理能力相结合实现了真正的零代码数据分析。4.1 PandasAI架构原理PandasAI的核心工作原理基于以下几个关键组件SmartDataframe封装Pandas DataFrame增加自然语言交互能力LLM集成支持OpenAI、Azure OpenAI、Google Gemini等多种大模型代码生成器将自然语言转换为可执行的Pandas代码安全执行器在受限环境中执行生成的代码结果解析器将执行结果转换为友好格式4.2 客户端轻量化实现客户端版本适合个人用户和快速原型验证具备离线分析能力import pandas as pd from pandasai import SmartDataframe from pandasai_openai import OpenAI import warnings warnings.filterwarnings(ignore) class ClientDataAgent: 客户端数据分析Agent def __init__(self, api_key, data_source): 初始化客户端Agent Args: api_key: OpenAI API密钥 data_source: 数据文件路径或DataFrame self.llm OpenAI(api_keyapi_key) # 加载数据 if isinstance(data_source, str): self.df pd.read_csv(data_source) else: self.df data_source # 创建智能DataFrame self.smart_df SmartDataframe(self.df, config{llm: self.llm}) # 查询历史记录 self.query_history [] def preprocess_data(self): 数据预处理 preprocessor DataPreprocessor(dfself.df) self.df (preprocessor .handle_missing_values() .remove_duplicates() .standardize_formats() .get_processed_data()) # 重新创建SmartDataframe self.smart_df SmartDataframe(self.df, config{llm: self.llm}) print(数据预处理完成) def analyze(self, query, visualizeTrue): 执行数据分析 Args: query: 自然语言查询 visualize: 是否生成可视化 Returns: 分析结果 print(f执行分析: {query}) try: # 执行分析 result self.smart_df.chat(query) # 记录查询历史 self.query_history.append({ query: query, timestamp: pd.Timestamp.now(), result_type: type(result).__name__ }) # 如果结果是图表保存图片 if visualize and hasattr(result, savefig): filename fchart_{len(self.query_history)}.png result.savefig(filename) print(f图表已保存: {filename}) return result except Exception as e: error_msg f分析执行失败: {str(e)} print(error_msg) return error_msg def batch_analyze(self, queries): 批量执行分析 results {} for query in queries: results[query] self.analyze(query, visualizeFalse) return results def get_analysis_summary(self): 获取分析摘要 summary { total_queries: len(self.query_history), data_shape: self.df.shape, columns: list(self.df.columns), recent_queries: self.query_history[-5:] if self.query_history else [] } return summary # 客户端使用示例 def demonstrate_client_agent(): 演示客户端Agent的使用 # 创建示例数据 data { date: pd.date_range(2024-01-01, periods100, freqD), region: [华东] * 40 [华北] * 30 [华南] * 30, product: [A, B] * 50, sales: np.random.normal(1000, 200, 100), quantity: np.random.randint(10, 100, 100) } df pd.DataFrame(data) # 保存示例数据 df.to_csv(sample_sales_data.csv, indexFalse) # 初始化Agent需要真实的API密钥 # agent ClientDataAgent(api_keyyour_api_key, data_sourcesample_sales_data.csv) # 数据预处理 # agent.preprocess_data() # 执行分析示例 queries [ 计算每个区域的平均销售额, 显示销售金额最高的5个记录, 绘制月度销售趋势图, 按产品类别统计销售数量 ] print(客户端Agent演示完成需要真实API密钥才能运行) if __name__ __main__: demonstrate_client_agent()4.3 云端进阶实现云端版本支持数据库直连和大规模数据分析具备企业级特性from pandasai import SmartDatalake from pandasai_openai import OpenAI import mysql.connector from sqlalchemy import create_engine class CloudDataAgent: 云端数据分析Agent def __init__(self, api_key, db_configNone): 初始化云端Agent Args: api_key: OpenAI API密钥 db_config: 数据库连接配置 self.llm OpenAI(api_keyapi_key) self.db_config db_config self.engine None self.smart_lake None if db_config: self._setup_database_connection() def _setup_database_connection(self): 建立数据库连接 try: self.engine create_engine( fmysqlmysqlconnector://{self.db_config[user]}:{self.db_config[password]} f{self.db_config[host]}:{self.db_config[port]}/{self.db_config[database]} ) print(数据库连接建立成功) except Exception as e: print(f数据库连接失败: {e}) def connect_to_database(self, db_config): 连接数据库 self.db_config db_config self._setup_database_connection() def analyze_from_table(self, table_name, query): 从数据库表执行分析 if not self.engine: raise ValueError(未建立数据库连接) try: # 读取表数据 df pd.read_sql_table(table_name, self.engine) # 创建SmartDatalake self.smart_lake SmartDatalake([df], config{llm: self.llm}) # 执行分析 result self.smart_lake.chat(query) return result except Exception as e: return f分析失败: {str(e)} def execute_sql_analysis(self, sql_query): 直接执行SQL分析 if not self.engine: raise ValueError(未建立数据库连接) try: result_df pd.read_sql_query(sql_query, self.engine) return result_df except Exception as e: return fSQL执行失败: {str(e)} def generate_smart_sql(self, natural_language_query): 将自然语言转换为SQL # 这里可以集成更智能的SQL生成逻辑 mapping_rules { 销售额: sales_amount, 用户数: user_count, 日期: date_column } # 简单的规则映射实际项目应使用LLM sql_template SELECT * FROM sales_data WHERE {conditions} return sql_template # 云端Agent配置示例 def demonstrate_cloud_agent(): 演示云端Agent的配置 # 数据库配置示例 db_config { host: localhost, user: your_username, password: your_password, database: sales_db, port: 3306 } # 初始化云端Agent # cloud_agent CloudDataAgent(api_keyyour_api_key, db_configdb_config) # 分析示例 # result cloud_agent.analyze_from_table(sales_data, 统计月度销售额趋势) print(云端Agent演示完成需要真实数据库配置) if __name__ __main__: demonstrate_cloud_agent()4.4 核心能力优势对比零代码分析能力业务人员无需编写任何Pandas或SQL代码通过自然语言即可完成复杂数据分析。智能数据源适配自动识别数据源类型为本地文件生成Pandas代码为数据库生成优化SQL查询。自动可视化根据分析场景智能选择最合适的图表类型无需手动配置绘图参数。执行效率优化大数据量场景优先使用SQL查询避免Pandas内存瓶颈提升分析速度。5. 安全防护机制实现数据分析Agent拥有代码执行权限必须建立完善的安全防护机制防止恶意代码执行和数据泄露风险。5.1 安全风险分析代码注入风险模型可能生成包含系统命令执行、文件删除等危险操作的代码。数据泄露风险分析结果可能包含敏感信息需要防止未经授权的数据导出。资源滥用风险死循环或大规模数据查询可能导致服务器资源耗尽。5.2 多层安全防护实现import ast import re from typing import List, Tuple class SecurityManager: 安全管理系统 def __init__(self, security_levelstrict): 初始化安全管理器 Args: security_level: 安全级别lenient, moderate, strict self.security_level security_level self._setup_security_rules() def _setup_security_rules(self): 设置安全规则 # 高危操作黑名单 self.dangerous_keywords [ os.system, subprocess, eval, exec, shutil.rmtree, os.remove, os.unlink, requests.get, requests.post, __import__, open(, rm -rf, del , drop table ] # 敏感数据模式 self.sensitive_patterns [ r\bpassword\b, r\bapi_key\b, r\bsecret\b, r\btoken\b, r\bcredit_card\b, r\bssn\b ] # 资源限制 self.resource_limits { max_execution_time: 30, # 秒 max_memory_usage: 1024, # MB max_result_size: 10000 # 行 } def validate_code_safety(self, code: str) - Tuple[bool, str]: 验证代码安全性 validation_steps [ self._check_dangerous_keywords, self._check_syntax_safety, self._check_sensitive_data, self._check_resource_usage ] for step in validation_steps: is_safe, message step(code) if not is_safe: return False, message return True, 代码安全验证通过 def _check_dangerous_keywords(self, code: str) - Tuple[bool, str]: 检查危险关键词 for keyword in self.dangerous_keywords: if keyword in code.lower(): return False, f检测到危险操作: {keyword} return True, 危险关键词检查通过 def _check_syntax_safety(self, code: str) - Tuple[bool, str]: 检查语法安全性 try: ast.parse(code) return True, 语法检查通过 except SyntaxError as e: return False, f语法错误: {e} def _check_sensitive_data(self, code: str) - Tuple[bool, str]: 检查敏感数据泄露风险 for pattern in self.sensitive_patterns: if re.search(pattern, code, re.IGNORECASE): return False, f可能包含敏感信息: {pattern} return True, 敏感数据检查通过 def _check_resource_usage(self, code: str) - Tuple[bool, str]: 检查资源使用风险 # 检查是否有明显的大规模数据操作 dangerous_patterns [ r\.copy\(\), r\.apply\(, r\.iterrows\(\), rfor.*in.*range\(10000\), rwhile True ] for pattern in dangerous_patterns: if re.search(pattern, code): return False, f可能造成资源滥用: {pattern} return True, 资源使用检查通过 class SecureDataAgent(ClientDataAgent): 带安全防护的数据分析Agent def __init__(self, api_key, data_source, security_levelstrict): super().__init__(api_key, data_source) self.security_manager SecurityManager(security_level) def secure_analyze(self, query, visualizeTrue): 安全执行分析 # 先获取生成的代码不执行 try: # 使用PandasAI生成代码但不执行 generated_code self._get_generated_code(query) # 安全验证 is_safe, message self.security_manager.validate_code_safety(generated_code) if not is_safe: return f安全拦截: {message} # 安全通过执行分析 return self.analyze(query, visualize) except Exception as e: return f安全分析失败: {str(e)} def _get_generated_code(self, query): 获取生成的代码模拟方法 # 在实际的PandasAI中需要hook代码生成过程 # 这里返回模拟的生成代码用于演示 return f # 生成的分析代码 result df.groupby(region)[sales].mean() print(result) # 安全防护演示 def demonstrate_security_features(): 演示安全防护功能 security_mgr SecurityManager() # 测试安全代码 safe_code import pandas as pd df pd.read_csv(data.csv) result df.groupby(category)[sales].sum() print(result) # 测试危险代码 dangerous_code import os os.system(rm -rf /) print(安全代码检查:) is_safe, msg security_mgr.validate_code_safety(safe_code) print(f结果: {is_safe}, 消息: {msg}) print(\n危险代码检查:) is_safe, msg security_mgr.validate_code_safety(dangerous_code) print(f结果: {is_safe}, 消息: {msg}) if __name__ __main__: demonstrate_security_features()5.3 沙箱执行环境对于云端生产环境需要建立完整的沙箱执行机制import multiprocessing import signal import resource class CodeSandbox: 代码沙箱执行环境 def __init__(self, timeout30, memory_limit1024): self.timeout timeout self.memory_limit memory_limit def execute_in_sandbox(self, code: str, globals_dict: dict, locals_dict: dict): 在沙箱中执行代码 def run_code(): try: # 设置内存限制 resource.setrlimit(resource.RLIMIT_AS, (self.memory_limit * 1024 * 1024, self.memory_limit * 1024 * 1024)) # 执行代码 exec(code, globals_dict, locals_dict) return 执行成功, None except Exception as e: return 执行失败, str(e) # 创建进程执行 process multiprocessing.Process(targetrun_code) process.start() process.join(timeoutself.timeout) if process.is_alive(): process.terminate() process.join() return 执行超时, None return run_code() # 沙箱使用示例 def demonstrate_sandbox(): 演示沙箱执行 sandbox CodeSandbox(timeout10, memory_limit100) code import pandas as pd import numpy as np # 创建测试数据 data {A: range(10000), B: np.random.randn(10000)} df pd.DataFrame(data) # 执行分析 result df.describe() print(分析完成) result, error sandbox.execute_in_sandbox(code, {}, {}) print(f沙箱执行结果: {result}) if error: print(f错误信息: {error}) if __name__ __main__: demonstrate_sandbox()6. 智能报告生成与业务解读数据分析的最终价值在于为业务决策提供支持智能报告生成将冰冷的数据结果转化为有洞察力的业务分析。6.1 报告生成架构设计数据提取层从分析结果中提取关键指标和趋势洞察发现层识别数据中的异常点、变化趋势、关联关系报告生成层基于模板生成结构化分析报告可视化整合层将图表嵌入报告增强可读性6.2 智能报告生成实现from langchain_openai import ChatOpenAI from datetime import datetime import json class ReportGenerator: 智能报告生成器 def __init__(self, api_key, template_pathNone): self.llm ChatOpenAI( modelgpt-3.5-turbo, temperature0.2, api_keyapi_key ) self.template_path template_path def generate_analysis_report(self, data_result, user_query, contextNone): 生成分析报告 # 准备报告数据 report_data self._prepare_report_data(data_result, user_query, context) # 生成报告内容 report_content self._generate_with_llm(report_data) # 格式化报告 formatted_report self._format_report(report_content, report_data) return formatted_report def _prepare_report_data(self, data_result, user_query, context): 准备报告数据 report_data { user_query: user_query, analysis_time: datetime.now().strftime(%Y-%m-%d %H:%M:%S), data_summary: self._extract_data_summary(data_result), key_insights: self._extract_insights(data_result), context: context or {} } return report_data def _extract_data_summary(self, data_result): 提取数据摘要 if isinstance(data_result, pd.DataFrame): return { row_count: len(data_result), column_count: len(data_result.columns), data_types: data_result.dtypes.to_dict(), basic_stats: data_result.describe().to_dict() if data_result.select_dtypes(include[np.number]).shape[1] 0 else {} } elif isinstance(data_result, (int, float)): return {value: data_result} else: return {result: str(data_result)} def _extract_insights(self, data_result): 提取数据洞察 insights [] if isinstance(data_result, pd.DataFrame): # 提取数值列的洞察 numeric_columns data_result.select_dtypes(include[np.number]).columns for col in numeric_columns: col_insights self._analyze_numeric_column(data_result[col]) insights.extend(col_