OpenCode 源码-Session 处理器——Prompt 编排、Compaction与工具执行循环

OpenCode 源码解析

第 18 讲：Session 处理器——Prompt 编排、Compaction 与工具执行循环

基于 dev 分支源码 · 2026-07-26

一、Session 模块：OpenCode 的 AI 大脑

前十七讲覆盖了 OpenCode 从 CLI 到事件总线的核心能力。这一讲深入Session 会话模块——OpenCode 的 AI 大脑，负责 Prompt 编排、LLM 流式调用、工具执行循环和上下文压缩。Session 模块横跨 21 个文件、约 7,129 行代码，是整个系统最复杂的业务逻辑层。

📦 本讲核心文件

session/prompt.ts（1630 行）— Prompt 编排与会话入口

session/processor.ts（716 行）— LLM 流式事件处理器

session/compaction.ts（562 行）— 上下文压缩引擎

session/session.ts（1018 行）— 会话 CRUD 与 Token 计费

二、SessionProcessor：12 种流式事件的管道

processor.ts（716 行）是 Session 的核心事件处理器。LLM 的流式响应被拆解为 12 种事件类型，每个事件触发对应的状态更新。

1. ProcessorContext：处理器上下文

📄 session/processor.ts (第 67-75 行)

interface ProcessorContext extends Input {
  toolcalls: Record<string, ToolCall>   // 活跃工具调用表
  shouldBreak: boolean                   // 是否中断循环
  snapshot: string | undefined          // 文件系统快照
  blocked: boolean                      // 是否被权限阻塞
  needsCompaction: boolean              // 是否需要压缩
  currentText: SessionV1.TextPart | undefined
  reasoningMap: Record<string,
    SessionV1.ReasoningPart>            // 推理过程映射
}

// 工具调用句柄
type ToolCall = {
  partID: SessionV1.ToolPart["id"]
  messageID: SessionV1.ToolPart["messageID"]
  sessionID: SessionV1.ToolPart["sessionID"]
  done: Deferred.Deferred<void>         // 完成信号
}

上下文设计：ProcessorContext 维护了整个 LLM 处理过程的状态。toolcalls 是一个活跃工具调用表，每个工具调用都有一个 Deferred 完成信号。reasoningMap 跟踪推理过程的增量更新。snapshot 记录文件系统状态，用于 step-finish 时生成 diff。

2. handleEvent：事件分发器

📄 session/processor.ts (第 276-509 行)

const handleEvent = Effect.fnUntraced(function* (value: StreamEvent) {
  switch (value.type) {
    // --- 推理事件 ---
    case "reasoning-start":
      // 创建推理 Part，记录开始时间
      ctx.reasoningMap[value.id] = {
        id: PartID.ascending(),
        type: "reasoning", text: "",
        time: { start: Date.now() },
      }
      yield* session.updatePart(ctx.reasoningMap[value.id])
      return

    case "reasoning-delta":
      // 增量更新推理文本（不重建整个 Part）
      ctx.reasoningMap[value.id].text += value.text
      yield* session.updatePartDelta({
        field: "text", delta: value.text,
      })
      return

    case "reasoning-end":
      yield* finishReasoning(value.id)
      return

    // --- 工具输入事件 ---
    case "tool-input-start":
    case "tool-input-delta":
    case "tool-input-end":
      if (ctx.assistantMessage.summary) {
        throw new Error(
          `Tool call not allowed while generating summary: ${value.name}`)
      }
      yield* ensureToolCall(value)
      return

    // --- 工具调用事件 ---
    case "tool-call":
      yield* ensureToolCall(value)
      yield* updateToolCall(value.id, (match) => ({
        ...match,
        state: { status: "running",
          input, time: { start: Date.now() } }
      }))
      // 死循环检测（见下文）
      return

    // --- 工具结果事件 ---
    case "tool-result":
      const rawOutput = toolResultOutput(value)
      // 图片附件自动 resize
      const normalized = yield* Effect.forEach(
        rawOutput.attachments ?? [], (attachment) =>
        attachment.mime.startsWith("image/")
          ? image.normalize(attachment)
          : Effect.succeed(Exit.succeed(attachment)))
      yield* completeToolCall(value.id, output)
      return

    // --- 步骤事件 ---
    case "step-start":
      if (!ctx.snapshot) ctx.snapshot = yield* snapshot.track()
      yield* session.updatePart({ type: "step-start" })
      return

    case "step-finish":
      // 计算 token 用量和成本
      const usage = Session.getUsage({
        model: ctx.model, usage: value.usage,
        metadata: value.providerMetadata,
      })
      ctx.assistantMessage.cost += usage.cost
      ctx.assistantMessage.tokens = usage.tokens
      // 生成文件 diff
      if (ctx.snapshot) {
        const patch = yield* snapshot.patch(ctx.snapshot)
        if (patch.files.length) {
          yield* session.updatePart({
            type: "patch", hash: patch.hash,
            files: patch.files,
          })
        }
      }
      // 检查是否需要压缩
      if (isOverflow({ cfg, tokens: usage.tokens, model })) {
        ctx.needsCompaction = true
      }
      return
  }
})

