OpenCode 实现原理 - 上下文工程

[Pines-Cheng]

上下文隔离机制通过多个层面的技术实现，确保每个 agent 执行环境完全独立：

1. 会话层隔离

独立会话空间

const session = await Session.create(ctx.sessionID, params.description + ` (@${agent.name} subagent)`)

每个 agent 都在独立的会话空间中运行，拥有自己的：

• 会话 ID (session.id)
• 消息历史
• 执行上下文

消息 ID 隔离

const messageID = Identifier.ascending("message")

使用递增的唯一标识符确保消息不会混淆。

2. 事件系统隔离

基于会话 ID 的事件过滤

const unsub = Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
  if (evt.properties.part.sessionID !== session.id) return  // 关键隔离点
  if (evt.properties.part.messageID === messageID) return
  // 只处理属于当前会话的事件
})

消息部分隔离

const parts: Record<string, MessageV2.ToolPart> = {}
// 每个会话维护独立的消息部分集合

3. 工具权限隔离

工具访问控制

tools: {
  todowrite: false,
  todoread: false,
  task: false,  // 防止递归和交叉调用
  ...agent.tools,  // 每个 agent 独立的工具集
}

权限边界

permission: {
  edit: merged.edit ?? "allow",
  webfetch: merged.webfetch ?? "allow",
  bash: mergedBash ?? { "*": "allow" },
}

每个 agent 有独立的权限边界，不能访问超出权限范围的资源。

4. 模型配置隔离

独立模型实例

const model = agent.model ?? {
  modelID: msg.info.modelID,
  providerID: msg.info.providerID,
}

每个 agent 可以使用不同的模型配置，包括：

• 不同的 AI 模型
• 独立的温度参数
• 专门的 top_p 设置

5. 状态管理隔离

无状态函数式设计

export async function get(agent: string) {
  return state().then((x) => x[agent])
}

通过函数式编程避免共享状态，每次调用都是独立的。

配置驱动隔离

const state = Instance.state(async () => {
  const cfg = await Config.get()
  // 基于配置重新计算，无共享状态
})

6. 生命周期隔离

独立的中止机制

ctx.abort.addEventListener("abort", () => {
  Session.abort(session.id)  // 只中止当前会话
})

资源清理隔离

unsub()  // 每个会话结束时清理自己的订阅

7. 数据流隔离

输入隔离

每个 agent 只接收专门为它准备的输入：

parts: [
  {
    id: Identifier.ascending("part"),
    type: "text", 
    text: params.prompt,  // 独立的任务描述
  },
]

输出隔离

return {
  title: params.description,
  metadata: {
    summary: result.parts.filter((x: any) => x.type === "tool"),
  },
  output: (result.parts.findLast((x: any) => x.type === "text") as any)?.text ?? "",
}

每个 agent 的输出完全独立，不会影响其他 agent。

这种多层次的隔离机制确保了系统的高可靠性和安全性，即使在高并发场景下也能保证各个 agent 之间完全独立运行。

[Pines-Cheng]

subagent 采用的正是单向最终报告机制。从代码和文档中可以清楚地看到这种设计：

1. 单次交互模式

一次性执行

const result = await Session.prompt({
  messageID,
  sessionID: session.id,
  // ... 配置
  parts: [
    {
      id: Identifier.ascending("part"),
      type: "text",
      text: params.prompt,  // 完整任务描述一次性传入
    },
  ],
})

subagent 接收完整的任务描述，执行后直接返回最终结果，没有中间交互。

2. 最终报告返回

结构化输出

return {
  title: params.description,
  metadata: {
    summary: result.parts.filter((x: any) => x.type === "tool"),
  },
  output: (result.parts.findLast((x: any) => x.type === "text") as any)?.text ?? "",
}

返回包含标题、元数据和输出的完整报告，这是 subagent 与外界的唯一通信结果。

3. 文档明确说明

在 task.txt 中明确描述了这种机制：

"Each agent invocation is stateless. You will not be able to send additional messages to the agent, nor will the agent be able to communicate with you outside of its final report."

"When the agent is done, it will return a single message back to you. The result returned by the agent is not visible to the user."

4. 设计约束

防止双向通信

tools: {
  todowrite: false,
  todoread: false,
  task: false,  // 防止递归调用
  ...agent.tools,
}

通过禁用某些工具（如 task 工具）防止 subagent 创建新的交互循环。

完整任务描述要求

文档强调：

"your prompt should contain a highly detailed task description for the agent to perform autonomously and you should specify exactly what information the agent should return back to you in its final and only message to you."

5. 实时状态 vs 最终报告

虽然有实时状态更新：

const unsub = Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
  // 实时状态更新
})

但这只是进度反馈，不是双向通信。最终的业务结果仍然是单向的最终报告。

6. 机制优势

• 简化复杂性: 避免了复杂的对话状态管理
• 提高可靠性: 减少了通信失败的可能性
• 便于并发: 单向机制天然支持并发执行
• 明确边界: 清晰的输入输出边界，便于测试和调试

这种单向最终报告机制确保了系统的简洁性和可预测性，同时保持了 subagent 的自主性和无状态特性。

