OpenCode Agent 核心实现原理#

// Agent 核心信息结构
Agent.Info = {
  name: string,                    // Agent 唯一标识名
  description?: string,            // Agent 描述（供用户/LLM 选择）
  mode: "subagent" | "primary" | "all",  // 运行模式
  native?: boolean,               // 是否内置 Agent
  hidden?: boolean,                // 是否对用户隐藏
  topP?: number,                  // 采样参数
  temperature?: number,            // 生成温度
  color?: string,                 // UI 显示颜色
  permission: PermissionNext.Ruleset,  // 权限规则集
  model?: {                        // 可选：指定模型
    modelID: string,
    providerID: string,
  },
  variant?: string,               // 模型变体
  prompt?: string,                 // 自定义系统提示
  welcome_message?: string,        // 欢迎消息
  options: Record<string, any>,   // 扩展选项
  steps?: number,                  // 最大迭代步数
}

// packages/opencode/src/agent/agent.ts 第 74-200 行

// 1. build - 默认主 Agent，拥有所有工具权限
build: {
  name: "build",
  description: "The default agent. Executes tools based on configured permissions.",
  mode: "primary",
  native: true,
  permission: PermissionNext.merge(
    defaults,
    { question: "allow", plan_enter: "allow" },
    user
  )
}

// 2. plan - 计划模式，拒绝所有编辑工具
plan: {
  name: "plan",
  description: "Plan mode. Disallows all edit tools.",
  mode: "primary",
  native: true,
  permission: PermissionNext.merge(
    defaults,
    {
      question: "allow",
      plan_exit: "allow",
      edit: { "*": "deny", "*.md": "allow" }  // 只允许编辑 markdown
    }
  )
}

// 3. explore - 快速代码探索子 Agent（只读）
explore: {
  name: "explore",
  mode: "subagent",
  native: true,
  permission: PermissionNext.merge(
    defaults,
    {
      "*": "deny",
      grep: "allow", glob: "allow", list: "allow",
      bash: "allow", webfetch: "allow", read: "allow"
    }
  )
}

// 4. general - 通用任务子 Agent
general: {
  name: "general",
  mode: "subagent",
  native: true,
  permission: PermissionNext.merge(
    defaults,
    { todoread: "deny", todowrite: "deny" }
  )
}

// 用户通过 @agentname 显式指定 Agent
const lastUserMsg = msgs.findLast((m) => m.info.role === "user")
const bypassAgentCheck = lastUserMsg?.parts.some((p) => p.type === "agent") ?? false

// TaskTool.execute() 核心逻辑
const result = await SessionPrompt.prompt({
  messageID,
  sessionID: session.id,     // 新建子会话
  model: { ... },
  agent: agent.name,          // 指定子 Agent
  tools: {                    // 禁用递归调用
    todowrite: false,
    todoread: false,
    task: false,              // 防止子 Agent 继续调用子 Agent
  },
  parts: promptParts,
})

export const loop = fn(Identifier.schema("session"), async (sessionID) => {
  const abort = start(sessionID)
  let step = 0

  while (true) {
    step++

    // 1. 加载消息历史
    let msgs = await MessageV2.filterCompacted(MessageV2.stream(sessionID))

    // 2. 提取最后一条用户消息和助手消息
    let lastUser, lastAssistant, lastFinished
    let tasks: (CompactionPart | SubtaskPart)[] = []
    for (const msg of msgs) {
      if (!lastUser && msg.info.role === "user") lastUser = msg.info
      if (!lastAssistant && msg.info.role === "assistant") lastAssistant = msg.info
      if (!lastFinished && msg.info.finish) lastFinished = msg.info
      const task = msg.parts.filter(p => p.type === "compaction" || p.type === "subtask")
      if (task.length) tasks.push(...task)
    }

    // 3. 检查是否退出循环
    if (lastFinished && !["tool-calls", "unknown"].includes(lastFinished.finish)) {
      break  // 正常完成
    }

    // 4. 处理 pending subtask（子 Agent 调用）
    const task = tasks.pop()
    if (task?.type === "subtask") {
      await executeSubtask(task, lastUser)
      continue
    }

    // 5. 处理 pending compaction（上下文压缩）
    if (task?.type === "compaction") {
      await SessionCompaction.process({ messages: msgs, ... })
      continue
    }

    // 6. 检查是否需要自动压缩
    if (await SessionCompaction.isOverflow({ tokens: lastFinished.tokens, model })) {
      await SessionCompaction.create({ auto: true })
      continue
    }

    // 7. 正常处理流程
    const agent = await Agent.get(lastUser.agent)
    const processor = SessionProcessor.create({ ... })
    const tools = await resolveTools({ agent, ... })

    const result = await processor.process({
      user: lastUser,
      agent,
      system: [...],
      messages: [...],
      tools,
      model,
    })

    if (result === "stop") break
    if (result === "compact") { /* 触发压缩 */ }
  }
})

