Agent执行代码时如果每次都用临时文件配合subprocess,会丢失REPL的连续性。更合理的做法是给模型挂载一个常驻的Python解释器,让不同代码块共享同一个命名空间。这样变量、函数、导入的模块都能跨轮复用,模型只需要关注增量修改,工具返回结果也能保持结构化。

一、REPL与单次exec的本质差异
标准 Python REPL 由 code 模块提供,核心是 InteractiveInterpreter。它内部维护一个字典作为全局命名空间,每次调用 runsource 或 runcode 时,都基于这个字典执行新代码。相比之下,exec 函数虽然也能接收 globals 字典,但大多数封装只调用一次,执行完就丢弃引用。要持久化状态,必须显式把同一个字典传回下一次调用。
这种差异在多轮 Agent 调用中会被放大。假设第一轮用户要求读取 CSV 文件并给前五行加一列,第二轮要求继续按新列分组统计。如果每次都用新进程执行,第二轮必须重新读取 CSV、重新定义清洗函数;而常驻 REPL 可以直接访问上一轮留下的 DataFrame 和列名。对模型来说,这意味着提示词可以更短,函数调用参数可以只包含增量代码,出错时也只需要回滚到最近的命名空间快照。
除了变量持久化,REPL 还能自动求值表达式。当输入不是完整的语句块,而是单个表达式时,解释器会尝试用 eval 执行并把结果返回。例如执行 3 + 4 不会只得到 None,而是可以得到 7。这个特性非常适合 Agent 做数值计算、对象探测或快速验证逻辑,因为模型经常生成类似 df.shape、len(col) 这样的表达式来检查数据状态。
二、实现一个可观测的Python REPL执行器
最简实现不必直接复用 code.InteractiveInterpreter,但要解决三个问题:捕获标准输出和标准错误、区分表达式与语句、返回结构化结果。下面是一个基础实现,用 ast.parse 判断输入类型,用 StringIO 替换 stdout 和 stderr,再把执行结果包装成字典返回给上层 Agent:
import ast
import contextlib
import io
import traceback
class AgentREPL:
def __init__(self):
self.namespace = {"__builtins__": __builtins__}
self.last_result = None
def run(self, source):
stdout = io.StringIO()
stderr = io.StringIO()
error = None
try:
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
try:
# 先尝试按表达式求值
expr = ast.parse(source, mode="eval")
self.last_result = eval(compile(expr, "<agent-repl>", "eval"), self.namespace)
except SyntaxError:
# 再按语句块执行
block = compile(source, "<agent-repl>", "exec")
exec(block, self.namespace)
except Exception:
error = traceback.format_exc(limit=3)
return {
"status": "ok" if error is None else "error",
"output": stdout.getvalue(),
"stderr": stderr.getvalue(),
"result": repr(self.last_result),
"error": error,
}
这个实现把 sys.stdout 和 sys.stderr 重定向到内存缓冲区,因此任何 print、警告、异常追踪都不会直接打到 Agent 进程的控制台。result 字段使用 repr 序列化,避免返回不可 JSON 化的对象。当输入以表达式形式给出时,last_result 会保存求值结果;输入是完整语句块时,正常执行代码但不会强制把最后一个赋值当结果返回,避免把整个数据结构塞给模型。
超时控制不能直接加在这个执行器上,因为 exec 和 eval 没有提供执行时间限制。常见的做法有两种:如果 Agent 运行在 Unix 环境且 REPL 在主线程,可以用 signal.alarm 设置软中断,超时触发 TimeoutError;跨平台更稳妥的方案是使用 ThreadPoolExecutor 或 multiprocessing.Process,在独立线程或进程中执行 run 方法,主线程用 future.result(timeout=5) 等待。子进程方案隔离性更好,但要考虑对象序列化成本,线程方案则要注意命名空间并发访问。
三、安全加固与Agent交互设计
让模型直接执行任意 Python 代码风险很大。即使模型本身可信,也可能因为提示词注入、数据文件中的恶意字符串或模型幻觉生成危险调用。最低限度的加固是限制 builtins,把 import、open、eval、exec、compile、input 等高风险函数从命名空间中移除,只保留数据转换、数学计算和类型判断相关的内置函数。不过这样做会牺牲不少灵活性,因此更推荐在子进程或容器中运行 REPL,并通过资源限制控制 CPU、内存和文件系统访问。
下面是一个用 subprocess 启动受限 Python 的示例。外部通过 stdin 发送代码,子进程用 -I 参数进入隔离模式,site-packages 不会被自动加载,再配合 ulimit 或操作系统的资源限制工具,可以避免内存耗尽和磁盘写入。示例代码省略了输入输出协议细节,只展示核心参数:
import subprocess
def create_sandbox():
return subprocess.Popen(
["python3", "-I", "-q"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
cwd="/tmp/agent-sandbox",
)
def run_in_sandbox(proc, code: str, timeout: float = 5.0):
proc.stdin.write(code + "\n")
proc.stdin.flush()
try:
stdout, stderr = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
proc.kill()
stdout, stderr = proc.communicate()
return {"status": "timeout", "output": stdout, "error": stderr}
return {"status": "ok", "output": stdout, "error": stderr}
subprocess 方案中,每个代码块仍然通过同一个进程的 stdin 输入,进程不会退出,因此保留了变量状态。这与直接使用解释器的区别在于,隔离边界从 Python 命名空间上升到操作系统进程,即使代码调用了 os.system 或触发进程炸弹,也只会影响子进程,不会直接拖垮 Agent 主服务。还可以进一步用 Docker、gVisor 或 WebAssembly 沙箱来限制网络访问和文件路径。
Agent 交互侧需要把 REPL 返回的结构化结果转换成模型能理解的工具消息。最直接的方式是返回 JSON 字符串,包含 status、output、stderr、result 和 error 五个字段。对于正常执行,模型只需要看 output 和 result;对于异常,把 stderr 中 traceback 的最后三到五帧截取出来,避免超长上下文。不要在工具返回中重复完整代码,模型已经生成过代码,重复只会浪费 token。如果检测到输入不完整,应提示模型继续输出剩余代码,而不是当作错误重试。
为了让解释器在长会话中稳定运行,还应该实现命名空间快照和回滚。可以在每个任务开始前深拷贝一份当前命名空间,或保存为 pickle 文件,任务执行失败时恢复快照,避免错误代码污染后续交互。对 DataFrame、模型权重等大对象,深拷贝开销很高,可以改为记录已导入模块和关键变量的版本,让模型自行决定是否需要重置。
综合来看,Agent 代码解释器采用 Python REPL 集成,核心收益是有状态的增量执行、自动表达式求值以及清晰的输出捕获。落地时先把执行器封装成可观测的组件,再用超时和沙箱策略控制风险,最后通过结构化工具消息把结果回传给模型。这个方案既能支撑数据分析、代码调试、数值计算等常见 Agent 场景,也为后续引入更严格的隔离机制留出了接口。

