附录 J：分布式 Agent — 当你的框架从笔记本搬到 K8s

// → 有状态 Agent 的会话管理
import { Redis } from "ioredis"

class DistributedAgentManager {
  constructor(
    private redis: Redis,
    private framework: AgentFramework,
  ) {}

  async getOrCreateAgent(sessionId: string): Promise<AgentLoop> {
    // 1. 尝试从 Redis 恢复
    const serialized = await this.redis.get(`session:${sessionId}`)
    if (serialized) {
      return AgentLoop.deserialize(JSON.parse(serialized), this.framework)
    }

    // 2. 创建新 Agent
    const agent = this.framework.createAgent()

    // 3. 每次状态变化后异步保存
    agent.onStateChange(async (state) => {
      await this.redis.setex(
        `session:${sessionId}`,
        3600,  // 1 小时 TTL
        JSON.stringify(state.serialize()),
      )
    })

    return agent
  }

  async deleteSession(sessionId: string): Promise<void> {
    await this.redis.del(`session:${sessionId}`)
  }
}

// → 使用 BullMQ 做子 Agent 调度
import { Queue, Worker } from "bullmq"

// 主 Agent 创建子任务
const subAgentQueue = new Queue("subagent-tasks", { connection: redis })

async function spawnSubAgent(task: SubAgentTask): Promise<SubAgentResult> {
  const job = await subAgentQueue.add("execute", task, {
    timeout: 60_000,  // 子 Agent 超时 1 分钟
    attempts: 2,       // 失败重试 2 次
  })

  // 等待结果
  return job.waitUntilFinished(subAgentQueueEvents)
}

// Worker 端消费任务
const worker = new Worker("subagent-tasks", async (job) => {
  const agent = framework.createAgent({ /* 子 Agent 配置 */ })
  return agent.runSync(job.data.prompt)  // 同步执行
}, { connection: redis, concurrency: 5 })  // 5 个并发子 Agent

// → 幂等工具执行包装
class IdempotentToolExecutor {
  private results = new Map<string, ToolResult>()  // 生产环境用 Redis

  async execute(
    toolName: string,
    input: Record<string, unknown>,
    idempotencyKey: string,  // 由调用方生成
  ): Promise<ToolResult> {
    // 1. 检查是否已执行
    const cached = this.results.get(idempotencyKey)
    if (cached) {
      console.log(`[idempotent] ${toolName} 已执行过，返回缓存结果`)
      return cached
    }

    // 2. 执行工具
    const result = await this.executeTool(toolName, input)

    // 3. 缓存结果（24h TTL）
    this.results.set(idempotencyKey, result)

    return result
  }
}

// 幂等键的生成
function generateIdempotencyKey(
  sessionId: string,
  turn: number,
  toolName: string,
): string {
  return `${sessionId}:turn-${turn}:${toolName}`
}

// → Session 序列化格式
interface SerializedSession {
  id: string
  messages: Array<{
    role: "user" | "assistant"
    content: string | ContentBlockParam[]
  }>
  metadata: {
    model: string
    turnCount: number
    totalTokens: { input: number; output: number }
    sessionCost: number
    startedAt: string
    lastActivityAt: string
  }
}

// 代理层：透明持久化
class PersistentAgentLoop {
  constructor(
    private inner: AgentLoop,
    private store: SessionStore,
    private sessionId: string,
  ) {}

  async *run(input: string): AsyncGenerator<AgentEvent> {
    for await (const event of this.inner.run(input)) {
      yield event
      // 每个事件后增量保存
      if (event.type === "tool_result" || event.type === "response") {
        await this.store.save(this.sessionId, this.inner.serialize())
      }
    }
  }
}