第 6 章 · 01 自研 AST 解释器 ★

本节摘要:全书最硬核的一节。local_python_executor.py(1768 行)是 CodeAgent 的心脏:为什么不直接 exec——exec 一旦启动就无法逐步拦截、无法计数操作、无法限制时长、无法控制导入;smolagents 的答案是把 LLM 生成的代码 ast.parse 成语法树,再用自己的 evaluate_ast 逐节点解释执行:赋值、函数调用、for 循环、if 分支、推导式、属性访问各有专属处理函数。配合 MAX_OPERATIONS 操作计数、MAX_WHILE_ITERATIONS 循环上限、30 秒线程超时三重资源闸门,以及工具函数注入 state/static_tools 的命名空间设计、DANGEROUS_MODULES 导入黑名单,构成入口函数 evaluate_python_code。本节把这条链路从头走到尾。

内容来源:src/smolagents/local_python_executor.py、src/smolagents/utils.py(BASE_BUILTIN_MODULES)

⚠️ 注意:这套解释器是「防意外」设施,不是「防恶意」沙箱——官方文档明确写着 It is not a security sandbox。安全边界的完整讨论留给 6.2 节,请勿跳读。

学习目标

说清不直接 exec 的四个理由。
读懂 evaluate_ast 的节点分发与五类核心处理函数。
掌握三重资源限制的实现方式。
理解命名空间三层结构(state/static_tools/custom_tools)与导入检查。
走通 evaluate_python_code 从源码字符串到 CodeOutput 的全流程。

一、为什么不直接 exec

CodeAgent 让 LLM 写 Python 代码当动作,执行前你面对的是一段来历可疑的字符串。最 naive 的方案:


exec(llm_generated_code)  # 危险且不可控

exec 的问题不在「能不能跑」,而在「跑起来之后你两手空空」:

无法逐步拦截。exec 把控制权整个交给解释器,代码跑在原生 Python 里,你在门外连一行日志都插不进去;而逐节点解释让每个 ast.Assign、每个 ast.Call 都过一遍你的检查函数。
无法计数操作。一段 while True: pass 或巨大循环会无限吃 CPU;解释器模式下每求值一个节点计数 +1,到 MAX_OPERATIONS(1000 万)直接叫停。
无法限制时长。exec 没有超时参数;解释器外面可以套线程超时(30 秒)。
无法控制导入与调用面。exec 里 import os; os.system("rm -rf ~") 畅通无阻;解释器可以逐条检查 import 语句是否在白名单、函数名是否在工具表里。

一句话:exec 是「黑盒放行」,AST 解释是「白盒放行」。注意它采用的并非继承 ast.NodeVisitor,而是在 evaluate_ast 里用一长串 isinstance 分发到各自的 evaluate_xxx 函数——本质是手写的访问者模式,好处是每个处理函数签名统一、方便在分发点统一计数。

二、三重资源闸门与顶层常量

文件开头的常量定义了所有闸门:


DEFAULT_MAX_LEN_OUTPUT = 50000
MAX_OPERATIONS = 10000000
MAX_WHILE_ITERATIONS = 1000000
MAX_EXECUTION_TIME_SECONDS = 30
ALLOWED_DUNDER_METHODS = ["__init__", "__str__", "__repr__"]

闸门一:操作计数。evaluate_ast 的第一件事:


if state.setdefault("_operations_count", {"counter": 0})["counter"] >= MAX_OPERATIONS:
    raise InterpreterError(
        f"Reached the max number of operations of {MAX_OPERATIONS}. Maybe there is an infinite loop somewhere in the code, or you're just asking too many calculations."
    )
state["_operations_count"]["counter"] += 1

计数器藏在 state 字典里,每解释一个 AST 节点 +1——while True 转不成 AST 里的「一亿次操作」也逃不过计数。闸门二:while 迭代上限,evaluate_while 里每轮 +1,超过 MAX_WHILE_ITERATIONS 抛错。闸门三:线程超时:


