Agent 沙盒机制学习笔记

一、本章概览

本模块演示如何让 AI Agent 在隔离环境中安全执行代码，包含三种实现方案：

二、为什么需要沙盒

Agent 需要执行代码来完成计算、文件操作等任务。但直接在宿主机执行代码有严重风险：

沙盒解决了这些问题：代码在隔离环境中执行，即使出错也不影响宿主机。

三、Docker 本地沙盒（核心实现）

3.1 工作原理

每次执行代码时，DockerSandbox 会：

1. 创建一个临时 Docker 容器
2. 将代码作为 python -c 参数传入
3. 捕获 stdout/stderr 输出
4. 自动销毁容器

3.2 DockerSandbox 类

import docker
import uuid

class DockerSandbox:
    def __init__(
        self,
        image: str = "python:3.12-slim",   # Docker 镜像
        mem_limit: str = "256m",             # 内存限制
        cpu_quota: int = 50000,              # CPU 配额（50% 单核）
        network_disabled: bool = False,      # 是否禁用网络
        timeout: int = 30,                   # 超时时间（秒）
    ):
        self.client = docker.from_env()      # 连接本地 Docker

    def execute(self, code: str) -> str:
        """在沙盒中执行代码"""
        container = self.client.containers.create(
            image=self.image,
            command=["python", "-c", code],
            mem_limit=self.mem_limit,
            network_disabled=self.network_disabled,
            detach=True,
        )
        container.start()
        result = container.wait(timeout=self.timeout)
        stdout = container.logs(stdout=True, stderr=False).decode("utf-8")
        stderr = container.logs(stdout=False, stderr=True).decode("utf-8")
        container.remove(force=True)  # 自动清理
        return stdout.strip() or stderr.strip() or "(无输出)"

3.3 安全隔离机制

3.4 LangChain Tool 封装

将 DockerSandbox 封装为 LangChain Tool，供 Agent 调用：

from langchain_core.tools import tool

def create_sandbox_tool(sandbox: DockerSandbox = None):
    if sandbox is None:
        sandbox = DockerSandbox()

    @tool
    def run_in_sandbox(code: str) -> str:
        """在隔离的 Docker 沙盒中执行 Python 代码。
        入参是纯 Python 代码字符串，不要包裹在 ```python 标记中。"""
        return sandbox.execute(code)

    return run_in_sandbox

四、Agent 集成测试

4.1 完整代码流程

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

# 1. 创建 Docker 沙盒（禁用网络，完全隔离）
sandbox = DockerSandbox(
    image="python:3.12-slim",
    mem_limit="256m",
    network_disabled=True,
    timeout=30,
)
sandbox_tool = create_sandbox_tool(sandbox)

# 2. 创建 Agent（带内存记忆）
memory = MemorySaver()
agent = create_react_agent(
    model=llm.llm,
    tools=[sandbox_tool],
    checkpointer=memory,
)

# 3. 发送任务，Agent 自主决定在沙盒中执行代码
result = await agent.ainvoke(
    {"messages": [("user", "在沙盒中计算 1 到 100 的累加和")]},
    config={"configurable": {"thread_id": "sandbox-001"}},
)

4.2 Agent 决策流程

4.3 测试验证结果

4.4 工具调用日志追踪

通过追踪 AIMessage.tool_calls 和 ToolMessage，验证 Agent 确实通过沙盒执行：

from langchain_core.messages import AIMessage, ToolMessage

for msg in result["messages"]:
    if isinstance(msg, AIMessage) and msg.tool_calls:
        for tc in msg.tool_calls:
            print(f"调用: {tc['name']}({tc['args']})")
    elif isinstance(msg, ToolMessage):
        print(f"返回: {msg.content}")

日志输出示例：

[沙盒工具调用日志]
  调用[1]: run_in_sandbox({'code': 'total = sum(range(1, 101))\nprint(total)'})
    沙盒返回: 5050
