agent loop を durable workflow で安定化する実装

[plan] → checkpoint → [execute:step1] → checkpoint → [execute:step2] → checkpoint → [verify] → checkpoint → [done]

// checkpoint.ts
interface CheckpointRecord {
  runId: string;
  stepName: string;
  stepIndex: number;
  status: "running" | "completed" | "failed";
  output: unknown;
  createdAt: Date;
}

class AgentCheckpointStore {
  constructor(private db: DatabaseClient) {}

  async save(runId: string, stepName: string, stepIndex: number, output: unknown): Promise<void> {
    await this.db.upsert("agent_checkpoints", {
      run_id: runId,
      step_name: stepName,
      step_index: stepIndex,
      status: "completed",
      output: JSON.stringify(output),
      created_at: new Date(),
    });
  }

  async load(runId: string): Promise<CheckpointRecord | null> {
    return this.db.findOne("agent_checkpoints", {
      run_id: runId,
      status: "completed",
      order: { step_index: "desc" },
    });
  }
}

// durable-agent-loop.ts
async function durableAgentLoop(runId: string, task: Task): Promise<Result> {
  const store = new AgentCheckpointStore(db);
  const steps: Step[] = [planStep, executeStep, verifyStep];

  const lastCheckpoint = await store.load(runId);
  const startIndex = lastCheckpoint ? lastCheckpoint.stepIndex + 1 : 0;

  let context = lastCheckpoint?.output ?? {};

  for (let i = startIndex; i < steps.length; i++) {
    const step = steps[i];
    context = await step.run(context, task);
    await store.save(runId, step.name, i, context);
  }

  return context as Result;
}

// retry.ts
interface RetryOptions {
  maxAttempts: number;
  baseDelayMs: number;
  maxDelayMs: number;
  retryableErrors: string[];
}

async function withRetry<T>(
  fn: () => Promise<T>,
  opts: RetryOptions
): Promise<T> {
  let lastError: Error;

  for (let attempt = 1; attempt <= opts.maxAttempts; attempt++) {
    try {
      return await fn();
    } catch (err) {
      lastError = err as Error;

      const isRetryable = opts.retryableErrors.some((code) =>
        lastError.message.includes(code)
      );
      if (!isRetryable || attempt === opts.maxAttempts) throw lastError;

      const delay = Math.min(
        opts.baseDelayMs * 2 ** (attempt - 1),
        opts.maxDelayMs
      );
      await sleep(delay);
    }
  }

  throw lastError!;
}

// 使用例
const result = await withRetry(() => llm.complete(prompt), {
  maxAttempts: 3,
  baseDelayMs: 1000,
  maxDelayMs: 30_000,
  retryableErrors: ["rate_limit_exceeded", "timeout", "service_unavailable"],
});

// saga.ts
interface SagaStep<T> {
  name: string;
  execute: (ctx: T) => Promise<T>;
  compensate: (ctx: T) => Promise<void>;
}

async function executeSaga<T>(steps: SagaStep<T>[], initialCtx: T): Promise<T> {
  const completed: SagaStep<T>[] = [];
  let ctx = initialCtx;

  try {
    for (const step of steps) {
      ctx = await step.execute(ctx);
      completed.push(step);
    }
    return ctx;
  } catch (err) {
    // 補償処理を逆順で実行
    for (const step of completed.reverse()) {
      await step.compensate(ctx).catch((compensateErr) => {
        console.error(`compensation failed for ${step.name}:`, compensateErr);
      });
    }
    throw err;
  }
}

// agent loopへの適用例
const result = await executeSaga(
  [
    {
      name: "createDraftDocument",
      execute: async (ctx) => {
        const doc = await docStore.create(ctx.content);
        return { ...ctx, docId: doc.id };
      },
      compensate: async (ctx) => {
        if (ctx.docId) await docStore.delete(ctx.docId);
      },
    },
    {
      name: "publishToExternalAPI",
      execute: async (ctx) => {
        await externalApi.publish(ctx.docId, ctx.metadata);
        return { ...ctx, published: true };
      },
      compensate: async (ctx) => {
        if (ctx.published) await externalApi.unpublish(ctx.docId);
      },
    },
  ],
  { content: task.content, metadata: task.metadata }
);

// tracing.ts
import { trace, SpanStatusCode } from "@opentelemetry/api";

const tracer = trace.getTracer("agent-loop", "1.0.0");

async function tracedStep<T>(stepName: string, fn: () => Promise<T>): Promise<T> {
  return tracer.startActiveSpan(`agent.step.${stepName}`, async (span) => {
    span.setAttribute("agent.step.name", stepName);
    span.setAttribute("agent.run.id", currentRunId);

    try {
      const result = await fn();
      span.setAttribute("agent.step.output.size", JSON.stringify(result).length);
      span.setStatus({ code: SpanStatusCode.OK });
      return result;
    } catch (err) {
      span.setStatus({ code: SpanStatusCode.ERROR, message: (err as Error).message });
      span.recordException(err as Error);
      throw err;
    } finally {
      span.end();
    }
  });
}

// 使用例
const planResult = await tracedStep("plan", () => planner.run(task));
const execResult = await tracedStep("execute", () => executor.run(planResult));
const verifyResult = await tracedStep("verify", () => verifier.run(execResult));

// structured logging
logger.info({
  event: "step.completed",
  runId,
  stepName,
  stepIndex,
  durationMs: Date.now() - startMs,
  outputTokens: result.usage?.totalTokens,
});