代码即 SQL:校验、清洗与安全执行

30 秒导读: PandasAI 让 LLM 写 Python 来回答你的数据问题,但有一条铁规矩——这段 Python 不许自己碰 pandas 去读数,它取数据的唯一入口是一个叫 execute_sql_query 的函数。本章讲这条规矩是怎么被强制的:先用 AST 检查代码里"到底有没有调它",再用 AST 把代码改写干净(剥掉 LLM 偷偷重定义的函数、把 SQL 里的表名换成白名单里授权的名字、把出图路径改到临时文件),最后在一个只塞了 pd/plt/np 的字典里 exec,并校验它吐出的 result 结构对不对。

本章属于 PandasAI 子库,聚焦"生成代码之后、真正执行之前"这一段安全管道。它的上游(Agent 主循环怎么把这一段串起来)见 01-agent-loop.md;喂给 LLM 的 prompt 文本见 03-prompts.md;数据集 schema 怎么编译成 SQL 见 04-semantic-layer.md。这里不重复那几章的内容。

---

1. 这是什么(零基础也能懂)

一句话定义

PandasAI 让 LLM 生成一小段 Python 来回答数据问题,但这段 Python 只能通过一个受控函数 execute_sql_query(sql) 拿数据——本章就是保证"它真的这么干、而且没夹带私货"的那套校验与清洗管道。

为什么要这么设计

假设你把一张 DataFrame 交给 AI,问它"哪个产品线 Q4 卖得最好"。最直接的做法是让 LLM 直接写 df.groupby(...) 这类 pandas 代码然后执行。PandasAI 刻意不这么做,原因有二:

• 可控性。 LLM 直接写 pandas,想象空间太大——它可能读文件、发网络请求、import os 干任何事。把取数收窄成"只能写一句 SELECT,交给 execute_sql_query 去跑",攻击面一下就小了。
• 一致性。 不管底层是本地 CSV(用 DuckDB 在内存里跑)还是远程 Postgres,LLM 看到的都是"写 SQL"。取数路由的差异被 execute_sql_query 这层抹平了(详见 04-semantic-layer.md)。

一句话直觉

把 LLM 生成的 Python 想成一个信封:信封外面(Python 逻辑、画图、组织 result)你随便写,但信封里真正"去数据库取东西"的那张纸,必须是通过 execute_sql_query 递出去的一句 SELECT。本章的所有机制,都是在信封封口前拆开检查这张纸。

用起来什么样

LLM 被要求生成的代码,长得大概是这样:

# LLM 生成的典型代码(示意)
sql_query = "SELECT product_line, SUM(revenue) AS total FROM sales GROUP BY product_line"
df = execute_sql_query(sql_query)          # 唯一的取数入口
top = df.sort_values("total", ascending=False).head(1)
result = {"type": "dataframe", "value": top}   # 必须返回 result

注意两件事:取数走 execute_sql_query;结尾必须有个 result 字典。本章接下来讲的每一步,都是围绕这两条铁规矩展开的检查。

---

2. 顶层全景(这套管道怎么转)

从"LLM 吐出一段字符串"到"拿到最终结构化响应",代码要穿过五道关卡。方向从上到下,任何一关不过就抛异常、进重试(重试逻辑见 01-agent-loop.md)。

   LLM 生成的代码字符串
          │
          ▼
 ┌─────────────────────────────┐
 │ ① 校验:必须调用             │  CodeRequirementValidator.validate
 │   execute_sql_query          │  没调用 → 抛 ExecuteSQLQueryNotUsed
 └─────────────────────────────┘
          │
          ▼
 ┌─────────────────────────────┐
 │ ② 清洗:AST 改写             │  CodeCleaner.clean_code
 │  · 剥掉 LLM 重定义的函数     │  · 表名白名单替换(否则抛 MaliciousQueryError)
 │  · 出图路径改成 temp_chart   │  · 去掉 plt.show()
 └─────────────────────────────┘
          │
          ▼
 ┌─────────────────────────────┐
 │ ③ 装配环境                   │  Agent.execute_code
 │  把真正的 execute_sql_query  │  注入受限环境(只有 pd/plt/np)
 │  和 skills 注入 env          │
 └─────────────────────────────┘
          │
          ▼
 ┌─────────────────────────────┐
 │ ④ 执行:受限 exec            │  CodeExecutor.execute_and_return_result
 │  exec(code, env)             │  没有 result → 抛 NoResultFoundError
 └─────────────────────────────┘
          │
          ▼ (代码里调 execute_sql_query 时,实时发生)
 ┌─────────────────────────────┐
 │ ⑤ 运行期取数 + 安全网        │  Agent._execute_sql_query
 │  路由 DuckDB(本地)/远程    │  远程执行前再过 is_sql_query_safe
 └─────────────────────────────┘
          │
          ▼
 ┌─────────────────────────────┐
 │ ⑥ 解析 result                │  ResponseParser.parse
 │  映射成 number/string/       │  结构不对 → 抛 InvalidOutputValueMismatch
 │  dataframe/plot 四类          │
 └─────────────────────────────┘

