智能数据分析Agent是当前AI技术在企业级应用中最具实用价值的落地项目之一。这个项目基于PandasAI开源框架实现了从自然语言到数据分析的全流程自动化让不懂SQL和Pandas的业务人员也能通过简单的对话完成专业级数据分析。最值得关注的是该项目采用双端设计架构客户端版本支持本地CSV/Excel文件分析无需数据库环境适合个人快速验证云端版本支持MySQL/PostgreSQL等数据库连接具备企业级安全防护能力。无论是个人用户还是企业团队都能找到适合自己的部署方案。本文将从环境准备、核心功能实现、安全防护到实战演示完整介绍如何搭建一个可商用的智能数据分析Agent。重点涵盖PandasAI的集成使用、Schema映射策略、代码安全防护以及自然语言报告生成等关键技术点。1. 核心能力速览能力项说明项目类型智能数据分析AI Agent核心技术PandasAI 大语言模型数据处理支持CSV/Excel本地文件 MySQL/PostgreSQL数据库分析能力数值统计、数据查询、可视化图表生成部署方式客户端轻量化部署 云端生产级部署安全防护语法黑名单拦截 沙箱隔离执行输出形式数据结果 可视化图表 自然语言分析报告2. 适用场景与使用边界智能数据分析Agent主要面向两类用户群体一是需要快速进行数据探索的业务人员二是希望提升数据分析效率的开发团队。对于日常的数据统计、趋势分析、报表生成等场景该工具能够显著降低技术门槛。在客户端模式下适合处理中小规模的数据文件通常小于100MB进行快速的数据探索和可视化分析。云端版本则适合企业级应用能够连接数据库处理海量数据并具备完整的权限管理和安全审计功能。需要注意的是该工具不适合处理高度敏感的商业数据除非部署在完全隔离的内网环境中。对于需要复杂业务逻辑推理的分析任务仍需要专业数据分析师的介入。3. 环境准备与前置条件3.1 基础软件环境Python版本: 3.8及以上操作系统: Windows 10/11, macOS 10.15, Ubuntu 18.04内存要求: 至少8GB RAM处理大型数据集时建议16GB存储空间: 至少2GB可用空间包含依赖包和模型文件3.2 Python依赖包核心依赖包包括# 基础数据分析包 pandas1.5.0 numpy1.21.0 # 智能数据分析核心框架 pandasai3.0.0b2 pandasai-openai1.0.0 # 数据库连接云端版本需要 mysql-connector-python8.0.0 psycopg2-binary2.9.0 # 报告生成可选 langchain-openai0.1.03.3 API密钥配置项目需要配置大语言模型的API密钥支持OpenAI、Anthropic等主流模型# 在代码中配置API密钥 import os os.environ[OPENAI_API_KEY] 你的API密钥4. 安装部署与启动方式4.1 一键安装命令# 安装核心依赖 pip install pandasai3.0.0b2 pandasai-openai pandas numpy # 如果需要数据库支持 pip install mysql-connector-python psycopg2-binary # 如果需要报告生成功能 pip install langchain-openai4.2 客户端轻量化启动客户端版本适合本地文件分析启动简单快捷# client_agent.py import pandas as pd from pandasai import SmartDataframe from pandasai_openai import OpenAI # 初始化大语言模型 llm OpenAI(api_key你的OpenAI密钥) # 加载本地数据文件 df pd.read_csv(sales_data.csv) # 创建智能数据帧 sdf SmartDataframe(df, config{llm: llm}) print(客户端智能数据分析Agent启动成功)4.3 云端服务化部署云端版本支持数据库连接和API服务# server_agent.py from pandasai import SmartDatalake from pandasai_openai import OpenAI import mysql.connector from flask import Flask, request, jsonify app Flask(__name__) # 数据库连接配置 db_config { host: localhost, user: root, password: 密码, database: sales_db } app.route(/analyze, methods[POST]) def analyze_data(): query request.json.get(query) # 连接数据库 db_conn mysql.connector.connect(**db_config) llm OpenAI(api_key你的OpenAI密钥) # 创建数据湖分析实例 dl SmartDatalake([db_conn], config{llm: llm}) result dl.chat(query) return jsonify({result: result}) if __name__ __main__: app.run(host0.0.0.0, port5000)5. 数据预处理与Schema映射实战5.1 标准化数据预处理流程数据质量直接影响分析结果的准确性必须建立标准化的预处理流程import pandas as pd import numpy as np def preprocess_data(df): 标准化数据预处理函数 包含缺失值处理、异常值过滤、重复数据清理等功能 # 缺失值处理 numeric_columns df.select_dtypes(include[np.number]).columns for col in numeric_columns: df[col] df[col].fillna(df[col].mean()) text_columns df.select_dtypes(include[object]).columns for col in text_columns: df[col] df[col].fillna(未知) # 异常值处理基于3σ原则 for col in numeric_columns: mean_val df[col].mean() std_val df[col].std() df df[(df[col] mean_val - 3*std_val) (df[col] mean_val 3*std_val)] # 重复数据清理 df df.drop_duplicates() # 日期格式统一 date_columns [date, time, created_at] # 根据实际列名调整 for col in date_columns: if col in df.columns: df[col] pd.to_datetime(df[col], errorscoerce) return df # 使用示例 df pd.read_csv(your_data.csv) cleaned_df preprocess_data(df) print(f数据预处理完成原始数据{len(df)}行清洗后{len(cleaned_df)}行)5.2 Schema语义映射配置建立自然语言与数据字段的映射关系是智能分析的关键# schema_mapping.py class SchemaMapper: def __init__(self): self.schema_map {} def load_static_mapping(self, mapping_dict): 加载静态字段映射配置 self.schema_map.update(mapping_dict) def generate_dynamic_mapping(self, df, llm): 使用LLM动态生成字段映射 column_descriptions \n.join([f{col}: 示例值 {df[col].iloc[0] if len(df) 0 else 无数据} for col in df.columns]) prompt f 根据以下数据表字段信息为每个字段生成中文描述 {column_descriptions} 请返回JSON格式{{字段名: 中文描述}} # 调用LLM生成映射简化示例 # 实际实现需要完整的LLM调用逻辑 return {sales: 销售额, quantity: 销售数量, region: 区域} def map_user_query(self, user_query): 将用户查询中的自然语言映射到实际字段 mapped_query user_query for chinese_name, field_name in self.schema_map.items(): mapped_query mapped_query.replace(chinese_name, field_name) return mapped_query # 使用示例 mapper SchemaMapper() static_mapping { 销售额: sales, 销售数量: quantity, 销售日期: date, 区域: region, 产品名称: product_name } mapper.load_static_mapping(static_mapping) user_query 计算各区域销售额平均值 mapped_query mapper.map_user_query(user_query) print(f映射后查询: {mapped_query})6. 核心分析功能测试与验证6.1 基础数值统计测试测试智能Agent的基础统计分析能力# test_basic_analysis.py def test_basic_analysis(sdf): 测试基础分析功能 test_cases [ 销售总额是多少, 每个区域的平均销售额是多少, 销量最高的产品是什么, 最近30天的销售趋势如何 ] results {} for query in test_cases: try: result sdf.chat(query) results[query] { status: success, result: result, code_executed: sdf.last_code_executed } print(f✓ {query} - 执行成功) except Exception as e: results[query] { status: error, error: str(e) } print(f✗ {query} - 执行失败: {e}) return results # 执行测试 df pd.read_csv(sales_data.csv) sdf SmartDataframe(df, config{llm: llm}) test_results test_basic_analysis(sdf)6.2 可视化图表生成测试验证自动图表生成能力# test_visualization.py def test_visualization(sdf): 测试可视化功能 visualization_queries [ 绘制各区域销售额柱状图, 显示月度销售趋势折线图, 生成产品销量分布饼图, 创建销售额与销量散点图 ] for query in visualization_queries: try: print(f执行可视化查询: {query}) result sdf.chat(query) # 检查是否生成图表文件 if hasattr(sdf, last_chart_path) and sdf.last_chart_path: print(f图表已保存至: {sdf.last_chart_path}) else: print(查询执行成功但未检测到图表文件) except Exception as e: print(f可视化查询失败: {e}) # 执行可视化测试 test_visualization(sdf)6.3 复杂查询逻辑测试测试多条件组合查询能力# test_complex_queries.py def test_complex_queries(sdf): 测试复杂查询逻辑 complex_cases [ 找出销售额超过10万且销量大于100的产品, 计算每个区域Q1和Q2销售额的增长率, 分析哪些产品的销售额在最近三个月持续下降, 找出销量前十的产品并计算它们占总销售额的比例 ] for query in complex_cases: try: start_time time.time() result sdf.chat(query) execution_time time.time() - start_time print(f复杂查询: {query}) print(f执行时间: {execution_time:.2f}秒) print(f结果: {result}\n) except Exception as e: print(f复杂查询失败: {query} - 错误: {e}) test_complex_queries(sdf)7. 安全防护机制实现7.1 代码安全校验围栏生产环境必须部署严格的安全防护# security_manager.py class SecurityManager: def __init__(self): self.blacklist [ os.system, subprocess, shutil.rmtree, requests.get, requests.post, __import__, eval(, exec(, compile(, open(, file(, rm -, del , drop table, delete from, update , insert into, alter table ] self.whitelist [ pd.read_, df., plt., sns., np., df[, df.loc, df.iloc, groupby, merge, join, sum(), mean(), count(), plot., hist(, bar(, line(, scatter( ] def security_check(self, code: str) - tuple[bool, str]: 代码安全校验 code_lower code.lower() # 黑名单检查 for forbidden in self.blacklist: if forbidden in code_lower: return False, f检测到高危操作: {forbidden} # 白名单验证可选严格模式开启 if self.strict_mode: has_valid_operation any(valid in code for valid in self.whitelist) if not has_valid_operation: return False, 未包含允许的安全操作 # 嵌套深度检查 if code_lower.count(() 20 or code_lower.count()) 20: return False, 代码嵌套深度过大可能存在风险 return True, 安全校验通过 def sandbox_execute(self, code: str, timeout30): 沙箱环境执行代码 try: # 创建限制环境 restricted_globals { pd: pd, np: np, plt: plt, __builtins__: {**__builtins__} } # 移除危险函数 for func in [open, file, eval, exec, compile, __import__]: if func in restricted_globals[__builtins__]: del restricted_globals[__builtins__][func] # 超时执行 with timeout(timeout): exec(code, restricted_globals) return True, 执行成功 except TimeoutError: return False, 执行超时 except Exception as e: return False, f执行错误: {e} # 集成到PandasAI流程中 security_mgr SecurityManager() def safe_chat(sdf, query): 安全版本的chat方法 result sdf.chat(query) executed_code sdf.last_code_executed is_safe, message security_mgr.security_check(executed_code) if not is_safe: raise SecurityError(f安全拦截: {message}) return result7.2 数据库操作安全防护针对云端版本的数据库安全防护# database_security.py class DatabaseSecurity: def __init__(self, db_connection): self.conn db_connection self.max_rows 10000 # 最大返回行数 self.allowed_tables [sales, products, users] # 允许访问的表 def validate_sql_query(self, sql_query): 验证SQL查询安全性 sql_lower sql_query.lower() # 禁止危险操作 dangerous_keywords [drop, delete, update, insert, alter, truncate] for keyword in dangerous_keywords: if keyword in sql_lower: return False, f禁止执行{keyword}操作 # 检查表访问权限 for table in self.allowed_tables: if table in sql_lower: break else: return False, 未授权访问数据表 # 限制查询结果大小 if limit not in sql_lower: sql_query sql_query.rstrip(;) f LIMIT {self.max_rows} return True, sql_query def execute_safe_query(self, sql_query): 安全执行SQL查询 is_valid, result self.validate_sql_query(sql_query) if not is_valid: raise DatabaseSecurityError(result) if isinstance(result, str): # 返回的是修改后的SQL sql_query result cursor self.conn.cursor() cursor.execute(sql_query) return cursor.fetchall()8. 自然语言报告生成8.1 智能报告生成器将数据分析结果转化为业务洞察报告# report_generator.py from langchain_openai import ChatOpenAI import json class ReportGenerator: def __init__(self, api_key): self.llm ChatOpenAI( modelgpt-3.5-turbo, temperature0.2, api_keyapi_key ) def generate_analysis_report(self, data_result, user_query, contextNone): 生成结构化分析报告 prompt_template 作为数据分析专家请基于以下信息生成专业的数据分析报告 用户分析需求{user_query} 原始分析结果{data_result} 业务背景信息{context} 报告需要包含以下章节 1. 数据概览数据基本情况、统计口径说明 2. 核心发现最重要的数据洞察和结论 3. 详细分析关键指标的趋势、对比和分布情况 4. 业务建议基于数据提出的 actionable 建议 请用专业、简洁的商业语言撰写避免技术术语。 prompt prompt_template.format( user_queryuser_query, data_resultstr(data_result), contextcontext or 无额外背景信息 ) response self.llm.invoke(prompt) return self._format_report(response.content) def _format_report(self, raw_report): 格式化报告输出 # 简单的报告格式化逻辑 sections { 数据概览: , 核心发现: , 详细分析: , 业务建议: } current_section None for line in raw_report.split(\n): line line.strip() if not line: continue # 检测章节标题 for section in sections.keys(): if section in line: current_section section break else: if current_section and line: sections[current_section] line \n return sections # 使用示例 report_gen ReportGenerator(api_key你的API密钥) # 模拟分析结果 analysis_result { 总销售额: 1500000, 平均销售额: 50000, 区域分布: {华东: 600000, 华北: 450000, 华南: 450000}, 趋势: 季度环比增长15% } report report_gen.generate_analysis_report( analysis_result, 分析各区域销售表现, 公司本季度销售数据分析 ) for section, content in report.items(): print(f## {section}) print(content) print()8.2 报告模板定制化支持不同场景的报告模板# report_templates.py class ReportTemplates: staticmethod def sales_performance_template(): 销售业绩分析报告模板 return { introduction: 本报告对销售数据进行全面分析旨在识别业务机会和优化方向。, metrics: [销售额, 销售量, 增长率, 市场份额], visualizations: [趋势图, 分布图, 对比图], recommendation_focus: [产品优化, 区域策略, 促销活动] } staticmethod def customer_analysis_template(): 客户行为分析报告模板 return { introduction: 基于客户行为数据的深度分析揭示客户特征和偏好。, metrics: [客户数量, 复购率, 客单价, 生命周期价值], visualizations: [分布图, 行为路径图, 聚类分析], recommendation_focus: [客户细分, 留存策略, 个性化推荐] } staticmethod def operational_efficiency_template(): 运营效率分析报告模板 return { introduction: 评估业务流程效率识别优化点和资源分配策略。, metrics: [处理时长, 成功率, 资源利用率, 成本效益], visualizations: [时序图, 效率矩阵, 瓶颈分析], recommendation_focus: [流程优化, 自动化, 资源调配] } # 模板选择器 def select_report_template(user_query): 根据用户查询选择合适
智能数据分析Agent是当前AI技术在企业级应用中最具实用价值的落地项目之一。这个项目基于PandasAI开源框架实现了从自然语言到数据分析的全流程自动化让不懂SQL和Pandas的业务人员也能通过简单的对话完成专业级数据分析。最值得关注的是该项目采用双端设计架构客户端版本支持本地CSV/Excel文件分析无需数据库环境适合个人快速验证云端版本支持MySQL/PostgreSQL等数据库连接具备企业级安全防护能力。无论是个人用户还是企业团队都能找到适合自己的部署方案。本文将从环境准备、核心功能实现、安全防护到实战演示完整介绍如何搭建一个可商用的智能数据分析Agent。重点涵盖PandasAI的集成使用、Schema映射策略、代码安全防护以及自然语言报告生成等关键技术点。1. 核心能力速览能力项说明项目类型智能数据分析AI Agent核心技术PandasAI 大语言模型数据处理支持CSV/Excel本地文件 MySQL/PostgreSQL数据库分析能力数值统计、数据查询、可视化图表生成部署方式客户端轻量化部署 云端生产级部署安全防护语法黑名单拦截 沙箱隔离执行输出形式数据结果 可视化图表 自然语言分析报告2. 适用场景与使用边界智能数据分析Agent主要面向两类用户群体一是需要快速进行数据探索的业务人员二是希望提升数据分析效率的开发团队。对于日常的数据统计、趋势分析、报表生成等场景该工具能够显著降低技术门槛。在客户端模式下适合处理中小规模的数据文件通常小于100MB进行快速的数据探索和可视化分析。云端版本则适合企业级应用能够连接数据库处理海量数据并具备完整的权限管理和安全审计功能。需要注意的是该工具不适合处理高度敏感的商业数据除非部署在完全隔离的内网环境中。对于需要复杂业务逻辑推理的分析任务仍需要专业数据分析师的介入。3. 环境准备与前置条件3.1 基础软件环境Python版本: 3.8及以上操作系统: Windows 10/11, macOS 10.15, Ubuntu 18.04内存要求: 至少8GB RAM处理大型数据集时建议16GB存储空间: 至少2GB可用空间包含依赖包和模型文件3.2 Python依赖包核心依赖包包括# 基础数据分析包 pandas1.5.0 numpy1.21.0 # 智能数据分析核心框架 pandasai3.0.0b2 pandasai-openai1.0.0 # 数据库连接云端版本需要 mysql-connector-python8.0.0 psycopg2-binary2.9.0 # 报告生成可选 langchain-openai0.1.03.3 API密钥配置项目需要配置大语言模型的API密钥支持OpenAI、Anthropic等主流模型# 在代码中配置API密钥 import os os.environ[OPENAI_API_KEY] 你的API密钥4. 安装部署与启动方式4.1 一键安装命令# 安装核心依赖 pip install pandasai3.0.0b2 pandasai-openai pandas numpy # 如果需要数据库支持 pip install mysql-connector-python psycopg2-binary # 如果需要报告生成功能 pip install langchain-openai4.2 客户端轻量化启动客户端版本适合本地文件分析启动简单快捷# client_agent.py import pandas as pd from pandasai import SmartDataframe from pandasai_openai import OpenAI # 初始化大语言模型 llm OpenAI(api_key你的OpenAI密钥) # 加载本地数据文件 df pd.read_csv(sales_data.csv) # 创建智能数据帧 sdf SmartDataframe(df, config{llm: llm}) print(客户端智能数据分析Agent启动成功)4.3 云端服务化部署云端版本支持数据库连接和API服务# server_agent.py from pandasai import SmartDatalake from pandasai_openai import OpenAI import mysql.connector from flask import Flask, request, jsonify app Flask(__name__) # 数据库连接配置 db_config { host: localhost, user: root, password: 密码, database: sales_db } app.route(/analyze, methods[POST]) def analyze_data(): query request.json.get(query) # 连接数据库 db_conn mysql.connector.connect(**db_config) llm OpenAI(api_key你的OpenAI密钥) # 创建数据湖分析实例 dl SmartDatalake([db_conn], config{llm: llm}) result dl.chat(query) return jsonify({result: result}) if __name__ __main__: app.run(host0.0.0.0, port5000)5. 数据预处理与Schema映射实战5.1 标准化数据预处理流程数据质量直接影响分析结果的准确性必须建立标准化的预处理流程import pandas as pd import numpy as np def preprocess_data(df): 标准化数据预处理函数 包含缺失值处理、异常值过滤、重复数据清理等功能 # 缺失值处理 numeric_columns df.select_dtypes(include[np.number]).columns for col in numeric_columns: df[col] df[col].fillna(df[col].mean()) text_columns df.select_dtypes(include[object]).columns for col in text_columns: df[col] df[col].fillna(未知) # 异常值处理基于3σ原则 for col in numeric_columns: mean_val df[col].mean() std_val df[col].std() df df[(df[col] mean_val - 3*std_val) (df[col] mean_val 3*std_val)] # 重复数据清理 df df.drop_duplicates() # 日期格式统一 date_columns [date, time, created_at] # 根据实际列名调整 for col in date_columns: if col in df.columns: df[col] pd.to_datetime(df[col], errorscoerce) return df # 使用示例 df pd.read_csv(your_data.csv) cleaned_df preprocess_data(df) print(f数据预处理完成原始数据{len(df)}行清洗后{len(cleaned_df)}行)5.2 Schema语义映射配置建立自然语言与数据字段的映射关系是智能分析的关键# schema_mapping.py class SchemaMapper: def __init__(self): self.schema_map {} def load_static_mapping(self, mapping_dict): 加载静态字段映射配置 self.schema_map.update(mapping_dict) def generate_dynamic_mapping(self, df, llm): 使用LLM动态生成字段映射 column_descriptions \n.join([f{col}: 示例值 {df[col].iloc[0] if len(df) 0 else 无数据} for col in df.columns]) prompt f 根据以下数据表字段信息为每个字段生成中文描述 {column_descriptions} 请返回JSON格式{{字段名: 中文描述}} # 调用LLM生成映射简化示例 # 实际实现需要完整的LLM调用逻辑 return {sales: 销售额, quantity: 销售数量, region: 区域} def map_user_query(self, user_query): 将用户查询中的自然语言映射到实际字段 mapped_query user_query for chinese_name, field_name in self.schema_map.items(): mapped_query mapped_query.replace(chinese_name, field_name) return mapped_query # 使用示例 mapper SchemaMapper() static_mapping { 销售额: sales, 销售数量: quantity, 销售日期: date, 区域: region, 产品名称: product_name } mapper.load_static_mapping(static_mapping) user_query 计算各区域销售额平均值 mapped_query mapper.map_user_query(user_query) print(f映射后查询: {mapped_query})6. 核心分析功能测试与验证6.1 基础数值统计测试测试智能Agent的基础统计分析能力# test_basic_analysis.py def test_basic_analysis(sdf): 测试基础分析功能 test_cases [ 销售总额是多少, 每个区域的平均销售额是多少, 销量最高的产品是什么, 最近30天的销售趋势如何 ] results {} for query in test_cases: try: result sdf.chat(query) results[query] { status: success, result: result, code_executed: sdf.last_code_executed } print(f✓ {query} - 执行成功) except Exception as e: results[query] { status: error, error: str(e) } print(f✗ {query} - 执行失败: {e}) return results # 执行测试 df pd.read_csv(sales_data.csv) sdf SmartDataframe(df, config{llm: llm}) test_results test_basic_analysis(sdf)6.2 可视化图表生成测试验证自动图表生成能力# test_visualization.py def test_visualization(sdf): 测试可视化功能 visualization_queries [ 绘制各区域销售额柱状图, 显示月度销售趋势折线图, 生成产品销量分布饼图, 创建销售额与销量散点图 ] for query in visualization_queries: try: print(f执行可视化查询: {query}) result sdf.chat(query) # 检查是否生成图表文件 if hasattr(sdf, last_chart_path) and sdf.last_chart_path: print(f图表已保存至: {sdf.last_chart_path}) else: print(查询执行成功但未检测到图表文件) except Exception as e: print(f可视化查询失败: {e}) # 执行可视化测试 test_visualization(sdf)6.3 复杂查询逻辑测试测试多条件组合查询能力# test_complex_queries.py def test_complex_queries(sdf): 测试复杂查询逻辑 complex_cases [ 找出销售额超过10万且销量大于100的产品, 计算每个区域Q1和Q2销售额的增长率, 分析哪些产品的销售额在最近三个月持续下降, 找出销量前十的产品并计算它们占总销售额的比例 ] for query in complex_cases: try: start_time time.time() result sdf.chat(query) execution_time time.time() - start_time print(f复杂查询: {query}) print(f执行时间: {execution_time:.2f}秒) print(f结果: {result}\n) except Exception as e: print(f复杂查询失败: {query} - 错误: {e}) test_complex_queries(sdf)7. 安全防护机制实现7.1 代码安全校验围栏生产环境必须部署严格的安全防护# security_manager.py class SecurityManager: def __init__(self): self.blacklist [ os.system, subprocess, shutil.rmtree, requests.get, requests.post, __import__, eval(, exec(, compile(, open(, file(, rm -, del , drop table, delete from, update , insert into, alter table ] self.whitelist [ pd.read_, df., plt., sns., np., df[, df.loc, df.iloc, groupby, merge, join, sum(), mean(), count(), plot., hist(, bar(, line(, scatter( ] def security_check(self, code: str) - tuple[bool, str]: 代码安全校验 code_lower code.lower() # 黑名单检查 for forbidden in self.blacklist: if forbidden in code_lower: return False, f检测到高危操作: {forbidden} # 白名单验证可选严格模式开启 if self.strict_mode: has_valid_operation any(valid in code for valid in self.whitelist) if not has_valid_operation: return False, 未包含允许的安全操作 # 嵌套深度检查 if code_lower.count(() 20 or code_lower.count()) 20: return False, 代码嵌套深度过大可能存在风险 return True, 安全校验通过 def sandbox_execute(self, code: str, timeout30): 沙箱环境执行代码 try: # 创建限制环境 restricted_globals { pd: pd, np: np, plt: plt, __builtins__: {**__builtins__} } # 移除危险函数 for func in [open, file, eval, exec, compile, __import__]: if func in restricted_globals[__builtins__]: del restricted_globals[__builtins__][func] # 超时执行 with timeout(timeout): exec(code, restricted_globals) return True, 执行成功 except TimeoutError: return False, 执行超时 except Exception as e: return False, f执行错误: {e} # 集成到PandasAI流程中 security_mgr SecurityManager() def safe_chat(sdf, query): 安全版本的chat方法 result sdf.chat(query) executed_code sdf.last_code_executed is_safe, message security_mgr.security_check(executed_code) if not is_safe: raise SecurityError(f安全拦截: {message}) return result7.2 数据库操作安全防护针对云端版本的数据库安全防护# database_security.py class DatabaseSecurity: def __init__(self, db_connection): self.conn db_connection self.max_rows 10000 # 最大返回行数 self.allowed_tables [sales, products, users] # 允许访问的表 def validate_sql_query(self, sql_query): 验证SQL查询安全性 sql_lower sql_query.lower() # 禁止危险操作 dangerous_keywords [drop, delete, update, insert, alter, truncate] for keyword in dangerous_keywords: if keyword in sql_lower: return False, f禁止执行{keyword}操作 # 检查表访问权限 for table in self.allowed_tables: if table in sql_lower: break else: return False, 未授权访问数据表 # 限制查询结果大小 if limit not in sql_lower: sql_query sql_query.rstrip(;) f LIMIT {self.max_rows} return True, sql_query def execute_safe_query(self, sql_query): 安全执行SQL查询 is_valid, result self.validate_sql_query(sql_query) if not is_valid: raise DatabaseSecurityError(result) if isinstance(result, str): # 返回的是修改后的SQL sql_query result cursor self.conn.cursor() cursor.execute(sql_query) return cursor.fetchall()8. 自然语言报告生成8.1 智能报告生成器将数据分析结果转化为业务洞察报告# report_generator.py from langchain_openai import ChatOpenAI import json class ReportGenerator: def __init__(self, api_key): self.llm ChatOpenAI( modelgpt-3.5-turbo, temperature0.2, api_keyapi_key ) def generate_analysis_report(self, data_result, user_query, contextNone): 生成结构化分析报告 prompt_template 作为数据分析专家请基于以下信息生成专业的数据分析报告 用户分析需求{user_query} 原始分析结果{data_result} 业务背景信息{context} 报告需要包含以下章节 1. 数据概览数据基本情况、统计口径说明 2. 核心发现最重要的数据洞察和结论 3. 详细分析关键指标的趋势、对比和分布情况 4. 业务建议基于数据提出的 actionable 建议 请用专业、简洁的商业语言撰写避免技术术语。 prompt prompt_template.format( user_queryuser_query, data_resultstr(data_result), contextcontext or 无额外背景信息 ) response self.llm.invoke(prompt) return self._format_report(response.content) def _format_report(self, raw_report): 格式化报告输出 # 简单的报告格式化逻辑 sections { 数据概览: , 核心发现: , 详细分析: , 业务建议: } current_section None for line in raw_report.split(\n): line line.strip() if not line: continue # 检测章节标题 for section in sections.keys(): if section in line: current_section section break else: if current_section and line: sections[current_section] line \n return sections # 使用示例 report_gen ReportGenerator(api_key你的API密钥) # 模拟分析结果 analysis_result { 总销售额: 1500000, 平均销售额: 50000, 区域分布: {华东: 600000, 华北: 450000, 华南: 450000}, 趋势: 季度环比增长15% } report report_gen.generate_analysis_report( analysis_result, 分析各区域销售表现, 公司本季度销售数据分析 ) for section, content in report.items(): print(f## {section}) print(content) print()8.2 报告模板定制化支持不同场景的报告模板# report_templates.py class ReportTemplates: staticmethod def sales_performance_template(): 销售业绩分析报告模板 return { introduction: 本报告对销售数据进行全面分析旨在识别业务机会和优化方向。, metrics: [销售额, 销售量, 增长率, 市场份额], visualizations: [趋势图, 分布图, 对比图], recommendation_focus: [产品优化, 区域策略, 促销活动] } staticmethod def customer_analysis_template(): 客户行为分析报告模板 return { introduction: 基于客户行为数据的深度分析揭示客户特征和偏好。, metrics: [客户数量, 复购率, 客单价, 生命周期价值], visualizations: [分布图, 行为路径图, 聚类分析], recommendation_focus: [客户细分, 留存策略, 个性化推荐] } staticmethod def operational_efficiency_template(): 运营效率分析报告模板 return { introduction: 评估业务流程效率识别优化点和资源分配策略。, metrics: [处理时长, 成功率, 资源利用率, 成本效益], visualizations: [时序图, 效率矩阵, 瓶颈分析], recommendation_focus: [流程优化, 自动化, 资源调配] } # 模板选择器 def select_report_template(user_query): 根据用户查询选择合适
