ITADN
pydantic/monty
README.md
以下内容由 AI 翻译,如有问题请点此提交 issue 反馈

Monty

一个用 Rust 编写的极简、安全的 Python 解释器,供 AI 使用。

CI Codspeed Coverage PyPI versions license Join Slack

实验性 - 本项目仍处于开发阶段,尚未达到生产就绪状态。

一个用 Rust 编写的极简、安全的 Python 解释器,供 AI 使用。

Monty 避免了使用基于容器的完整沙箱来运行 LLM 生成代码所带来的成本、延迟、复杂性以及一般性的繁琐操作。

相反,它允许你在智能体中安全地运行由 LLM 编写的 Python 代码,启动时间以个位数微秒计,而非数百毫秒。

Monty 能够做到的:

  • 运行合理的 Python 代码子集 - 足以让你的 agent 表达它想要做什么
  • 完全阻止访问宿主环境:文件系统、环境变量和网络访问均通过开发者可控制的外部函数调用实现
  • 调用宿主上的函数 - 仅限你授予其访问权限的函数
  • 运行类型检查 - monty 支持完整的现代 Python 类型提示,并附带 ty 包含在单个二进制文件中以运行类型检查
  • 在外部函数调用时快照为字节,这意味着你可以将解释器状态存储在文件或数据库中,并稍后恢复
  • 启动极快(从代码到执行结果 <1μs),且运行时性能与 CPython 相似(通常介于快 5 倍到慢 5 倍之间)
  • 可从 Rust、Python 或 Javascript 调用 - 因为 Monty 不依赖 cpython,你可以在任何能运行 Rust 的地方使用它
  • 控制资源使用 - Monty 可以跟踪内存使用、栈深度和执行时间,并在超过预设限制时取消执行
  • 收集 stdout 和 stderr 并将其返回给调用者
  • 通过宿主上的异步或同步代码在宿主上运行异步或同步代码
  • 使用标准库的一个小子集:sys, os, typing, asyncio, re, datetime, json, dataclasses (即将推出)

Monty 不能 做的事情:

  • 使用标准库的其余部分
  • 使用第三方库(如 Pydantic),对外部 python 库的支持并非目标
  • 定义类(支持即将推出)
  • 使用 match 语句(同样,支持即将推出)

简而言之,Monty 功能极其有限,专为一个用例而设计:

运行由智能体编写的代码。

关于为何可能需要这样做的动机,请参阅:

用最简单的话来说,上述所有方案的理念是:如果要求 LLM 编写 Python(或 Javascript)代码,而不是依赖传统的工具调用,它们可以工作得更快、更便宜且更可靠。Monty 使得这成为可能,而无需沙箱的复杂性,也无需直接在主机上运行代码的风险。

注意: Monty 将(很快)用于在 Pydantic AI 中实现 codemode

用法

Monty 可以从 Python、JavaScript/TypeScript 或 Rust 中调用。

Python

安装方法:

uv add pydantic-monty

(或者 pip install pydantic-monty 给婴儿潮一代)

用法:

from typing import Any

import pydantic_monty

code = """
async def agent(prompt: str, messages: Messages):
    while True:
        print(f'messages so far: {messages}')
        output = await call_llm(prompt, messages)
        if isinstance(output, str):
            return output
        messages.extend(output)

await agent(prompt, [])
"""

type_definitions = """
from typing import Any

Messages = list[dict[str, Any]]

async def call_llm(prompt: str, messages: Messages) -> str | Messages:
    raise NotImplementedError()

prompt: str = ''
"""


Messages = list[dict[str, Any]]


async def call_llm(prompt: str, messages: Messages) -> str | Messages:
    if len(messages) < 2:
        return [{'role': 'system', 'content': 'example response'}]
    else:
        return f'example output, message count {len(messages)}'


async def main():
    async with pydantic_monty.AsyncMonty() as pool:
        async with pool.checkout(
            script_name='agent.py',
            type_check=True,
            type_check_stubs=type_definitions,
        ) as session:
            output = await session.feed_run(
                code,
                inputs={'prompt': 'testing'},
                external_lookup={'call_llm': call_llm},
            )
    print(output)
    #> example output, message count 2


if __name__ == '__main__':
    import asyncio

    asyncio.run(main())