[Agent 回复] 1 到 100 的累加和为 5050

五、本地 Docker 沙盒 vs 云沙盒（Daytona）

推荐：日常开发使用本地 Docker 沙盒，生产环境按需选择。

六、Docker 沙盒独立测试

local_sandbox.py 可以独立运行，不需要 Agent：

python -m app.sandbox_agent_test.local_sandbox

测试项目及结果：

退出码 137 表示容器因超出内存限制被系统强制终止（OOM Killed），证明资源限制生效。

七、环境准备

7.1 安装依赖

pip install docker

7.2 安装 Docker Desktop

• Windows/Mac：下载 Docker Desktop
• Linux：sudo apt install docker.io

7.3 拉取 Python 镜像

# 直接拉取（需科学上网）
docker pull python:3.12-slim

# 国内网络使用华为云镜像
docker pull swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/library/python:3.12-slim
docker tag swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/library/python:3.12-slim python:3.12-slim

八、关键 API 速查

# ---- Docker 沙盒 ----
from app.sandbox_agent_test.local_sandbox import DockerSandbox, create_sandbox_tool

sandbox = DockerSandbox(
    image="python:3.12-slim",
    mem_limit="256m",
    network_disabled=True,
    timeout=30,
)
result = sandbox.execute("print('hello')")
sandbox.cleanup()

# ---- 作为 LangChain Tool ----
tool = create_sandbox_tool(sandbox)
# tool.name = "run_in_sandbox"
# tool.description = "在隔离的 Docker 沙盒中执行 Python 代码"

# ---- Agent 集成 ----
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

agent = create_react_agent(
    model=llm.llm,
    tools=[create_sandbox_tool(sandbox)],
    checkpointer=MemorySaver(),
)
result = await agent.ainvoke(
    {"messages": [("user", "在沙盒中执行代码")]},
    config={"configurable": {"thread_id": "001"}},
)

附录：完整源码

A.1 local_sandbox.py — Docker 沙盒核心实现

"""
Docker 本地沙盒核心实现

基于 Docker 构建完全隔离的代码执行环境，无需任何云服务。

核心能力：
    - 在 Docker 容器中执行 Python 代码，与宿主机完全隔离
    - 自动限制资源（CPU、内存），防止恶意代码耗尽系统资源
    - 执行完毕自动清理容器
    - 封装为 LangChain Tool，供 Agent 调用

架构图：
    Agent → run_in_sandbox(code) → Docker 容器 → 返回执行结果

依赖：
    pip install docker
    本地需安装 Docker Desktop（Windows/Mac）或 Docker Engine（Linux）
"""

from __future__ import annotations

import docker
import uuid
from typing import Optional
from langchain_core.tools import tool


# ============================================================
# Docker 沙盒类
# ============================================================