def timeout(timeout_seconds: int):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            with ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(func, *args, **kwargs)
                try:
                    return future.result(timeout=timeout_seconds)
                except FuturesTimeoutError:
                    raise ExecutionTimeoutError(
                        f"Code execution exceeded the maximum execution time of {timeout_seconds} seconds"
                    )
        return wrapper
    return decorator

为什么用线程而不是 signal?源码注释答了:跨平台(Windows 没有 SIGALRM)且线程安全。注释也坦承局限——Python 杀不掉线程,超时后那个线程仍在后台跑完,但调用方已拿到 TimeoutError 继续。三重闸门覆盖三种失控:节点级(计数)、循环级(迭代上限)、时间级(30 秒)。

三、evaluate_ast:节点分发中枢

分发中枢是一个巨型 if-elif,节选主干感受结构:


@safer_eval
def evaluate_ast(expression, state, static_tools, custom_tools, authorized_imports=BASE_BUILTIN_MODULES):
    if state.setdefault("_operations_count", {"counter": 0})["counter"] >= MAX_OPERATIONS:
        raise InterpreterError(...)
    state["_operations_count"]["counter"] += 1
    common_params = (state, static_tools, custom_tools, authorized_imports)
    if isinstance(expression, ast.Assign):
        return evaluate_assign(expression, *common_params)
    elif isinstance(expression, ast.Call):
        return evaluate_call(expression, *common_params)
    elif isinstance(expression, ast.Constant):
        return expression.value
    elif isinstance(expression, ast.ListComp):
        return evaluate_listcomp(expression, *common_params)
    elif isinstance(expression, ast.For):
        return evaluate_for(expression, *common_params)
    elif isinstance(expression, ast.If):
        return evaluate_if(expression, *common_params)
    elif isinstance(expression, ast.Attribute):
        return evaluate_attribute(expression, *common_params)
    ...
    else:
        raise InterpreterError(f"{expression.__class__.__name__} is not supported.")

注意装饰器 @safer_eval——每次求值返回值都要过 check_safer_result(模块白名单、危险函数名单检查,6.2 节细讲)。兜底分支很关键:没实现的语法直接报「not supported」,绝不静默降级。

3.1 赋值:set_value 与工具名保护


def evaluate_assign(assign, state, static_tools, custom_tools, authorized_imports) -> Any:
    result = evaluate_ast(assign.value, state, static_tools, custom_tools, authorized_imports)
    if len(assign.targets) == 1:
        target = assign.targets[0]
        set_value(target, result, state, static_tools, custom_tools, authorized_imports)
    ...
    return result

def set_value(target, value, state, ...):
    if isinstance(target, ast.Name):
        if target.id in static_tools:
            raise InterpreterError(f"Cannot assign to name '{target.id}': doing this would erase the existing tool!")
        state[target.id] = value
    elif isinstance(target, ast.Tuple):
        ...  # 解包:检查可迭代性与长度,递归 set_value
    elif isinstance(target, ast.Subscript):
        obj = evaluate_ast(target.value, ...); key = evaluate_ast(target.slice, ...)
        obj[key] = value
    elif isinstance(target, ast.Attribute):
        obj = evaluate_ast(target.value, ...)
        setattr(obj, target.attr, value)

最妙的一行:给 static_tools 里的名字赋值直接报错——防止 LLM 写 web_search = lambda q: "假装搜过了" 把真工具偷梁换柱。state 就是「变量表」:所有赋值都落到这个字典,跨执行持久(LocalPythonExecutor 每步复用同一个 state,上一步算的变量这一步还在)。

3.2 函数调用:白名单查找链


def evaluate_call(call, state, static_tools, custom_tools, authorized_imports) -> Any:
    ...
    elif isinstance(call.func, ast.Name):
        func_name = call.func.id
        if func_name in state:
            func = state[func_name]
        elif func_name in static_tools:
            func = static_tools[func_name]
        elif func_name in custom_tools:
            func = custom_tools[func_name]
        elif func_name in ERRORS:
            func = ERRORS[func_name]
        else:
            raise InterpreterError(
                f"Forbidden function evaluation: '{call.func.id}' is not among the explicitly allowed tools or defined/imported in the preceding code"
            )