执行发生在 monty 个工作子进程的池中,因此即使由对抗性代码触发的内存 错误(栈溢出、分配器中止)也永远不会导致你的进程崩溃 —— 工作进程会终止,抛出 MontyCrashedError,并 被替换。此外,还有一个完全同步的 API:

import pydantic_monty

with pydantic_monty.Monty() as pool:
    with pool.checkout() as session:
        # session state persists between feed_run calls
        session.feed_run('x = 21')
        print(session.feed_run('x * 2'))
        #> 42

JavaScript / TypeScript

安装方法:

npm install @pydantic/monty

JS 包是对 Python 包所使用的同一 Rust worker 池的原生 (napi) 绑定——该绑定和 monty worker 二进制文件通过平台特定的 npm 包分发:

import { Monty } from '@pydantic/monty'

await using pool = await Monty.create()
await using session = await pool.checkout()

// session state persists between feedRun calls
await session.feedRun('x = 21')
console.log(await session.feedRun('x * 2')) // 42

// external functions may be async
const result = await session.feedRun('await fetch_data()', {
  externalLookup: { fetch_data: async () => 'data' },
})

对于浏览器(或任何无法使用子进程的环境),相同的包 在 @pydantic/monty/wasm 子路径下暴露了一个进程内 WebAssembly 构建(无崩溃隔离:沙箱崩溃即为主机崩溃)。

Rust

对于从 Rust 运行不受信任的代码,我们推荐 使用 monty-pool crate,而不是下面的进程内 API。 monty-pool 仅在 monty worker 子进程中运行代码,这提供了额外的保护: 由对抗性代码触发的崩溃(栈溢出、分配器中止)只会终止 worker —— 池会检测到死亡并替换 worker —— 并且父进程端的看门狗可以终止 超过硬性超时的 worker。它是上述 Python 和 JavaScript 包 所基于的同一引擎。请参阅 monty-pool README 以了解用法。

monty crate 本身提供了进程内解释器:

use monty::MontyRun;
use monty_types::{CompileOptions, ResourceTracker, MontyObject, PrintWriter, ResourceLimits};

let code = r#"
def fib(n):
    if n <= 1:
        return n
    return fib(n - 1) + fib(n - 2)

fib(x)
"#;

let runner = MontyRun::new(code.to_owned(), "fib.py", vec!["x".to_owned()], CompileOptions::default()).unwrap();
let result = runner.run(vec![MontyObject::Int(10)], ResourceTracker::default(), PrintWriter::Stdout).unwrap();
assert_eq!(result, MontyObject::Int(55));

序列化

MontyRunRunProgress 可以使用 dump()load() 方法进行序列化:

use monty::MontyRun;
use monty_types::{CompileOptions, ResourceTracker, MontyObject, PrintWriter, ResourceLimits};

// Serialize parsed code
let runner = MontyRun::new("x + 1".to_owned(), "main.py", vec!["x".to_owned()], CompileOptions::default()).unwrap();
let bytes = runner.dump().unwrap();

// Later, restore and run
let runner2 = MontyRun::load(&bytes).unwrap();
let result = runner2.run(vec![MontyObject::Int(41)], ResourceTracker::default(), PrintWriter::Stdout).unwrap();
assert_eq!(result, MontyObject::Int(42));

工作进程中的内存限制

会话的 max_memory 由工作进程的分配器进行测量。当超过软限制后,解释器会报告一次优雅的 MemoryError;如果某次分配在检查点之间跳跃过大,则更高的硬限制会终止并替换该工作进程。

请参阅 limitations/resource_limits.md 了解超出限制如何呈现给宿主,以及 monty-alloc 了解子进程和 WebAssembly 工作进程所运行的分配器。

PydanticAI 集成

Monty 将驱动 Pydantic AI 中的代码模式。LLM 不再进行 顺序的工具调用,而是编写调用你的工具作为函数的 Python 代码,并由 Monty 安全地执行它。

import asyncio
import json

import logfire
from httpx import AsyncClient
from pydantic_ai import Agent, RunContext
from pydantic_ai.toolsets.code_mode import CodeModeToolset
from pydantic_ai.toolsets.function import FunctionToolset
from typing_extensions import TypedDict

