Chuoer47's Blog

Appearance

LeetCode Agent Lab - 工具系统与沙箱执行 ​

工具系统（src/tool_executor/）和沙箱执行（src/execution/）是项目的基础设施层。Agent 不管是多 Agent 还是 ReAct，都通过统一的工具接口调用底层能力。本文深入分析 6 层工具架构、@tool_method 装饰器、沙箱安全执行等设计。

1. 工具系统：6 层架构 ​

工具系统采用分层设计，每层职责清晰：

┌─────────────────────────────────────┐
│  6. service.py — 业务门面            │
│     @tool_method 装饰器              │
├─────────────────────────────────────┤
│  5. adapters.py — 工具适配器         │
│     将现有模块包装为标准工具接口       │
├─────────────────────────────────────┤
│  4. loader.py — 配置驱动注册         │
│     从 tools.v2.json 加载工具定义     │
├─────────────────────────────────────┤
│  3. executor.py — 执行编排           │
│     重试、审批、沙箱选择              │
├─────────────────────────────────────┤
│  2. router.py — 名称路由             │
│     source:name / source.name 前缀   │
├─────────────────────────────────────┤
│  1. registry.py — 注册与查找         │
│     ToolDefinition + 别名支持        │
└─────────────────────────────────────┘

1.1 第 1 层：Registry — 注册与查找 ​

ToolRegistry 管理所有工具的注册和查找，支持别名：

python
class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, ToolDefinition] = {}
        self._name_bindings: dict[str, _ToolNameBinding] = {}

    def register(self, definition: ToolDefinition, overwrite: bool = False) -> None:
        primary = definition.name
        self._tools[primary] = definition
        # 注册主名称和所有别名
        for name in [primary, *definition.aliases]:
            self._name_bindings[name] = _ToolNameBinding(primary_name=primary)

    def get(self, name: str) -> ToolDefinition:
        key = name.strip()
        primary = self._name_bindings[key].primary_name
        return self._tools[primary]

设计要点：

  • 别名支持：同一个工具可以有多个名字（如 run_solution_tests 和 run_tests）
  • 冻结数据类：_ToolNameBinding 用 @dataclass(frozen=True) 保证不可变
  • 查找时自动 trim 空白

1.2 第 2 层：Router — 名称路由 ​

ToolRouter 解析工具名称中的来源提示（source hint），支持两种格式：

python
class ToolRouter:
    @staticmethod
    def _split_source_hint(tool_name: str) -> tuple[str | None, str]:
        name = tool_name.strip()
        # 支持 "source:name" 和 "source.name" 两种格式
        if ":" in name:
            source, rest = name.split(":", 1)
            if source.strip().lower() in SUPPORTED_TOOL_SOURCES:
                return source.strip().lower(), rest.strip()
        if "." in name:
            source, rest = name.split(".", 1)
            if source.strip().lower() in SUPPORTED_TOOL_SOURCES:
                return source.strip().lower(), rest.strip()
        return None, name

路由逻辑按优先级尝试多个候选名称：

python
def route(self, request: ToolCallRequest) -> ToolRouteDecision:
    source_hint, base_name = self._split_source_hint(request.tool_name)

    candidates = [request.tool_name.strip(), base_name]
    if source_hint:
        candidates.insert(0, f"{source_hint}:{base_name}")
        candidates.insert(1, f"{source_hint}.{base_name}")

    for name in candidates:
        if self.registry.has(name):
            definition = self.registry.get(name)
            if source_hint and definition.source != source_hint:
                continue
            return ToolRouteDecision(
                requested_tool_name=request.tool_name,
                resolved_tool_name=definition.name,
                source=definition.source,
                definition=definition,
            )
    raise KeyError(f"无法路由工具: {request.tool_name}")

为什么需要路由？ 因为 LLM 输出的工具名可能不完全一致——有时带 source 前缀，有时不带。路由层做模糊匹配，让 Agent 不需要精确记住工具名。

1.3 第 3 层：Executor — 执行编排 ​

ToolExecutor 负责工具的实际执行，包括重试、审批、沙箱选择。