调用裸函数名时按 state(用户变量)→ static_tools(agent 工具+基础函数)→ custom_tools(代码内定义的函数)→ ERRORS(异常类,允许 raise ValueError 之类)的顺序查找;四层都 miss 直接拒绝。调用前的两道安检:


if (inspect.getmodule(func) == builtins) and inspect.isbuiltin(func) and (func not in static_tools.values()):
    raise InterpreterError(
        f"Invoking a builtin function that has not been explicitly added as a tool is not allowed ({func_name})."
    )
if (
    hasattr(func, "__name__")
    and func.__name__.startswith("__") and func.__name__.endswith("__")
    and (func.__name__ not in static_tools)
    and (func.__name__ not in ALLOWED_DUNDER_METHODS)
):
    raise InterpreterError(f"Forbidden call to dunder function: {func.__name__}")

内置函数(如 eval/exec 本尊)没进工具表不许调;dunder 函数(__import__ 等)一律禁止,白名单只放行 __init__/__str__/__repr__ 三个。另外 print 被特判:输出追加进 state["_print_outputs"](一个 PrintContainer),而不是打到控制台——这样执行日志能完整回流进 ActionStep 的 observations。

3.3 for、if 与 break/continue 的异常实现


def evaluate_for(for_loop, state, static_tools, custom_tools, authorized_imports) -> Any:
    result = None
    iterator = evaluate_ast(for_loop.iter, state, static_tools, custom_tools, authorized_imports)
    for counter in iterator:
        set_value(for_loop.target, counter, state, static_tools, custom_tools, authorized_imports)
        for node in for_loop.body:
            try:
                line_result = evaluate_ast(node, state, static_tools, custom_tools, authorized_imports)
                if line_result is not None:
                    result = line_result
            except BreakException:
                return result
            except ContinueException:
                break
    return result

外层还是真 for,内层逐节点解释;循环变量经 set_value 落进 state。break/continue 用专门异常实现——evaluate_ast 里 ast.Break 节点 raise BreakException(),由循环处理器捕获。同理函数的 return:


class ReturnException(Exception):
    def __init__(self, value):
        self.value = value

create_function 把 def 编译成闭包:调用时拷贝 state、绑定参数、逐语句执行、捕获 ReturnException 拿返回值。用异常实现控制流是树遍历解释器的经典手法——break/continue/return 会穿越多层递归的 evaluate_ast 直达对应的处理器,不用层层传递标志。

3.4 推导式与属性访问

推导式统一由 _evaluate_comprehensions 递归处理(支持嵌套 for 与 if 过滤):


def _evaluate_comprehensions(comprehensions, evaluate_element, state, ...):
    if not comprehensions:
        yield evaluate_element(state)
        return
    comprehension = comprehensions[0]
    iter_value = evaluate_ast(comprehension.iter, state, ...)
    for value in iter_value:
        new_state = state.copy()
        set_value(comprehension.target, value, new_state, ...)
        if all(evaluate_ast(if_clause, new_state, ...) for if_clause in comprehension.ifs):
            yield from _evaluate_comprehensions(comprehensions[1:], evaluate_element, new_state, ...)

注意 state.copy()——推导式变量在副本上求值,不污染外层作用域,语义与 CPython 一致。list/set/dict 推导式与生成器表达式分别用 list()/set()/dict()/generator() 包装这个生成器。属性访问则先过 dunder 检查:


def evaluate_attribute(expression, state, ...):
    if expression.attr.startswith("__") and expression.attr.endswith("__"):
        raise InterpreterError(f"Forbidden access to dunder attribute: {expression.attr}")
    value = evaluate_ast(expression.value, state, ...)
    return getattr(value, expression.attr)

obj.__class__、obj.__globals__ 这类逃逸口全部焊死——为什么必须焊死,6.2 节展开。补一个贴心设计:evaluate_name 里变量名打错时,用 difflib 找相近名字提示「Maybe you meant one of these」——LLM 拼错变量名能立刻自我纠正,省一整轮重试。