各关卡的职责与落点:

关卡	干什么	文件 · 符号
① 校验	强制代码里出现 execute_sql_query 调用	pandasai/core/code_generation/code_validation.py · CodeRequirementValidator.validate
② 清洗	AST 改写:剥重定义、表名白名单、改图路径	pandasai/core/code_generation/code_cleaning.py · CodeCleaner.clean_code
③ 装配	把真 execute_sql_query / skills 注入 env	pandasai/agent/base.py · Agent.execute_code
④ 执行	受限 exec,取回 result 变量	pandasai/core/code_execution/code_executor.py · CodeExecutor
⑤ 取数	路由本地 DuckDB / 远程 executor + 安全网	pandasai/agent/base.py · Agent._execute_sql_query
⑥ 解析	result 字典 → 四类响应对象	pandasai/core/response/parser.py · ResponseParser

①②由 CodeGenerator.generate_code 在生成阶段串起来(先 validate 后 clean,pandasai/core/code_generation/base.py:54 validate_and_clean_code);③④⑤⑥在执行阶段发生。

---

3. 核心原理(逐个机制,由浅入深)

3.1 校验:凭什么断定"它真调了 execute_sql_query"

它要解决的小问题: LLM 可能答非所问——比如没写 SQL,直接编了个 result = {"type":"string","value":"我不知道"}。这种代码语法没错、也能跑,但它绕过了取数。得有办法把这类代码在执行前就拦下来。

思路: 不能靠字符串里 grep "execute_sql_query"(注释里、字符串里出现都会误判)。正确做法是把代码解析成 AST(抽象语法树),遍历所有函数调用节点,看名字集合里有没有它。

原理演示:

# 示意,非源码:用 AST 收集所有被调用的函数名
import ast

class CallCollector(ast.NodeVisitor):
    def __init__(self):
        self.calls = []
    def visit_Call(self, node):
        if isinstance(node.func, ast.Name):        # foo(...)
            self.calls.append(node.func.id)
        self.generic_visit(node)                   # 继续往下走

tree = ast.parse(code)
c = CallCollector(); c.visit(tree)
assert "execute_sql_query" in c.calls   # 不在就说明没取数

真实实现: CodeRequirementValidator 内嵌一个 _FunctionCallVisitor,visit_Call 同时处理两种形式——普通调用 foo()(记 node.func.id)和属性调用 obj.foo()(记成 "obj.foo")pandasai/core/code_generation/code_validation.py:20-30。收集完后做一句判断:

# code_validation.py:62-65
if "execute_sql_query" not in func_call_visitor.function_calls:
    raise ExecuteSQLQueryNotUsed(
        "The code must execute SQL queries using the `execute_sql_query` function, ..."
    )

validate 通过则返回 True,pandasai/core/code_generation/code_validation.py:41-67。抛出的 ExecuteSQLQueryNotUsed 会被上层捕获并触发重试。

关键细节: 这一步只保证"名字出现过",不保证参数是安全 SQL——那是后面清洗和运行期安全网的事。校验早于清洗:validate_and_clean_code 先 validate 再 clean(pandasai/core/code_generation/base.py:57-63)。

3.2 清洗:AST 改写,把代码"洗干净"再放行