1.4 第 4 层：Loader — 配置驱动注册 ​

loader.py 从 tools.v2.json 配置文件加载工具定义，实现配置驱动的工具注册。

1.5 第 5 层：Adapters — 工具适配器 ​

adapters.py（约 866 行）是工具系统的"翻译层"，将现有模块的功能包装为标准工具接口。注册了约 30+ 个工具，涵盖：

类别工具
题目加载load_problem, fetch_leetcode_problem
代码执行run_solution_tests, run_test_generator
对拍generate_duel_dataset, run_duel
记忆retrieve_memories, list_runs, score_trend
搜索serp_search（SerpAPI）
LeetCodeprepare_submission, submit_solution, check_submission_status
Shellshell_list_directory, shell_read_file, shell_search_text, shell_write_file
Workspaceworkspace_write_solution, workspace_read_solution, workspace_apply_unified_diff 等

1.6 第 6 层：Service — 业务门面 ​

ToolExecutionService 是面向业务的门面，提供 @tool_method 装饰器。

2. @tool_method 装饰器 ​

2.1 设计动机 ​

在引入装饰器之前，每个工具方法都有大量样板代码：提取参数、构建 arguments dict、调用 invoke_or_raise、校验返回格式。service.py 一度膨胀到 735 行。

@tool_method 装饰器解决了这个问题：

python
def tool_method(fn=None, *, deserialize=None):
    """标记一个方法为 tool_executor 便捷方法，自动生成样板代码。"""
    def decorator(fn):
        sig = inspect.signature(fn)

        @functools.wraps(fn)
        def wrapper(self, *args, **kwargs):
            bound = sig.bind_partial(self, *args, **kwargs)
            bound.apply_defaults()
            context = bound.arguments.get("context")
            # 自动构建 arguments dict，过滤 None 值
            arguments = {
                k: v for k, v in bound.arguments.items()
                if k not in ("self", "context") and v is not None
            }
            tool_name = fn.__name__
            payload = self.invoke_or_raise(
                tool_name=tool_name, arguments=arguments, context=context
            )
            if not isinstance(payload, dict):
                raise ToolInvocationError(f"{tool_name} 返回结果格式异常")
            if deserialize is not None:
                return deserialize(payload)
            return payload
        return wrapper
    # 支持 @tool_method 和 @tool_method(deserialize=xxx) 两种用法
    if fn is not None:
        return decorator(fn)
    return decorator

2.2 使用效果 ​

Before（手动样板代码）：

python
def load_problem(self, slug: str, context=None) -> Problem:
    arguments = {}
    if slug is not None:
        arguments["slug"] = slug
    payload = self.invoke_or_raise("load_problem", arguments, context)
    if not isinstance(payload, dict):
        raise ToolInvocationError("load_problem 返回结果格式异常")
    return _deserialize_problem(payload)

After（装饰器）：

python
@tool_method(deserialize=_deserialize_problem)
def load_problem(self, slug: str, context=None) -> Problem:
    ...

代码量减少了约 35%（735 → 472 行），而且新增工具方法只需一行装饰器。

2.3 反序列化回调 ​

装饰器支持 deserialize 参数，用于将原始 dict 转换为领域对象：

python
def _deserialize_problem(payload: dict[str, Any]) -> Problem:
    return Problem.model_validate(payload)

def _deserialize_memories(payload: dict[str, Any]) -> list[RetrievedMemory]:
    raw_items = payload.get("items", [])
    return [RetrievedMemory.model_validate(item) for item in raw_items if isinstance(item, dict)]

3. 沙箱执行 ​

3.1 为什么需要沙箱 ​

Agent 生成的代码需要在本地执行测试。但直接执行 LLM 生成的代码有安全风险——它可能包含恶意操作（删除文件、网络请求等）。沙箱的作用是：隔离执行环境，限制资源使用。

3.2 三种后端 ​

python
_VALID_BACKENDS = {"auto", "process", "nsjail"}
后端说明安全性
nsjail轻量级容器隔离高（内存限制、CPU 限制、网络禁用）
process直接子进程执行低（仅用于开发）
auto自动选择（默认）有 nsjail 用 nsjail，否则降级