事件管道详解：handleEvent 是一个 12 种事件类型的 switch 分发器。推理事件（reasoning-start/delta/end）跟踪模型思考过程；工具事件（tool-input/tool-call/tool-result/tool-error）管理工具生命周期；步骤事件（step-start/step-finish）标记一轮 LLM 调用的边界；文本事件（text-start/text-delta）处理流式文本输出。

三、死循环检测：DOOM_LOOP_THRESHOLD 机制

当 LLM 连续 3 次调用相同工具且参数完全相同时，触发死循环检测。

检测逻辑

📄 session/processor.ts (第 354-378 行)

const DOOM_LOOP_THRESHOLD = 3

// 在 tool-call 事件中检测
const recentParts = parts.slice(-DOOM_LOOP_THRESHOLD)

if (
  recentParts.length !== DOOM_LOOP_THRESHOLD ||
  !recentParts.every(
    (part) =>
      part.type === "tool" &&
      part.tool === value.name &&                   // 相同工具
      part.state.status !== "pending" &&
      JSON.stringify(part.state.input) ===           // 相同参数
        JSON.stringify(input)
  )
) {
  return  // 未触发死循环
}

// 触发死循环：请求用户授权
const agent = yield* agents.get(ctx.assistantMessage.agent)
yield* permission.ask({
  permission: "doom_loop",
  patterns: [value.name],
  sessionID: ctx.assistantMessage.sessionID,
  metadata: { tool: value.name, input },
  always: [value.name],
  ruleset: agent.permission,
})

检测条件：最近 3 个 Part 必须都是 tool 类型、相同工具名、非 pending 状态、输入参数 JSON 完全一致。四个条件同时满足才触发。触发后通过 Permission 服务请求用户授权，用户可以选择 Allow Always 或拒绝。

四、Compaction：上下文压缩引擎

compaction.ts（562 行）实现了三级压缩策略：Prune（工具输出清理）→ Select（消息选择）→ Summarize（LLM 摘要）。

1. 压缩参数矩阵

📄 session/compaction.ts (第 28-34 行)

export const PRUNE_MINIMUM = 20_000       // 清理阈值：至少省 20K token
export const PRUNE_PROTECT = 40_000       // 保护阈值：保留最近 40K token
const TOOL_OUTPUT_MAX_CHARS = 2_000       // 工具输出截断到 2K 字符
const PRUNE_PROTECTED_TOOLS = ["skill"]   // skill 工具输出受保护
const DEFAULT_TAIL_TURNS = 2              // 默认保留最后 2 轮对话
const MIN_PRESERVE_RECENT_TOKENS = 2_000  // 最小保留 2K token
const MAX_PRESERVE_RECENT_TOKENS = 8_000  // 最大保留 8K token

预算计算：保留预算 = min(8000, max(2000, 可用窗口 × 0.25))。即模型可用上下文窗口的 25%，范围在 2K-8K token 之间。

2. Turn 分割算法

📄 session/compaction.ts (第 87-128 行)

// 将消息按用户消息分割成 Turn
function turns(messages: SessionV1.WithParts[]) {
  const result: Turn[] = []
  for (let i = 0; i < messages.length; i++) {
    const msg = messages[i]
    if (msg.info.role !== "user") continue
    // 跳过已压缩的消息
    if (msg.parts.some((part) => part.type === "compaction")) continue
    result.push({ start: i, end: messages.length, id: msg.info.id })
  }
  // 计算每个 Turn 的结束边界
  for (let i = 0; i < result.length - 1; i++) {
    result[i].end = result[i + 1].start
  }
  return result
}

