ツール呼び出しは成功しているのに前進しない — 無人エージェントの停滞を検知して止める設計

import hashlib, json
from dataclasses import dataclass, field
from collections import deque
 
VOLATILE_KEYS = {"timestamp", "request_id", "trace_id", "elapsed_ms", "nonce"}
 
def _strip_volatile(obj):
    if isinstance(obj, dict):
        return {k: _strip_volatile(v) for k, v in sorted(obj.items())
                if k not in VOLATILE_KEYS}
    if isinstance(obj, list):
        return [_strip_volatile(v) for v in obj]
    return obj
 
def action_fingerprint(tool_name: str, tool_input: dict) -> str:
    canonical = json.dumps(
        {"tool": tool_name, "input": _strip_volatile(tool_input)},
        sort_keys=True, ensure_ascii=False,
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
 
def result_digest(tool_result: dict) -> str:
    # 結果のうち「意味のある変化」だけを残す。本文が長い場合は要約ハッシュにする
    salient = _strip_volatile(tool_result)
    blob = json.dumps(salient, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]

@dataclass
class Step:
    fp: str          # action_fingerprint
    digest: str      # result_digest
    score: float     # その時点の進捗オラクル値
 
@dataclass
class StagnationGuard:
    window: int = 8                 # 直近何ステップを見るか
    max_exact_repeats: int = 3      # 同一(行動+結果)の許容回数
    novelty_floor: float = 0.34     # 新規フィンガープリント比率の下限
    progress_patience: int = 6      # 進捗が伸びないまま許すステップ数
    _hist: deque = field(default_factory=lambda: deque(maxlen=64))
    _best_score: float = float("-inf")
    _since_improved: int = 0
 
    def observe(self, step: Step) -> None:
        self._hist.append(step)
        if step.score > self._best_score + 1e-9:
            self._best_score = step.score
            self._since_improved = 0
        else:
            self._since_improved += 1
 
    def _exact_repeats(self) -> int:
        if not self._hist:
            return 0
        last = self._hist[-1]
        key = (last.fp, last.digest)
        return sum(1 for s in self._hist if (s.fp, s.digest) == key)
 
    def _novelty_ratio(self) -> float:
        recent = list(self._hist)[-self.window:]
        if len(recent) < self.window:
            return 1.0  # サンプル不足のうちは判定しない
        uniq = len({s.fp for s in recent})
        return uniq / len(recent)
 
    def _oscillating(self) -> bool:
        # A-B-A-B のような短周期サイクルを検出する
        recent = [s.fp for s in list(self._hist)[-self.window:]]
        for period in (2, 3):
            if len(recent) >= period * 2:
                tail = recent[-period * 2:]
                if tail[:period] == tail[period:]:
                    return True
        return False
 
    def verdict(self) -> str | None:
        if self._exact_repeats() >= self.max_exact_repeats:
            return "exact_repeat"
        if self._oscillating():
            return "oscillation"
        if self._novelty_ratio() < self.novelty_floor:
            return "low_novelty"
        if self._since_improved >= self.progress_patience:
            return "no_progress"
        return None

class StagnationHalt(Exception):
    def __init__(self, reason: str, evidence: dict):
        super().__init__(f"stagnation halt: {reason}")
        self.reason = reason
        self.evidence = evidence
 
def build_evidence(guard: StagnationGuard, reason: str) -> dict:
    recent = list(guard._hist)[-guard.window:]
    return {
        "reason": reason,
        "best_score": guard._best_score,
        "steps_since_improved": guard._since_improved,
        "novelty_ratio": round(guard._novelty_ratio(), 3),
        "recent_fingerprints": [s.fp for s in recent],
        "recent_scores": [round(s.score, 3) for s in recent],
    }

def run_agent_loop(client, messages, tools, oracle,
                   guard: StagnationGuard, max_turns: int = 50):
    for turn in range(max_turns):
        resp = client.run_turn(messages, tools)   # モデル呼び出し（簡略化）
        messages.append(resp.assistant_message)
 
        if not resp.tool_calls:                    # ツールを呼ばない＝完了
            return {"status": "done", "turns": turn + 1}
 
        for call in resp.tool_calls:
            result = call.execute()
            messages.append(result.as_tool_message())
            guard.observe(Step(
                fp=action_fingerprint(call.name, call.input),
                digest=result_digest(result.payload),
                score=oracle.score(),
            ))
 
        reason = guard.verdict()
        if reason:
            evidence = build_evidence(guard, reason)
            raise StagnationHalt(reason, evidence)
 
    raise StagnationHalt("turn_limit", {"turns": max_turns})

try:
    out = run_agent_loop(client, msgs, tools, oracle, StagnationGuard())
except StagnationHalt as halt:
    log.warning("agent halted: %s", halt.evidence)
    # 無人運用なら: 証拠を残して当該ジョブはクリーンに失敗扱い。
    # 次回の再実行に向けて、窓幅・patience・プロンプト制約の見直し対象として記録する。
    record_failure(reason=halt.reason, evidence=halt.evidence)