logfire.configure()
logfire.instrument_pydantic_ai()


class LatLng(TypedDict):
    lat: float
    lng: float


weather_toolset: FunctionToolset[AsyncClient] = FunctionToolset()


@weather_toolset.tool
async def get_lat_lng(
    ctx: RunContext[AsyncClient], location_description: str
) -> LatLng:
    """Get the latitude and longitude of a location."""
    # NOTE: the response here will be random, and is not related to the location description.
    r = await ctx.deps.get(
        'https://demo-endpoints.pydantic.workers.dev/latlng',
        params={'location': location_description},
    )
    r.raise_for_status()
    return json.loads(r.content)


@weather_toolset.tool
async def get_temp(ctx: RunContext[AsyncClient], lat: float, lng: float) -> float:
    """Get the temp at a location."""
    # NOTE: the responses here will be random, and are not related to the lat and lng.
    r = await ctx.deps.get(
        'https://demo-endpoints.pydantic.workers.dev/number',
        params={'min': 10, 'max': 30},
    )
    r.raise_for_status()
    return float(r.text)


@weather_toolset.tool
async def get_weather_description(
    ctx: RunContext[AsyncClient], lat: float, lng: float
) -> str:
    """Get the weather description at a location."""
    # NOTE: the responses here will be random, and are not related to the lat and lng.
    r = await ctx.deps.get(
        'https://demo-endpoints.pydantic.workers.dev/weather',
        params={'lat': lat, 'lng': lng},
    )
    r.raise_for_status()
    return r.text


agent = Agent(
    'gateway/anthropic:claude-sonnet-4-5',
    # toolsets=[weather_toolset],
    toolsets=[CodeModeToolset(weather_toolset)],
    deps_type=AsyncClient,
)


async def main():
    async with AsyncClient() as client:
        await agent.run('Compare the weather of London, Paris, and Tokyo.', deps=client)


if __name__ == '__main__':
    asyncio.run(main())

社区绑定

替代方案

当人们看到 Monty 时,通常会有两种反应:

  1. 天哪,这解决了很多问题,我想要它。
  2. 为什么不用 X?

其中 X 是某种替代技术。奇怪的是,这些反应经常结合在一起,表明人们尚未找到适合他们的替代方案,但对于从头创建一个完整的 Python 实现竟然没有好的替代方案这一事实感到难以置信。

我将尝试梳理最明显的替代方案,以及为什么它们不适合我们的需求。

注意:所有这些技术都令人印象深刻且应用广泛,关于它们在特定用例中局限性的评论不应被视为批评。大多数这些解决方案在设计之初并未以提供 LLM 沙箱为目标,因此它们在这方面并不一定表现优异。

技术语言完整性安全性启动延迟开源配置复杂度文件挂载快照
Monty部分严格0.06ms免费 / 开源简单简单简单
Docker完整良好195ms免费 / 开源中等简单中等
Pyodide完整2800ms免费 / 开源中等简单困难
starlark-rust非常有限良好1.7ms免费 / 开源简单不可用?不可能?
WASI / Wasmer部分,接近完整严格66ms免费 *中等简单中等
沙箱服务完整严格1033ms非免费中等困难中等
YOLO Python完整不存在0.1ms / 30ms免费 / 开源简单简单 / 危险困难

参见 ./scripts/startup_performance.py 用于计算启动性能数据的脚本。

以下是每行的详细信息:

Monty

  • 语言完整性:尚无类(yet),标准库有限,无第三方库
  • 安全性:显式控制文件系统、网络和环境访问,对执行时间和内存使用有严格限制
  • 启动延迟:微秒级启动
  • 配置复杂度:仅需 pip install pydantic-montynpm install @pydantic/monty,下载大小约 4.5MB
  • 文件挂载:严格受控,参见 #85
  • 快照:Monty 的暂停和恢复功能配合 dump()load() 使得暂停、恢复和分叉执行变得轻而易举