const parameters = z.object({
  description: z.string(),     // 任务简短描述
  prompt: z.string(),         // 任务详细提示
  subagent_type: z.string(),  // 子 Agent 类型名
  session_id: z.string().optional(),  // 继续已有会话
  command: z.string().optional(),
})

export const TaskTool = Tool.define("task", async (ctx) => {
  // 1. 获取可用的子 Agent 列表
  const agents = await Agent.list().then(x => x.filter(a => a.mode !== "primary"))

  // 2. 过滤基于权限可访问的 Agent
  const caller = ctx?.agent
  const accessibleAgents = caller
    ? agents.filter(a => PermissionNext.evaluate("task", a.name, caller.permission).action !== "deny")
    : agents

  return {
    description: description.replace("{agents}", accessibleAgents.map(...).join("\n")),
    parameters,

    async execute(params, ctx) {
      // 3. 权限检查（除非用户显式调用）
      if (!ctx.extra?.bypassAgentCheck) {
        await ctx.ask({
          permission: "task",
          patterns: [params.subagent_type],
          always: ["*"],
        })
      }

      // 4. 获取目标 Agent
      const agent = await Agent.get(params.subagent_type)

      // 5. 创建子会话
      const session = await Session.create({
        parentID: ctx.sessionID,  // 关联父会话
        title: params.description + ` (@${agent.name} subagent)`,
        permission: [/* 限制权限 */],
      })

      // 6. 调用子 Agent
      const result = await SessionPrompt.prompt({
        messageID,
        sessionID: session.id,
        model: { modelID: agent.model?.modelID, providerID: agent.model?.providerID },
        agent: agent.name,
        tools: { todowrite: false, todoread: false, task: false },
        parts: promptParts,
      })

      // 7. 返回结果
      return {
        title: params.description,
        output: text + "\n\n<task_metadata>...",
        metadata: { sessionId: session.id, summary, model },
      }
    }
  }
})

// packages/opencode/src/agent/agent.ts 第 54-71 行

const defaults = PermissionNext.fromConfig({
  "*": "allow",                 // 默认允许所有
  doom_loop: "ask",             // 死循环检测需要询问
  external_directory: {        // 外部目录访问
    "*": "ask",
    [Truncate.GLOB]: "allow",  // glob 模式允许
  },
  question: "deny",            // 禁止提问
  plan_enter: "deny",          // 禁止进入计划模式
  plan_exit: "deny",           // 禁止退出计划模式
  read: {                      // 读文件权限
    "*": "allow",
    "*.env": "ask",            // .env 文件需要确认
    "*.env.*": "ask",
    "*.env.example": "allow",
  },
})

// packages/opencode/src/permission/next.ts 第 231-238 行

export function evaluate(
  permission: string,
  pattern: string,
  ...rulesets: Ruleset[]
): Rule {
  const merged = merge(...rulesets)

  // 从后往前查找最后一个匹配的规则
  const match = merged.findLast(
    rule => Wildcard.match(permission, rule.permission) &&
            Wildcard.match(pattern, rule.pattern)
  )

  // 没有匹配的规则时，默认询问
  return match ?? { action: "ask", permission, pattern: "*" }
}

