07 Python 沙箱执行与分析——让 Agent 具备二次计算能力

起凡2026/3/30大约 4 分钟

上一章我们已经打通了 SQL_GENERATION -> SQL_EXECUTION 的主干链路，Agent 能查数了。
但很多真实问题并不止于“查出来”，而是要继续做比例、排名、统计、异常检测甚至聚合摘要。

这篇我们把 Agent 推进到下一阶段：让它在 Python 沙箱里对 SQL 结果做二次计算，并把分析结论回填到执行上下文。

🎯 本文目标

让 Agent 在 SQL 结果之上执行深度计算。
通过 Docker 沙箱保证 Python 依赖与运行环境稳定一致。
打通 PYTHON_GENERATION -> PYTHON_EXECUTION -> PYTHON_ANALYSIS 编排闭环。
明确前端如何可视化展示 Python 节点产物。

1. 为什么要引入 Python 阶段

纯 SQL 在很多场景会变得笨重，例如：

复杂业务指标（多步转换后再统计）。
灵活的数据清洗（缺失值、类型推断、边界值处理）。
需要程序化输出结构化 JSON 给后续报告节点消费。

因此我们把“SQL 查询”与“Python 分析”分层：

SQL 负责把原始数据拉出来
Python 负责做业务计算和表达

分层价值

SQL 保持可审计、可复现；Python 保持灵活表达，两者组合比“让 SQL 做一切”更稳健。

2. 图编排：从计划调度到 Python 三连节点

在图配置中，我们把 PLAN_EXECUTION 的条件分支扩展为可跳到 Python 路径：

.addConditionalEdges(
    DataAgentSpec.Graph.Node.PLAN_EXECUTION, edge_async(PlanExecutorEdge()),
    mapOf(
        DataAgentSpec.Graph.Node.SQL_GENERATION to DataAgentSpec.Graph.Node.SQL_GENERATION,
        DataAgentSpec.Graph.Node.PYTHON_GENERATION to DataAgentSpec.Graph.Node.PYTHON_GENERATION,
        END to END,
    )
)
.addEdge(DataAgentSpec.Graph.Node.PYTHON_GENERATION, DataAgentSpec.Graph.Node.PYTHON_EXECUTION)
.addEdge(DataAgentSpec.Graph.Node.PYTHON_EXECUTION, DataAgentSpec.Graph.Node.PYTHON_ANALYSIS)
.addEdge(DataAgentSpec.Graph.Node.PYTHON_ANALYSIS, END)

对应状态键：

object Execution {
    const val PYTHON_GENERATION_RESULT = "PYTHON_GENERATE_NODE_OUTPUT"
    const val PYTHON_EXECUTION_RESULT = "PYTHON_EXECUTE_NODE_OUTPUT"
}

3. 代码生成节点：`PythonGeneratorNode`

PythonGeneratorNode 的职责是根据：

当前执行步骤参数（plan_description）
数据库结构（database_schema）
SQL 执行样例数据（sample_input）

生成“可直接运行”的 Python 脚本，并写入 PYTHON_GENERATE_NODE_OUTPUT。

核心输入模板来自 python-generator.st。这里建议的约束是：

必须从 sys.stdin 读取 JSON 输入。
输出必须是一个 JSON 对象（print(json.dumps(result))）。
允许使用 continuumio/anaconda3 默认库（如 pandas）。
严禁文件、网络、系统调用等越权行为。

这样一来，模型生成代码时就“知道”它跑在受控沙箱里，而不是宿主机的任意 Python 环境。

4. 执行节点：`PythonExecuteNode`

PythonExecuteNode 负责把 SQL 结果喂给 Python 代码执行：

val pythonCode = state.value(DataAgentSpec.Graph.StateKey.Execution.PYTHON_GENERATION_RESULT, "")
val result = state.value(DataAgentSpec.Graph.StateKey.Execution.SQL_EXECUTION_RESULT, SqlExecuteNode.Result::class.java).orElseThrow()
val inputDataJson = JsonUtil.toJson(result.resultSet.data) ?: throw RuntimeException("sql结果为空")
val output = SimplePythonExecutor.execute(pythonCode, inputDataJson)
return mapOf(DataAgentSpec.Graph.StateKey.Execution.PYTHON_EXECUTION_RESULT to output)

这里输入与输出的契约非常清晰：

输入：List<Map<String, Any?>>（SQL 结果行集合）
输出：PythonExecutionResult(success, output, error)

5. 沙箱执行器：`SimplePythonExecutor`

这是本章关键。我们将执行环境改为强制 Docker 沙箱，避免“本地能跑、线上缺包”的经典问题。

5.1 核心策略

永远使用 docker run 执行 Python。
默认镜像：continuumio/anaconda3:latest。
默认隔离策略：
- --network none
- --cpus 1
- --memory 512m
- --pids-limit 128
- --security-opt no-new-privileges
工作目录只读挂载，减少脚本对宿主机影响面。

执行片段示例：

val dockerResult = runCommand(
    listOf(
        "docker", "run", "--rm", "-i",
        "--network", "none",
        "--cpus", "1",
        "--memory", memoryLimit,
        "--pids-limit", "128",
        "--security-opt", "no-new-privileges",
        "-v", "${workDir.absolutePath}:/work:ro",
        "-w", "/work",
        image,
        "python",
        "/work/script.py",
    ),
    dataFile,
    timeoutSec
)

运行前提

服务节点必须安装并启动 Docker。若 Docker 不可用，执行器会返回明确错误，而不是静默降级。

5.2 可配置项

可通过环境变量覆盖默认值：

DATA_AGENT_PYTHON_DOCKER_IMAGE：自定义 Python 沙箱镜像。
DATA_AGENT_PYTHON_MEMORY_LIMIT：覆盖容器内存限制（如 1g）。

6. 分析节点：`PythonAnalyzeNode`

PythonAnalyzeNode 读取 Python 执行结果和用户问题，生成文字化分析结论并追加到 PLAN_EXECUTE_NODE_OUTPUT：

val rewriteQuery = state.value(DataAgentSpec.Graph.StateKey.Recall.REWRITE_QUERY, "")
val pythonOutput = state.value(DataAgentSpec.Graph.StateKey.Execution.PYTHON_EXECUTION_RESULT, "")
val map = state.value<MutableMap<String, String>>(DataAgentSpec.Graph.StateKey.Planning.EXECUTION_OUTPUT).orElseThrow()
// ...
map["step_${step}_analysis"] = analyze

这一步的意义是：把“结构化输出”转成“可读结论”，便于最终报告节点直接复用。

7. 前端展示：A2UI 节点卡片如何接入

前端需要三件事：

在节点组件映射中注册 Python 三节点卡片。
补全 state key 常量。
为执行输出和分析输出分别做可读渲染。

组件映射示例：

[DATA_AGENT_GRAPH_NODE.PYTHON_GENERATION]: markRaw(PythonGenerationNodeCard),
[DATA_AGENT_GRAPH_NODE.PYTHON_EXECUTION]: markRaw(PythonExecutionNodeCard),
[DATA_AGENT_GRAPH_NODE.PYTHON_ANALYSIS]: markRaw(PythonAnalysisNodeCard),