class DockerSandbox:
    """
    基于 Docker 的本地代码沙盒。

    原理：
        每次执行代码时，创建一个临时 Docker 容器，
        将代码写入容器内执行，捕获输出后销毁容器。

    安全隔离：
        - 文件系统隔离：容器内的文件操作不影响宿主机
        - 网络隔离：可选禁用网络访问
        - 资源限制：限制 CPU 和内存使用
        - 自动清理：执行完毕自动删除容器
    """

    def __init__(
        self,
        image: str = "python:3.12-slim",
        mem_limit: str = "256m",
        cpu_quota: int = 50000,
        network_disabled: bool = False,
        timeout: int = 30,
    ):
        self.image = image
        self.mem_limit = mem_limit
        self.cpu_quota = cpu_quota
        self.network_disabled = network_disabled
        self.timeout = timeout

        # 连接本地 Docker
        self.client = docker.from_env()
        print(f"[DockerSandbox] 镜像: {image}, 内存限制: {mem_limit}, "
              f"网络: {'禁用' if network_disabled else '启用'}")

    def execute(self, code: str) -> str:
        """在沙盒中执行 Python 代码"""
        container_name = f"sandbox-{uuid.uuid4().hex[:8]}"
        container = None

        try:
            # 创建容器（不启动）
            container = self.client.containers.create(
                image=self.image,
                name=container_name,
                command=["python", "-c", code],
                mem_limit=self.mem_limit,
                cpu_quota=self.cpu_quota,
                network_disabled=self.network_disabled,
                detach=True,
            )

            # 启动并等待完成
            container.start()
            result = container.wait(timeout=self.timeout)

            # 获取输出
            stdout = container.logs(stdout=True, stderr=False).decode("utf-8", errors="replace")
            stderr = container.logs(stdout=False, stderr=True).decode("utf-8", errors="replace")

            # 组装返回结果
            exit_code = result.get("StatusCode", -1)
            output_parts = []

            if stdout.strip():
                output_parts.append(stdout.strip())
            if stderr.strip():
                output_parts.append(f"[stderr] {stderr.strip()}")
            if exit_code != 0:
                output_parts.append(f"[退出码] {exit_code}")

            return "\n".join(output_parts) if output_parts else "(无输出)"

        except docker.errors.ContainerError as e:
            return f"沙盒执行错误: {e}"
        except Exception as e:
            return f"沙盒异常: {e}"
        finally:
            # 自动清理容器
            if container is not None:
                try:
                    container.remove(force=True)
                except Exception:
                    pass

    def cleanup(self):
        """释放 Docker 客户端资源"""
        self.client.close()


# ============================================================
# LangChain Tool 封装
# ============================================================

def create_sandbox_tool(sandbox: Optional[DockerSandbox] = None) -> tool:
    """创建沙盒执行工具，供 Agent 调用"""
    if sandbox is None:
        sandbox = DockerSandbox()

    @tool
    def run_in_sandbox(code: str) -> str:
        """在隔离的 Docker 沙盒中执行 Python 代码。
        沙盒环境与宿主机完全隔离，代码执行不会影响本地系统。
        入参是纯 Python 代码字符串，不要包裹在 ```python 标记中。"""
        return sandbox.execute(code)

    return run_in_sandbox


# ============================================================
# 独立运行测试
# ============================================================

if __name__ == "__main__":
    print("=" * 60)
    print("Docker 本地沙盒 - 独立测试")
    print("=" * 60)

    sb = DockerSandbox(network_disabled=True)

    # 测试1：基础输出
    print("\n--- 测试1: 基础输出 ---")
    result = sb.execute("print('Hello from Docker sandbox!')")
    print(f"结果: {result}")

    # 测试2：系统信息
    print("\n--- 测试2: 获取系统信息 ---")
    result = sb.execute("""
import sys, os
print(f'Python 版本: {sys.version}')
print(f'工作目录: {os.getcwd()}')
print(f'文件列表: {os.listdir(\\".\\")}')
""")
    print(f"结果: {result}")

    # 测试3：文件操作（沙盒内）
    print("\n--- 测试3: 沙盒内文件操作 ---")
    result = sb.execute("""
with open('sandbox_probe.txt', 'w') as f:
    f.write('hello from docker sandbox')
with open('sandbox_probe.txt', 'r') as f:
    content = f.read()
print(f'文件内容: {content}')
""")
    print(f"结果: {result}")

    # 测试4：资源限制验证
    print("\n--- 测试4: 内存限制验证（尝试分配 512MB）---")
    result = sb.execute("""
try:
    big_list = [0] * (512 * 1024 * 1024 // 8)
    print('分配成功（不应该发生）')
except MemoryError:
    print('内存不足，被沙盒限制成功拦截！')
""")
    print(f"结果: {result}")

    sb.cleanup()
    print("\n" + "=" * 60)
    print("测试完成")
    print("=" * 60)

A.2 docker_sandbox_test.py — Agent + Docker 沙盒集成测试

