夜間バッチの出力は、いつも通り正しく見えていました。生成された記事も、投稿されたログも、何ひとつ壊れていません。
気づいたのは請求のほうが先でした。一日あたりのツール呼び出し回数が、以前の記録と比べておよそ 1.6 倍になっていたのです。答えは合っている。けれど、そこに至る道筋が変わっている。
出力の品質だけを見ていると、この種の変化は永遠に見えません。私が Claude の品質回帰を eval ハーネスで自動検出する設計 で書いたような出力評価は、最終的な文字列の良し悪しを測るものです。軌跡(トラジェクトリ)は測っていません。
個人開発で無人のエージェントを回し続けている立場からすると、この「道筋の静かな変化」こそが厄介でした。壊れないので気づけない。そして気づいたときには、コストか実行時間のどちらかが確実に膨らんでいます。
「軌跡」として何を記録すれば足りるのか
最終出力ではなく、そこへ至る過程を記録します。私が実用上、これだけあれば足りると判断したのは次の 5 つです。
| 記録項目 | 何を守るために要るか |
|---|---|
| ツール名の並び | 順序の入れ替わり・不要なツールの混入 |
| 各呼び出しの引数(正規化後) | 同一引数での重複呼び出し |
| ターン数とツール呼び出し総数 | ループの膨張 |
| stop_reason の系列 | 打ち切りや一時停止の混入 |
| 書き込み系ツールの呼び出し回数 | 副作用の二重実行 |
引数を丸ごと保存しないのが要点です。タイムスタンプや UUID が入るとトレースは毎回変わってしまい、比較の役に立ちません。揺れる値を落としてから記録する、これだけで軌跡は安定した比較対象になります。
実行トレースを記録する薄いラッパー
エージェントループに割り込むのではなく、ツール実行の入口だけを包みます。既存のループに手を入れずに済むので、本番コードと検証コードの乖離が起きにくいのが利点です。
# trace.py
from __future__ import annotations
import hashlib
import json
import re
from dataclasses import dataclass, field, asdict
from typing import Any, Callable
# 実行ごとに変わる値は比較前に落とす(落とさないと毎回 diff が出る)
VOLATILE_KEYS = {"request_id", "timestamp", "trace_id", "nonce", "cursor"}
UUID_RE = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")
ISO_RE = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}")
def normalize(value: Any) -> Any:
if isinstance(value, dict):
return {k: normalize(v) for k, v in sorted(value.items()) if k not in VOLATILE_KEYS}
if isinstance(value, list):
return [normalize(v) for v in value]
if isinstance(value, str):
value = UUID_RE.sub("<uuid>", value)
return ISO_RE.sub("<ts>", value)
return value
@dataclass
class ToolCall:
name: str
args_digest: str
ok: bool
@dataclass
class Trace:
case_id: str
model: str
calls: list[ToolCall] = field(default_factory=list)
stop_reasons: list[str] = field(default_factory=list)
turns: int = 0
def record(self, name: str, args: dict, ok: bool) -> None:
payload = json.dumps(normalize(args), ensure_ascii=False, sort_keys=True)
digest = hashlib.sha256(payload.encode()).hexdigest()[:12]
self.calls.append(ToolCall(name=name, args_digest=digest, ok=ok))
@property
def tool_names(self) -> list[str]:
return [c.name for c in self.calls]
def to_json(self) -> str:
return json.dumps(asdict(self), ensure_ascii=False, indent=2, sort_keys=True)
def instrument(tool_fn: Callable[[str, dict], Any], trace: Trace) -> Callable[[str, dict], Any]:
"""既存のツールディスパッチャを包むだけ。ループ側は無改造で済む。"""
def wrapped(name: str, args: dict) -> Any:
try:
result = tool_fn(name, args)
except Exception:
trace.record(name, args, ok=False)
raise
trace.record(name, args, ok=True)
return result
return wrapped引数そのものではなくダイジェストを持つのは、トレースをリポジトリにコミットしても機密が漏れないようにするためです。差分を見たいときはハッシュの不一致だけで十分に足ります。
ループ側は、ディスパッチャを差し替えて stop_reason を追記するだけです。
# run_case.py
import anthropic
client = anthropic.Anthropic()
MODEL = "claude-sonnet-5"
def run_case(case_id: str, prompt: str, tools: list[dict], dispatch) -> Trace:
trace = Trace(case_id=case_id, model=MODEL)
traced_dispatch = instrument(dispatch, trace)
messages = [{"role": "user", "content": prompt}]
while trace.turns < 24: # 上限は不変条件のひとつ(後述)
trace.turns += 1
resp = client.messages.create(
model=MODEL, max_tokens=4096, tools=tools, messages=messages,
)
trace.stop_reasons.append(resp.stop_reason)
if resp.stop_reason != "tool_use":
break
messages.append({"role": "assistant", "content": resp.content})
results = []
for block in resp.content:
if block.type != "tool_use":
continue
output = traced_dispatch(block.name, block.input)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(output, ensure_ascii=False),
})
messages.append({"role": "user", "content": results})
return tracetool_result を必ず同一ターンで全件返す点は Claude API の Parallel Tool Use を本番で使いこなす で扱った通りです。ここではトレース記録がその構造を壊さないことを確認しておけば十分です。