Docker

  • 语言完整性:完整的 CPython,支持任意库
  • 安全性:进程和文件系统隔离,网络策略,但存在容器逃逸风险,可实现内存限制
  • 启动延迟:容器启动开销(实测约 195ms)
  • 配置复杂度:需要 Docker 守护进程、容器镜像、编排,python:3.14-alpine 为 50MB - docker 无法从 PyPI 安装
  • 文件挂载:卷挂载工作良好
  • 快照:可通过 Temporal 等持久化执行方案实现,或通过对镜像进行快照并将其保存为 Docker 镜像来实现。

Pyodide

  • 语言完整性:完整的 CPython 编译为 WASM,几乎所有库均可用
  • 安全性:依赖浏览器/WASM 沙箱 - 并非为服务端隔离而设计,Python 代码可以在 JS 运行时中执行任意代码,只有 deno 允许隔离,而在 deno 中内存限制难以/无法强制执行
  • 启动延迟:WASM 运行时加载缓慢(冷启动约 2800ms)
  • 配置复杂度:需要加载 WASM 运行时,处理异步初始化,pyodide NPM 包约为 12MB,deno 约为 50MB - Pyodide 无法仅通过 PyPI 包调用
  • 文件挂载:通过浏览器 API 实现虚拟文件系统
  • 快照:据推测可以通过 Temporal 等持久化执行方案实现,但难度较大

starlark-rust

参见 starlark-rust.

  • 语言完整性:配置语言,非 Python - 无类、异常、异步
  • 安全性:设计上具有确定性和封闭性
  • 启动延迟:像 Monty 一样嵌入在进程中运行,因此启动时间令人印象深刻
  • 配置复杂度:可通过 starlark-pyo3 在 Python 中使用
  • 文件挂载:据我所知,设计上不支持文件处理?
  • 快照:据我所知,不可能?

WASI / Wasmer

通过 Wasmer 在 WebAssembly 中运行 Python。

  • 语言完整性:完整的 CPython,纯 Python 外部包可通过挂载运行,带有 C 绑定的外部包无法运行
  • 安全性:原则上 WebAssembly 应提供强沙箱保证。
  • 启动延迟wasmer python 包已 3 年未更新,且我未找到从 Python 调用 wasmer 中 Python 的文档,因此我通过 subprocess 调用。启动延迟为 66ms。
  • 配置复杂度:wasmer 下载大小为 100mb,"python/python" 包为 50mb。
  • FOSS:我将其标记为 "free *",因为成本为零,但并非所有内容似乎都是开源的。截至 2026-02-10,python/python wasmer 包 包没有 readme,没有 license,没有 source 链接,也没有说明其构建方式,最近上传的版本显示大小为 "0B",尽管下载大小约为 ~50MB - Python 二进制的构建过程不清晰且不透明。(如果我在这里错了,请创建一个 issue 来纠正我)
  • 文件挂载:支持
  • 快照:通过日志记录支持

沙箱服务

类似 DaytonaE2BModal 的服务。

存在类似的挑战,使用 k8s 设置自己的沙箱环境时,配置复杂度更高,但网络延迟更低。

  • 语言完整性:完整的 CPython,支持任意库
  • 安全性:专业管理的容器隔离
  • 启动延迟:网络往返时间和容器启动时间。我在伦敦使用 Daytona EU 测得约 1 秒的冷启动时间,Daytona 宣称延迟低于 90 毫秒,推测这是针对已存在容器的情况,不清楚是否包含网络延迟
  • 开源软件(FOSS):按执行次数或计算时间计费,部分实现是开源的
  • 设置复杂度:API 集成、认证令牌——对初创公司来说没问题,但对企业来说通常难以接受
  • 文件挂载:通过 API 调用进行上传/下载
  • 快照:使用 Temporal 等持久化执行方案可以实现,这些服务也提供了一些相关解决方案,我认为基于 Docker 容器

YOLO Python

直接通过 exec()(约 0.1 毫秒)或子进程(约 30 毫秒)运行 Python。

  • 语言完整性:完整的 CPython,支持任意库
  • 安全性:无——完整的文件系统、网络、环境变量、系统命令访问权限
  • 启动延迟exec() 接近零,子进程约 30 毫秒
  • 设置复杂度:无
  • 文件挂载:直接文件系统访问(这就是问题所在)
  • 快照:使用 Temporal 等持久化执行方案可以实现

Pydantic 技术栈的一部分

Pydantic 技术栈包含您发布生产级 AI 智能体所需的一切: