对话式数据科学：把大模型装进 Jupyter Notebook 的智能分析与代码自愈实战

在数据科学家与分析师的日常工作中，至少有 60% 的时间被消耗在枯燥的“语法搬砖”上：

• 面对一份新数据集，要写几十行重复的 Pandas 代码进行缺失值填充、异常值处理与分布统计（EDA）；
• 调整 Matplotlib / Plotly 图表样式时，往往要在 Stack Overflow 上反复查询“如何旋转 X 轴标签”、“如何画双 Y 轴次坐标”；
• 执行特征工程与统计假设检验时，频繁遭遇 SettingWithCopyWarning 或 KeyError。

这种繁琐的语法摩擦严重打断了分析师的数据洞察心流。

将大语言模型（LLM）深度集成进 Jupyter Notebook / JupyterLab 运行时环境，正在将数据科学工作流从传统的“纯手工写代码”升级为**“自然语言引导 + 代码自动生成 + 结果自愈纠错（Self-Healing）”**的全新人机协作范式。

本文深入剖析基于 Jupyter Kernel 协议的运行时上下文探针、动态图表渲染、代码自愈闭环以及生产级 Python 助手实现。

一、系统架构：Kernel 运行时感知与自愈执行闭环

将大模型接入 Notebook 远非仅仅调用一次 Chat API，它必须与 Jupyter Kernel 运行时内存状态 深度联动：

+-----------------------------------------------------------------------------------+
| 1. Kernel 运行时状态反射 (Runtime Introspection)                                  |
|    - 自动探针: 扫描当前内存中的 DataFrame (列名、数据类型、缺失率、数值极值)       |
|    - 安全脱敏: 仅提取 Schema 与统计画像 (Profiling)，严禁将真实敏感数据外发       |
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 2. 结构化代码生成与 AST 安全审计 (LLM Code Generation & AST Audit)                 |
|    - 提示词注入当前上下文变量、目标图表库 (Plotly/Seaborn) 与分析指令              |
|    - 安全审查: AST 拦截危险系统调用 (禁止 `os.system`、文件删除与未经授权的网络连接)|
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 3. Jupyter Kernel 隔离执行与多模态渲染 (Execution & Multimodal Display)           |
|    - 在当前 Kernel 会话中动态执行代码                                             |
|    - 捕获 stdout 文本、数据表格 (HTML 表格) 以及渲染出的可视化图表 (PNG / Plotly)   |
+-----------------------------------------------------------------------------------+
                                          |
            +-----------------------------+-----------------------------+
            | (执行成功)                                                | (执行抛错: KeyError/TypeError)
            v                                                           v
+--------------------------------------+     +--------------------------------------+
| 4. 单元格落盘与交互回显              |     | 5. 错误上下文回流与自愈修正 (Self-Heal) |
| - 将可用代码写回 Notebook Cell       |     | - 提取 Traceback 报错行与异常信息    |
| - 分析师可继续手动编辑或追问         |     | - 触发 LLM 自我纠错 (最多重试 3 次)   |
+--------------------------------------+     +--------------------------------------+

二、核心挑战：如何在保护数据隐私的同时给模型足够信息？

大模型生成高质量数据分析代码的前提，是必须知道数据长什么样。但直接把百万行原始数据喂给模型，不仅会撑爆 Token 上下文，更会引发严重的数据隐私合规违规。

优雅的解决方案：紧凑型元数据画像（Compact Schema Profiling）

系统在后台静默执行轻量探针，为内存中的每个 DataFrame 提取结构化元数据：

# 探针自动提取的紧凑上下文示例 (仅消耗 < 300 Token，且零隐私泄露)
{
  "df_name": "df_sales",
  "shape": [150000, 5],
  "columns": {
    "order_id": {"dtype": "int64", "null_count": 0, "unique_count": 150000},
    "channel": {"dtype": "object", "top_values": ["TikTok", "Tmall", "JD"]},
    "pay_amount": {"dtype": "float64", "min": 9.9, "max": 8999.0, "mean": 156.4},
    "created_at": {"dtype": "datetime64[ns]", "min": "2026-08-01", "max": "2026-08-24"}
  }
}

三、生产级 Notebook 对话分析与自愈代码执行器实现

