在企业数据分析场景中业务人员与数据技术之间存在明显的鸿沟业务人员熟悉业务逻辑但不懂SQL和Pandas而技术人员擅长数据处理却不了解具体业务背景。智能数据分析Agent正是为了解决这一矛盾而设计的它能够将自然语言需求自动转换为可执行的数据分析代码并生成可视化图表和业务报告。本文将完整实现一个可商用的智能数据分析Agent涵盖从自然语言解析到最终报告生成的全流程。项目将区分客户端轻量级版本和云端生产级版本满足不同场景下的数据分析需求。1. 智能数据分析Agent的核心架构设计智能数据分析Agent的核心价值在于消除自然语言与结构化数据之间的技术壁垒。一个完整的数据分析Agent需要具备需求理解、数据预处理、代码生成、安全执行和结果解释五大核心能力。1.1 系统整体架构设计智能数据分析Agent的系统架构采用分层设计每层职责明确便于维护和扩展交互层接收自然语言输入展示分析结果和可视化图表解析层理解用户意图拆解分析需求选择执行引擎处理层数据预处理、Schema映射、代码生成与安全校验执行层Pandas本地执行或SQL数据库查询输出层结果可视化、报告生成、格式导出1.2 双端架构差异设计根据使用场景的不同我们需要设计两种架构方案客户端轻量级架构本地文件数据源CSV/ExcelPandas内存计算Matplotlib/Plotly本地绘图基础安全校验适合个人离线分析、快速验证云端生产级架构数据库连接MySQL/PostgreSQLSQL优先查询优化云端图表渲染服务多层安全防护体系支持多用户、权限管理、审计日志2. 数据预处理与Schema映射策略原始数据往往存在缺失值、异常值、格式不一致等问题直接进行分析会导致结果不准确。同时用户自然语言中的字段描述与数据表中的实际字段名存在语义鸿沟需要通过Schema映射建立对应关系。2.1 标准化数据预处理流程数据预处理是保障分析准确性的基础需要建立统一的清洗规则import pandas as pd import numpy as np from datetime import datetime class DataPreprocessor: def __init__(self, data_source): self.df self.load_data(data_source) self.cleaning_report {} def load_data(self, source): 支持多种数据源加载 if isinstance(source, str) and source.endswith((.csv, .xlsx)): if source.endswith(.csv): return pd.read_csv(source) else: return pd.read_excel(source) elif hasattr(source, execute): # 数据库连接对象 return pd.read_sql(SELECT * FROM data_table, source) else: raise ValueError(不支持的数据源类型) def standardize_preprocessing(self): 标准化预处理流程 original_shape self.df.shape # 1. 处理缺失值 self.handle_missing_values() # 2. 去除重复数据 self.remove_duplicates() # 3. 统一数据格式 self.standardize_formats() # 4. 过滤异常值 self.filter_outliers() final_shape self.df.shape self.cleaning_report { original_records: original_shape[0], final_records: final_shape[0], removed_duplicates: original_shape[0] - final_shape[0], cleaning_steps: [缺失值处理, 去重, 格式标准化, 异常值过滤] } return self.df def handle_missing_values(self): 智能处理缺失值 for column in self.df.columns: if self.df[column].dtype in [int64, float64]: # 数值列用中位数填充 self.df[column].fillna(self.df[column].median(), inplaceTrue) elif self.df[column].dtype object: # 文本列用众数填充 mode_value self.df[column].mode() self.df[column].fillna(mode_value[0] if len(mode_value) 0 else 未知, inplaceTrue) def remove_duplicates(self): 基于关键字段去重 key_columns [col for col in self.df.columns if col not in [id, timestamp]] self.df self.df.drop_duplicates(subsetkey_columns) def standardize_formats(self): 统一日期、数值格式 for column in self.df.columns: # 尝试转换为日期格式 if date in column.lower() or time in column.lower(): try: self.df[column] pd.to_datetime(self.df[column]) except: pass # 统一数值格式中的逗号分隔 if self.df[column].dtype object: self.df[column] self.df[column].astype(str).str.replace(,, )2.2 Schema语义映射方案Schema映射是连接用户自然语言与数据字段的关键桥梁class SchemaMapper: def __init__(self, df): self.df df self.schema_map {} self._build_initial_mapping() def _build_initial_mapping(self): 基于字段名自动构建初始映射 common_mappings { sale: [销售额, 销售金额, 营收], amount: [金额, 数额, 总值], quantity: [数量, 销量, 计量], date: [日期, 时间, 时间段], region: [区域, 地区, 地域], product: [产品, 商品, 品类] } for column in self.df.columns: column_lower column.lower() mapped False for eng_key, chn_terms in common_mappings.items(): if eng_key in column_lower: self.schema_map[column] chn_terms mapped True break if not mapped: # 无匹配时使用字段名本身 self.schema_map[column] [column] def enhance_mapping_with_llm(self, llm_client): 使用LLM增强语义映射 column_descriptions \n.join([f- {col}: 示例值 {self.df[col].iloc[0] if len(self.df[col]) 0 else 空} for col in self.df.columns]) prompt f 根据以下数据表字段信息为每个字段生成3-5个常见的中文业务描述 {column_descriptions} 请以JSON格式返回{{字段名: [中文描述1, 中文描述2, ...]}} try: enhanced_mapping llm_client.generate(prompt) self.schema_map.update(enhanced_mapping) except Exception as e: print(fLLM增强映射失败使用基础映射: {e}) def map_user_query_to_columns(self, user_query): 将用户查询映射到实际字段 matched_columns [] for column, terms in self.schema_map.items(): if any(term in user_query for term in terms): matched_columns.append(column) return matched_columns3. 基于PandasAI的智能代码生成与执行PandasAI是一个专门为智能数据分析设计的开源库它封装了自然语言到数据分析代码的转换能力大大简化了Agent的实现复杂度。3.1 PandasAI核心集成方案import pandas as pd from pandasai import SmartDataframe, SmartDatalake from pandasai_openai import OpenAI import matplotlib.pyplot as plt class PandasAIAgent: def __init__(self, data_source, api_key, environmentclient): self.environment environment self.llm OpenAI(api_keyapi_key) self.data_source data_source self.setup_agent() def setup_agent(self): 根据环境设置不同的Agent配置 if self.environment client: # 客户端配置侧重响应速度和本地化 self.config { llm: self.llm, enable_cache: True, max_retries: 3, custom_whitelisted_dependencies: [numpy, matplotlib] } self.df pd.read_csv(self.data_source) if isinstance(self.data_source, str) else self.data_source self.agent SmartDataframe(self.df, configself.config) else: # 云端环境 self.config { llm: self.llm, enable_cache: True, max_retries: 5, use_error_correction_framework: True, custom_whitelisted_dependencies: [] } self.agent SmartDatalake([self.data_source], configself.config) def analyze_data(self, user_query, visualization_typeauto): 执行数据分析并返回结果 try: # 添加可视化指令 if 图 in user_query or 可视化 in user_query: viz_query user_query else: # 根据查询类型自动添加可视化指令 if any(word in user_query for word in [趋势, 变化, 时间]): viz_query f{user_query}并生成折线图 elif any(word in user_query for word in [占比, 比例, 分布]): viz_query f{user_query}并生成饼图 elif any(word in user_query for word in [比较, 对比, 排名]): viz_query f{user_query}并生成柱状图 else: viz_query f{user_query}并生成合适的图表 result self.agent.chat(viz_query) return { success: True, result: result, query_type: self.classify_query_type(user_query), generated_code: self.agent.last_code_executed } except Exception as e: return { success: False, error: str(e), suggestion: 请尝试更明确地描述分析需求 } def classify_query_type(self, query): 分类查询类型以便后续优化 query_lower query.lower() if any(word in query_lower for word in [平均, 总和, 最大, 最小, 统计]): return 数值统计 elif any(word in query_lower for word in [筛选, 查找, 查询, 条件]): return 数据查询 elif any(word in query_lower for word in [趋势, 变化, 增长]): return 趋势分析 else: return 通用分析3.2 云端SQL优化执行器对于云端环境直接使用SQL查询比Pandas内存计算更高效class SQLQueryOptimizer: def __init__(self, db_connection): self.conn db_connection self.query_cache {} def generate_optimized_sql(self, analysis_intent, table_schema): 根据分析意图生成优化SQL intent_type analysis_intent.get(type, general) if intent_type 数值统计: return self._build_statistical_sql(analysis_intent, table_schema) elif intent_type 趋势分析: return self._build_trend_sql(analysis_intent, table_schema) elif intent_type 数据查询: return self._build_filter_sql(analysis_intent, table_schema) else: return self._build_general_sql(analysis_intent, table_schema) def _build_statistical_sql(self, intent, schema): 构建统计类SQL measures intent.get(measures, []) dimensions intent.get(dimensions, []) select_parts [] for dim in dimensions: select_parts.append(f{dim}) for measure in measures: if measure.get(agg) avg: select_parts.append(fAVG({measure[field]}) as avg_{measure[field]}) elif measure.get(agg) sum: select_parts.append(fSUM({measure[field]}) as total_{measure[field]}) group_by fGROUP BY {, .join(dimensions)} if dimensions else sql f SELECT {, .join(select_parts)} FROM {schema[table_name]} {group_by} ORDER BY {select_parts[1]} DESC LIMIT 1000 return sql4. 生产环境安全防护体系数据分析Agent需要执行生成的代码这在生产环境中存在严重安全风险。必须建立多层安全防护机制。4.1 代码安全校验系统import ast import re class CodeSecurityValidator: def __init__(self, environmentcloud): self.environment environment self.setup_security_rules() def setup_security_rules(self): 设置安全规则库 # 语法级黑名单 self.syntax_blacklist [ os.system, subprocess, shutil, requests.get, open(, file(, eval(, exec(, __import__, rmdir, remove, delete, drop, truncate ] # 模式黑名单正则表达式 self.pattern_blacklist [ rimport\sos, rimport\ssys, rimport\ssubprocess, rfrom\sos\simport, r__.*__, rlambda.*:.*import ] # 资源限制 self.resource_limits { max_execution_time: 30, # 秒 max_memory_mb: 512, max_result_rows: 10000 } def validate_code_safety(self, code_string): 全面代码安全校验 checks [ self._check_syntax_blacklist, self._check_pattern_blacklist, self._check_ast_safety, self._check_resource_usage ] for check_func in checks: is_safe, message check_func(code_string) if not is_safe: return False, message return True, 代码安全校验通过 def _check_syntax_blacklist(self, code): 检查语法黑名单 for forbidden in self.syntax_blacklist: if forbidden in code: return False, f检测到高危操作: {forbidden} return True, def _check_pattern_blacklist(self, code): 检查模式黑名单 for pattern in self.pattern_blacklist: if re.search(pattern, code): return False, f检测到危险模式: {pattern} return True, def _check_ast_safety(self, code): 使用AST进行语法树安全分析 try: tree ast.parse(code) for node in ast.walk(tree): # 检查危险函数调用 if isinstance(node, ast.Call): if isinstance(node.func, ast.Name): if node.func.id in [eval, exec, compile]: return False, f检测到危险函数调用: {node.func.id} # 检查危险属性访问 if isinstance(node, ast.Attribute): if node.attr in [system, popen, rmdir]: return False, f检测到危险属性访问: {node.attr} return True, except SyntaxError: return False, 代码语法错误4.2 沙箱执行环境对于云端生产环境必须使用沙箱隔离执行import tempfile import resource import signal import threading class CodeSandbox: def __init__(self, timeout30, memory_limit512): self.timeout timeout self.memory_limit memory_limit * 1024 * 1024 # 转换为字节 def execute_in_sandbox(self, code_string, globals_dictNone, locals_dictNone): 在沙箱中安全执行代码 if globals_dict is None: globals_dict {__builtins__: self._get_safe_builtins()} if locals_dict is None: locals_dict {} # 设置资源限制 self._set_resource_limits() # 使用线程超时控制 result {output: None, error: None, timed_out: False} def worker(): try: # 在临时目录中执行 with tempfile.TemporaryDirectory() as temp_dir: old_cwd os.getcwd() os.chdir(temp_dir) try: exec(code_string, globals_dict, locals_dict) result[output] locals_dict.get(result, 执行完成) finally: os.chdir(old_cwd) except Exception as e: result[error] str(e) thread threading.Thread(targetworker) thread.start() thread.join(self.timeout) if thread.is_alive(): result[timed_out] True result[error] f执行超时{self.timeout}秒 # 强制终止线程 # 注意实际生产环境需要更安全的线程终止机制 return result def _get_safe_builtins(self): 提供安全的builtins子集 safe_builtins {} allowed [len, range, str, int, float, list, dict, tuple] for name in allowed: if hasattr(__builtins__, name): safe_builtins[name] getattr(__builtins__, name) return safe_builtins def _set_resource_limits(self): 设置系统资源限制 try: # 设置内存限制 resource.setrlimit(resource.RLIMIT_AS, (self.memory_limit, self.memory_limit)) # 设置CPU时间限制 resource.setrlimit(resource.RLIMIT_CPU, (self.timeout, self.timeout 5)) except (ValueError, resource.error): # 在某些系统上可能不支持 pass5. 智能报告生成与业务解读数据分析的最终价值在于为业务决策提供支持。智能报告生成系统将冰冷的数字转化为有洞察力的业务建议。5.1 多维度报告生成引擎from langchain_openai import ChatOpenAI import json class AnalysisReportGenerator: def __init__(self, llm_api_key): self.llm ChatOpenAI( modelgpt-3.5-turbo, temperature0.3, openai_api_keyllm_api_key ) self.report_templates self._load_report_templates() def _load_report_templates(self): 加载报告模板库 return { sales_analysis: { sections: [数据概览, 销售表现, 区域对比, 趋势分析, 业务建议], focus_metrics: [销售额, 销售量, 增长率, 市场份额] }, user_behavior: { sections: [用户概况, 行为模式, 留存分析, 转化漏斗, 优化建议], focus_metrics: [活跃度, 留存率, 转化率, 使用时长] }, operational_efficiency: { sections: [效率概览, 瓶颈分析, 资源利用, 改进方向, 实施建议], focus_metrics: [处理时长, 成功率, 资源消耗, 吞吐量] } } def generate_comprehensive_report(self, analysis_results, user_query, report_typeauto): 生成综合分析报告 # 确定报告类型 if report_type auto: report_type self._detect_report_type(user_query, analysis_results) template self.report_templates.get(report_type, self.report_templates[sales_analysis]) # 构建报告生成提示词 prompt self._build_report_prompt(analysis_results, user_query, template) try: response self.llm.invoke(prompt) report_content self._structure_report(response.content, template) return self._add_visualization_suggestions(report_content, analysis_results) except Exception as e: return f报告生成失败: {str(e)} def _build_report_prompt(self, results, query, template): 构建报告生成提示词 return f 作为资深数据分析师请基于以下分析结果生成专业报告。 用户分析需求{query} 分析结果数据{json.dumps(results, ensure_asciiFalse, indent2)} 报告需要包含以下部分 {chr(10).join([- section for section in template[sections]])} 报告要求 1. 数据准确引用具体数值 2. 洞察深刻揭示数据背后的业务含义 3. 建议具体提供可落地的优化方案 4. 语言专业但易懂适合业务人员阅读 5. 字数控制在800-1200字 请直接生成完整的报告内容。 def _structure_report(self, raw_content, template): 结构化报告内容 structured_report { metadata: { generated_at: datetime.now().isoformat(), sections: template[sections], word_count: len(raw_content.split()) }, content: {} } # 按章节拆分内容 current_section 引言 for line in raw_content.split(\n): line line.strip() if not line: continue # 检测章节标题 if any(section in line for section in template[sections]): current_section next((s for s in template[sections] if s in line), current_section) structured_report[content][current_section] [] else: if current_section not in structured_report[content]: structured_report[content][current_section] [] structured_report[content][current_section].append(line) return structured_report5.2 报告可视化增强class ReportVisualizer: def __init__(self): self.chart_templates self._init_chart_templates() def _init_chart_templates(self): 初始化图表模板库 return { trend_analysis: { type: line, title: 趋势分析图, config: {width: 800, height: 400} }, comparison: { type: bar, title: 对比分析图, config: {width: 800, height: 400} }, distribution: { type: pie, title: 分布图, config: {width: 600, height: 400} } } def generate_chart_suggestions(self, analysis_results, report_content): 根据分析结果生成图表建议 suggestions [] # 检测趋势数据 if self._has_trend_data(analysis_results): suggestions.append({ type: trend_analysis, reason: 数据包含时间序列趋势, priority: high }) # 检测对比数据 if self._has_comparison_data(analysis_results): suggestions.append({ type: comparison, reason: 数据包含多维度对比, priority: high }) # 检测分布数据 if self._has_distribution_data(analysis_results): suggestions.append({ type: distribution, reason: 数据包含比例分布, priority: medium }) return suggestions def _has_trend_data(self, results): 检查是否包含趋势数据 # 实现趋势数据检测逻辑 return any(date in str(key).lower() or time in str(key).lower() for key in results.keys())6. 完整项目集成与部署方案将各个模块集成为完整的智能数据分析Agent并提供部署指南。6.1 系统集成与配置管理import yaml import logging from pathlib import Path class DataAnalysisAgent: def __init__(self, config_pathNone, environmentclient): self.environment environment self.config self._load_config(config_path) self.logger self._setup_logging() self._initialize_components() def _load_config(self, config_path): 加载配置文件 default_config { client: { max_file_size_mb: 100, allowed_formats: [.csv, .xlsx, .xls], enable_local_cache: True }, cloud: { db_connections: {}, max_query_timeout: 300, enable_audit_log: True }, security: { enable_sandbox: False, max_memory_mb: 512, allowed_dependencies: [pandas, numpy] } } if config_path and Path(config_path).exists(): with open(config_path, r, encodingutf-8) as f: user_config yaml.safe_load(f) # 深度合并配置 return self._deep_merge(default_config, user_config) return default_config def _initialize_components(self): 初始化各个组件 # 数据预处理组件 self.preprocessor DataPreprocessor() # Schema映射组件 self.schema_mapper SchemaMapper() # 安全校验组件 self.security_validator CodeSecurityValidator(self.environment) # 报告生成组件 self.report_generator AnalysisReportGenerator( self.config.get(llm_api_key, ) ) if self.environment cloud: self.sandbox CodeSandbox() def process_analysis_request(self, user_query, data_source): 处理分析请求的完整流程 try: # 1. 数据预处理 clean_data self.preprocessor.standardize_preprocessing(data_source) # 2. Schema映射 self.schema_mapper.enhance_mapping_with_llm(clean_data) mapped_columns self.schema_mapper.map_user_query_to_columns(user_query) # 3. 安全分析执行 if self.environment client: agent PandasAIAgent(clean_data, self.config[llm_api_key], client) else: agent PandasAIAgent(data_source, self.config[llm_api_key], cloud) # 4. 安全校验 analysis_result agent.analyze_data(user_query) if not analysis_result[success]: return analysis_result # 5. 安全代码执行云端需要沙箱 if self.environment cloud: safe_result self.sandbox.execute_in_sandbox( analysis_result[generated_code] ) if safe_result[error]: return {success: False, error: safe_result[error]} # 6. 生成报告 report self.report_generator.generate_comprehensive_report( analysis_result[result], user_query ) return { success: True, report: report, visualization: analysis_result.get(result, {}), execution_log: analysis_result.get(generated_code, ) } except Exception as e: self.logger.error(f分析处理失败: {str(e)}) return { success: False, error: f系统处理异常: {str(e)}, suggestion: 请检查数据格式和分析需求 }6.2 部署配置与运维指南客户端部署要求Python 3.8内存至少4GB存储500MB可用空间网络可选仅LLM API调用需要云端生产环境部署Docker容器化部署Kubernetes集群管理数据库连接池配置监控告警体系日志审计系统性能优化建议启用查询结果缓存数据库查询添加索引优化使用连接池管理数据库连接设置合理的超时时间和资源限制定期清理临时文件和缓存数据通过以上完整实现智能数据分析Agent已经具备了从自然语言理解到业务报告生成的全流程能力。在实际部署时需要根据具体业务需求调整配置参数特别是安全规则和资源限制需要根据实际环境进行调优。
在企业数据分析场景中业务人员与数据技术之间存在明显的鸿沟业务人员熟悉业务逻辑但不懂SQL和Pandas而技术人员擅长数据处理却不了解具体业务背景。智能数据分析Agent正是为了解决这一矛盾而设计的它能够将自然语言需求自动转换为可执行的数据分析代码并生成可视化图表和业务报告。本文将完整实现一个可商用的智能数据分析Agent涵盖从自然语言解析到最终报告生成的全流程。项目将区分客户端轻量级版本和云端生产级版本满足不同场景下的数据分析需求。1. 智能数据分析Agent的核心架构设计智能数据分析Agent的核心价值在于消除自然语言与结构化数据之间的技术壁垒。一个完整的数据分析Agent需要具备需求理解、数据预处理、代码生成、安全执行和结果解释五大核心能力。1.1 系统整体架构设计智能数据分析Agent的系统架构采用分层设计每层职责明确便于维护和扩展交互层接收自然语言输入展示分析结果和可视化图表解析层理解用户意图拆解分析需求选择执行引擎处理层数据预处理、Schema映射、代码生成与安全校验执行层Pandas本地执行或SQL数据库查询输出层结果可视化、报告生成、格式导出1.2 双端架构差异设计根据使用场景的不同我们需要设计两种架构方案客户端轻量级架构本地文件数据源CSV/ExcelPandas内存计算Matplotlib/Plotly本地绘图基础安全校验适合个人离线分析、快速验证云端生产级架构数据库连接MySQL/PostgreSQLSQL优先查询优化云端图表渲染服务多层安全防护体系支持多用户、权限管理、审计日志2. 数据预处理与Schema映射策略原始数据往往存在缺失值、异常值、格式不一致等问题直接进行分析会导致结果不准确。同时用户自然语言中的字段描述与数据表中的实际字段名存在语义鸿沟需要通过Schema映射建立对应关系。2.1 标准化数据预处理流程数据预处理是保障分析准确性的基础需要建立统一的清洗规则import pandas as pd import numpy as np from datetime import datetime class DataPreprocessor: def __init__(self, data_source): self.df self.load_data(data_source) self.cleaning_report {} def load_data(self, source): 支持多种数据源加载 if isinstance(source, str) and source.endswith((.csv, .xlsx)): if source.endswith(.csv): return pd.read_csv(source) else: return pd.read_excel(source) elif hasattr(source, execute): # 数据库连接对象 return pd.read_sql(SELECT * FROM data_table, source) else: raise ValueError(不支持的数据源类型) def standardize_preprocessing(self): 标准化预处理流程 original_shape self.df.shape # 1. 处理缺失值 self.handle_missing_values() # 2. 去除重复数据 self.remove_duplicates() # 3. 统一数据格式 self.standardize_formats() # 4. 过滤异常值 self.filter_outliers() final_shape self.df.shape self.cleaning_report { original_records: original_shape[0], final_records: final_shape[0], removed_duplicates: original_shape[0] - final_shape[0], cleaning_steps: [缺失值处理, 去重, 格式标准化, 异常值过滤] } return self.df def handle_missing_values(self): 智能处理缺失值 for column in self.df.columns: if self.df[column].dtype in [int64, float64]: # 数值列用中位数填充 self.df[column].fillna(self.df[column].median(), inplaceTrue) elif self.df[column].dtype object: # 文本列用众数填充 mode_value self.df[column].mode() self.df[column].fillna(mode_value[0] if len(mode_value) 0 else 未知, inplaceTrue) def remove_duplicates(self): 基于关键字段去重 key_columns [col for col in self.df.columns if col not in [id, timestamp]] self.df self.df.drop_duplicates(subsetkey_columns) def standardize_formats(self): 统一日期、数值格式 for column in self.df.columns: # 尝试转换为日期格式 if date in column.lower() or time in column.lower(): try: self.df[column] pd.to_datetime(self.df[column]) except: pass # 统一数值格式中的逗号分隔 if self.df[column].dtype object: self.df[column] self.df[column].astype(str).str.replace(,, )2.2 Schema语义映射方案Schema映射是连接用户自然语言与数据字段的关键桥梁class SchemaMapper: def __init__(self, df): self.df df self.schema_map {} self._build_initial_mapping() def _build_initial_mapping(self): 基于字段名自动构建初始映射 common_mappings { sale: [销售额, 销售金额, 营收], amount: [金额, 数额, 总值], quantity: [数量, 销量, 计量], date: [日期, 时间, 时间段], region: [区域, 地区, 地域], product: [产品, 商品, 品类] } for column in self.df.columns: column_lower column.lower() mapped False for eng_key, chn_terms in common_mappings.items(): if eng_key in column_lower: self.schema_map[column] chn_terms mapped True break if not mapped: # 无匹配时使用字段名本身 self.schema_map[column] [column] def enhance_mapping_with_llm(self, llm_client): 使用LLM增强语义映射 column_descriptions \n.join([f- {col}: 示例值 {self.df[col].iloc[0] if len(self.df[col]) 0 else 空} for col in self.df.columns]) prompt f 根据以下数据表字段信息为每个字段生成3-5个常见的中文业务描述 {column_descriptions} 请以JSON格式返回{{字段名: [中文描述1, 中文描述2, ...]}} try: enhanced_mapping llm_client.generate(prompt) self.schema_map.update(enhanced_mapping) except Exception as e: print(fLLM增强映射失败使用基础映射: {e}) def map_user_query_to_columns(self, user_query): 将用户查询映射到实际字段 matched_columns [] for column, terms in self.schema_map.items(): if any(term in user_query for term in terms): matched_columns.append(column) return matched_columns3. 基于PandasAI的智能代码生成与执行PandasAI是一个专门为智能数据分析设计的开源库它封装了自然语言到数据分析代码的转换能力大大简化了Agent的实现复杂度。3.1 PandasAI核心集成方案import pandas as pd from pandasai import SmartDataframe, SmartDatalake from pandasai_openai import OpenAI import matplotlib.pyplot as plt class PandasAIAgent: def __init__(self, data_source, api_key, environmentclient): self.environment environment self.llm OpenAI(api_keyapi_key) self.data_source data_source self.setup_agent() def setup_agent(self): 根据环境设置不同的Agent配置 if self.environment client: # 客户端配置侧重响应速度和本地化 self.config { llm: self.llm, enable_cache: True, max_retries: 3, custom_whitelisted_dependencies: [numpy, matplotlib] } self.df pd.read_csv(self.data_source) if isinstance(self.data_source, str) else self.data_source self.agent SmartDataframe(self.df, configself.config) else: # 云端环境 self.config { llm: self.llm, enable_cache: True, max_retries: 5, use_error_correction_framework: True, custom_whitelisted_dependencies: [] } self.agent SmartDatalake([self.data_source], configself.config) def analyze_data(self, user_query, visualization_typeauto): 执行数据分析并返回结果 try: # 添加可视化指令 if 图 in user_query or 可视化 in user_query: viz_query user_query else: # 根据查询类型自动添加可视化指令 if any(word in user_query for word in [趋势, 变化, 时间]): viz_query f{user_query}并生成折线图 elif any(word in user_query for word in [占比, 比例, 分布]): viz_query f{user_query}并生成饼图 elif any(word in user_query for word in [比较, 对比, 排名]): viz_query f{user_query}并生成柱状图 else: viz_query f{user_query}并生成合适的图表 result self.agent.chat(viz_query) return { success: True, result: result, query_type: self.classify_query_type(user_query), generated_code: self.agent.last_code_executed } except Exception as e: return { success: False, error: str(e), suggestion: 请尝试更明确地描述分析需求 } def classify_query_type(self, query): 分类查询类型以便后续优化 query_lower query.lower() if any(word in query_lower for word in [平均, 总和, 最大, 最小, 统计]): return 数值统计 elif any(word in query_lower for word in [筛选, 查找, 查询, 条件]): return 数据查询 elif any(word in query_lower for word in [趋势, 变化, 增长]): return 趋势分析 else: return 通用分析3.2 云端SQL优化执行器对于云端环境直接使用SQL查询比Pandas内存计算更高效class SQLQueryOptimizer: def __init__(self, db_connection): self.conn db_connection self.query_cache {} def generate_optimized_sql(self, analysis_intent, table_schema): 根据分析意图生成优化SQL intent_type analysis_intent.get(type, general) if intent_type 数值统计: return self._build_statistical_sql(analysis_intent, table_schema) elif intent_type 趋势分析: return self._build_trend_sql(analysis_intent, table_schema) elif intent_type 数据查询: return self._build_filter_sql(analysis_intent, table_schema) else: return self._build_general_sql(analysis_intent, table_schema) def _build_statistical_sql(self, intent, schema): 构建统计类SQL measures intent.get(measures, []) dimensions intent.get(dimensions, []) select_parts [] for dim in dimensions: select_parts.append(f{dim}) for measure in measures: if measure.get(agg) avg: select_parts.append(fAVG({measure[field]}) as avg_{measure[field]}) elif measure.get(agg) sum: select_parts.append(fSUM({measure[field]}) as total_{measure[field]}) group_by fGROUP BY {, .join(dimensions)} if dimensions else sql f SELECT {, .join(select_parts)} FROM {schema[table_name]} {group_by} ORDER BY {select_parts[1]} DESC LIMIT 1000 return sql4. 生产环境安全防护体系数据分析Agent需要执行生成的代码这在生产环境中存在严重安全风险。必须建立多层安全防护机制。4.1 代码安全校验系统import ast import re class CodeSecurityValidator: def __init__(self, environmentcloud): self.environment environment self.setup_security_rules() def setup_security_rules(self): 设置安全规则库 # 语法级黑名单 self.syntax_blacklist [ os.system, subprocess, shutil, requests.get, open(, file(, eval(, exec(, __import__, rmdir, remove, delete, drop, truncate ] # 模式黑名单正则表达式 self.pattern_blacklist [ rimport\sos, rimport\ssys, rimport\ssubprocess, rfrom\sos\simport, r__.*__, rlambda.*:.*import ] # 资源限制 self.resource_limits { max_execution_time: 30, # 秒 max_memory_mb: 512, max_result_rows: 10000 } def validate_code_safety(self, code_string): 全面代码安全校验 checks [ self._check_syntax_blacklist, self._check_pattern_blacklist, self._check_ast_safety, self._check_resource_usage ] for check_func in checks: is_safe, message check_func(code_string) if not is_safe: return False, message return True, 代码安全校验通过 def _check_syntax_blacklist(self, code): 检查语法黑名单 for forbidden in self.syntax_blacklist: if forbidden in code: return False, f检测到高危操作: {forbidden} return True, def _check_pattern_blacklist(self, code): 检查模式黑名单 for pattern in self.pattern_blacklist: if re.search(pattern, code): return False, f检测到危险模式: {pattern} return True, def _check_ast_safety(self, code): 使用AST进行语法树安全分析 try: tree ast.parse(code) for node in ast.walk(tree): # 检查危险函数调用 if isinstance(node, ast.Call): if isinstance(node.func, ast.Name): if node.func.id in [eval, exec, compile]: return False, f检测到危险函数调用: {node.func.id} # 检查危险属性访问 if isinstance(node, ast.Attribute): if node.attr in [system, popen, rmdir]: return False, f检测到危险属性访问: {node.attr} return True, except SyntaxError: return False, 代码语法错误4.2 沙箱执行环境对于云端生产环境必须使用沙箱隔离执行import tempfile import resource import signal import threading class CodeSandbox: def __init__(self, timeout30, memory_limit512): self.timeout timeout self.memory_limit memory_limit * 1024 * 1024 # 转换为字节 def execute_in_sandbox(self, code_string, globals_dictNone, locals_dictNone): 在沙箱中安全执行代码 if globals_dict is None: globals_dict {__builtins__: self._get_safe_builtins()} if locals_dict is None: locals_dict {} # 设置资源限制 self._set_resource_limits() # 使用线程超时控制 result {output: None, error: None, timed_out: False} def worker(): try: # 在临时目录中执行 with tempfile.TemporaryDirectory() as temp_dir: old_cwd os.getcwd() os.chdir(temp_dir) try: exec(code_string, globals_dict, locals_dict) result[output] locals_dict.get(result, 执行完成) finally: os.chdir(old_cwd) except Exception as e: result[error] str(e) thread threading.Thread(targetworker) thread.start() thread.join(self.timeout) if thread.is_alive(): result[timed_out] True result[error] f执行超时{self.timeout}秒 # 强制终止线程 # 注意实际生产环境需要更安全的线程终止机制 return result def _get_safe_builtins(self): 提供安全的builtins子集 safe_builtins {} allowed [len, range, str, int, float, list, dict, tuple] for name in allowed: if hasattr(__builtins__, name): safe_builtins[name] getattr(__builtins__, name) return safe_builtins def _set_resource_limits(self): 设置系统资源限制 try: # 设置内存限制 resource.setrlimit(resource.RLIMIT_AS, (self.memory_limit, self.memory_limit)) # 设置CPU时间限制 resource.setrlimit(resource.RLIMIT_CPU, (self.timeout, self.timeout 5)) except (ValueError, resource.error): # 在某些系统上可能不支持 pass5. 智能报告生成与业务解读数据分析的最终价值在于为业务决策提供支持。智能报告生成系统将冰冷的数字转化为有洞察力的业务建议。5.1 多维度报告生成引擎from langchain_openai import ChatOpenAI import json class AnalysisReportGenerator: def __init__(self, llm_api_key): self.llm ChatOpenAI( modelgpt-3.5-turbo, temperature0.3, openai_api_keyllm_api_key ) self.report_templates self._load_report_templates() def _load_report_templates(self): 加载报告模板库 return { sales_analysis: { sections: [数据概览, 销售表现, 区域对比, 趋势分析, 业务建议], focus_metrics: [销售额, 销售量, 增长率, 市场份额] }, user_behavior: { sections: [用户概况, 行为模式, 留存分析, 转化漏斗, 优化建议], focus_metrics: [活跃度, 留存率, 转化率, 使用时长] }, operational_efficiency: { sections: [效率概览, 瓶颈分析, 资源利用, 改进方向, 实施建议], focus_metrics: [处理时长, 成功率, 资源消耗, 吞吐量] } } def generate_comprehensive_report(self, analysis_results, user_query, report_typeauto): 生成综合分析报告 # 确定报告类型 if report_type auto: report_type self._detect_report_type(user_query, analysis_results) template self.report_templates.get(report_type, self.report_templates[sales_analysis]) # 构建报告生成提示词 prompt self._build_report_prompt(analysis_results, user_query, template) try: response self.llm.invoke(prompt) report_content self._structure_report(response.content, template) return self._add_visualization_suggestions(report_content, analysis_results) except Exception as e: return f报告生成失败: {str(e)} def _build_report_prompt(self, results, query, template): 构建报告生成提示词 return f 作为资深数据分析师请基于以下分析结果生成专业报告。 用户分析需求{query} 分析结果数据{json.dumps(results, ensure_asciiFalse, indent2)} 报告需要包含以下部分 {chr(10).join([- section for section in template[sections]])} 报告要求 1. 数据准确引用具体数值 2. 洞察深刻揭示数据背后的业务含义 3. 建议具体提供可落地的优化方案 4. 语言专业但易懂适合业务人员阅读 5. 字数控制在800-1200字 请直接生成完整的报告内容。 def _structure_report(self, raw_content, template): 结构化报告内容 structured_report { metadata: { generated_at: datetime.now().isoformat(), sections: template[sections], word_count: len(raw_content.split()) }, content: {} } # 按章节拆分内容 current_section 引言 for line in raw_content.split(\n): line line.strip() if not line: continue # 检测章节标题 if any(section in line for section in template[sections]): current_section next((s for s in template[sections] if s in line), current_section) structured_report[content][current_section] [] else: if current_section not in structured_report[content]: structured_report[content][current_section] [] structured_report[content][current_section].append(line) return structured_report5.2 报告可视化增强class ReportVisualizer: def __init__(self): self.chart_templates self._init_chart_templates() def _init_chart_templates(self): 初始化图表模板库 return { trend_analysis: { type: line, title: 趋势分析图, config: {width: 800, height: 400} }, comparison: { type: bar, title: 对比分析图, config: {width: 800, height: 400} }, distribution: { type: pie, title: 分布图, config: {width: 600, height: 400} } } def generate_chart_suggestions(self, analysis_results, report_content): 根据分析结果生成图表建议 suggestions [] # 检测趋势数据 if self._has_trend_data(analysis_results): suggestions.append({ type: trend_analysis, reason: 数据包含时间序列趋势, priority: high }) # 检测对比数据 if self._has_comparison_data(analysis_results): suggestions.append({ type: comparison, reason: 数据包含多维度对比, priority: high }) # 检测分布数据 if self._has_distribution_data(analysis_results): suggestions.append({ type: distribution, reason: 数据包含比例分布, priority: medium }) return suggestions def _has_trend_data(self, results): 检查是否包含趋势数据 # 实现趋势数据检测逻辑 return any(date in str(key).lower() or time in str(key).lower() for key in results.keys())6. 完整项目集成与部署方案将各个模块集成为完整的智能数据分析Agent并提供部署指南。6.1 系统集成与配置管理import yaml import logging from pathlib import Path class DataAnalysisAgent: def __init__(self, config_pathNone, environmentclient): self.environment environment self.config self._load_config(config_path) self.logger self._setup_logging() self._initialize_components() def _load_config(self, config_path): 加载配置文件 default_config { client: { max_file_size_mb: 100, allowed_formats: [.csv, .xlsx, .xls], enable_local_cache: True }, cloud: { db_connections: {}, max_query_timeout: 300, enable_audit_log: True }, security: { enable_sandbox: False, max_memory_mb: 512, allowed_dependencies: [pandas, numpy] } } if config_path and Path(config_path).exists(): with open(config_path, r, encodingutf-8) as f: user_config yaml.safe_load(f) # 深度合并配置 return self._deep_merge(default_config, user_config) return default_config def _initialize_components(self): 初始化各个组件 # 数据预处理组件 self.preprocessor DataPreprocessor() # Schema映射组件 self.schema_mapper SchemaMapper() # 安全校验组件 self.security_validator CodeSecurityValidator(self.environment) # 报告生成组件 self.report_generator AnalysisReportGenerator( self.config.get(llm_api_key, ) ) if self.environment cloud: self.sandbox CodeSandbox() def process_analysis_request(self, user_query, data_source): 处理分析请求的完整流程 try: # 1. 数据预处理 clean_data self.preprocessor.standardize_preprocessing(data_source) # 2. Schema映射 self.schema_mapper.enhance_mapping_with_llm(clean_data) mapped_columns self.schema_mapper.map_user_query_to_columns(user_query) # 3. 安全分析执行 if self.environment client: agent PandasAIAgent(clean_data, self.config[llm_api_key], client) else: agent PandasAIAgent(data_source, self.config[llm_api_key], cloud) # 4. 安全校验 analysis_result agent.analyze_data(user_query) if not analysis_result[success]: return analysis_result # 5. 安全代码执行云端需要沙箱 if self.environment cloud: safe_result self.sandbox.execute_in_sandbox( analysis_result[generated_code] ) if safe_result[error]: return {success: False, error: safe_result[error]} # 6. 生成报告 report self.report_generator.generate_comprehensive_report( analysis_result[result], user_query ) return { success: True, report: report, visualization: analysis_result.get(result, {}), execution_log: analysis_result.get(generated_code, ) } except Exception as e: self.logger.error(f分析处理失败: {str(e)}) return { success: False, error: f系统处理异常: {str(e)}, suggestion: 请检查数据格式和分析需求 }6.2 部署配置与运维指南客户端部署要求Python 3.8内存至少4GB存储500MB可用空间网络可选仅LLM API调用需要云端生产环境部署Docker容器化部署Kubernetes集群管理数据库连接池配置监控告警体系日志审计系统性能优化建议启用查询结果缓存数据库查询添加索引优化使用连接池管理数据库连接设置合理的超时时间和资源限制定期清理临时文件和缓存数据通过以上完整实现智能数据分析Agent已经具备了从自然语言理解到业务报告生成的全流程能力。在实际部署时需要根据具体业务需求调整配置参数特别是安全规则和资源限制需要根据实际环境进行调优。
