AI-NEWS

Appearance

Agent Loop ​

🎯 学习目标

  • 实现可序列化的 Agent State 结构
  • 拆分 Planner / Tool Executor / Observer 职责
  • 配置最大迭代、Token 预算与死循环检测
  • 支持 Checkpoint 恢复与 trace 输出

引言 ​

ReAct 伪代码能跑 Demo；生产 Agent 需要显式状态、可测试模块、可中断恢复。本节给出一套 TypeScript 风格的工程骨架，你可以映射到 LangGraph 或自研运行时。

章节正文 ​

第 1 步：Agent State：可序列化是底线 ​

typescript
type AgentState = {
  sessionId: string
  messages: ChatMessage[]
  plan?: string
  toolResults: Array<{ callId: string; name: string; output: unknown }>
  iteration: number
  tokenUsed: number
  status: 'running' | 'done' | 'aborted'
  abortReason?: string
}

function createInitialState(sessionId: string, userQuery: string): AgentState {
  return {
    sessionId,
    messages: [
      { role: 'system', content: SYSTEM_PROMPT },
      { role: 'user', content: userQuery },
    ],
    toolResults: [],
    iteration: 0,
    tokenUsed: 0,
    status: 'running',
  }
}

State 应可 JSON 化存 Redis/DB，支持 crash 后 resume（Checkpoint）。

第 2 步：主 Loop：Planner → Executor → Observer ​

typescript
const MAX_ITERATIONS = 12
const TOKEN_BUDGET = 80_000

async function runAgentLoop(state: AgentState, deps: Deps): Promise<AgentState> {
  const actionSignatures: string[] = []

  while (state.status === 'running') {
    if (state.iteration >= MAX_ITERATIONS) {
      return abort(state, 'MAX_ITERATIONS')
    }
    if (state.tokenUsed >= TOKEN_BUDGET) {
      state.messages.push({
        role: 'system',
        content: 'Token 预算将尽，请基于已有信息给出结论，勿再调用工具。',
      })
    }

    // Planner：一次 LLM 调用，可能产出 tool_calls 或最终答案
    const plannerOut = await deps.planner(state)
    state.tokenUsed += plannerOut.usage.totalTokens
    state.messages.push(plannerOut.message)
    state.iteration += 1

    if (!plannerOut.message.tool_calls?.length) {
      state.status = 'done'
      return state
    }

    // Executor
    for (const call of plannerOut.message.tool_calls) {
      const sig = `${call.function.name}:${call.function.arguments}`
      if (actionSignatures.includes(sig)) {
        return abort(state, 'REpeated_ACTION')
      }
      actionSignatures.push(sig)

      const output = await deps.toolExecutor.execute(call, { sessionId: state.sessionId })
      state.toolResults.push({ callId: call.id, name: call.function.name, output })
      state.messages.push({
        role: 'tool',
        tool_call_id: call.id,
        content: JSON.stringify(output),
      })
    }

    // Observer：可选，更新 plan、metrics、external trace
    state = deps.observer(state)
  }
  return state
}

function abort(state: AgentState, reason: string): AgentState {
  return { ...state, status: 'aborted', abortReason: reason }
}

第 3 步：Tool Executor：超时、权限、并行 ​

typescript
class ToolExecutor {
  constructor(private registry: ToolRegistry, private harness: Harness) {}

  async execute(call: ToolCall, ctx: { sessionId: string }) {
    await this.harness.beforeToolCall(call, ctx)
    const def = this.registry.get(call.function.name)
    this.harness.assertPermission(def, ctx)

    const args = JSON.parse(call.function.arguments)
    const result = await Promise.race([
      def.handler(args),
      timeout(30_000),
    ]).catch(err => ({ error: true, message: String(err) }))

    await this.harness.afterToolCall(call, result, ctx)
    return result
  }
}

无依赖的工具调用可 Promise.all；有顺序依赖的（先 search 再 read_file）应让 Planner 分步或 Executor 识别 DAG。

第 4 步：Observer 与 Checkpoint ​

typescript
function observer(state: AgentState): AgentState {
  deps.metrics.gauge('agent.iteration', state.iteration)
  deps.trace.emit({ sessionId: state.sessionId, step: state.iteration, toolCount: state.toolResults.length })
  if (state.iteration % 3 === 0) {
    deps.checkpoint.save(state.sessionId, state)
  }
  return state
}

恢复：state = checkpoint.load(sessionId) 后继续 Loop，注意 idempotency（勿重复发邮件）。

第 5 步：Planner 与 Plan-and-ReAct 混合 ​

复杂任务可先 plan = planner.generatePlan(userQuery) 写入 state，再在 system 中注入：

text
当前计划：
1. 搜索文档
2. 对比表格
3. 输出结论
请按计划在必要时调用工具，完成一步可在回复中标注 [Step 1 done]。

计划可随 Observation 修订（dynamic replan），但要有「计划漂移」检测，避免无限改计划逃避执行。

第 6 步：测试策略 ​

  • 单元测试：mock LLM 固定 tool_calls 序列，断言 State 变迁
  • 契约测试：每个 Tool handler 输入输出
  • Golden trace replay：录制真实 trace，回归比较 abortReason 与最终 status
  • 混沌：随机 tool 超时，验证 Harness 降级

Loop 逻辑不应依赖「真实 LLM 今天心情好不好」。

动手练习 ​

  1. 实现 createInitialState + abort，并为 MAX_ITERATIONS 写单测。
  2. mock planner 返回两轮 tool_calls 再三轮 stop，断言 iteration 与 toolResults 长度。
  3. 在 ToolExecutor 加 1s timeout mock，验证 error 形状进入 messages。
  4. 设计 checkpoint JSON 样例，说明 resume 时如何防止重复副作用。
  5. 为 runAgentLoop 输出结构化 trace（每步一行 JSON），供 jq 分析。

常见问题 ​

Q：Planner 和 LLM 是一回事吗？

工程上 Planner 是「调用 LLM 的那一层模块」，可换模型（小模型规划、大模型生成）。也可拆成独立 plan 模型与 act 模型。

Q：Observer 必须吗？

最小 Loop 可省略。但生产建议至少有 metrics + trace + checkpoint，否则故障无法 replay。

Q：LangGraph 和这套结构的关系？

LangGraph 用图节点显式表达 State 与边；本节是概念同构的自研版。选型看团队是否要可视化图与内置 persistence。

本节小结 ​

工程化 Agent Loop = 可序列化 State + Planner/Executor/Observer 拆分 + 终止/预算/重复检测 + Harness 钩子 + Checkpoint。先 mock 测 Loop，再接真 LLM。