下面的 Python 实现结合了运行时上下文探针、AST 安全语法树检查、代码隔离求值以及捕获 Traceback 的多轮自愈修正引擎。

"""
notebook_copilot_engine.py
生产级 Jupyter Notebook 智能分析与代码自愈执行引擎
"""

import ast
import io
import json
import sys
import traceback
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
import pandas as pd


@dataclass
class DataFrameProfile:
    name: str
    shape: Tuple[int, int]
    columns_info: Dict[str, Dict[str, Any]]


class RuntimeInspector:
    """运行时上下文探针：安全提取当前命名空间中的 DataFrame 画像"""

    @staticmethod
    def inspect_namespace(local_env: Dict[str, Any]) -> List[DataFrameProfile]:
        profiles = []
        for var_name, var_val in local_env.items():
            if isinstance(var_val, pd.DataFrame):
                cols_meta = {}
                for col in var_val.columns:
                    col_series = var_val[col]
                    meta = {
                        "dtype": str(col_series.dtype),
                        "null_ratio": float(round(col_series.isnull().mean(), 4)),
                    }
                    if pd.api.types.is_numeric_dtype(col_series):
                        meta["min"] = float(col_series.min()) if not col_series.empty else None
                        meta["max"] = float(col_series.max()) if not col_series.empty else None
                    elif pd.api.types.is_object_dtype(col_series) or pd.api.types.is_string_dtype(col_series):
                        meta["top_categories"] = col_series.dropna().unique()[:3].tolist()
                    cols_meta[col] = meta

                profiles.append(DataFrameProfile(
                    name=var_name,
                    shape=var_val.shape,
                    columns_info=cols_meta
                ))
        return profiles


class SafeCodeValidator(ast.NodeVisitor):
    """AST 静态安全审查器：拦截破坏性操作"""

    FORBIDDEN_MODULES = {"os", "subprocess", "shutil", "socket", "sys"}

    def __init__(self):
        self.is_safe = True
        self.violations: List[str] = []

    def visit_Import(self, node: ast.Import):
        for alias in node.names:
            if alias.name.split('.')[0] in self.FORBIDDEN_MODULES:
                self.is_safe = False
                self.violations.append(f"安全拦截: 禁止导入高危系统模块 '{alias.name}'")
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom):
        if node.module and node.module.split('.')[0] in self.FORBIDDEN_MODULES:
            self.is_safe = False
            self.violations.append(f"安全拦截: 禁止从高危模块 '{node.module}' 导入函数")
        self.generic_visit(node)


class SelfHealingNotebookCopilot:
    """具备自愈纠错能力的 Notebook 助手核心"""

    def __init__(self, llm_mock_responder=None):
        self.llm_responder = llm_mock_responder

    def execute_in_sandbox(self, code: str, execution_env: Dict[str, Any]) -> Dict[str, Any]:
        """
        在受控环境中执行代码，捕获 stdout 输出与异常栈
        """
        # 1. 静态 AST 审查
        try:
            tree = ast.parse(code)
            validator = SafeCodeValidator()
            validator.visit(tree)
            if not validator.is_safe:
                return {"success": False, "error_type": "SecurityViolation", "traceback": "\n".join(validator.violations)}
        except SyntaxError as e:
            return {"success": False, "error_type": "SyntaxError", "traceback": f"代码语法解析错误: {e}"}

        # 2. 捕获标准输出
        stdout_capture = io.StringIO()
        old_stdout = sys.stdout
        sys.stdout = stdout_capture

        try:
            # 在同一运行时命名空间中执行
            exec(code, execution_env)
            output_text = stdout_capture.getvalue()
            return {"success": True, "output": output_text, "error_type": None, "traceback": None}
        except Exception as e:
            err_type = type(e).__name__
            tb_lines = traceback.format_exc()
            return {"success": False, "error_type": err_type, "traceback": tb_lines}
        finally:
            sys.stdout = old_stdout

    def run_analysis_loop(self, user_prompt: str, execution_env: Dict[str, Any], max_attempts: int = 3) -> Dict[str, Any]:
        """自愈分析主循环"""
        profiles = RuntimeInspector.inspect_namespace(execution_env)
        
        # 构造上下文画像描述
        profiles_desc = []
        for p in profiles:
            profiles_desc.append(f"- 变量 `{p.name}`: 形状 {p.shape}, 列结构: {json.dumps(p.columns_info, ensure_ascii=False)}")
        schema_context = "\n".join(profiles_desc)

        current_prompt = f"【当前内存数据画像】:\n{schema_context}\n【分析指令】: {user_prompt}\n请生成 Python 数据分析代码。"
        
        for attempt in range(1, max_attempts + 1):
            print(f"🔄 正在进行第 {attempt} 次生成与执行尝试...")
            
            # 调用大模型生成代码 (此处使用模拟响应器演示自愈逻辑)
            generated_code = self.llm_responder(current_prompt, attempt)
            print(f"📝 生成代码片段:\n```python\n{generated_code}\n```")

            exec_result = self.execute_in_sandbox(generated_code, execution_env)

            if exec_result["success"]:
                print("✅ 代码执行成功！输出结果:")
                print(exec_result["output"])
                return {
                    "status": "SUCCESS",
                    "final_code": generated_code,
                    "output": exec_result["output"],
                    "attempts": attempt
                }

            # 执行失败，组装错误信息触发自愈
            print(f"❌ 执行发生异常 [{exec_result['error_type']}]，准备触发自我纠错...")
            current_prompt += f"\n\n【上次生成的代码执行报错】:\n{exec_result['traceback']}\n请修正代码中存在的错误 (如修正列名、处理缺失值或类型转换)。"

        return {"status": "FAILED", "error": "超出最大自愈重试次数", "attempts": max_attempts}