它要解决的小问题: 校验只是"看",清洗是"改"。LLM 生成的代码有四类脏东西必须处理掉,而且都得在语法层面精确地改,不能靠字符串暴力替换。CodeCleaner.clean_code 把它们一次性做完(pandasai/core/code_generation/code_cleaning.py:138-169)。

清洗做的四件事:

脏东西	危害	怎么洗
LLM 重新定义 execute_sql_query	它可能覆盖掉宿主注入的真函数,绕过取数路由	AST 遍历时跳过这个 FunctionDef 节点
LLM 重新定义某个 skill 函数	同上,覆盖宿主提供的技能	同上,跳过
SQL 里的表名(大小写/未授权)	引用了不存在或未授权的表	表名白名单替换,不在白名单就抛 MaliciousQueryError
出图路径 / plt.show()	写到任意路径 / 阻塞式弹窗	路径改写成临时 temp_chart;plt.show() 删掉

(a) 剥掉 LLM 重定义的函数

宿主会在执行前把真正的 execute_sql_query 注入环境(见 3.3)。如果 LLM 代码里自己又 def execute_sql_query(...),exec 时后者会覆盖前者——取数路由就被劫持了。清洗的做法是在遍历 AST 顶层节点时,遇到这种定义就 continue 跳过,不放进新代码:

# code_cleaning.py:156-166(节选)
for node in tree.body:
    if self._check_direct_sql_func_def_exists(node):   # def execute_sql_query
        continue
    if self._check_if_skill_func_def_exists(node):     # def <某个 skill 名>
        continue
    node = self._validate_and_make_table_name_case_sensitive(node)
    new_body.append(node)

_check_direct_sql_func_def_exists 判断节点是否是 FunctionDef 且名叫 execute_sql_query(pandasai/core/code_generation/code_cleaning.py:27-31);_check_if_skill_func_def_exists 则拿节点名去和已注册 skill 名逐一比对(pandasai/core/code_generation/code_cleaning.py:33-40)。

(b) 表名白名单替换:最硬的一道防线

思路: LLM 写的 SQL 里的表名不能照单全收——可能大小写不对,也可能引用了根本没授权的表。清洗要把每一个表名都换成"授权表名字典"里的规范写法;只要有一个表名不在字典里,直接抛 MaliciousQueryError。

处理路径是 _validate_and_make_table_name_case_sensitive → _clean_sql_query → _replace_table_names:

• 在哪找 SQL: _validate_and_make_table_name_case_sensitive 认三种承载 SQL 的 AST 形态——赋值给名为 sql_query/query 的字符串常量(pandasai/core/code_generation/code_cleaning.py:79-87)、x = execute_sql_query("...") 里的字符串实参(:88-97)、以及独立表达式 execute_sql_query("...") 里的实参(:99-108)。
• 提表名 + 建白名单: _clean_sql_query 先 rstrip(";") 去掉尾分号,用 SQLParser.extract_table_names 按方言解析出所有表名,再把授权数据集的 schema 名做成字典(裸名和带引号名都收)pandasai/core/code_generation/code_cleaning.py:62-73:

# code_cleaning.py:69-71
allowed_table_names = {
    df.schema.name: df.schema.name for df in self.context.dfs
} | {f'"{df.schema.name}"': df.schema.name for df in self.context.dfs}

• 逐个替换或拒绝: _replace_table_names 对每个表名用词边界正则 \btable\b 替换成授权名;遇到不在白名单里的表名,立刻抛异常 pandasai/core/code_generation/code_cleaning.py:52-59:

# code_cleaning.py:52-59
for table_name in table_names:
    if table_name in allowed_table_names:
        quoted_table_name = allowed_table_names[table_name]
        sql_query = regex_patterns[table_name].sub(quoted_table_name, sql_query)
    else:
        raise MaliciousQueryError(f"Query uses unauthorized table: {table_name}.")

改写后的 SQL 字符串被写回 AST 节点(如 node.value.value = sql_query,pandasai/core/code_generation/code_cleaning.py:87),再由 astor.to_source 重新生成 Python 源码(:169)。

(c) 出图路径改写 + 去掉 plt.show

