AIエージェントのデバッグ：ループ・幻覚・汚染の対処法

エージェントが期待どおりに動かない
│
├─ 同じアクションを繰り返しているか？
│   YES → [障害1: 無限ループ] → ステップカウンタを確認
│   NO  ↓
│
├─ 存在しないリソースを参照しているか？
│   YES → [障害2: ハルシネーション] → ツール戻り値を確認
│   NO  ↓
│
├─ 指示と無関係な行動をとっているか？
│   YES → [障害3: コンテキスト汚染] → メッセージ履歴の長さを確認
│   NO  ↓
│
└─ 外部サービスの障害・レート制限を確認

Step 1: ファイルを読み込む → エラー
Step 2: ファイルを読み込む → エラー
Step 3: ファイルを読み込む → エラー
... (繰り返し)

class AgentLoop:
    def __init__(self, max_steps: int = 50):
        self.step_count = 0
        self.max_steps = max_steps
    
    def step(self, action):
        self.step_count += 1
        if self.step_count > self.max_steps:
            raise LoopLimitExceeded(f"最大ステップ数 {self.max_steps} を超えました")
        return self._execute(action)

import hashlib
from collections import deque

class LoopDetector:
    def __init__(self, window: int = 5):
        self.history = deque(maxlen=window)
    
    def check(self, action: dict) -> bool:
        """Trueならループ検知"""
        h = hashlib.md5(str(action).encode()).hexdigest()
        if h in self.history:
            return True
        self.history.append(h)
        return False

# プロンプトへのexit条件追加例
SYSTEM_PROMPT = """
あなたはファイル操作エージェントです。

## 停止条件（必ず守ること）
- 同じ操作が3回失敗した場合: 失敗理由を報告して停止する
- ステップ数が50を超えた場合: 現在の進捗を報告して停止する
- ツールエラーが連続5回発生した場合: エラー内容を報告して停止する
"""

# エージェントが生成したコード（ハルシネーション例）
result = client.files.upload_and_process(
    file_path="data.csv",
    processing_mode="intelligent"  # このメソッドは存在しない
)

from pydantic import BaseModel, ValidationError

class FileReadInput(BaseModel):
    file_path: str
    encoding: str = "utf-8"

def validate_tool_call(tool_name: str, params: dict) -> bool:
    schemas = {
        "read_file": FileReadInput,
    }
    if tool_name not in schemas:
        return False  # 未知のツール名 → 幻覚の可能性
    try:
        schemas[tool_name](**params)
        return True
    except ValidationError:
        return False

def execute_tool(tool_name: str, params: dict) -> dict:
    result = run_tool(tool_name, params)
    
    # 戻り値の構造を検証
    if "error" in result:
        # エラー内容をLLMに正確に伝え、捏造を防ぐ
        return {
            "success": False,
            "error_code": result["error"]["code"],
            "message": result["error"]["message"],
            "suggestion": "ツール名・パラメータを確認してください"
        }
    return {"success": True, "data": result}

# 幻覚率モニタリング
class HallucinationMetrics:
    def __init__(self):
        self.total_calls = 0
        self.validation_failures = 0
    
    @property
    def hallucination_rate(self) -> float:
        if self.total_calls == 0:
            return 0.0
        return self.validation_failures / self.total_calls
    
    def record(self, is_valid: bool):
        self.total_calls += 1
        if not is_valid:
            self.validation_failures += 1

メッセージ1: 「Aファイルを編集してください」
メッセージ50: 「Aの編集結果を確認...」
...エラーログが大量に蓄積...
メッセージ120: 「（エラーログを見て）まずエラーを解消します」← 元の指示から逸脱

import tiktoken

class ContextMonitor:
    def __init__(self, model: str = "gpt-4o", threshold: float = 0.8):
        self.enc = tiktoken.encoding_for_model(model)
        self.max_tokens = 128_000  # gpt-4o のコンテキスト長
        self.threshold = threshold
    
    def get_usage_ratio(self, messages: list) -> float:
        total = sum(
            len(self.enc.encode(m["content"]))
            for m in messages
        )
        return total / self.max_tokens
    
    def is_contaminated(self, messages: list) -> bool:
        return self.get_usage_ratio(messages) > self.threshold

def sliding_window(messages: list, keep_last: int = 20) -> list:
    # システムプロンプト + 最新N件のユーザー/アシスタントメッセージを保持
    system = [m for m in messages if m["role"] == "system"]
    others = [m for m in messages if m["role"] != "system"]
    return system + others[-keep_last:]

async def summarize_old_context(messages: list, llm) -> list:
    """コンテキストが長くなったら前半を要約する"""
    if len(messages) < 40:
        return messages
    
    old_messages = messages[1:-20]  # システムプロンプトと最新20件は除く
    summary_prompt = f"""
    以下の会話履歴を3-5行で要約してください。
    完了したタスクと判明した情報のみを含めてください。
    
    会話履歴:
    {format_messages(old_messages)}
    """
    summary = await llm.complete(summary_prompt)
    
    return [
        messages[0],  # システムプロンプト
        {"role": "assistant", "content": f"[これまでの要約] {summary}"},
        *messages[-20:]  # 最新20件
    ]

from opentelemetry import trace

tracer = trace.get_tracer("agent")

def agent_step(action: dict):
    with tracer.start_as_current_span("agent.step") as span:
        span.set_attribute("action.type", action["type"])
        span.set_attribute("action.tool", action.get("tool", ""))
        
        result = execute_action(action)
        
        span.set_attribute("result.success", result["success"])
        if not result["success"]:
            span.set_attribute("result.error", result.get("error", ""))
        
        return result

import json
from datetime import datetime

class ReplayableLogger:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.events = []
    
    def log_llm_call(self, messages: list, response: str):
        self.events.append({
            "type": "llm_call",
            "timestamp": datetime.utcnow().isoformat(),
            "messages": messages,
            "response": response,
        })
    
    def log_tool_call(self, tool: str, params: dict, result: dict):
        self.events.append({
            "type": "tool_call",
            "timestamp": datetime.utcnow().isoformat(),
            "tool": tool,
            "params": params,
            "result": result,
        })
    
    def save(self, path: str):
        with open(path, "w") as f:
            json.dump({"session_id": self.session_id, "events": self.events}, f, indent=2)