从一个看似简单的请求开始

你在 WhatsApp 上发了一条消息：

这条消息经过前两篇文章描述的 Gateway 和路由系统，最终到达了 Agent。然后呢？

这个请求背后需要的是一连串精密操作：理解意图、找到会议记录文件、读取内容、生成摘要、再写入文件、最后报告结果。每一步都可能出问题：文件找不到、API 限速、上下文太长……

Agent 执行引擎的任务，就是可靠地走完这整个过程，无论中间发生什么。

第一个挑战：防止并发冲突

在讨论 AI 怎么「思考」之前，得先解决一个基础问题：同一个会话能同时处理两条消息吗？

答案是不能。同一个 AI 对话上下文（SessionKey）只有一段历史记录，如果两条消息并发写入，历史就会错乱，AI 的上下文也就跟着崩了。

OpenClaw 的解决方案是 Lane（执行通道）：

// src/agents/pi-embedded-runner/lanes.ts
export function resolveSessionLane(key: string) {
    // 每个 SessionKey 对应一条独立的命令队列
    return `session:${key}`;
}

每个 SessionKey 有且仅有一条 Lane，Lane 内的任务严格按序执行：

// src/agents/pi-embedded-runner/run.ts
const sessionLane = resolveSessionLane(params.sessionKey ?? params.sessionId);
return enqueueSession(() =>
    // ① 先排队到 session lane
    enqueueGlobal(async () => {
        // ② 再排队到 global lane
        // 实际的 AI 执行逻辑
    })
);

这里两层 enqueue 嵌套的设计很有讲究：

• session lane：确保同一会话的消息串行，防止并发写入
• global lane：让跨会话的资源（模型连接、文件句柄）也能公平排队，避免单会话独占资源

可以理解为一种多级队列模式——内层管并发安全，外层管资源公平。

进入正题：一次执行的五个阶段

进入 Lane 后，核心函数 runEmbeddedAttempt 开始接力。它依次做五件事：

第一阶段：准备工作区和技能环境

// src/agents/pi-embedded-runner/run/attempt.ts
const sandbox = await resolveSandboxContext({ config, sessionKey, workspaceDir });
const effectiveWorkspace = sandbox?.enabled ? sandbox.workspaceDir : resolvedWorkspace;
// 切换工作目录到 workspace（AI 的文件系统视角）
process.chdir(effectiveWorkspace);
// 加载技能（Skills）并应用环境变量覆盖
const skillEntries = loadWorkspaceSkillEntries(effectiveWorkspace);
restoreSkillEnv = applySkillEnvOverrides({ skills: skillEntries, config });

Skills（技能）是 OpenClaw 的扩展机制——类似于给 AI 安装的「应用」。一个技能可以提供：

• 专属的环境变量（比如 GITHUB_TOKEN）
• 文档说明（注入系统提示，告诉 AI 这个工具怎么用）
• 预定义的任务模板

技能文档会在下一步注入到系统提示中，让 AI 知道自己拥有哪些能力。

第二阶段：构建系统提示

系统提示是 AI「人格」的来源——它决定 AI 如何行动、什么能做、什么不能做。OpenClaw 的系统提示是动态拼装出来的：

// src/agents/pi-embedded-runner/run/attempt.ts
const appendPrompt = buildEmbeddedSystemPrompt({
    workspaceDir: effectiveWorkspace,      // AI 的工作目录在哪里
    defaultThinkLevel: params.thinkLevel,   // 是否开启深度思考
    skillsPrompt,     // 已安装的技能说明
    docsPath,          // 文档路径
    sandboxInfo,       // 沙盒限制信息
    tools,             // 可用工具列表
    runtimeInfo: {     // 运行时环境
        host: machineName,
        os: `${os.type()} ${os.release()}`,
        model: `${params.provider}/${params.modelId}`,
        channel: runtimeChannel,
        capabilities: runtimeCapabilities,  // 当前渠道支持哪些功能（如 Telegram 的 inline buttons）
    },
    reactionGuidance,  // Telegram/Signal 的 emoji reaction 指导
    messageToolHints,  // 消息发送工具的使用建议
    // ...更多参数
});

