DeepSeek Harness 系列（04）：Agent Loop——一次对话是怎么跑起来的

Turn 开始
  ↓
Step 1：模型回复 + 调用 search_files
  ↓
Step 2：模型分析搜索结果 + 调用 read_file × 5
  ↓
Step 3：模型写测试 + 调用 write_file × 3
  ↓
模型回复："已完成，共补全 12 个函数的测试。"
  ↓
Turn 结束

// packages/core/agent-loop/src/agent.ts
 
/** 主循环：持续执行 Turn，直到没有更多待处理工作 */
private async kick(): Promise<void> {
  try {
    // turn() 返回 true 表示还有待处理工作，继续下一个 Turn
    while (await this.turn()) {}
  } catch (_error) {
    // 错误在这里被隔离，不会崩溃外层
  } finally {
    this.setPhase({ kind: 'idle', lastTurn: turn })
    // 如果有被 latch 的唤醒请求，继续处理
    if (wakeRequested && this.inbox.hasPending) this.wakeDriver()
  }
}

// 三种输入方式，对应三种调度语义
 
// 1. followup：普通用户消息，进入下一个 Turn
agent.followup(message)   // → next-turn 队列，唤醒驱动器
 
// 2. steer：在最近的 Step 边界注入，进入下一个 Step
agent.steer(message)      // → next-step 队列，唤醒驱动器
 
// 3. inject：注入上下文，不唤醒驱动器
agent.inject(message)     // → next-step 队列，不唤醒

turn/start（记入 Session 日志）
  │
  ├── 领取消息（从 inbox 取出）
  ├── 组装 System Prompt（调用 ctx.systemPrompt.assemble）
  │
  ▼
[agent/pre-step waterfall]  ← 扩展点 ①
  │  返回 reject 或 enter(messages)
  │  reject → Turn 空结束（不留 Step 记录）
  │
  ├── step/start（记入 Session 日志）
  │
  ▼
[agent/request waterfall]  ← 扩展点 ②
  │  替换 LLM 调用配置（模型、maxTokens 等）
  │
  ├── 提交 system/message 和 user/message
  ├── 从 Session 日志派生并冻结请求
  │
  ▼
[llm/stream waterfall]  → 模型流式输出
  │  → agent/assistant-stream（实时推送 UI）
  │  → 记录 assistant/message 或 assistant/attempt
  │
  ▼
工具调用（并行池 + 独占屏障）
  │  → tool/call 记入 Session
  │  → tools/pre-execute → tools/execute → tools/post-execute
  │  → tool/result 记入 Session
  │
  ├── step/end（记入 Session 日志）
  │
  ├── 还有工具结果要处理，或 next-step 有新消息？
  │   是 → 继续下一个 Step
  │   否 → 检查是否结束 Turn
  │
  ▼
[agent/turn-stopping serial]  ← 扩展点 ③
  │  监听器可以调用 agent.steer() 让 Turn 继续
  │  没有新工作 → 关闭 Turn
  │
turn/end（记入 Session 日志）

type PreStepDecision =
  | { kind: 'reject' }              // 拒绝这个 Step，Turn 空结束
  | {
    kind: 'enter'
    messages: UserMessage[]         // 实际进入模型的消息（可以改写）
    startsRequestSeries?: true      // 是否开始新的请求序列
  }

// 示例：拦截包含敏感路径的指令
ctx.on('agent/pre-step', async (payload, next) => {
  const { messages } = payload
  
  for (const msg of messages) {
    if (messageContainsSensitivePath(msg)) {
      // 拒绝这个 Step
      return { kind: 'reject' }
    }
  }
  
  // 没问题，继续
  return next()
})

ctx.on('agent/request', async (payload, next) => {
  const config = await next()       // 拿到默认配置
  
  // 根据当前任务调整参数
  return {
    ...config,
    maxTokens: 8192,               // 给这个会话更大的输出空间
    reasoningEffort: 'medium',     // 降低推理强度节省 token
  }
})

ctx.on('agent/turn-stopping', async (payload) => {
  const { agent, turn, signal } = payload
  
  // 检查是否还有待确认的任务
  const pendingTask = await db.getPendingTask(agent.session.id)
  if (pendingTask) {
    // 注入新工作，Turn 会继续
    agent.steer({
      type: 'user',
      content: [{ type: 'text', text: `还有一项任务：${pendingTask.description}` }],
    })
  }
  // 如果什么都不做，Turn 自然关闭
})

// 某个工具认为任务已完成，主动结束 Turn
async execute(args, exec) {
  const result = await doSomething()
  exec.concludeTurn()    // 告诉 Loop：这个 Turn 到此结束
  return result
}

export const name = 'agent-tracer'
export const inject = ['tools']
 
export function apply(ctx: Context): void {
  // 监听 Turn 状态变化
  ctx.on('agent/status', (payload) => {
    console.log(`[Agent] status → ${payload.status}`)
  })
  
  // 监听每个 Step 开始
  ctx.on('agent/pre-step', async (payload, next) => {
    const { turn, step, messages } = payload
    console.log(`[Step] turn=${turn} step=${step} messages=${messages.length}`)
    return next()
  })
  
  // 监听工具调用结果
  ctx.on('tools/result', (exec, result) => {
    const icon = result.isError ? '✗' : '✓'
    console.log(`[Tool] ${icon} ${exec.name}`)
  })
  
  // 监听 Turn 结束
  ctx.on('agent/turn-stopping', (payload) => {
    console.log(`[Turn] ${payload.turn} stopping`)
  })
}