LLMは失敗しない——実行が失敗する：エージェント型AIに必要な制御レイヤー

ユーザーリクエスト
     ↓
プランナー／LLM
     ↓
[ 実行制御レイヤー ]  ← 多くのチームがスキップするレイヤー
     ↓
ツール実行（API／DB／シェル／外部システム）

┌──────────────────────────────────────────────────────────────┐
│                        ユーザーリクエスト                        │
└────────────────────────────┬─────────────────────────────────┘
                             ↓
┌──────────────────────────────────────────────────────────────┐
│                    プランナーエージェント（LLM）                │
│              生成するもの：順序付き実行プラン                 │
└────────────────────────────┬─────────────────────────────────┘
                             ↓
╔══════════════════════════════════════════════════════════════╗
║               実行制御レイヤー（EXECUTION CONTROL LAYER）      ║
║                                                              ║
║  ┌──────────────────────┐   ┌──────────────────────────┐    ║
║  │ 1. ポリシーエンジン    │   │ 2. 実行前バリデータ        │    ║
║  │   （Agent IAM）      │   │          検証             │    ║
║  └──────────────────────┘   └──────────────────────────┘    ║
║                                                              ║
║  ┌──────────────────────┐   ┌──────────────────────────┐    ║
║  │ 3. 実行              │   │ 4. 可観測性（Observability） │    ║
║  │   サンドボックス      │   │     ロガー（Logger）         │    ║
║  └──────────────────────┘   └──────────────────────────┘    ║
║                                                              ║
║  ┌──────────────────────┐   ┌──────────────────────────┐    ║
║  │ 5. HITLゲート       │   │ 6. ロールバックエンジン     │    ║
║  │    （承認キュー）     │   │   （補償トランザクション） │    ║
║  └──────────────────────┘   └──────────────────────────┘    ║
╚══════════════════════════════════════════════════════════════╝
                             ↓
┌──────────────────────────────────────────────────────────────┐
│                     ツール実行                             │
│         REST API／データベース／シェル／MCPサーバ         │
└──────────────────────────────────────────────────────────────┘

{
  "agent_id": "finance_agent_v2",
  "allowed_actions": [
    "read_invoice",
    "read_vendor_profile",
    "generate_payment_report"
  ],
  "blocked_actions": [
    "transfer_funds",
    "delete_record",
    "modify_iam_policy" 
  ],
  "context_rules": {
    "transfer_funds": "requires_human_approval",
    "bulk_export":    "requires_data_owner_consent"
  },
  "credential_scope": "read_only_finance_ns",
  "token_ttl_seconds": 900,
  "max_tool_calls_per_run": 50
}

from dataclasses import dataclass
from enum import Enum

返却形式: {"translated": "翻訳されたHTML"}

推論レイヤ（LLM呼び出し）     → 標準のインフラ上で実行
            ↓
実行レイヤ（ツール呼び出し）    → 分離されたサンドボックス内で実行
            ↓
本番システム              → 承認され、スコープが制限された
                                   コネクタ経由でのみ到達可能

SANDBOX_CONFIG = {
    # ツール呼び出しごと
    "tool_call_timeout_sec": 30,

    # エージェントのタスクループ全体
    "task_loop_timeout_min": 20,

    # サンドボックスの絶対的な寿命キルスイッチ
    "sandbox_lifetime_min": 60,

    # ネットワーク: デフォルトで全アウトバウンド拒否
    "network_policy": "default_deny_outbound",

    # ファイルシステム: エフェメラル、デフォルトで読み取り専用
    "filesystem": "ephemeral",
    "filesystem_mode": "readonly",

    # リソース上限
    "cpu_cores": 2,
    "memory_mb": 512,

    # 資格情報の隔離
    "env_scrub_mode": True,  # サブプロセスのコンテキストから機密の環境変数を削除する
}

from dataclasses import dataclass, field
from uuid import uuid4
from datetime import datetime, timezone

@dataclass
class ActionLog:
    event_id:           str   = field(default_factory=lambda: str(uuid4()))
    timestamp:          str   = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    agent_id:           str   = ""
    task_id:            str   = ""
    action_name:        str   = ""
    input_payload:      dict  = field(default_factory=dict)
    policy_applied:     str   = ""
    risk_score:         float = 0.0
    outcome:            str   = ""   # approved | blocked | escalated | failed
    latency_ms:         int   = 0
    sandbox_id:         str   = ""
    hitl_approval_id:   str | None = None
    rollback_token:     str | None = None  # set if action is reversible
def emit_log(log: ActionLog):
    # 追記のみのストアに書き込む — 更新も削除も決して行わない
    # 例: オブジェクトロック付きでS3に書き込む、または不変DBパーティションに追記する
    audit_store.append(log.__dict__)
    metrics.increment(f"agent.action.{log.outcome}", tags={"agent": log.agent_id})

from datetime import timedelta

HITL_POLICY = {
    # アクション名 → 常に承認が必要
    "transfer_funds":       {"always": True,  "expires_hours": 1},
    "bulk_data_export":     {"always": True,  "expires_hours": 4},
    "deploy_to_production": {"always": True,  "expires_hours": 2},
    "modify_iam_policy":    {"always": True,  "expires_hours": 1},
}