这两步在 clean_code 一开头、AST 解析之前用正则做:

• _replace_output_filenames_with_temp_chart 把代码里任何 '...png' 字符串统一替换成 exports/charts/temp_chart_<uuid>.png,不让 LLM 把图写到任意路径 pandasai/core/code_generation/code_cleaning.py:171-182。路径前缀 DEFAULT_CHART_DIRECTORY = exports/charts(pandasai/constants.py:10)。
• plt.show() 直接正则删空 pandasai/core/code_generation/code_cleaning.py:151——服务端跑图不需要弹窗,留着会阻塞。

关键细节: 图路径用 uuid 命名避免并发冲突;路径里的反斜杠会被转义两次(:177)以适配 Windows 与后续字符串嵌入。

3.3 装配 + 执行:受限环境里的 exec

它要解决的小问题: 清洗后的代码要跑,但不能给它一个能 import os 干任何事的完整 Python 环境。同时,代码里那个 execute_sql_query 调用得真的连到宿主的取数逻辑。

思路: exec(code, env) 的第二个参数 env 是这段代码能看到的全部全局符号。只往 env 里塞该塞的东西,就等于给了它一个"白名单命名空间"。

装配(注入真函数): Agent.execute_code 新建一个 CodeExecutor,然后把 _execute_sql_query(宿主的真实取数方法)以 "execute_sql_query" 之名注入,再逐个注入 skillspandasai/agent/base.py:123-135:

# agent/base.py:127-130
code_executor = CodeExecutor(self._state.config)
code_executor.add_to_env("execute_sql_query", self._execute_sql_query)
for skill in self._state.skills:
    code_executor.add_to_env(skill.name, skill.func)

受限环境的底子: CodeExecutor 初始化时把环境设成 get_environment() 的返回——一个只有三个键的字典:pd(pandas)、plt(matplotlib.pyplot)、np(numpy)pandasai/core/code_execution/environment.py:22-34:

# environment.py:28-32
env = {
    "pd": import_dependency("pandas"),
    "plt": import_dependency("matplotlib.pyplot"),
    "np": import_dependency("numpy"),
}

于是 LLM 代码能用的全局符号,就是 pd/plt/np + 注入的 execute_sql_query + skills,别的没有。

执行 + 取结果: execute_and_return_result 调 exec(code, self._environment),任何异常包成 CodeExecutionError;跑完检查环境里有没有 result,没有就抛 NoResultFoundError(错误信息直接教 LLM 该返回什么格式,便于重试自愈)pandasai/core/code_execution/code_executor.py:27-46。

关键细节(边界要诚实): exec 加一个自定义 globals 字典是软隔离,不是真沙箱——它挡不住 __import__("os") 之类。真正的隔离要靠可选的 Sandbox(见 3.5)。所以 3.2 的 AST 清洗和 3.4 的运行期安全网,是这层软隔离的必要补强。

3.4 运行期取数:路由 + SQL 安全网

它要解决的小问题: LLM 代码里那句 execute_sql_query(sql) 真被调用时,数据到底从哪来?本地 CSV 和远程数据库走的是完全不同的路。而且——清洗阶段的白名单只管表名,不管 SQL 语句本身是不是只读。真正执行前还得再拦一道:确保它是纯 SELECT。

路由逻辑: Agent._execute_sql_query 遍历所有注册的 DataFrame,分两类处理 pandasai/agent/base.py:137-169:

DataFrame 类型	判据	取数方式
语义层数据集(远程)	有 query_builder 属性	用它的 execute_sql_query(远程 executor)
CSV 等本地数据	无 query builder	register 进本地 DuckDB,内存里跑

# agent/base.py:155-169(节选)
for df in self._state.dfs:
    if hasattr(df, "query_builder"):
        table_mapping[df.schema.name] = df.query_builder._get_table_expression()
        df_executor = df.execute_sql_query
    else:
        db_manager.register(df.schema.name, df)

final_query = SQLParser.replace_table_and_column_names(query, table_mapping)
if not df_executor:
    return db_manager.sql(final_query).df()   # 本地 DuckDB
else:
    return df_executor(final_query)           # 远程

