Antigravity Agent の出力に「自信スコア」を付ける — 確実な処理は自動承認、曖昧な判断だけ人に渡す設計

# 自信スコア計算: モデルに自己評価させる方式
import anthropic
from typing import Literal
 
client = anthropic.Anthropic()
 
SELF_EVAL_PROMPT = """以下は、AI エージェントが生成した出力です。
あなたはこの出力を厳しく評価する査読者です。
 
出力:
{output}
 
元のタスク:
{task}
 
次の各項目について、該当すれば該当する数だけ減点してください（初期値: 100）:
- 事実誤認・存在しない情報を含む: -40
- 元のタスクに直接答えていない: -25
- 不要な前置き・冗長な説明: -10
- スキーマ違反・形式エラー: -30
- 前提を勝手に追加している: -15
 
最終スコアを 0-100 の整数 1 つだけで返してください。それ以外の文字は禁止。
"""
 
def self_evaluation_score(task: str, output: str) -> float:
    msg = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 評価には軽量モデルで十分
        max_tokens=10,
        messages=[{
            "role": "user",
            "content": SELF_EVAL_PROMPT.format(task=task, output=output),
        }],
    )
    raw = msg.content[0].text.strip()
    try:
        score = int(raw)
    except ValueError:
        # パース失敗は最低スコアとして扱う（保守的）
        return 0.0
    return max(0.0, min(1.0, score / 100.0))
 
# 使用例
score = self_evaluation_score(
    task="ユーザーの請求書PDFから、金額・税率・支払期限を JSON で抽出して",
    output='{"amount": 12000, "tax_rate": 0.1, "due_date": "2026-06-15"}',
)
# 期待出力: 0.85〜0.95（情報抽出として妥当な内容のため）

# 自信スコア計算: スキーマ + ルールベースで測る方式
from pydantic import BaseModel, Field, ValidationError
from datetime import date
 
class InvoiceExtraction(BaseModel):
    amount: float = Field(gt=0, description="請求金額（正の数）")
    tax_rate: float = Field(ge=0, le=1, description="税率 0〜1")
    due_date: date = Field(description="支払期限")
 
def verifiability_score(output_dict: dict, expected_amount_range: tuple = None) -> float:
    """
    スキーマ検証とドメインルールでスコアを算出。
    完全パスで 1.0、一部違反で 0.5、構造破壊で 0.0。
    """
    score = 1.0
 
    # Layer 1: スキーマ検証（重め）
    try:
        parsed = InvoiceExtraction(**output_dict)
    except ValidationError:
        return 0.0  # スキーマ違反は致命的
 
    # Layer 2: ドメインルール検証（軽め）
    if expected_amount_range:
        lo, hi = expected_amount_range
        if not (lo <= parsed.amount <= hi):
            score -= 0.3  # 想定範囲外は要注意
    if parsed.due_date < date.today():
        score -= 0.4  # 過去日付は明らかにおかしい
    if parsed.tax_rate not in (0.0, 0.08, 0.1):
        score -= 0.2  # 日本の標準税率以外は要注意
 
    return max(0.0, score)
 
# 使用例
data = {"amount": 12000, "tax_rate": 0.1, "due_date": "2026-06-15"}
v_score = verifiability_score(data, expected_amount_range=(1000, 1_000_000))
# 期待出力: 1.0（全ルール通過）

# 自信スコア計算: 過去ログの類似タスク承認率を返す
import numpy as np
from typing import List, Tuple
 
class HistoricalScorer:
    """
    過去の (タスク埋め込み, 承認フラグ) のログを保持し、
    新規タスクとの類似度上位 K 件の承認率を返す。
    """
    def __init__(self, top_k: int = 10):
        self.embeddings: List[np.ndarray] = []
        self.approvals: List[bool] = []
        self.top_k = top_k
 
    def add_record(self, task_embedding: np.ndarray, approved: bool) -> None:
        self.embeddings.append(task_embedding)
        self.approvals.append(approved)
 
    def historical_score(self, new_task_embedding: np.ndarray) -> float:
        if len(self.embeddings) < self.top_k:
            # 学習データが少ないうちは中立スコア
            return 0.5
        sims = [
            np.dot(new_task_embedding, e) /
            (np.linalg.norm(new_task_embedding) * np.linalg.norm(e) + 1e-9)
            for e in self.embeddings
        ]
        top_indices = np.argsort(sims)[-self.top_k:]
        approval_rate = np.mean([self.approvals[i] for i in top_indices])
        return float(approval_rate)
 
# 使用例
scorer = HistoricalScorer(top_k=10)
# 起動時に過去 1000 件のログから埋め込みを再生成して投入する想定
# scorer.add_record(emb, approved=True) を 1000 回呼ぶ
new_emb = np.random.rand(1024)  # 実際は埋め込みモデルで生成
h_score = scorer.historical_score(new_emb)
# 期待出力: 学習データが揃っていれば 0.7〜0.95 程度（タスク種別による）

# 合成自信スコア
def composite_confidence(
    self_eval: float,
    verifiability: float,
    historical: float,
    weights: tuple = (0.3, 0.4, 0.3),
) -> float:
    """
    重み付き合成スコア。
    verifiability に最も重みを置くのは、決定論的で改ざん耐性があるため。
    """
    w1, w2, w3 = weights
    composite = w1 * self_eval + w2 * verifiability + w3 * historical
    # ただし、verifiability が 0 のときは強制的に低スコアに丸める
    if verifiability == 0.0:
        composite = min(composite, 0.2)
    return composite
 