// 在预算约束下找到分割点
function splitTurn(input) {
  return Effect.gen(function* () {
    if (input.budget <= 0) return undefined
    if (input.turn.end - input.turn.start <= 1) return undefined
    // 线性扫描找到预算内的分割点
    for (let start = input.turn.start + 1; start < input.turn.end; start++) {
      const size = yield* input.estimate({
        messages: input.messages.slice(start, input.turn.end),
        model: input.model,
      })
      if (size > input.budget) continue
      return { start, id: input.messages[start]!.info.id }
    }
    return undefined
  })
}

算法流程：turns() 将消息按用户消息分割成 Turn 数组，每个 Turn 包含一个用户消息和对应的助手回复。splitTurn() 在预算约束下线性扫描找到最佳分割点——从 Turn 末尾往前找，找到第一个 token 数在预算内的分割位置。

3. Prune 工具输出清理

📄 session/compaction.ts (第 243-287 行)

const prune = Effect.fn("SessionCompaction.prune")(
  function* (input: { sessionID: SessionID }) {
    const cfg = yield* config.get()
    if (!cfg.compaction?.prune) return
    yield* Effect.logInfo("pruning")

    const msgs = yield* session.messages({ sessionID: input.sessionID })
    let total = 0, pruned = 0, turns = 0
    const toPrune: SessionV1.ToolPart[] = []

    // 从后往前遍历
    loop: for (let msgIndex = msgs.length - 1; msgIndex >= 0; msgIndex--) {
      const msg = msgs[msgIndex]
      if (msg.info.role === "user") turns++
      if (turns < 2) continue               // 保护最近 2 轮
      if (msg.info.role === "assistant" && msg.info.summary) break loop  // 遇到摘要停止
      for (let partIndex = msg.parts.length - 1; partIndex >= 0; partIndex--) {
        const part = msg.parts[partIndex]
        if (part.type !== "tool") continue
        if (part.state.status !== "completed") continue
        if (PRUNE_PROTECTED_TOOLS.includes(part.tool)) continue  // 保护 skill
        if (part.state.time.compacted) break loop                // 已压缩过
        const estimate = Token.estimate(part.state.output)
        total += estimate
        if (total <= PRUNE_PROTECT) continue                     // 保留 40K
        pruned += estimate
        toPrune.push(part)
      }
    }

    // 只有当清理量 > 20K token 时才执行
    if (pruned > PRUNE_MINIMUM) {
      for (const part of toPrune) {
        part.state.time.compacted = Date.now()
        yield* session.updatePart(part)
      }
    }
  })

Prune 策略详解：从后往前遍历消息，保护最近 2 轮对话（turns < 2）、遇到摘要消息停止、保护 skill 工具输出。累积工具输出超过 40K token 后开始标记清理，只有当清理量超过 20K token 时才执行，避免频繁清理。

4. 压缩流程：Summarize + Auto-continue

📄 session/compaction.ts (第 343-509 行)

// 插件钩子：允许注入上下文或替换压缩 prompt
const compacting = yield* plugin.trigger(
  "experimental.session.compacting",
  { sessionID: input.sessionID },
  { context: [], prompt: undefined },
)
const nextPrompt = compacting.prompt ?? buildPrompt({
  previousSummary, context: compacting.context
})

// 创建 compaction 模式的 Assistant 消息
const msg: SessionV1.Assistant = {
  mode: "compaction", agent: "compaction",
  summary: true,  // 标记为摘要消息
  // ...
}
yield* session.updateMessage(msg)

// 通过 Processor 调用 LLM 生成摘要
const processor = yield* processors.create({
  assistantMessage: msg, sessionID: input.sessionID, model,
})
const result = yield* processor.process({
  user: userMessage, agent, sessionID: input.sessionID,
  tools: {}, system: [],  // 压缩时不传工具
  messages: [
    ...modelMessages,  // 压缩后的历史消息
    { role: "user", content: [{ type: "text", text: nextPrompt }] },
  ],
  model,
})