"""
Docker 本地沙盒 Agent 集成测试

让 Agent 在本地 Docker 沙盒中执行代码，无需任何云服务。

架构：
    用户指令 → LangGraph Agent → Docker 沙盒工具 → Docker 容器 → 返回结果
"""

from __future__ import annotations

import sys
import os
import asyncio

_project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
if _project_root not in sys.path:
    sys.path.insert(0, _project_root)

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import AIMessage, ToolMessage

from app.common import llm
from app.sandbox_agent_test.local_sandbox import DockerSandbox, create_sandbox_tool


# ============================================================
# 工具调用日志（追踪 Agent 是否真的通过沙盒执行代码）
# ============================================================

_printed_indices: set[int] = set()


def print_tool_call_logs(messages):
    """从 Agent 的 messages 中提取工具调用记录"""
    print("  [沙盒工具调用日志]")
    call_count = 0

    for idx, msg in enumerate(messages):
        if idx in _printed_indices:
            continue
        _printed_indices.add(idx)

        if isinstance(msg, AIMessage) and msg.tool_calls:
            for tc in msg.tool_calls:
                call_count += 1
                args_str = str(tc["args"])
                if len(args_str) > 150:
                    args_str = args_str[:150] + "..."
                print(f"  调用[{call_count}]: {tc['name']}({args_str})")

        elif isinstance(msg, ToolMessage):
            content = str(msg.content)
            if len(content) > 300:
                content = content[:300] + "..."
            print(f"    沙盒返回: {content}")

    if call_count == 0:
        print("  （本次未调用工具，Agent 直接回复）")
    print()


# ============================================================
# 测试函数
# ============================================================

async def run_sandbox_test():
    """运行 Docker 沙盒 Agent 测试"""

    print(f"[INFO] 当前使用模型: {llm.LLM_MODEL_NAME}")
    print(f"[INFO] 当前 Base URL: {llm.LLM_BASE_URL}")

    # 创建 Docker 沙盒和工具
    sandbox = DockerSandbox(
        image="python:3.12-slim",
        mem_limit="256m",
        network_disabled=True,  # 禁用网络，完全隔离
        timeout=30,
    )
    sandbox_tool = create_sandbox_tool(sandbox)

    # 创建 Agent（带内存记忆）
    memory = MemorySaver()
    agent = create_react_agent(
        model=llm.llm,
        tools=[sandbox_tool],
        checkpointer=memory,
    )

    config = RunnableConfig(configurable={"thread_id": "docker-sandbox-001"})

    # 预定义测试任务
    tasks = [
        "请在沙盒中执行：打印当前 Python 版本和工作目录",
        "在沙盒中创建一个文件 sandbox_test.txt，写入 'hello from local docker sandbox'，然后读取并打印内容",
        "在沙盒中计算 1 到 100 的累加和，并打印结果",
    ]

    print("\n" + "=" * 60)
    print("Docker 本地沙盒 Agent 测试")
    print("=" * 60)

    for i, task in enumerate(tasks, 1):
        print(f"\n--- 任务 {i}/{len(tasks)}: {task} ---")

        result = await agent.ainvoke(
            input={"messages": [("user", task)]},
            config=config,
        )

        # 打印工具调用日志
        print_tool_call_logs(result["messages"])

        # 打印 Agent 最终回复
        ai_message = result["messages"][-1].content
        print(f"[Agent 回复] {ai_message}")

    # 清理沙盒
    sandbox.cleanup()

    print("\n" + "=" * 60)
    print("测试完成")
    print("=" * 60)


if __name__ == "__main__":
    print("=" * 60)
    print("Docker 本地沙盒 Agent 集成测试")
    print("=" * 60)
    asyncio.run(run_sandbox_test())

A.3 sandbox_test.py — Daytona 云沙盒 Agent 测试（可选）