# 使用例
final_score = composite_confidence(
    self_eval=0.9,
    verifiability=1.0,
    historical=0.85,
)
# 期待出力: 0.3*0.9 + 0.4*1.0 + 0.3*0.85 = 0.925

# 段階的承認の振り分け
from enum import Enum
 
class Decision(Enum):
    AUTO_APPROVE = "auto_approve"
    HUMAN_REVIEW = "human_review"
    AUTO_REJECT = "auto_reject"
 
def decide(
    confidence: float,
    upper_threshold: float = 0.85,
    lower_threshold: float = 0.55,
) -> Decision:
    if confidence >= upper_threshold:
        return Decision.AUTO_APPROVE
    if confidence >= lower_threshold:
        return Decision.HUMAN_REVIEW
    return Decision.AUTO_REJECT
 
# 使用例
for c in [0.92, 0.70, 0.40]:
    print(c, decide(c))
# 期待出力:
# 0.92 Decision.AUTO_APPROVE
# 0.70 Decision.HUMAN_REVIEW
# 0.40 Decision.AUTO_REJECT

# 閾値キャリブレーション: ROC 曲線から最適閾値を求める
import numpy as np
 
def calibrate_upper_threshold(
    scores: list, true_labels: list, target_precision: float = 0.95
) -> float:
    """
    auto-approve に流すスコアの上限を、目標精度から逆算する。
    true_labels: 1 = 後で人が見て正解だった, 0 = 不正解だった
    target_precision: auto-approve した中での正解率目標
    """
    pairs = sorted(zip(scores, true_labels), key=lambda x: -x[0])
    cum_correct = 0
    cum_total = 0
    for s, label in pairs:
        cum_total += 1
        cum_correct += label
        precision = cum_correct / cum_total
        if precision < target_precision and cum_total >= 20:
            # 直前のスコアが上限
            return s + 0.01
    return 0.5  # データ不足
 
# 使用例（実データ想定）
historical_scores = [0.95, 0.90, 0.88, 0.85, 0.80, 0.70, 0.55, 0.40]
historical_labels = [1, 1, 1, 1, 0, 0, 0, 0]
upper = calibrate_upper_threshold(
    historical_scores, historical_labels, target_precision=0.95
)
# 期待出力: 0.86 前後（精度 95% を維持できる最低スコア）

# Antigravity Agent から呼ばれる graduated approval サービス
import json
import logging
from dataclasses import dataclass
 
logger = logging.getLogger("graduated-approval")
 
@dataclass
class ApprovalResult:
    decision: Decision
    confidence: float
    output: dict
    reason: str
 
def graduated_approval_pipeline(
    task: str,
    raw_output: str,
    expected_amount_range: tuple = None,
) -> ApprovalResult:
    """
    Antigravity Agent の出力を受け取り、振り分け結果を返すメインパイプライン。
    """
    # Step 1: パースを試みる（パース不能 = 即拒否）
    try:
        output_dict = json.loads(raw_output)
    except json.JSONDecodeError:
        return ApprovalResult(
            Decision.AUTO_REJECT, 0.0, {}, "JSON パース失敗"
        )
 
    # Step 2: 3 つのスコアを並行算出
    s_eval = self_evaluation_score(task, raw_output)
    v_score = verifiability_score(output_dict, expected_amount_range)
    # 過去ログスコアは事前に初期化済みの scorer を使用する想定
    # h_score = scorer.historical_score(embed_task(task))
    h_score = 0.7  # サンプル値
 
    # Step 3: 合成スコア
    confidence = composite_confidence(s_eval, v_score, h_score)
 
    # Step 4: 振り分け
    decision = decide(confidence)
 
    # Step 5: 監査ログに必ず残す（後でキャリブレーションに使う）
    logger.info(json.dumps({
        "task": task[:200],
        "self_eval": s_eval,
        "verifiability": v_score,
        "historical": h_score,
        "confidence": confidence,
        "decision": decision.value,
    }))
 
    return ApprovalResult(
        decision=decision,
        confidence=confidence,
        output=output_dict,
        reason=f"s={s_eval:.2f} v={v_score:.2f} h={h_score:.2f}",
    )
 
# 使用例: Antigravity Agent の Walkthrough から呼ばれる想定
result = graduated_approval_pipeline(
    task="請求書PDFから金額・税率・支払期限を抽出して",
    raw_output='{"amount": 12000, "tax_rate": 0.1, "due_date": "2026-06-15"}',
    expected_amount_range=(1000, 1_000_000),
)
print(result.decision, result.confidence)
# 期待出力: Decision.AUTO_APPROVE 0.91 (前後)

# 緊急停止スイッチ付きの振り分け
import os
 
def decide_with_kill_switch(
    confidence: float,
    upper_threshold: float = 0.85,
    lower_threshold: float = 0.55,
    force_human: bool = False,
) -> Decision:
    # 外部フラグ（環境変数 / KV / フィーチャーフラグ）で一括上書き
    kill_switch = force_human or os.environ.get("APPROVAL_FORCE_HUMAN") == "1"
    if kill_switch:
        # auto-reject だけは破棄してよい（再試行に回るだけ）
        if confidence < lower_threshold:
            return Decision.AUTO_REJECT
        # それ以外は確信度に関わらず人間へ
        return Decision.HUMAN_REVIEW
    return decide(confidence, upper_threshold, lower_threshold)
 
# 使用例: モデル切替直後の1日だけ全件を人間に戻す
os.environ["APPROVAL_FORCE_HUMAN"] = "1"
print(decide_with_kill_switch(0.97))   # Decision.HUMAN_REVIEW
print(decide_with_kill_switch(0.30))   # Decision.AUTO_REJECT