私のAIエージェントの仕事は1つだけだった。記事を書いて公開プラットフォームに投稿し、その後クリーンアップすることだ。10回に3回、実行途中でクラッシュし、下書きが孤立したままになり、一時ファイルが失われ、同じ記事を最初から再度公開しようとした。その結果、重複した下書き、半分失敗した投稿、そして最悪の場合、同じフィードに2つの同一記事が公開されることになった。
パイプラインをクラッシュセーフにする方法が必要だった。「結果整合性」ではなく、アトミックなクラッシュセーフだ。それを実現したステートマシンパターンを紹介する。
解決策
パイプラインを各々が単一の責任を持つフェーズに分割した。run IDですべてを結びつける。マニフェストファイルでどのテンポラリファイルがどのrunに属するかを追跡する。ステートマシンは、現在のフェーズのゲートがすべて緑にならない限り進まない。
from __future__ import annotations
import json
from pathlib import Path
from enum import Enum
from datetime import datetime
from typing import Optional
class Phase(Enum):
LOAD = "loading"
SELECT = "topic_selection"
DEDUP = "topic_dedup"
WRITE = "article_generation"
GATE = "de_ai_gate"
VALIDATE = "pre_publish_validate"
PUBLISH = "publish"
VERIFY = "post_publish_verify"
CLEAN = "cleanup"
class RunState:
def __init__(self, run_id: str, manifest_path: Path):
self.run_id = run_id
self.manifest_path = manifest_path
self.current_phase = None
self.phase_results = {}
def save(self):
data = {
"run_id": self.run_id,
"current_phase": self.current_phase.value if self.current_phase else None,
"phase_results": self.phase_results,
"updated_at": datetime.now().isoformat(),
}
self.manifest_path.write_text(json.dumps(data, ensure_ascii=False, indent=2))
@classmethod
def load(cls, manifest_path: Path):
if not manifest_path.exists():
return None
data = json.loads(manifest_path.read_text(encoding="utf-8"))
state = cls(data["run_id"], manifest_path)
state.current_phase = Phase(data["current_phase"]) if data["current_phase"] else None
state.phase_results = data["phase_results"]
return state
Enter fullscreen mode Exit fullscreen mode
ステートマシン自体はシンプルなループだ。各フェーズはTrue（進む）または例外を発生（停止）する関数だ。進む前に、状態はディスクに保存される。
class Pipeline:
def __init__(self, state: RunState, temp_files: list[Path]):
self.state = state
self.temp_files = temp_files
self.guard = Guard(temp_files)
def run(self, phases: list[Phase]):
for phase in phases:
print(f"[{self.state.run_id}] Phase: {phase.value}")
self.state.current_phase = phase
self.state.save()
handler = getattr(self, f"_phase_{phase.value}")
handler()
self.state.phase_results[phase.value] = True
self.state.save()
def _phase_cleanup(self):
for f in self.temp_files:
if f.exists():
f.unlink()
print(f" cleaned: {f.name}")
self.state.manifest_path.unlink(missing_ok=True)
Enter fullscreen mode Exit fullscreen mode
Guardクラスはパイプラインとファイルシステムの間に位置する。どのファイルが一時的な公開成果物かを正確に把握し、スキル整合性チェック中にそれらを無視する。
class Guard:
def __init__(self, allowed_temps: list[Path]):
self.allowed = {p.resolve() for p in allowed_temps}
def is_temp_artifact(self, path: Path) -> bool:
return path.resolve() in self.allowed
def verify_no_untracked_writes(self, workspace: Path):
for f in workspace.iterdir():
if f.is_file() and not self.is_temp_artifact(f):
raise RuntimeError(
f"Untracked file: {f.name}. "
"Add to temp manifest or refactor."
)
Enter fullscreen mode Exit fullscreen mode
なぜこれが機能するのか
このパターンをクラッシュセーフにする3つの設計上の決定がある。
第一に、ディスクファーストの状態。ステートマシンは各フェーズハンドラーの前後でディスクに書き込む。エージェントが_phase_validate()中にクラッシュした場合、次の実行はmanifest.jsonを読み込み、current_phase: "pre_publish_validate"を見て、どこから再開するかを正確に知ることができる。「前どこまでやったっけ？」という推測は不要だ。
第二に、一時ファイルマニフェスト。パイプラインが開始する前に、作成されるすべてのファイルが登録される。Guardクラスはこれを強制する。マニフェストにないファイルへの書き込みが発生した場合、パイプラインは停止する。これが2回検出された — 1回はPythonでopen()ではなくwrite_fileを使った時、もう1回はログライブラリが作業ディレクトリに.logファイルを作成した時だ。
第三に、冪等なフェーズハンドラー。各フェーズ関数は、作業が既に完了しているかをチェックしてから再度実行する。_phase_dedup()は既存の記事リストを読み込み、未完了の場合のみ新しいデータを書き込む。つまり、クラッシュ後にPhase 2を再実行しても安全で、二重チェックや二重書き込みは発生しない。
def _phase_dedup(self):
manifest = self.state.manifest_path
dedup_flag = manifest.parent / f"{self.state.run_id}_dedup_done"
if dedup_flag.exists():
print(" dedup already done, skipping")
return
# ... actual dedup logic ...
dedup_flag.write_text("done")
Enter fullscreen mode Exit fullscreen mode
注意点
Windowsのファイルロックは異なる。Linuxではfcntl.flock()が正常に動作する。Windowsではmsvcrt.locking()が必要だ。2つのパイプライン実行が偶然重複し、同じマニフェストファイルに書き込もうとした時に遭遇した。クロスプラットフォームロックで修正した。
import sys
def safe_write(path: Path, data: str):
if sys.platform == "win32":
import msvcrt
with open(path, "w", encoding="utf-8") as f:
msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)
f.write(data)
msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
else:
import fcntl
with open(path, "w", encoding="utf-8") as f:
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
f.write(data)
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
Enter fullscreen mode Exit fullscreen mode
マニフェストファイル自体が一時ファイル。一時ファイルマニフェストに含めれば、Guardは通過させる。ただし、_phase_cleanupハンドラーは他のすべてのテンポラリが削除された後に最後にマニフェストを削除するため、クリーンアップ中のクラッシュで状態レコードが失われることはない。
ステートマシンはフェーズ境界の良さで決まる。当初はフェーズを細かくしすぎた（10ステップのパイプラインに14フェーズ）。状態ファイルは絶えず上書きされ、I/Oオーバーヘッドが1実行あたり約2秒追加された。関連するステップを統合した。de_ai_gate + validateを1つのフェーズに、publish + verifyを別のフェーズに。7フェーズに削減し、体感できるオーバーヘッドはなくなった。
皆さんはどうですか？
パイプライン状態にSQLiteを使う人、分散パイプラインにRedisを使う人、S3上のJSONファイルを使う人を見てきた。私のアプローチは意図的に最小限 — 単一のJSONファイル、ガードクラス、そしてループだ。パイプラインがシングルスレッドで1台のマシンで実行されるため機能する。
クラッシュセーフな自動化に対する皆さんのアプローチは何ですか？ワークフローエンジン（Temporal、Prefect）に頼りますか、それとも独自のステートマシンを作りますか？皆さんの成功体験をぜひ聞かせてください。
0 Comments
Log in to join the conversation.No comments yet. Be the first to share your thoughts.