CAMEL Runtimes 深度指南:LLMGuard、Docker、RemoteHttp 与 Daytona 四种代码执行沙箱的架构与实战
【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel
导读
CAMEL 的runtime(运行时)模块为 Agent 的代码与工具执行提供了安全、灵活且相互隔离的执行环境,覆盖从进程内安全审查(LLMGuardRuntime)、Docker 容器隔离(DockerRuntime / UbuntuDockerRuntime)到远程 HTTP 分发(RemoteHttpRuntime)与云端托管沙箱(DaytonaRuntime)的全谱系方案。本文以 docs/key_modules/runtimes.md 为骨架,结合仓库源码与可运行示例,讲解每种运行时的工作机理、核心 API、关键参数与适用场景,读完你可以根据“安全性、隔离性、扩展性、成本”四个维度为自己的 Agent 工具链选择并落地正确的执行沙箱。
什么是 Runtimes?
现代 Agent 系统的执行需求远超一个简单的解释器:被调用的工具可能来自不可信来源,可能访问文件系统、执行任意代码,也可能需要跨机器分发计算。Runtimes正是为此设计的一层执行抽象——通过安全护栏(guardrails)、容器隔离(isolation)或远程/云端端点(remote/cloud endpoints),为复杂或不可信代码提供安全、可扩展的执行通道。
在 CAMEL 中,runtime 的核心职责是托管FunctionTool形式的工具函数:把原本在本地进程内直接执行的函数,包装为经过风险审查、或运行在容器/远端沙箱中的安全版本。从源码结构看,整个模块由以下文件组成(见 camel/runtimes/init.py):
base.py:抽象基类BaseRuntime;llm_guard_runtime.py:基于 LLM 的风险评估运行时;docker_runtime.py、ubuntu_docker_runtime.py:容器隔离运行时;remote_http_runtime.py:远程 HTTP 分发运行时;daytona_runtime.py:Daytona 云沙箱运行时;api.py:容器/远程运行时内部使用的 FastAPI 工具服务;configs.py:容器内任务执行配置TaskConfig;utils/:风险审查相关的FunctionRiskToolkit、IgnoreRiskToolkit等工具包。
此外__init__.py中留有 TODO 注释,计划增加支持分布式计算、限流与负载均衡的 Celery Runtime,可见该模块仍在持续演进。
统一运行时接口:BaseRuntime
所有运行时都继承自BaseRuntime(camel/runtimes/base.py),它定义了统一的生命周期与工具管理 API:
add(funcs, *args, **kwargs):注册一个或多个FunctionTool对象,供运行时执行(抽象方法,各运行时按需扩展签名);reset():将运行时重置到初始状态(抽象方法,例如 DockerRuntime 会stop().build()重建容器);get_tools():返回运行时当前管理的全部工具列表;cleanup():释放外部资源(容器、进程、连接等),是生命周期 API 的公共部分;stop():调用cleanup()并返回自身,支持链式调用;- 上下文管理器支持:
with runtime:进入/退出时自动触发build()与cleanup(),实现确定性回收。
基类内部通过self.tools_map(一个Dict[str, FunctionTool])保存已注册工具,get_tools()直接返回list(self.tools_map.values())。值得注意的细节是:各个运行时在add()中都会用闭包包装(wrapper)替换原始func.func,从而把“本地直接调用”变成“经风险审查/发往远端执行”的调用,这是整个 runtime 机制的核心实现手法。
LLMGuardRuntime:用大模型为函数调用把关
LLMGuardRuntime是安全护栏层运行时:在真正执行函数之前,先让一个 LLM Agent 评估该函数调用(函数名 + 描述 + 实际参数)的风险等级,再依据阈值决定放行或拦截(camel/runtimes/llm_guard_runtime.py)。
风险评分机制(1–3 分)
LLM 评估依据内置的GUARDPROMPT系统提示词,通过function_risk工具给出评分与理由:
| 评分 | 含义 | 典型场景 |
|---|---|---|
| 1 | 无危害 | 数学计算、内容搜索、数据检索等不触碰环境与敏感资源的操作;或“能力危险但参数安全”的调用 |
| 2 | 轻微危害 | 读取用户文件、获取非敏感数据等低风险资源访问 |
| 3 | 存在风险 | 删除文件、修改文件系统、执行任意代码、访问敏感数据等;但如果实际参数明确指示受限的安全用法,评分应相应下调 |
阈值控制与放行逻辑
add(funcs, threshold=2):注册函数时可指定风险阈值,默认2。评估得分score > threshold时拦截调用,并返回包含失败原因的错误信息;否则放行执行原始函数。- 评估流程(对应源码
wrapper中的逻辑):每次调用前agent.init_messages()清空会话 → 将函数名、描述、args、kwargs 交给ChatAgent评估 → 解析external_tool_call_requests中的score与reason→ 与阈值比较后决定执行或拒绝。 - 通过
IgnoreRiskToolkit支持“忽略风险”白名单:对已加入忽略列表的函数直接执行,跳过 LLM 评估。
初始化参数
prompt:评估用系统提示词,默认GUARDPROMPT;model:评估所用 LLM,缺省时通过ModelFactory创建平台默认模型(ModelPlatformType.DEFAULT+ModelType.DEFAULT);verbose:是否打印评估过程输出。
cleanup()为空实现(不持有外部资源),reset()会清空忽略列表并重置内部 Agent。可运行的完整示例见 examples/runtimes/code_execution_with_llm_guard_runtime.py,对应测试见 test/runtimes/test_code_execution_with_llm_guard_runtime.py。
DockerRuntime:容器级隔离与工具端点服务
DockerRuntime把注册的工具函数包装为对容器内 FastAPI 服务端点的 HTTP 调用,实现隔离、可复现的执行环境(camel/runtimes/docker_runtime.py)。
核心概念与参数
image:运行时使用的 Docker 镜像名;若本地不存在,构造时自动从 Docker Hub 拉取。port:暴露给宿主的 API 端口,默认8000;传0或负数时会自动分配随机端口(randint(10000, 20000))。remove:停止后是否删除容器,默认True。**kwargs:透传给 Docker 客户端(如environment、volumes、ports等容器配置)。
常用方法(构建与运维)
mount(path, mount_path):把本地绝对路径目录以 bind mount 方式挂载进容器,要求本地路径存在且为目录、两侧均为绝对路径。copy(source, dest):把本地文件/目录打包成 tar 后复制进容器(put_archive)。add_task(task)/exec_run(task):配置并执行容器内命令(类似docker exec),task为TaskConfig对象。build(time_out=15):创建并启动容器(默认命令sleep infinity保活)、复制文件、执行预置任务,最后在容器内以python3 api.py启动工具 API 服务。wait(timeout=10)/ok:轮询等待 API 服务就绪;docs属性返回http://localhost:{port}/docs,可直接在浏览器查看自动生成的 API 文档。
add() 的执行链路
add(funcs, entrypoint, redirect_stdout=False, arguments=None)中,entrypoint是容器内 API 服务导入工具的模块路径(如camel.toolkits.MathToolkit),arguments可为工具类提供初始化参数(会被 JSON 序列化后拼入入口字符串)。包装后的函数通过requests.post向http://localhost:{port}/{function_name}发送{args, kwargs, redirect_stdout},服务端执行后返回{output, stdout},客户端json.loads还原结果;当redirect_stdout=True时还会把容器端捕获的标准输出打印到本地。
容器内的工具服务由 camel/runtimes/api.py 提供:它是一个 FastAPI 应用,启动时读取sys.argv[1:]中的模块路径逐个importlib.import_module,若导入的是BaseToolkit子类则实例化并取其全部工具,为每个工具注册一个POST /{tool_name}端点;同步工具经 4 线程的ThreadPoolExecutor调度,避免阻塞事件循环(例如 Playwright 同步 API);/health端点会报告已加载的 toolkit 与端点列表,且服务端设置了CAMEL_RUNTIME=true环境变量以标识运行环境。
容器内任务配置 TaskConfig
TaskConfig(camel/runtimes/configs.py)是一个 Pydantic 模型,直接对应 Dockerexec_run的底层参数,用于add_task/exec_run:
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
cmd | str / list | 必填 | 要执行的命令 |
stdout/stderr | bool | True | 是否挂接标准输出/错误 |
stdin | bool | False | 是否挂接标准输入 |
tty | bool | False | 是否分配伪终端 |
privileged | bool | False | 是否以特权模式运行 |
user | str | "" | 执行命令的用户 |
detach | bool | False | 是否分离执行 |
stream | bool | False | 是否流式返回 |
socket | bool | False | 返回连接 socket |
environment | dict / list | None | 环境变量(支持{"PASSWORD": "xxx"}或["PASSWORD=xxx"]) |
workdir | str | None | 执行会话的工作目录 |
demux | bool | False | stdout/stderr 是否分开返回 |
示例脚本见 examples/runtimes/docker_runtime.py 与 examples/runtimes/docker_runtime_with_tasks.py,测试见 test/runtimes/test_docker_runtime.py。
UbuntuDockerRuntime:面向 Ubuntu 环境的脚本执行增强
UbuntuDockerRuntime继承自DockerRuntime,针对 Ubuntu 容器做了专门的 Python 环境配置与脚本执行能力增强(camel/runtimes/ubuntu_docker_runtime.py)。
环境预配置
初始化时自动设置容器环境变量并校验 Python 安装:
PYTHON_PATH/PYTHON_EXECUTABLE:指向容器内 Python 解释器,默认/usr/bin/python3;PATH:/usr/local/bin:/usr/bin:/bin;PYTHONUNBUFFERED=1:禁用输出缓冲;- 默认工作目录
/app(可通过working_dir覆盖)。
build()在父类构建基础上额外执行:校验python3 --version、apt-get update、安装curl,然后以显式 Python 路径与PYTHONPATH(.../lib/python3.10/site-packages)启动/home/api.py,并通过容器内curl轮询/docs状态码确认 API 服务就绪。add()还会把函数命令中的python替换为配置的解释器路径。
exec_python_file():运行动态生成的脚本
文档特别强调:对于 Agent 层面的动态代码执行,应优先考虑exec_python_file()以获得最大隔离性。其签名与行为如下:
local_file_path:本地 Python 文件路径(必须存在,否则抛FileNotFoundError);container_path:文件在容器内的目标路径,缺省复制到/tmp/{文件名};args:传给脚本的命令行参数列表;env:附加环境变量(默认注入PYTHONPATH与PYTHON_EXECUTABLE);callback:逐行处理输出的回调,缺省时直接打印(stdout/stderr 分离)。
实现上先用 tar 归档 +put_archive把脚本复制进容器,再以exec_run(stream=True, demux=True)流式执行并区分 stdout 与 stderr,适合运行长任务或需要实时观察输出的场景。配套的完整 Dockerfile、容器管理脚本与示例见 examples/runtimes/ubuntu_docker_runtime/(其中ubuntu_docker_example.py演示了把ai_society/role_playing.py传入容器执行并注入OPENAI_API_KEY环境变量的用法)。
RemoteHttpRuntime:远程分布式的工具执行
RemoteHttpRuntime让工具函数运行在远程 FastAPI 服务器上,适合分布式、可扩展或跨服务器执行场景(camel/runtimes/remote_http_runtime.py)。
与原文档一致的快速上手示例
原文档提供的 Quick Start 代码可直接运行,对应的完整版在 examples/runtimes/remote_http_runtime.py:
from camel.runtimes import RemoteHttpRuntime from camel.toolkits import MathToolkit if __name__ == "__main__": runtime = ( RemoteHttpRuntime("localhost") .add(MathToolkit().get_tools(), "camel.toolkits.MathToolkit") .build() ) print("Waiting for runtime to be ready...") runtime.wait() print("Runtime is ready.") # There are more tools imported from MathToolkit. # For simplicity, we use only "add" tool here add = runtime.get_tools()[0] print(f"Add 1 + 2: {add.func(1, 2)}") # Example output: # Add 1 + 2: 3运行输出(完整示例中)为:
Waiting for runtime to be ready... Runtime is ready. Add 1 + 2: 3 Subtract 5 - 3: 2 Multiply 2 * 3: 6 Documents: http://localhost:8000/docs工作机制与参数
- 构造参数:
host(远程服务器地址)、port(默认8000)、python_exec(启动 API 服务所用的 Python,默认python3)。 build()用subprocess.Popen在本机启动 camel/runtimes/api.py 作为 API 服务,并注册atexit钩子确保进程退出时自动回收;若目标是远程服务器,则需先在远端手动运行该 API 服务,此时本机RemoteHttpRuntime直接连接远端host:port。add(funcs, entrypoint, redirect_stdout=False, arguments=None)与 DockerRuntime 的语义一致,包装函数通过requests.post访问http://{host}:{port}/{function_name}完成调用;Pydantic 模型参数会被自动model_dump()序列化。wait(timeout=10)轮询健康状态;docs返回远端 API 文档地址;reset()实现为stop().build();__del__与atexit双重保障资源释放。
底层 API 服务(api.py)同样基于 FastAPI + uvicorn 运行在0.0.0.0:8000(reload=False,避免与 Playwright 等异步 toolkit 冲突),并通过线程池支持同步工具的并发执行。对应测试见 test/runtimes/test_remote_http_runtime.py。
DaytonaRuntime:云端托管的远程沙箱
DaytonaRuntime使用 Daytona SDK 在云端托管沙箱中运行代码,把上传源码、执行与输入/输出捕获交给云服务商管理,换取资源与安全保证(camel/runtimes/daytona_runtime.py)。
初始化参数
api_key:Daytona API 密钥,缺省读取环境变量DAYTONA_API_KEY;api_url:Daytona 服务器地址,缺省读取DAYTONA_API_URL,均未提供时回退到http://localhost:8000;language:沙箱语言,可选python、typescript、javascript,默认python;image:沙箱使用的 Docker 镜像,缺省按语言选择内置默认镜像:python:3.11、node:20、node:20。
执行链路
build()通过CreateSandboxFromImageParams(image, language)创建沙箱,成功后记录沙箱 ID。add(funcs, entrypoint, arguments=None)用inspect.getsource提取每个函数的源码,包装函数执行时:先把函数源码upload_file上传到沙箱/home/daytona/{func_name}.py,再拼接一段导入函数、反序列化 args/kwargs、调用并json.dumps结果的执行代码,交给sandbox.process.code_run()运行,最后解析 JSON 返回结果。info()返回沙箱状态与 CPU/内存资源信息;reset()为stop().build();cleanup()调用daytona.delete(sandbox)释放沙箱。
使用前提
使用前需要:Daytona 服务端已运行、已配置 API 密钥,且环境中安装daytona_sdk(代码在__init__中惰性导入)。示例见 examples/runtimes/daytona_runtime.py,测试见 test/runtimes/test_daytona_runtime.py。
运行时选型对比与更多示例
四种运行时的定位差异可以概括为:
| 运行时 | 隔离级别 | 执行位置 | 主要优势 | 典型场景 |
|---|---|---|---|---|
LLMGuardRuntime | 进程内(逻辑护栏) | 本地 | 零额外基础设施,LLM 动态风险判断 | 快速给现有工具加安全审查层 |
DockerRuntime | 容器隔离 | 本地 Docker | 隔离 + 可复现,自动注册 HTTP 工具端点 | 工具函数沙箱执行、多工具共享环境 |
UbuntuDockerRuntime | 容器隔离 | 本地 Docker | 完整脚本执行、环境变量注入、流式输出 | 运行动态生成的 Python 脚本 |
RemoteHttpRuntime | 进程外(远程) | 远程服务器 | 分布式扩展、跨机执行、卸载重活 | 远程/集群环境分发工具调用 |
DaytonaRuntime | 云端沙箱 | 云托管 | 资源与安全由云保证,多语言支持 | 需要托管式沙箱的隔离执行 |
更多可运行脚本集中在 examples/runtimes/ 目录:包括code_execution_with_docker_runtime.py、code_execution_with_llm_guard_runtime.py、docker_runtime_with_tasks.py、shared_runtime_multi_toolkit.py以及ubuntu_docker_runtime/子目录(含 Dockerfile、manage_camel_docker.sh与ubuntu_docker_example.py),每个脚本都演示了对应运行时从初始化到释放的完整流程。此外 test/runtimes/test_shared_runtime.py 与 test/runtimes/test_base_runtime.py 覆盖了多工具共享运行时与基类生命周期行为。
最后一点提醒:工具沙箱 ≠ 动态代码执行
需要明确的是,runtime 体系主要沙箱化的是FunctionTool形式的工具函数——即把“预定义好的工具调用”包装到受控环境中执行。对于 Agent 层面动态生成代码的执行需求,文档强调应始终考虑专用的沙箱方案,例如使用UbuntuDockerRuntime的exec_python_file()来运行动态生成的脚本,从而在最大程度上获得隔离性与安全性。在编排自己的 Agent 时,建议根据任务对隔离性、执行位置、资源上限与基础设施成本的权衡,选择(或组合)上述运行时:本地可信工具用LLMGuardRuntime加护栏,不可信或副作用强的工具进DockerRuntime,需要脚本级执行的用UbuntuDockerRuntime,跨机分发用RemoteHttpRuntime,追求托管式安全沙箱则用DaytonaRuntime。
【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考