AI 驱动数据分析引擎：用 LangChain + Ollama 构建智能 EDA 助手
传统数据分析依赖人工编写 `df.describe()`、逐列绘制分布图，不仅效率低，且难以从海量统计指标中提取业务洞察。本文不讨论理论，而是**纯工程实战**：基于本地开源大模型 Llama 3，结合 LangChain 构建一个能自动生成 Pandas 代码、执行统计分析并产出可视化图表的 AI 数据分析代理（Agent）。我们将完整实现“自然语言需求 → 可执行 Python 代码 → 数据结果与洞察”的闭环。 --- ## 1. 技术选型与环境初始化 选用以下工具链（全部本地部署，无需 OpenAI API Key）： - **LLM 服务**：Ollama + Llama 3 (8B)，兼顾性能与代码生成准确率 - **编排框架**：LangChain（提示模板、输出解析器、链式调用） - **数据处理**：Pandas + NumPy - **可视化**：Matplotlib + Seaborn 安装依赖并拉取模型： ``` pip install langchain langchain-community pandas matplotlib seaborn openpyxl ollama ollama pull llama3:latest ``` 创建项目结构： ``` ai_analyst/ ├── analyst.py # 核心引擎 ├── prompts.py # 提示词模板 └── data/sample.csv # 待分析数据集 ``` --- ## 2. 构建安全的代码生成提示链 LLM 生成 Python 代码存在安全风险（如 `os.system()`）。我们通过**强约束提示词 + 白名单函数限制**来规避： ``` # prompts.py from langchain_core.prompts import ChatPromptTemplate CODE_GENERATOR_PROMPT = ChatPromptTemplate.from_messages([ ("system", """你是一名资深数据分析工程师。用户提供 DataFrame 变量名为 `df`，请根据需求生成纯 Python 代码。 严格遵循： 1. 仅允许使用 pandas、numpy、matplotlib.pyplot、seaborn 库 2. 禁止使用 os、subprocess、eval、exec、open 等危险函数 3. 若需绘图，必须调用 plt.show() 或 plt.savefig() 4. 只返回代码块，不要包含任何解释性文字"""), ("human", "当前数据集列名：{columns}\n前5行预览：{head}\n用户需求：{query}") ]) ``` LangChain 解析器将 LLM 输出提取为纯净代码字符串： ``` from langchain_core.output_parsers import StrOutputParser from langchain_community.llms import Ollama llm = Ollama(model="llama3", temperature=0.1, num_predict=2048) chain = CODE_GENERATOR_PROMPT | llm | StrOutputParser() ``` --- ## 3. 数据分析引擎核心类 构建 `AIAnalyst` 类，封装数据加载、代码执行与结果回传。这里采用 `exec` 执行生成代码，但通过**显式限制全局作用域**来隔离危险操作： ``` import pandas as pd import matplotlib.pyplot as plt import io import sys class AIAnalyst: def __init__(self, file_path: str): self.df = pd.read_csv(file_path) self.llm = Ollama(model="llama3", temperature=0) def _safe_exec(self, code: str, return_df: bool = False): # 构建安全的执行环境 safe_globals = { 'pd': pd, 'plt': plt, 'np': __import__('numpy'), 'seaborn': __import__('seaborn'), '__builtins__': {'print': print, 'range': range, 'len': len, 'int': int, 'float': float, 'str': str} } local_scope = {'df': self.df.copy()} try: exec(code, safe_globals, local_scope) if return_df and 'df' in local_scope: return local_scope['df'] return True except Exception as e: return str(e) def query(self, user_question: str, return_df: bool = False): """对外暴露的查询接口""" columns = ', '.join(self.df.columns) head = self.df.head(3).to_string() # 调用第一步的 chain 生成代码 generated_code = chain.invoke({ 'columns': columns, 'head': head, 'query': user_question }) # 清理 markdown 标记（如有） clean_code = generated_code.replace('```python', '').replace('```', '').strip() print(f"[AI 生成代码]\n{clean_code}") result = self._safe_exec(clean_code, return_df) if return_df and isinstance(result, pd.DataFrame): return result return result ``` --- ## 4. 自动化 EDA（探索性数据分析）流水线 我们通过两次链式调用来实现“统计计算 → 自然语言洞察”： ``` def auto_insights(self) -> str: # 第一步：生成统计摘要代码 stats_code = chain.invoke({ 'columns': ', '.join(self.df.columns), 'head': self.df.head(2).to_string(), 'query': '计算所有数值列的均值、中位数、标准差、最大值、最小值，并将结果保存为 stats_df 变量' }) clean_stats = stats_code.replace('```python', '').replace('```', '').strip() # 执行统计代码（本地作用域捕获 stats_df） local_scope = {'df': self.df, 'pd': pd, 'np': __import__('numpy')} exec(clean_stats, {'pd': pd, 'np': __import__('numpy'), '__builtins__': {}}, local_scope) stats_df = local_scope.get('stats_df', pd.DataFrame()) # 第二步：将统计表传给 LLM，生成业务洞察 insight_prompt = f"以下为数据集的统计汇总：\n{stats_df.to_string()}\n请用中文总结 3 条核心业务洞察，并指出潜在异常点。" insight = self.llm.invoke(insight_prompt) return insight ``` --- ## 5. 智能可视化生成 当用户提出“画图”需求时，我们让 AI 生成绘图代码并自动执行。在提示词中强制要求输出图片保存路径，便于查看： ``` def plot(self, plot_query: str, save_path: str = 'output.png'): query_with_path = f"{plot_query}，将图形保存为 {save_path}，dpi=150" code = chain.invoke({ 'columns': ', '.join(self.df.columns), 'head': self.df.head(2).to_string(), 'query': query_with_path }) clean_code = code.replace('```python', '').replace('```', '').strip() self._safe_exec(clean_code) print(f"图表已生成：{save_path}") return save_path ``` --- ## 6. 集成测试：完整运行示例 假设 `sample.csv` 包含电商订单数据（订单ID、金额、数量、地区、品类）。我们编写主程序： ``` # main.py from analyst import AIAnalyst analyst = AIAnalyst('data/sample.csv') # 测试 1：数据清洗 analyst.query("将金额列中缺失值填充为平均值，数量列小于0的替换为1") # 测试 2：复杂分析 result_df = analyst.query("按地区分组，计算每个地区的总金额和平均数量，返回结果", return_df=True) print(result_df) # 测试 3：自动洞察 print(analyst.auto_insights()) # 测试 4：可视化 analyst.plot("绘制各品类订单数量的箱线图，按地区区分颜色") ``` 运行日志将显示 AI 生成的中间代码，例如： ``` df['金额'].fillna(df['金额'].mean(), inplace=True) df['数量'] = df['数量'].apply(lambda x: 1 if x < 0 else x) ``` 整个流程无需人工编写任何分析逻辑，完全由 LLM 理解数据列名后动态生成。 --- ## 7. 安全加固与性能优化 - **代码沙箱**：生产环境建议使用 `RestrictedPython` 或 Docker 隔离执行。本文用 `exec` 配合空 `__builtins__` 已过滤大部分高危函数。 - **缓存机制**：对相同的 `(列名摘要, 查询)` 进行缓存，减少重复调用 LLM 的延迟（约 3~5 秒/次）。 - **错误自愈**：捕获 `exec` 异常后，将错误信息重新输入 LLM，让其修正代码（重试 2 次）。 ``` # 自愈重试示例 for attempt in range(2): res = self._safe_exec(clean_code) if isinstance(res, str) and "Error" in res: clean_code = self.llm.invoke(f"以下代码报错：{res}，请修正：{clean_code}") else: break ``` --- ## 结语 我们仅用 200 余行 Python 代码便构建了一个具备“理解-生成-执行-解读”全链路的 AI 数据分析引擎。其核心价值在于**将 LLM 的语义理解能力与 Pandas 的计算能力解耦**，使分析师可以专注于提问而非写脚本。该架构可轻松扩展至数据库查询（Text-to-SQL）、异常检测自动报告等场景。当然，当前方案对复杂业务逻辑和多表 Join 的支持仍有局限，但结合 **Function Calling** 和 **ReAct Agent**，AI 数据分析正从“玩具”走向真正的生产力工具。
