長時間エージェント実行を途中から立て直す — Gemini 3.5 Flash の long-horizon タスクをステップ台帳で運用する設計

import sqlite3
import hashlib
import json
import time
from contextlib import contextmanager
 
class StepLedger:
    def __init__(self, db_path: str = "ledger.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS steps (
                run_id     TEXT NOT NULL,
                step_id    TEXT NOT NULL,
                idem_key   TEXT NOT NULL,
                status     TEXT NOT NULL,
                output_ref TEXT,
                created_at REAL NOT NULL,
                PRIMARY KEY (run_id, step_id)
            )
        """)
        self.conn.commit()
 
    def idem_key(self, step_id: str, payload: dict) -> str:
        raw = step_id + "|" + json.dumps(payload, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
 
    def already_done(self, run_id: str, step_id: str):
        row = self.conn.execute(
            "SELECT output_ref FROM steps WHERE run_id=? AND step_id=? AND status='done'",
            (run_id, step_id),
        ).fetchone()
        return row[0] if row else None
 
    def _write(self, run_id, step_id, idem_key, status, output_ref):
        self.conn.execute(
            "INSERT OR REPLACE INTO steps VALUES (?,?,?,?,?,?)",
            (run_id, step_id, idem_key, status, output_ref, time.time()),
        )
        self.conn.commit()

    @contextmanager
    def run_step(self, run_id: str, step_id: str, payload: dict):
        cached = self.already_done(run_id, step_id)
        if cached is not None:
            # すでに成功済み。実体は実行せず、保存済み参照を返す
            yield {"skipped": True, "output_ref": cached}
            return
 
        key = self.idem_key(step_id, payload)
        self._write(run_id, step_id, key, "started", None)
        box = {"skipped": False, "output_ref": None}
        try:
            yield box
            self._write(run_id, step_id, key, "done", box["output_ref"])
        except Exception:
            self._write(run_id, step_id, key, "failed", None)
            raise

from google import genai
 
client = genai.Client()
ledger = StepLedger()
 
def summarize_article(run_id: str, article_id: str, text: str, store) -> str:
    step_id = f"summarize:{article_id}"
    with ledger.run_step(run_id, step_id, {"article_id": article_id}) as box:
        if box["skipped"]:
            return store.load(box["output_ref"])
        resp = client.models.generate_content(
            model="gemini-3.5-flash",
            contents=f"次の記事を3文で要約してください。\n\n{text}",
        )
        ref = store.save(f"summary/{article_id}.txt", resp.text)
        box["output_ref"] = ref
        return resp.text

def publish_draft(run_id: str, draft_id: str, body: str, store, publisher) -> str:
    step_id = f"publish:{draft_id}"
    payload = {"draft_id": draft_id, "body_hash": hashlib.sha256(body.encode()).hexdigest()[:12]}
    with ledger.run_step(run_id, step_id, payload) as box:
        if box["skipped"]:
            return box["output_ref"]
        key = ledger.idem_key(step_id, payload)
        # publisher 側が同じ key の二重受付を拒否する設計にしておく
        result_ref = publisher.publish(body, idempotency_key=key)
        box["output_ref"] = result_ref
        return result_ref

def explain_run(ledger: StepLedger, run_id: str):
    rows = ledger.conn.execute(
        "SELECT step_id, status, created_at FROM steps "
        "WHERE run_id=? ORDER BY created_at",
        (run_id,),
    ).fetchall()
    for step_id, status, ts in rows:
        mark = {"done": "OK", "failed": "NG", "started": "..."}.get(status, "?")
        print(f"[{mark}] {step_id}")