生产自愈纠错演练（模拟列名写错自动修复）

# 1. 初始化用户的分析环境 (包含一个销售事实表)
mock_env = {
    "df_orders": pd.DataFrame({
        "order_sn": ["S001", "S002", "S003", "S004"],
        "channel": ["TikTok", "Tmall", "TikTok", "JD"],
        "pay_fee": [120.0, 350.0, 99.0, 480.0]  # 注意列名叫 pay_fee，而不是 pay_amount
    })
}

# 2. 模拟大模型首次犯错 (把 pay_fee 写成了 pay_amount)，在第二轮自愈中纠正
def mock_llm_with_healing(prompt: str, attempt: int) -> str:
    if attempt == 1:
        # 第一次尝试：犯经典错误 KeyError: 'pay_amount'
        return """
channel_summary = df_orders.groupby('channel')['pay_amount'].sum().reset_index()
print(channel_summary)
"""
    else:
        # 第二次尝试：感知到报错并修正为正确的列名 'pay_fee'
        return """
channel_summary = df_orders.groupby('channel')['pay_fee'].sum().reset_index()
print("【各渠道成交金额汇总】:")
print(channel_summary)
"""

# 3. 运行自愈分析助手
copilot = SelfHealingNotebookCopilot(llm_mock_responder=mock_llm_with_healing)
final_result = copilot.run_analysis_loop(
    user_prompt="统计各渠道的成交总金额并打印",
    execution_env=mock_env
)

print(f"\n【最终自愈状态】: {final_result['status']} (耗费尝试次数: {final_result['attempts']})")

四、生产避坑与安全合规防线

在企业内网部署 Notebook 对话式助手时，必须筑牢三道安全底线：

1. 严格限制沙箱文件系统与网络权限：
   通过 Linux 容器化（Docker / cgroups）或 Jupyter Enterprise Gateway 将代码执行环境与宿主机完全隔离，挂载只读文件系统，严禁访问内网未授权的敏感元数据库。
2. 生成的代码必须“显式落盘至单元格”供人复核：
   严禁将 AI 变成了“黑盒代理”。所有执行成功的代码必须以可见的 Code Cell 形式追加在 Notebook 中，数据科学家必须能够逐行 Review、微调参数并提交至 Git 仓库留存。
3. 敏感数据不出域与本地模型部署：
   对于金融、医疗等强合规场景，严禁使用公网商业 API。应在企业内部私有 GPU 集群上部署开源代码模型（如 CodeLlama、Qwen-Coder 或 DeepSeek-Coder），确保所有数据画像与代码生成全程在内网闭环流转。

通过将运行时元数据感知、AST 语法审查与自愈纠错机制深度融合，Notebook 智能助手能够真正成为数据科学家手中敏捷、安全且高度可靠的“分析副驾驶”。