Antigravity エージェントの判断を追跡する — 意思決定ログと説明可能性の実装設計

# ❌ 悪い例: ツールに理由フィールドがない
tools_bad = [
    {
        "name": "write_file",
        "description": "ファイルに内容を書き込む",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        }
    }
]
# → エージェントはなぜそのファイルを選んだか外部から判断できない
# → デバッグ時に「何をしたか」はわかるが「なぜしたか」が追跡不能
# → 問題が起きたとき、プロンプトを修正すべきかツール定義を修正すべきか判断できない

# ✅ 良い例: reason と scope フィールドを必須化し、意図の言語化を促す
tools_good = [
    {
        "name": "write_file",
        "description": "ファイルに内容を書き込む。副作用があるため reason と scope を必ず記入する",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "変更するファイルのパス"},
                "content": {"type": "string", "description": "新しい内容"},
                "reason": {
                    "type": "string",
                    "description": "このファイルを変更する理由を50文字以上で説明する"
                },
                "scope": {
                    "type": "string",
                    "enum": ["specific_file", "pattern_match"],
                    "description": "変更範囲（specific_file: 指定ファイルのみ、pattern_match: 条件に合致する全ファイル）"
                }
            },
            "required": ["path", "content", "reason", "scope"]
        }
    }
]
# ✅ ログから後で確認できること:
#   - どのファイルを変更したか（path）
#   - なぜそのファイルを選んだか（reason フィールド）
#   - 意図した変更範囲（scope フィールド）
#   - 理由の充実度から推定した信頼度スコア

from dataclasses import dataclass, field
import re
from typing import Optional
 
@dataclass
class ReasoningRecord:
    """ReActパターンの思考ステップを記録する構造体"""
    step: int
    thought: str                                          # エージェントが考えたこと
    observation: str                                      # 環境から得た情報
    action_chosen: str                                    # 選択したアクション
    alternatives_considered: list[str] = field(default_factory=list)  # 検討した他の選択肢
    decision_factors: list[str] = field(default_factory=list)         # 決定に影響した要因
    uncertainty_flags: list[str] = field(default_factory=list)        # 不確実だった点
 
    def to_dict(self) -> dict:
        return {
            "step": self.step,
            "thought": self.thought,
            "observation": self.observation,
            "action_chosen": self.action_chosen,
            "alternatives": self.alternatives_considered,
            "factors": self.decision_factors,
            "uncertainties": self.uncertainty_flags
        }
 
def build_structured_logging_prompt() -> str:
    """構造化ログを促すシステムプロンプトを構築する"""
    return """各ステップで以下の形式で思考してください:
 
<思考>
現在の状況: [観察した内容]
検討した選択肢: [A案, B案, C案]
選択した理由: [なぜこのアクションを選んだか]
不確実な点: [確信が持てない要素があれば記録]
</思考>
<行動>
[実際のツール呼び出しまたは回答]
</行動>
 
副作用のあるアクション（書き込み・削除・外部API呼び出し）の前には
必ず「不確実な点」を明示してください。
不確実な点がある場合は、最も保守的な選択をしてください。"""
 
def parse_reasoning_from_response(response_text: str, step: int) -> Optional[ReasoningRecord]:
    """レスポンスから構造化された思考ステップを抽出する"""
    thought_match = re.search(r'<思考>(.*?)</思考>', response_text, re.DOTALL)
    action_match = re.search(r'<行動>(.*?)</行動>', response_text, re.DOTALL)
 
    if not thought_match:
        return None
 
    thought_text = thought_match.group(1).strip()
    action_text = action_match.group(1).strip() if action_match else "（行動未記録）"
 
    # 各要素を抽出
    uncertainties = re.findall(r'不確実な点[：:]\s*(.+)', thought_text)
    alternatives = re.findall(r'検討した選択肢[：:]\s*(.+)', thought_text)
    factors = re.findall(r'選択した理由[：:]\s*(.+)', thought_text)
    observation = re.search(r'現在の状況[：:]\s*(.+)', thought_text)
 
    return ReasoningRecord(
        step=step,
        thought=thought_text,
        observation=observation.group(1) if observation else "",
        action_chosen=action_text[:100],
        alternatives_considered=alternatives,
        decision_factors=factors,
        uncertainty_flags=uncertainties
    )