// Agent 配置在 opencode.json 中的格式
Agent = {
  model?: string,              // "provider/model"
  variant?: string,            // 模型变体
  temperature?: number,        // 生成温度
  top_p?: number,              // top-p 采样
  prompt?: string,             // 自定义系统提示
  disable?: boolean,          // 禁用内置 Agent
  description?: string,        // Agent 描述
  mode?: "subagent" | "primary" | "all",  // 运行模式
  hidden?: boolean,            // 对用户隐藏
  options?: Record<string, any>,  // 扩展选项
  color?: string,               // "#FF5733" 颜色
  steps?: number,               // 最大迭代步数
  permission?: Permission      // 权限规则
}

// Permission 格式
Permission = {
  "tool-name": "allow" | "ask" | "deny",
  "tool-name": { "pattern": "allow" | "ask" | "deny" },
}

// minimal_agent.ts - 最小化 Agent 实现

import { z } from "zod"

// ============ 1. Agent 定义 ============

interface AgentInfo {
  name: string
  mode: "primary" | "subagent"
  permission: Rule[]
}

interface Rule {
  permission: string
  action: "allow" | "deny" | "ask"
}

// 内置 Agent
const agents: Record<string, AgentInfo> = {
  build: {
    name: "build",
    mode: "primary",
    permission: [{ permission: "*", action: "allow" }]
  },
  explore: {
    name: "explore",
    mode: "subagent",
    permission: [
      { permission: "read", action: "allow" },
      { permission: "grep", action: "allow" },
      { permission: "task", action: "deny" }  // 子 Agent 不能调用其他子 Agent
    ]
  },
  general: {
    name: "general",
    mode: "subagent",
    permission: [{ permission: "*", action: "allow" }]
  }
}

// 获取 Agent
function getAgent(name: string): AgentInfo | undefined {
  return agents[name]
}

// 获取默认主 Agent
function defaultAgent(): AgentInfo {
  return agents["build"]
}

// ============ 2. 工具定义 ============

type ToolExecute = (args: any, ctx: ToolContext) => Promise<ToolResult>

interface ToolContext {
  agent: string
  sessionID: string
  abort: AbortSignal
}

interface ToolResult {
  output: string
  metadata?: any
}

interface Tool {
  id: string
  description: string
  parameters: z.ZodType
  execute: ToolExecute
}

// Task 工具 - 用于调用子 Agent
const taskTool: Tool = {
  id: "task",
  description: "调用子 Agent 执行任务",

  parameters: z.object({
    description: z.string(),
    prompt: z.string(),
    subagent_type: z.string()
  }),

  async execute(args, ctx) {
    const { description, prompt, subagent_type } = args

    // 1. 获取目标 Agent
    const agent = getAgent(subagent_type)
    if (!agent) throw new Error(`Unknown agent: ${subagent_type}`)

    // 2. 权限检查
    const canCall = evaluatePermission("task", agent.name, ctx.agent)
    if (canCall === "deny") throw new Error("Permission denied")

    if (canCall === "ask") {
      // 实际场景中这里会弹出用户确认
      console.log(`[权限] 询问是否允许调用 ${subagent_type}`)
    }

    // 3. 创建子会话并执行
    const subSession = {
      id: `sess_${Date.now()}`,
      parentID: ctx.sessionID,
      agent: agent.name
    }

    console.log(`[子Agent] 创建会话 ${subSession.id}, agent: ${agent.name}`)

    // 4. 递归调用主循环（子 Agent 模式）
    const result = await mainLoop(prompt, agent, subSession)

    return {
      output: result + `\n\n<subagent_result session="${subSession.id}">`,
      metadata: { sessionId: subSession.id }
    }
  }
}

// 其他工具
const tools: Record<string, Tool> = {
  task: taskTool,
  read: { id: "read", description: "读取文件", parameters: z.any(), async execute() {} },
  bash: { id: "bash", description: "执行命令", parameters: z.any(), async execute() {} },
}