留意 runtimeCapabilities：不同渠道的 AI 行为会有差异。Telegram 支持 inline buttons，AI 就知道可以发送交互按钮；WhatsApp 不支持，AI 就只发纯文本。系统提示会根据当前渠道动态调整 AI 的能力描述。

第三阶段：加载会话历史

AI 需要知道「之前说了什么」才能接着对话：

// src/agents/pi-embedded-runner/run/attempt.ts
await repairSessionFileIfNeeded({ sessionFile: params.sessionFile });
const sessionManager = guardSessionManager(
    (await createAgentSession({ sessionFile, ... })).session,
    { sessionId: params.sessionId }
);
// 历史长度限制：DM 会话有单独的上限（避免单用户独占上下文）
const historyLimit = getDmHistoryLimitFromSessionKey(params.sessionKey, params.config);
if (historyLimit) {
    await limitHistoryTurns(sessionManager, historyLimit);
}

会话历史存储在 JSONL 文件中（~/.openclaw/agents//sessions/），由 @mariozechner/pi-coding-agent 的 SessionManager 管理。OpenClaw 在外面套了一层 guardSessionManager，拦截并检查每次写操作的合法性——比如 tool_use 和 tool_result 必须正确配对。

第四阶段：注册工具

AI 的「手」就是工具。所有可用工具在这里注册：

// src/agents/pi-embedded-runner/run/attempt.ts
const toolsRaw = createOpenClawCodingTools({
    agentId: sessionAgentId,
    exec: { ...params.execOverrides, elevated: params.bashElevated },
    sandbox,
    messageProvider: params.messageChannel,
    sessionKey: params.sessionKey ?? params.sessionId,
    workspaceDir: effectiveWorkspace,
    config: params.config,
    abortSignal: runAbortController.signal,
    // ... 更多上下文
});
// 工具策略过滤
const tools = sanitizeToolsForGoogle({ tools: toolsRaw, provider: params.provider });
const allowedToolNames = collectAllowedToolNames({ tools, clientTools: params.clientTools });

工具集包括：文件读写、bash 执行、消息发送、网络请求、媒体处理……工具策略下一节会详述。

第五阶段：订阅流式输出

// src/agents/pi-embedded-runner/run/attempt.ts
const subscribeResult = await subscribeEmbeddedPiSession({
    session: sessionManager,
    prompt: params.prompt,
    onBlockReply: params.onBlockReply,       // 每当 AI 完成一个文本块，回调这里
    onReasoningStream: params.onReasoningStream,
    // ...
});

subscribeEmbeddedPiSession 是 AI 实际执行的入口，接收来自 SDK 的流式事件并处理。

流式订阅：AI 的思考过程是如何被捕获的

subscribeEmbeddedPiSession 处理来自 @mariozechner/pi-agent-core SDK 的三类事件：

事件 1：文本流

// 每次 token 到来
text_delta → deltaBuffer 累积 → 检测  标签 → 过滤/输出
const THINKING_TAG_SCAN_RE = /<\s*(\/)?\s*(?:think(?:ing)?|thought|antthinking)\s*>/gi;

遇到 ... 时，内容根据 reasoningMode 决定去留：

• off：过滤掉，用户看不到 AI 的「思考链」
• on：把思考过程作为单独消息发送
• stream：实时推送思考流（实验性）

文本块的「发送时机」由 blockReplyBreak 控制：

• text_end（默认）：一个文本块完成才发送——避免频繁打断
• paragraph：遇到段落换行就发送——让用户更快看到进展

这背后涉及一个代码块感知分块器（EmbeddedBlockChunker）：分割文本时会检测是否在代码块内，避免把代码块从中劈开，破坏 Markdown 渲染。

事件 2：工具调用

// 工具调用事件序列：
tool_use_start → 分发到对应工具执行器
tool_use_result → 把结果写回 SessionManager

工具调用前会经过 runBeforeToolCallHook：