表名重写: SQLParser.replace_table_and_column_names 用 sqlglot 把查询解析成表达式树,把每个 Table 节点按 table_mapping 换成新表名或子查询,再统一给标识符加引号 pandasai/query_builders/sql_parser.py:12-57。这一步把 LLM 眼里的"逻辑表名"翻译成底层真实的表/视图表达式(schema→SQL 编译的细节见 04-semantic-layer.md)。

运行期安全网 is_sql_query_safe: 这是只读保证的最后一道。它做两件事 pandasai/helpers/sql_sanitizer.py:40-111:

• 只允许 SELECT: sqlglot 解析后,顶层语句类型不是 SELECT 直接返回 False(pandasai/helpers/sql_sanitizer.py:89-91)。
• 关键字黑名单: 一长串写操作/元数据/探测关键字(INSERT/UPDATE/DELETE/DROP/ALTER/CREATE/EXEC/GRANT/SLEEP/-- 注释/块注释 等),主查询和子查询里出现任何一个都返回 False``pandasai/helpers/sql_sanitizer.py:43-79、:99-106。解析失败(ParseError)也判不安全 :110-111。

这道网在远程 SQL loader 执行前被调用:SQLDatasetLoader.execute_query 里 transpile 完方言就检查,不安全就抛 MaliciousQueryError``pandasai/data_loader/sql_loader.py:46-49(view/local loader 里也各有一处调用)。

关键细节: 白名单表名替换(3.2,生成期)和 is_sql_query_safe(运行期)是两道独立的防线——前者管"表能不能碰",后者管"语句是不是只读"。两者叠加才覆盖住"未授权表 + 危险语句"两类风险。

3.5 可选沙箱:把 exec 挪到隔离环境

它要解决的小问题: 3.3 说过,给 exec 一个受限 globals 挡不住 __import__。对更强的隔离需求,PandasAI 留了一个 Sandbox 抽象基类,让代码在真正隔离的进程/容器里跑。

接口: Sandbox 是抽象类,start / stop / _exec_code / transfer_file 都留给子类实现;execute 负责"没启动就先 start,再 _exec_code"pandasai/sandbox/sandbox.py:4-27。它还自带两个工具:_extract_sql_queries_from_code 用 AST 从代码里抠出 SQL 字符串(:29-72)、_compile_code 编译成模块以校验语法(:74-89)。

接入点: Agent.execute_code 里,若构造 Agent 时传了 sandbox,就走沙箱执行、把清洗后的代码和环境交给它;否则才用普通的 CodeExecutor``pandasai/agent/base.py:132-135:

# agent/base.py:132-135
if self._sandbox:
    return self._sandbox.execute(code, code_executor.environment)
return code_executor.execute_and_return_result(code)

关键细节: 沙箱是可选加固,不是默认路径。默认路径靠的仍是 AST 校验/清洗 + 受限 env + SQL 安全网这套组合拳。

3.6 响应解析:result 字典 → 四类对象

它要解决的小问题: 代码执行吐出的 result 只是个裸字典。得校验它结构合法,再映射成强类型的响应对象,好让调用方拿到 number / string / dataframe / plot 之一。

四类映射: ResponseParser.parse 先校验后生成 pandasai/core/response/parser.py:16-30:

result["type"]	value 合法类型	生成对象
"number"	int / float / np.int64	NumberResponse
"string"	str	StringResponse
"dataframe"	pd.DataFrame / pd.Series / dict	DataFrameResponse
"plot"	str(路径)/ dict / base64	ChartResponse

校验逻辑: _validate_response 先确认 result 是含 type+value 的字典(否则抛 InvalidOutputValueMismatch),再按 type 逐项检查 value 类型 pandasai/core/response/parser.py:32-73。plot 类型最讲究:允许 base64 图或 dict 直接放行(:63-67),否则要求 value 匹配一个"合法文件路径"正则(:69-73)——这跟 3.2 里把图写到 temp_chart 路径的约定对上了。类型不认识就抛异常(:29-30),同样会触发重试。

---

4. 巧妙之处(可借鉴的技术)