四、命名空间与导入检查

解释器的可见世界由三个字典 + 一张白名单构成:

state:变量表。初始含 "__name__": "__main__" 与 _print_outputs、_operations_count 等内部槽位;agent 每步注入的任务变量也在这里。
static_tools:静态工具表 = agent 工具 + BASE_PYTHON_TOOLS + 额外函数。BASE_PYTHON_TOOLS 白名单化了约 50 个基础函数(print/isinstance/range/len/sorted/map...)加全套 math 函数,其中 getattr 被替换成 nodunder_getattr(拦 dunder 访问);只读——代码里赋值覆盖会被 set_value 拒绝。
custom_tools:代码里 def 定义的函数,可覆盖。
authorized_imports:导入白名单。BASE_BUILTIN_MODULES 是默认集:collections、datetime、itertools、math、queue、random、re、stat、statistics、time、unicodedata——全是纯计算,无一能碰网络或文件系统之外的世界。

导入的处理在 evaluate_import,以 check_import_authorized 把关(把白名单构建成前缀树,支持 "*" 通配与 os.path 这类前缀授权):


def evaluate_import(expression, state, authorized_imports):
    if isinstance(expression, ast.Import):
        for alias in expression.names:
            if check_import_authorized(alias.name, authorized_imports):
                raw_module = import_module(alias.name)
                state[alias.asname or alias.name] = get_safe_module(raw_module, authorized_imports)
            else:
                raise InterpreterError(
                    f"Import of {alias.name} is not allowed. Authorized imports are: {str(authorized_imports)}"
                )
    ...

白名单外的导入当场抛 InterpreterError(黑名单机制与 DANGEROUS_MODULES 清单见 6.2 节)。放行的模块还要过 get_safe_module——重建一个浅拷贝模块,只复制属性引用(遇到加载失败或循环引用跳过),拿不到原模块对象本身,堵住「改模块属性反噬主进程」的路。CodeAgent(additional_authorized_imports=["pandas"]) 就是在这层扩白名单,LocalPythonExecutor 初始化时会先检查这些模块确实装了(_check_authorized_imports_are_installed),免得跑到一半才 ImportError。

五、evaluate_python_code:入口与出口

一切汇于入口函数:


def evaluate_python_code(
    code: str,
    static_tools: dict[str, Callable] | None = None,
    custom_tools: dict[str, Callable] | None = None,
    state: dict[str, Any] | None = None,
    authorized_imports: list[str] = BASE_BUILTIN_MODULES,
    max_print_outputs_length: int = DEFAULT_MAX_LEN_OUTPUT,
    timeout_seconds: int | None = MAX_EXECUTION_TIME_SECONDS,
):
    try:
        expression = ast.parse(code)
    except SyntaxError as e:
        raise InterpreterError(
            f"Code parsing failed on line {e.lineno} due to: {type(e).__name__}: {str(e)}\n"
            f"{e.text}"
            f"{' ' * (e.offset or 0)}^"
        )
    ...
    state["_print_outputs"] = PrintContainer()
    state["_operations_count"] = {"counter": 0}

    if "final_answer" in static_tools:
        previous_final_answer = static_tools["final_answer"]

        def final_answer(*args, **kwargs):
            raise FinalAnswerException(previous_final_answer(*args, **kwargs))

        static_tools["final_answer"] = final_answer

    def _execute_code():
        result = None
        try:
            for node in expression.body:
                result = evaluate_ast(node, state, static_tools, custom_tools, authorized_imports)
            ...
            return result, False
        except FinalAnswerException as e:
            ...
            return e.value, True
        except Exception as e:
            ...
            raise InterpreterError(
                f"Code execution failed at line '{ast.get_source_segment(code, node)}' due to: {type(e).__name__}: {e}"
            )

    if timeout_seconds is not None:
        _execute_code = timeout(timeout_seconds)(_execute_code)
    return _execute_code()