[Pines-Cheng]

从当前的代码分析来看，系统没有明确实现 Token 预算管理机制。让我检查相关的配置和模型管理部分来确认：

从代码分析来看，系统没有实现明确的 Token 预算管理机制。以下是相关的发现：

1. 模型配置中的限制信息

在 Provider 代码中可以看到模型定义包含了限制信息：

limit: model.limit ??
  existing?.limit ?? {
    context: 0,
    output: 0,
  },

但这些限制信息似乎只是模型的元数据描述，而不是实际的预算控制机制。

2. 成本信息但无预算控制

模型配置中包含成本信息：

cost: {
  input: 0,
  output: 0,
  cache_read: 0,
  cache_write: 0,
  ...existing?.cost,
  ...model.cost,
},

但这只是成本计算的基础数据，没有看到实际的预算限制或消费跟踪机制。

3. Agent 配置中缺少预算参数

在 Agent 的配置中，只有模型参数控制：

temperature: z.number().optional(),
topP: z.number().optional(),

没有 token 预算、最大消费限制等相关配置。

4. Session 执行中无预算检查

在 Task Tool 的执行过程中：

const result = await Session.prompt({
  messageID,
  sessionID: session.id,
  model: {
    modelID: model.modelID,
    providerID: model.providerID,
  },
  // 没有预算相关参数
})

没有看到任何预算检查或限制机制。

结论

当前系统没有实现 Token 预算管理，这意味着：

• 无消费限制: subagent 可以无限制地使用 token
• 无成本控制: 没有单次任务或总体的成本上限
• 无预算分配: 不同 agent 之间没有预算隔离
• 无消费监控: 缺少实时的 token 使用量跟踪

这可能是系统的一个潜在风险点，特别是在生产环境中运行大量并发 subagent 时，可能导致意外的高额 API 调用费用。

[Pines-Cheng]

基于对代码的深入分析，系统实现了一套完整的自动上下文压缩机制：

1. 触发条件

Token 阈值检测

// auto summarize if too long
if (previous && previous.tokens) {
  const tokens =
    previous.tokens.input + previous.tokens.cache.read + previous.tokens.cache.write + previous.tokens.output
  if (model.info.limit.context && tokens > Math.max((model.info.limit.context - outputLimit) * 0.9, 0)) {
    state().autoCompacting.set(input.sessionID, true)
    
    await summarize({
      sessionID: input.sessionID,
      providerID: model.providerID,
      modelID: model.info.id,
    })
    return prompt(input)
  }
}

当累计 token 数超过模型上下文限制的 90% 时，自动触发压缩。

2. 压缩实现机制

智能消息过滤

const lastSummary = msgs.findLast((msg) => msg.info.role === "assistant" && msg.info.summary === true)
const filtered = msgs.filter((msg) => !lastSummary || msg.info.id >= lastSummary.info.id)

只对上次摘要之后的消息进行压缩，避免重复压缩。

专用摘要提示词

const system = [
  ...SystemPrompt.summarize(model.providerID),
  ...(await SystemPrompt.environment()),
  ...(await SystemPrompt.custom()),
]

使用专门的摘要系统提示词来指导压缩过程。

结构化摘要请求

{
  role: "user",
  content: [
    {
      type: "text",
      text: "Provide a detailed but concise summary of our conversation above. Focus on information that would be helpful for continuing the conversation, including what we did, what we're doing, which files we're working on, and what we're going to do next.",
    },
  ],
}

明确要求 AI 生成包含关键信息的结构化摘要。

3. 压缩后处理

摘要消息标记

const next: MessageV2.Info = {
  // ...
  summary: true,  // 标记为摘要消息
  // ...
}

递归调用

await summarize(...)
return prompt(input)  // 递归调用，使用压缩后的上下文

压缩完成后重新执行原始请求，确保使用压缩后的上下文。

上下文过滤

const lastSummary = msgs.findLast((msg) => msg.info.role === "assistant" && msg.info.summary === true)
if (lastSummary) msgs = msgs.filter((msg) => msg.info.id >= lastSummary.info.id)

在后续对话中，只保留最新摘要之后的消息。

4. 状态管理

自动压缩状态跟踪

state().autoCompacting.set(input.sessionID, true)
// ...
const isAutoCompacting = state().autoCompacting.get(sessionID) ?? false
if (isAutoCompacting) {
  state().autoCompacting.delete(sessionID)
  return
}

跟踪会话的自动压缩状态，避免重复操作。

5. 系统提示词优化

缓存优化

// max 2 system prompt messages for caching purposes
const [first, ...rest] = system
system = [first, rest.join("\n")]

将系统提示词合并为最多 2 条消息，优化缓存效果。

这种上下文压缩机制的优势：

• 自动化: 无需手动干预，系统自动检测并执行压缩
• 智能化: 保留关键信息，丢弃冗余内容
• 递增式: 只压缩新增内容，避免重复处理
• 透明性: 对用户完全透明，不影响对话体验
• 效率优化: 通过缓存和状态管理提高性能