• 用 AST 而非字符串做安全检查。 无论"有没有调 execute_sql_query"(3.1)还是"有没有偷偷重定义它"(3.2),都在语法树层面判定,避开了注释/字符串误伤和字符串替换的脆弱。pandasai/core/code_generation/code_validation.py:20-30、pandasai/core/code_generation/code_cleaning.py:27-40。
• 白名单默认拒绝。 表名替换是"不在名单就抛异常",而不是"尽量修一修"——安全默认值应当是 fail-closed。pandasai/core/code_generation/code_cleaning.py:57-59。
• 纵深防御,多道独立防线。 生成期白名单(管表)+ 运行期 is_sql_query_safe(管只读)+ 受限 env(管命名空间)+ 可选 Sandbox(管进程隔离),任一层被绕过还有下一层。pandasai/helpers/sql_sanitizer.py:40-111。
• 错误信息本身是"给 LLM 的修复提示"。 NoResultFoundError 直接告诉模型 result 该长啥样(pandasai/core/code_execution/code_executor.py:41-44),配合 01-agent-loop.md 的重试形成自愈闭环。

---

5. 边界与局限(诚实)

• 受限 env 不是真沙箱。 exec(code, {"pd":..,"plt":..,"np":..}) 只是限制了可见符号,挡不住 __import__("os")、().__class__.__bases__ 这类逃逸。要真隔离得自己实现 Sandbox 子类——仓库里 Sandbox 只是抽象基类,_exec_code 未提供默认实现(pandasai/sandbox/sandbox.py:21-22)。
• 关键字黑名单是启发式。 is_sql_query_safe 靠正则匹配一串关键字(pandasai/helpers/sql_sanitizer.py:43-79)。黑名单式防御天然可能被绕过(编码、方言差异、大小写混淆),它是"额外一层网"而非唯一保证——真正的边界收窄靠"只允许顶层 SELECT"这条结构判定。
• 表名白名单只校验表名,不校验列/函数。 3.2 的替换只覆盖 Table 节点;SQL 里调用的函数、列的合法性不在这一步管。
• 校验只认名字,不认调用是否"有效"。 3.1 只要求 execute_sql_query 出现在调用集合里;它不检查那次调用的参数是不是真 SQL、结果有没有被用上。

---

6. 代码地图(导航索引)

主题	文件路径	符号名
AST 校验:强制调用 execute_sql_query	pandasai/core/code_generation/code_validation.py	CodeRequirementValidator.validate · _FunctionCallVisitor.visit_Call
生成期"先校验后清洗"串接	pandasai/core/code_generation/base.py	CodeGenerator.validate_and_clean_code
AST 清洗主流程	pandasai/core/code_generation/code_cleaning.py	CodeCleaner.clean_code
剥掉重定义的函数	pandasai/core/code_generation/code_cleaning.py	_check_direct_sql_func_def_exists · _check_if_skill_func_def_exists
SQL 表名白名单替换	pandasai/core/code_generation/code_cleaning.py	_validate_and_make_table_name_case_sensitive · _clean_sql_query · _replace_table_names
出图路径改 temp_chart	pandasai/core/code_generation/code_cleaning.py	_replace_output_filenames_with_temp_chart
受限执行环境(pd/plt/np)	pandasai/core/code_execution/environment.py	get_environment
exec + 取 result	pandasai/core/code_execution/code_executor.py	CodeExecutor.execute_and_return_result
注入真 execute_sql_query / skills	pandasai/agent/base.py	Agent.execute_code
运行期取数路由(DuckDB / 远程)	pandasai/agent/base.py	Agent._execute_sql_query
SQL 表名重写(sqlglot)	pandasai/query_builders/sql_parser.py	SQLParser.replace_table_and_column_names · extract_table_names
运行期 SQL 安全网(只允许 SELECT)	pandasai/helpers/sql_sanitizer.py	is_sql_query_safe
远程执行前的安全网调用点	pandasai/data_loader/sql_loader.py	SQLDatasetLoader.execute_query
可选沙箱抽象	pandasai/sandbox/sandbox.py	Sandbox.execute · _exec_code · _extract_sql_queries_from_code
响应解析(四类)	pandasai/core/response/parser.py	ResponseParser.parse · _validate_response · _generate_response