// 压缩完成后自动继续
if (result === "continue" && input.auto) {
  if (replay) {
    // Replay：重新发送被压缩的用户消息
    const replayMsg = yield* session.updateMessage({
      role: "user", sessionID: input.sessionID,
    })
    for (const part of replay.parts) {
      if (part.type === "compaction") continue
      yield* session.updatePart({ ...replayPart,
        messageID: replayMsg.id,
      })
    }
  } else {
    // Auto-continue：发送继续指令
    const text = "Continue if you have next steps, or stop and ask for clarification if you are unsure how to proceed."
    yield* session.updatePart({
      type: "text", synthetic: true,
      metadata: { compaction_continue: true },
      text,
    })
  }
}

压缩流程详解：1) 插件钩子允许自定义压缩 prompt；2) 创建 compaction 模式的 Assistant 消息；3) 通过 Processor 调用 LLM 生成摘要（不传工具）；4) 压缩完成后 replay 被压缩的用户消息或发送 auto-continue 指令。

五、Token 计费与成本计算

session.ts 中的 getUsage 函数实现了复杂的 Token 计费和成本计算，支持多层 Provider 的缓存 token 统计。

计费逻辑

📄 session/session.ts (第 338-406 行)

export function getUsage(input: {
  model: Provider.Model; usage: Usage;
  metadata?: ProviderMetadata
}) {
  // 从多个 Provider 元数据中获取缓存 token
  const cacheWriteInputTokens = safe(
    input.usage.cacheWriteInputTokens ??
    input.metadata?.["anthropic"]?.["cacheCreationInputTokens"] ??
    input.metadata?.["vertex"]?.["cacheCreationInputTokens"] ??
    input.metadata?.["bedrock"]?.["usage"]?.["cacheWriteInputTokens"] ??
    input.metadata?.["venice"]?.["usage"]?.["cacheCreationInputTokens"] ?? 0
  )
  // AI SDK v6 将缓存 token 计入了 inputTokens，需要减去
  const adjustedInputTokens = safe(
    inputTokens - cacheReadInputTokens - cacheWriteInputTokens)

  // 成本计算：使用 Decimal 避免浮点精度问题
  return {
    cost: new Decimal(0)
      .add(new Decimal(tokens.input)
        .mul(costInfo?.input ?? 0).div(1_000_000))
      .add(new Decimal(tokens.output)
        .mul(costInfo?.output ?? 0).div(1_000_000))
      .add(new Decimal(tokens.cache.read)
        .mul(costInfo?.cache?.read ?? 0).div(1_000_000))
      .add(new Decimal(tokens.cache.write)
        .mul(costInfo?.cache?.write ?? 0).div(1_000_000))
      .add(new Decimal(tokens.reasoning)
        .mul(costInfo?.output ?? 0).div(1_000_000))
      .toNumber(),
    tokens: { input, output, reasoning,
      cache: { read, write } },
  }
}

计费设计：支持 Anthropic、Vertex、Bedrock、Venice 等多个 Provider 的缓存 token 统计。使用 Decimal.js 避免浮点精度问题。Reasoning token 按 output 价格计费。成本按每百万 token 计算。

六、总结

🔹 ProcessorContext 维护 7 个状态字段：toolcalls/shouldBreak/snapshot/blocked/needsCompaction/currentText/reasoningMap

🔹 handleEvent 处理 12 种事件：reasoning-start/delta/end、tool-input-start/delta/end、tool-call、tool-result、tool-error、step-start/finish、text-start/delta

🔹 DOOM_LOOP 检测：连续 3 次相同工具+相同参数 → 请求用户授权

🔹 Compaction 三级策略：Prune（40K/20K 阈值）→ Select（Turn 分割）→ Summarize（LLM 摘要）

🔹 Auto-continue：压缩完成后自动 replay 用户消息或发送继续指令

🔹 Token 计费：Decimal.js 精度 + 多 Provider 缓存 token 支持

🔹 21 个文件、7,129 行代码构成 OpenCode 最复杂的业务模块

← 系列导航 →

← 第 17 讲：Event 事件总线 | 第 19 讲：Agent 核心系统 →

关注公众号获取更多 OpenCode 源码解析干货

源码：https://github.com/opencode-ai/opencode