"""
Daytona 沙盒 Agent 集成测试

使用 Daytona 沙盒隔离代码执行环境，Agent 在沙盒内运行 Python 脚本，
不会影响宿主机文件系统。

核心组件：
    - Daytona: 沙盒基础设施，提供隔离的执行环境
    - DaytonaSandbox: langchain-daytona 适配器，将沙盒封装为 LangChain 工具
    - deepagents: Agent 框架，自动赋予 Agent 沙盒执行能力

依赖安装：
    pip install daytona-sdk langchain-daytona deepagents

必需环境变量：
    DAYTONA_API_KEY     Daytona API Key（在 Daytona 控制台获取）
"""

from __future__ import annotations

import sys
import os
import traceback
from typing import Any

_project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
if _project_root not in sys.path:
    sys.path.insert(0, _project_root)

from daytona import Daytona
from deepagents import create_deep_agent
from langchain_daytona import DaytonaSandbox

from app.common import llm


# ============================================================
# 沙盒任务：在隔离环境中执行 Python 脚本
# ============================================================

SANDBOX_TASK = """
请在沙盒里执行一个小型 Python 脚本，完成以下动作：
1. 打印当前 Python 版本。
2. 打印当前工作目录。
3. 在沙盒当前目录创建 sandbox_probe.txt。
4. 写入内容：hello from daytona sandbox。
5. 读取 sandbox_probe.txt 并打印读取结果。
6. 打印当前目录文件列表。
请明确说明这些操作是在沙盒内完成的。
""".strip()


# ============================================================
# 沙盒执行函数
# ============================================================

def invoke_agent(agent: Any) -> Any:
    """调用 Agent 执行沙盒任务"""
    return agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": SANDBOX_TASK,
                }
            ]
        }
    )


# ============================================================
# 主函数
# ============================================================

def main() -> int:
    sandbox = None

    try:
        # 检查 Daytona API Key
        if not os.getenv("DAYTONA_API_KEY"):
            raise RuntimeError("缺少必需环境变量：DAYTONA_API_KEY")

        print(f"[INFO] 当前使用模型: {llm.LLM_MODEL_NAME}")
        print(f"[INFO] 当前 Base URL: {llm.LLM_BASE_URL}")

        # 第一步：创建 Daytona 沙盒实例
        print("\n[1/4] 正在创建 Daytona 沙盒...")
        sandbox = Daytona().create()
        backend = DaytonaSandbox(sandbox=sandbox)

        # 第二步：创建具备沙盒能力的 Agent（使用项目统一的 LLM）
        print("[2/4] 正在创建具备沙盒能力的 Agent...")
        agent = create_deep_agent(
            model=llm.llm,
            system_prompt=(
                "你是一个拥有 Daytona 沙盒访问权限的 Python 编程助手。"
                "所有需要执行的代码都必须在沙盒中运行，不要在宿主机执行。"
                "执行完成后，返回关键命令输出和生成文件验证结果。"
            ),
            backend=backend,
        )

        # 第三步：调用 Agent 执行沙盒任务
        print("[3/4] 正在调用 Agent 执行沙盒任务...")
        result = invoke_agent(agent)

        # 第四步：输出执行结果
        print("[4/4] Agent 执行结果：")
        print(result)
        return 0

    except Exception as exc:
        # 捕获所有异常，完整输出失败原因
        print(f"沙盒 Agent 测试失败：{exc}", file=sys.stderr)
        traceback.print_exc()
        return 1

    finally:
        # 无论成功失败，都要清理沙盒资源
        if sandbox is not None:
            print("\n正在清理 Daytona 沙盒...")
            try:
                sandbox.stop()
                print("Daytona 沙盒已停止。")
            except Exception as cleanup_exc:
                print(f"停止 Daytona 沙盒失败：{cleanup_exc}", file=sys.stderr)


if __name__ == "__main__":
    print("=" * 60)
    print("Daytona 沙盒 Agent 集成测试")
    print("=" * 60)
    raise SystemExit(main())