// src/agents/pi-tools.before-tool-call.ts
export async function runBeforeToolCallHook(args: {
    toolName: string;
    params: unknown;
    toolCallId?: string;
    ctx?: HookContext;
}): Promise {
    // 1. 工具循环检测（防止 AI 死循环重复调用同一工具）
    // 2. 插件钩子（before_tool_call hook，可以拦截或修改参数）
    // 3. 返回 blocked=true 时，把错误作为工具结果返回给 AI
}

工具循环检测的逻辑很直白：如果 AI 在 10 次调用内用相同参数反复调用同一工具，那就说明陷入了循环——向 AI 提示「检测到重复调用，请换个思路」。

事件 3：会话压缩信号

// 当 pi-agent-core 内部触发压缩时
compaction_start → 设置 compactionInFlight = true
compaction_done → 清除标志，继续流式

工具策略：AI 被允许做什么？

AI 拥有大量工具，但不是所有场景都适合开放所有工具。工具策略是安全边界的关键实现。

工具过滤：Deny/Allow 模式

// src/agents/pi-tools.policy.ts
function makeToolPolicyMatcher(policy: SandboxToolPolicy) {
    const deny = compileGlobPatterns({ raw: expandToolGroups(policy.deny ?? []) });
    const allow = compileGlobPatterns({ raw: expandToolGroups(policy.allow ?? []) });
    return (name: string) => {
        if (matchesAnyGlobPattern(normalized, deny)) return false;  // 拒绝列表优先
        if (allow.length === 0) return true;  // 无允许列表 = 全部允许
        return matchesAnyGlobPattern(normalized, allow);
    };
}

工具名支持 Glob 模式：exec:* 匹配所有 exec 系列工具，bash 只匹配 bash 工具。

子 Agent 的额外限制

当主 Agent 派生一个子 Agent（subagent）来处理子任务时，子 Agent 的工具集会受到额外限制：

// src/agents/pi-tools.policy.ts — 永远拒绝给子 Agent 的工具
const SUBAGENT_TOOL_DENY_ALWAYS = [
    "gateway",          // 系统管理——危险
    "agents_list",      // 系统管理
    "whatsapp_login",   // 交互式设置——不是任务
    "session_status",   // 状态/调度——主 Agent 负责协调
    "cron",             // 定时任务——不归子 Agent 管
    "memory_search",    // 记忆——主 Agent 通过 spawn prompt 传递相关信息
    "memory_get",
    "sessions_send",    // 直接会话发送——子 Agent 通过 announce 链通信
];
// 叶子子 Agent（不能再派生子 Agent 的最深层）额外限制
const SUBAGENT_TOOL_DENY_LEAF = [
    "sessions_list",
    "sessions_history",
    "sessions_spawn",   // 叶子不能再派生
];

这个设计背后是一个清晰的原则：每个 Agent 只做它该做的事。子 Agent 是执行者，不是管理者；记忆查询和任务调度是编排者（主 Agent）的职责。

子 Agent 的深度可以配置（maxSpawnDepth），深度越大，工具限制就越严格：

• 深度 1（第一层子 Agent）且 maxSpawnDepth >= 2：可以是「编排者」，允许派生孙子 Agent
• 深度 >= maxSpawnDepth（叶子）：只能执行，不能派生

外层重试循环：对抗现实世界的不可靠

内层的「单次执行」偶尔会失败：API 限速、认证过期、上下文溢出……外层有一个专门处理这些情况的重试循环。

// src/agents/pi-embedded-runner/run.ts
const MAX_RUN_LOOP_ITERATIONS = resolveMaxRunRetryIterations(profileCandidates.length);
// 32~160 次，根据认证 Profile 数量动态调整
while (true) {
    if (runLoopIterations >= MAX_RUN_LOOP_ITERATIONS) {
        return { error: "Exceeded retry limit after N attempts" };
    }
    const attempt = await runEmbeddedAttempt({ ... });
    if (attempt 成功) {
        markAuthProfileGood(profileId);  // 标记这个 Profile 可用
        return 成功结果;
    }
    if (isRateLimitError(attempt)) {
        markAuthProfileFailure(profileId, "rate_limit");  // 标记限速，进入冷却
        const advanced = await advanceAuthProfile();  // 切换到下一个 Profile
        if (!advanced) return 失败;
        continue;  // 用新 Profile 重试
    }
    if (isContextOverflowError(attempt)) {
        if (overflowCompactionAttempts < 3) {
            await compactEmbeddedPiSession( ... );  // 压缩会话历史
            overflowCompactionAttempts++;
            continue;  // 压缩后重试
        }
        return 上下文溢出失败;
    }
    if (isAuthError(attempt)) {
        markAuthProfileFailure(profileId, "auth");
        advanceAuthProfile();
        continue;
    }
    // ... 其他错误处理
}