// ============ 3. 权限检查 ============

function evaluatePermission(permission: string, pattern: string, agentName: string): "allow" | "ask" | "deny" {
  const agent = getAgent(agentName)
  if (!agent) return "ask"

  // 从后往前找匹配的规则
  for (let i = agent.permission.length - 1; i >= 0; i--) {
    const rule = agent.permission[i]
    if (matches(rule.permission, permission) && matches(rule.pattern, pattern)) {
      return rule.action
    }
  }

  return "ask"  // 默认询问
}

function matches(pattern: string, value: string): boolean {
  if (pattern === "*") return true
  return pattern === value
}

// ============ 4. 主循环 ============

async function mainLoop(
  userInput: string,
  agent: AgentInfo,
  session: { id: string; parentID?: string }
): Promise<string> {
  console.log(`[主循环] session=${session.id}, agent=${agent.name}`)
  console.log(`[用户] ${userInput}`)

  // 模拟 LLM 决定调用工具
  // 实际场景中这里调用 LLM API

  // 检查是否需要调用子 Agent
  if (userInput.includes("@explore") || userInput.includes("@general")) {
    const match = userInput.match(/@(explore|general)\s+(.*)/)
    if (match) {
      const [, agentName, prompt] = match

      const result = await taskTool.execute({
        description: "子任务",
        prompt,
        subagent_type: agentName
      }, {
        agent: agent.name,
        sessionID: session.id,
        abort: new AbortController().signal
      })

      return `[父Agent] 子Agent ${agentName} 结果: ${result.output}`
    }
  }

  // 正常处理
  return `[${agent.name}] 处理: ${userInput}`
}

// ============ 5. 入口点 ============

async function main() {
  // 场景 1: 主 Agent 单独处理
  console.log("\n=== 场景 1: 主 Agent 单独处理 ===")
  const result1 = await mainLoop(
    "帮我读取文件 src/index.ts",
    defaultAgent(),
    { id: "main_1" }
  )
  console.log(`[结果] ${result1}\n`)

  // 场景 2: 主 Agent 调用子 Agent
  console.log("=== 场景 2: 主 Agent 调用子 Agent ===")
  const result2 = await mainLoop(
    "@explore 查找所有包含 'Agent' 的文件",
    defaultAgent(),
    { id: "main_2" }
  )
  console.log(`[结果] ${result2}\n`)

  // 场景 3: 嵌套调用（子 Agent 再调用其他）
  console.log("=== 场景 3: 子 Agent 嵌套调用测试 ===")
  try {
    await mainLoop(
      "调用 task tool",
      getAgent("explore")!,
      { id: "sub_1", parentID: "main_2" }
    )
  } catch (e) {
    console.log(`[预期错误] ${e.message}`)
  }
}

main()

// TaskTool.execute() 中
tools: {
  todowrite: false,
  todoread: false,
  task: false,  // ◀ 禁用 task 工具，防止递归
}

// 创建子会话时
const session = await Session.create({
  parentID: ctx.sessionID,  // ◀ 关联父会话
  ...
})

// MessageV2 核心类型
namespace MessageV2 {
  interface User {
    id: string
    role: "user"
    agent: string      // 当前使用的 Agent
    model: Model
    parts: Part[]
  }

  interface Assistant {
    id: string
    role: "assistant"
    agent: string
    mode: string       // Agent 名称
    finish?: string    // 完成状态
    parts: Part[]
  }

  type Part = TextPart | ToolPart | SubtaskPart | AgentPart | ...

  interface SubtaskPart {
    type: "subtask"
    agent: string       // 子 Agent 名称
    prompt: string
    description: string
    command?: string
  }

  interface AgentPart {
    type: "agent"
    name: string        // @mentioned agent
  }
}

// Session 核心类型
namespace Session {
  interface Info {
    id: string
    parentID?: string    // 父会话 ID（子会话才有）
    agent?: string
    permission: PermissionNext.Ruleset
    // ...
  }
}