流程六步:ast.parse(语法错误连同行号、箭头定位一起报给 LLM)→ 初始化 state 内部槽位 → 猴子补丁 final_answer(调用即抛 FinalAnswerException,它继承 BaseException 正是为了不被 agent 代码里的 except Exception 吞掉)→ 逐语句 evaluate_ast → 捕获终止/异常 → 套 30 秒超时后执行。返回 (result, is_final_answer) 二元组。真正的终止信号机制至此闭环:第 5.3 节那个五行空壳工具,在解释器这里变成控制流炸弹。

LocalPythonExecutor 把这些打包成可复用的执行器对象:


class LocalPythonExecutor(PythonExecutor):
    def __init__(self, additional_authorized_imports, ...):
        self.custom_tools = {}
        self.state = {"__name__": "__main__"}
        self.authorized_imports = list(set(BASE_BUILTIN_MODULES) | set(self.additional_authorized_imports))
        ...

    def __call__(self, code_action: str) -> CodeOutput:
        output, is_final_answer = evaluate_python_code(
            code_action,
            static_tools=self.static_tools,
            custom_tools=self.custom_tools,
            state=self.state,
            authorized_imports=self.authorized_imports,
            ...
        )
        logs = str(self.state["_print_outputs"])
        return CodeOutput(output=output, logs=logs, is_final_answer=is_final_answer)

    def send_tools(self, tools: dict[str, Tool]):
        self.static_tools = {**tools, **BASE_PYTHON_TOOLS.copy(), **self.additional_functions}

state 与 custom_tools 挂在实例上跨调用存活——agent 第一步算出的 DataFrame,第二步直接用。CodeAgent 每步的代码动作就是调 self.python_executor(code_action),拿回 CodeOutput(结果/日志/是否最终答案),再渲染进 ActionStep 的 observations。

💡 阶梯要点:本阶是全书技术含量的顶点。心智模型:LLM 的代码先被 parse 成树,再由 evaluate_ast 这台「逐节点安检传送带」执行——每个节点过三重资源闸门,每次求值过 safer_eval 返回值检查,每个调用过四层白名单查找,每个 import 过授权树。用异常实现 break/continue/return/final_answer,用 state 字典实现跨步记忆,用线程实现超时。但记住:这套体系防的是失控与误操作,不是蓄意攻击——6.2 节讲它挡不住什么。

本节要点回顾

不 exec 的四理由:无法逐步拦截/计数操作/限制时长/控制导入;逐节点解释把每个 AST 节点变成检查点。
三重闸门:MAX_OPERATIONS=1000 万节点计数(state 内计数器)、MAX_WHILE_ITERATIONS=100 万 while 上限、30 秒 ThreadPoolExecutor 超时(跨平台,但线程无法强杀)。
节点分发:巨型 isinstance 链(手写访问者模式),未支持语法直接报错;赋值经 set_value(禁改 static_tools 名),调用按 state→static_tools→custom_tools→ERRORS 四层查找,内置函数与 dunder 调用被拒。
控制流全靠异常:BreakException/ContinueException/ReturnException/FinalAnswerException(继承 BaseException 防被吞)。
推导式在 state.copy() 副本上求值不污染作用域;属性访问禁 dunder;变量名打错有 difflib 相近名提示。
命名空间:state 变量表 + static_tools(工具+BASE_PYTHON_TOOLS,含 nodunder_getattr)+ custom_tools;导入过 check_import_authorized 前缀树 + get_safe_module 浅拷贝。
evaluate_python_code:parse→初始化→补丁 final_answer→逐语句执行→(result, is_final_answer);LocalPythonExecutor 持久化 state 跨步复用,返回 CodeOutput。

下一节:02 安全边界:黑名单与"不是沙箱"警告——黑名单清单、dunder 逃逸路径、官方警告原文,以及「什么时候本地解释器够用、什么时候必须上沙箱」的决策框架。

作者与出处

原作者：灏天文库

整理：灏天文库整理

本站整理收录，版权归原作者/开源协议所有；欢迎通过原文链接访问源仓库。