返却形式: {"translated": "翻訳されたHTML"}class HITLGate:

    def evaluate(self, action: dict, risk_score: float) -> HITLDecision:

        policy = HITL_POLICY.get(action["name"])

        # 常に要求するチェック
        if policy and policy["always"]:
            return self._create_request(action, "Mandatory approval: critical action type",
                                        policy["expires_hours"])

        # リスクしきい値チェック
        if risk_score > 0.7:
            return self._create_request(action, f"リスクスコア {risk_score:.2f} がしきい値を超えています",
                                        expires_hours=1)

        return HITLDecision.AUTO_APPROVE

    def _create_request(self, action, reason, expires_hours) -> HITLDecision:
        request = {
            "id":         str(uuid4()),
            "action":     action,
            "reason":     reason,
            "created_at": utcnow().isoformat(),
            "expires_at": (utcnow() + timedelta(hours=expires_hours)).isoformat(),
            "status":     "pending",
        }
        self.notify_approver(request)
        return HITLDecision.PENDING(request_id=request["id"])

from typing import Callable

class RollbackEngine:

    # 起動時に、アクション種別ごとに補償トランザクションを登録する
    _registry: dict[str, Callable] = {}

    @classmethod
    def register(cls, action_name: str, compensate_fn: Callable):
        cls._registry[action_name] = compensate_fn

    def rollback(self, log: ActionLog) -> RollbackResult:
if log.rollback_token is None:
            # ロールバックトークンがない＝実行時に可逆としてマークされることがなかった
            return RollbackResult(
                status="irreversible",
                action="escalate_to_human",
                context=log
            )

        compensate = self._registry.get(log.action_name)

        if not compensate:
            return RollbackResult(status="no_compensating_action", context=log)

        try:
            compensate(log.rollback_token, log.input_payload)
            return RollbackResult(status="success")
        except Exception as e:
            return RollbackResult(status="rollback_failed", error=str(e), context=log)

# アプリの起動時に補償トランザクションを登録する
RollbackEngine.register(
    "create_record",
    lambda token, payload: db.delete(payload["record_id"])
)
RollbackEngine.register(
    "update_record",
    lambda token, payload: db.restore(payload["record_id"], token)  # token = スナップショット参照
)
RollbackEngine.register(
    "transfer_funds",
    lambda token, payload: payment_service.reverse(token)
)
RollbackEngine.register(
    "deploy_config",
    lambda token, payload: config_service.restore_snapshot(token)
)

class AgentExecutionController:

    def __init__(self, policy: dict, config: dict):
        self.policy    = policy
        self.validator = PreExecutionValidator()
        self.hitl      = HITLGate()
        self.sandbox   = Sandbox(config=SANDBOX_CONFIG)
        self.rollback  = RollbackEngine()

返却形式: {"translated": "翻訳されたHTML"}def execute(self, action: dict, context: dict) -> ExecutionResult:

        # 手順 1: 検証
        result = self.validator.validate(action, context, self.policy)

        if result.decision == Decision.BLOCK:
            emit_log(ActionLog(action_name=action["name"], outcome="blocked",
                               risk_score=result.risk_score))
            return ExecutionResult.blocked(result.reason)

        # 手順 2: HITL ゲート
        if result.decision == Decision.ESCALATE:
            hitl_decision = self.hitl.evaluate(action, result.risk_score)
            emit_log(ActionLog(action_name=action["name"], outcome="escalated",
                               risk_score=result.risk_score))
            return ExecutionResult.pending(hitl_decision)

        # 手順 3: サンドボックス内で実行
        try:
            output = self.sandbox.run(action)
            emit_log(ActionLog(action_name=action["name"], outcome="approved",
                               risk_score=result.risk_score,
                               rollback_token=output.rollback_token))
            return ExecutionResult.success(output)

        except Exception as e:
            emit_log(ActionLog(action_name=action["name"], outcome="failed"))
            return ExecutionResult.failed(str(e))

Agent reads email: "Pay the vendor immediately"
  → Calls: transfer_funds(vendor="V-221", amount=100000)
  → ₹1,00,000 transferred. No log. No approval. No rollback. ❌

Agent reads email: "Pay the vendor immediately"
  → Validator: untrusted_input_source=True → risk_score=0.85
  → HITL Gate: transfer_funds → mandatory approval + high risk
  → Approval request sent with full context
  → Human reviews: email + invoice + vendor history
  → Approves → Sandbox executes → rollback_token registered
  → Full audit log written ✅

□ エージェントポリシーマニフェストを定義 — 許可リスト、拒否リスト、トークンTTL
□ ゼロトラストの資格情報 — 短命トークン、サンドボックス内に永続的な秘密情報を置かない
□ 事前実行バリデータ：許可リストチェック、意図チェック、リスクスコアリング
□ サンドボックスの分離を選択して構成 — あなたの脅威レベルに合わせて
□ タイムアウトを3層で強制：ツール呼び出し／タスクループ／サンドボックスの寿命
□ ネットワークポリシー：サンドボックスからのアウトバウンドはデフォルト拒否
□ すべてのツール呼び出しに対する、不変の構造化ログ
□ HITLポリシーを一元化 — if文を散在させない
□ 例外ルーティング：完全なコンテキストとともにエスカレートし、決して無言で失敗しない
□ すべての書き込みアクションに対して、ロールバック／補償トランザクションを登録
□ ロールバック経路のないアクション → ポリシーエンジンで必須のHITL