认证 Profile 轮转

这是 OpenClaw 处理 API 限速的核心机制。你可以配置多个 API key（或 OAuth 账号），称为「认证 Profile」：

# openclaw.yml
auth:
  profiles:
    - id: primary
      provider: anthropic
      apiKey: sk-ant-...
    - id: backup-1
      provider: anthropic
      apiKey: sk-ant-...
    - id: backup-2
      provider: anthropic
      apiKey: sk-ant-...

当 primary 触发速率限制时：

1. markAuthProfileFailure(primary, "rate_limit") —— 进入冷却期
2. advanceAuthProfile() —— 切换到 backup-1
3. 用 backup-1 重试
4. 如果 backup-1 也限速，切到 backup-2
5. 所有 Profile 都在冷却期 → 向用户报告「API 暂时不可用」

这解决了一个个人 AI 助手的实际痛点：深夜你发起一个需要大量 API 调用的复杂任务，如果只有一个 key，触发限速就只能干等；有了轮转，系统会自动用其他 key 继续工作。

上下文溢出与压缩（Compaction）

LLM 的上下文窗口是有限的（比如 Claude 200k tokens）。长时间的对话、大量工具结果，迟早会填满。

当 API 返回「上下文超出」错误时：

// src/agents/pi-embedded-runner/run.ts
if (isLikelyContextOverflowError(attempt)) {
    const compacted = await compactEmbeddedPiSession({
        sessionFile: params.sessionFile,
        trigger: "overflow",
        model: compactionModelId,  // 使用轻量模型做摘要（而非当前的大模型）
        // ...
    });
    // 压缩成功后，重新运行 attempt，历史已经被摘要化
}

压缩的过程：

1. 读取完整会话历史
2. 用 AI 生成一段「对话摘要」
3. 把历史消息替换为这段摘要
4. 用压缩后的历史重新发起请求

这不是简单的截断——截断会导致 AI「失忆」，压缩保留了关键上下文。用小模型做摘要也是个聪明的选择：摘要任务不需要复杂推理，用便宜快速的模型能节省时间和成本。

压缩最多重试 3 次（MAX_OVERFLOW_COMPACTION_ATTEMPTS = 3），防止陷入无限压缩循环。

全流程图

把以上所有内容串联起来：

消息抵达 Agent
↓
runEmbeddedPiAgent()→ 入队 session lane（保证同一会话串行）
                 → 入队 global lane（资源公平排队）
↓
外层重试循环（最多 160 次）
↓                         ↑ (失败时：认证轮转 / 上下文压缩 / 模型 failover)
runEmbeddedAttempt()
① 准备工作区 + 技能环境
② 动态构建系统提示（含渠道能力、技能文档）
③ 加载会话历史（含历史长度限制）
④ 注册工具（含策略过滤）
⑤ subscribeEmbeddedPiSession()
   ↓
pi-agent-core SDK 内层循环：
[模型生成文本]
   ↓ text_delta 事件
     检测  标签 → 过滤/单独输出
     EmbeddedBlockChunker 分块（代码块感知）
     onBlockReply → 推送给 Gateway → 广播到所有客户端
   ↓
[模型决定调用工具]
   ↓ tool_use 事件
     runBeforeToolCallHook（循环检测 + 插件钩子）
     ↓
     工具执行（bash / 文件读写 / 消息发送 / ...）
     tool_result → 写回 SessionManager
     ↓
     结果送回模型，继续下一轮推理
   ↓
[模型完成]
     最终回复通过渠道发回用户

总结