def analyze_decision_logs(log_entries: list[dict]) -> dict:
    """判断ログを分析してダッシュボード用メトリクスを算出する"""
 
    total = len(log_entries)
    if total == 0:
        return {"error": "ログエントリが空です"}
 
    # 信頼度分布
    confidence_counts = {"high": 0, "medium": 0, "low": 0}
    for entry in log_entries:
        c = entry.get("confidence", "medium")
        confidence_counts[c] = confidence_counts.get(c, 0) + 1
 
    low_confidence_rate = confidence_counts["low"] / total
 
    # ツール使用パターン
    tool_sequence = [
        e["tool"]["name"]
        for e in log_entries
        if e.get("tool")
    ]
 
    # 不確実フラグ密度
    uncertainty_steps = sum(
        1 for e in log_entries
        if e.get("uncertainty_flags") and len(e["uncertainty_flags"]) > 0
    )
    uncertainty_density = uncertainty_steps / total
 
    # 品質スコア（0〜100）
    quality_score = max(0, min(100, int(
        100
        - (low_confidence_rate * 40)     # 低信頼度が多いと減点
        - (uncertainty_density * 20)      # 不確実フラグが多いと減点
    )))
 
    return {
        "quality_score": quality_score,
        "total_steps": total,
        "confidence_distribution": confidence_counts,
        "low_confidence_rate": f"{low_confidence_rate:.1%}",
        "uncertainty_density": f"{uncertainty_density:.1%}",
        "tool_sequence": tool_sequence,
        "recommendations": generate_recommendations(
            low_confidence_rate,
            uncertainty_density
        )
    }
 
def generate_recommendations(
    low_confidence_rate: float,
    uncertainty_density: float
) -> list[str]:
    """メトリクスから改善推奨事項を生成する"""
    recs = []
 
    if low_confidence_rate > 0.20:
        recs.append(
            "低信頼度判断が20%を超えています。"
            "タスク指示を具体化するか、ツールの説明を詳細化することを推奨します"
        )
 
    if uncertainty_density > 0.30:
        recs.append(
            "不確実フラグの密度が高いです。"
            "エージェントに渡すコンテキスト情報を増やすか、"
            "不確実なケースの判断基準をシステムプロンプトに追記することを推奨します"
        )
 
    if not recs:
        recs.append("品質指標は良好です。このプロンプト設計を参考に他のタスクを構築してください")
 
    return recs

import json
import sqlite3
from datetime import datetime
from pathlib import Path
 
LOG_DIR = Path("logs/decisions")
 
def append_entry(entry: dict) -> None:
    """実行中はJSONLに追記するだけ。スキーマ検証は読み込み側で行う"""
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    path = LOG_DIR / f"{datetime.now():%Y-%m-%d}.jsonl"
    with path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
 
def load_into_sqlite(db_path: str = "decisions.db") -> int:
    """分析前にJSONLをSQLiteへ取り込む。壊れた行はスキップして件数を返す"""
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS decisions (
            session_id TEXT, step INTEGER, tool TEXT,
            confidence TEXT, reason TEXT, ts TEXT,
            PRIMARY KEY (session_id, step)
        )
    """)
    imported = 0
    for jsonl in sorted(LOG_DIR.glob("*.jsonl")):
        for line in jsonl.read_text(encoding="utf-8").splitlines():
            try:
                e = json.loads(line)
                conn.execute(
                    "INSERT OR IGNORE INTO decisions VALUES (?,?,?,?,?,?)",
                    (e["session_id"], e["step"],
                     (e.get("tool") or {}).get("name"),
                     e.get("confidence"), e.get("reason"), e.get("timestamp")),
                )
                imported += 1
            except (json.JSONDecodeError, KeyError):
                continue  # 壊れた行で全体を止めない
    conn.commit()
    conn.close()
    return imported

def estimate_reason_specificity(reason: str) -> str:
    """理由テキストの具体性から信頼度を推定する"""
    import re
    specificity_markers = [
        r'\b[a-zA-Z_]+\.(py|ts|json|md|mdx|txt)\b',    # ファイル名
        r'\b(function|class|def|const|let|var)\s+\w+',  # コード要素
        r'\b(because|since|given that|due to)\b',        # 因果表現（英語）
        r'(ため|から|ので|なぜなら)',                      # 因果表現（日本語）
        r'\d+',                                           # 数値
    ]
    matched = sum(1 for p in specificity_markers if re.search(p, reason, re.IGNORECASE))
 
    if matched >= 3:
        return "high"
    elif matched >= 1:
        return "medium"
    else:
        return "low"