3.3 后端解析逻辑 ​

python
def resolve_backend_info(policy: SandboxPolicy) -> tuple[str, bool, str]:
    backend = (policy.backend or "auto").strip().lower()

    if backend == "process":
        return "process", False, ""

    nsjail_available = _resolve_nsjail_binary(policy.nsjail_path) is not None
    if backend == "nsjail":
        if not nsjail_available:
            raise SandboxConfigError(f"nsjail not found: {policy.nsjail_path}")
        return "nsjail", False, ""

    # auto 模式
    if nsjail_available:
        return "nsjail", False, ""
    reason = f"nsjail not found: {policy.nsjail_path}"
    if bool(getattr(policy, "fail_on_degrade", False)):
        raise SandboxConfigError(f"sandbox degrade blocked: {reason}")
    return "process", True, reason  # 降级到 process

设计要点：

  • auto 模式默认尝试 nsjail，不可用时降级到 process
  • fail_on_degrade=True 可以阻止静默降级——如果 nsjail 不可用直接报错，而不是悄悄用 process 跑
  • 降级时返回 is_degraded=True 和降级原因，上层可以记录告警

3.4 nsjail 配置 ​

nsjail 模式下的资源限制：

python
# 简化的 nsjail 命令构建
def _build_nsjail_prefix(policy, *, timeout_seconds, cwd):
    return [
        "nsjail",
        "--mode", "l",                          # l = once (执行一次)
        "--time_limit", str(int(timeout_seconds)),  # 超时
        "--rlimit_as", str(policy.memory_limit_mb),  # 内存限制
        "--disable_proc",                        # 禁用 /proc
        "--disable_clone_newnet" if not policy.allow_network else "",
        "--cwd", str(cwd),
        # ... 更多限制
    ]

3.5 三种 Runner ​

Runner作用
code_runner.py执行 Python 解答，跑测试用例
test_script_runner.py执行测试生成脚本
duel_runner.py对拍：候选代码 vs 参考解

4. Workspace 工具 ​

4.1 设计理念 ​

Workspace 工具是 ReAct Agent 的文件操作层。核心理念是：模型只看到抽象工具，不看到真实文件路径。

模型视角：workspace_write_solution(content="...")
实际操作：写入 /tmp/runs/<run_id>/solution.py

4.2 隐藏 run_root ​

run_root 是每次运行的真实工作目录，但模型不知道它的路径。所有文件操作通过工具名间接完成：

工具操作
workspace_write_solution写入 solution.py
workspace_read_solution读取 solution.py
workspace_write_oracle写入 oracle.py（参考解）
workspace_write_gen_cases写入 gen_cases.py（测试生成）
workspace_apply_unified_diff应用 diff 补丁
workspace_run_solution_tests执行解答测试
workspace_leetcode_submit在线提交到 LeetCode

4.3 审计追踪 ​

所有文件操作都有审计记录：

run_root/
├── solution.py          # 当前解答
├── oracle.py            # 参考解
├── gen_cases.py         # 测试生成脚本
├── patches/             # 补丁历史
│   ├── 001.diff
│   └── 002.diff
└── observations/        # 执行观测
    ├── 001_base_run.json
    └── 002_duel_run.json

5. Shell 工具 ​

Shell 工具提供文件系统操作能力：

工具作用
shell_list_directory列出目录内容
shell_read_file读取文件
shell_search_text文本搜索
shell_write_file写入文件
shell_environment_info获取环境信息

这些工具让 Agent 可以查看文件结构、读取已有代码、搜索参考实现等。

总结 ​

知识点要点
工具架构6 层：Registry → Router → Executor → Loader → Adapters → Service
@tool_method自动提取参数、构建 arguments、调用 invoke_or_raise
沙箱后端nsjail（安全）/ process（开发）/ auto（自动选择）
降级策略fail_on_degrade 阻止静默降级
Workspace模型只看抽象工具，不看真实文件路径
审计追踪patches/ 和 observations/ 记录所有操作

项目源码：leetcode-coder-agent