🧪 非決定的な AI エージェントの検証 - PTA + Dominator で Trust Layer を構築する

成功トレース例: A -> X -> B -> Y -> C
essential states: A, B, C (X, Y はオプション変動として無視)
-> A, B, C が規定順序で出現すれば合格

version: "3.9"
services:
  agent-runner:
    image: mcr.microsoft.com/devcontainers/universal:2-linux
    environment:
      - DISPLAY=:99
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./traces:/workspace/traces
      - ./scripts:/workspace/scripts
    command: >
      bash -c "
        Xvfb :99 -screen 0 1280x800x24 &
        sleep 1
        python /workspace/scripts/collect_traces.py
      "
    shm_size: "2gb"

traces/
├── run_001/
│   ├── states/          # スクリーンショット (PNG)
│   ├── actions.jsonl    # アクションログ
│   └── metadata.json    # 実行メタデータ
├── run_002/
│   └── ...
└── reference/           # 学習用成功トレース (3-10 件)

pip install imagehash scikit-image networkx openai pillow

# pta_builder.py — 実装案
import networkx as nx
import imagehash
import json
from pathlib import Path
from PIL import Image
from skimage.metrics import structural_similarity as ssim
import numpy as np
import openai

class State:
    def __init__(self, state_id: str, screenshot_path: str, action: dict):
        self.state_id = state_id
        self.screenshot_path = screenshot_path
        self.action = action
        self.phash = imagehash.phash(Image.open(screenshot_path))

class PTABuilder:
    PHASH_THRESHOLD = 10
    SSIM_THRESHOLD = 0.95

    def __init__(self, llm_client: openai.OpenAI, model: str = "gpt-4o"):
        self.graph = nx.DiGraph()
        self.llm_client = llm_client
        self.model = model
        self._node_counter = 0

    def add_trace(self, trace_dir: Path) -> None:
        actions = self._load_actions(trace_dir / "actions.jsonl")
        states = sorted((trace_dir / "states").glob("*.png"))
        prev_node = "START"
        if not self.graph.has_node("START"):
            self.graph.add_node("START", label="start")
        for img_path, action in zip(states, actions):
            state = State(f"s{self._node_counter}", str(img_path), action)
            merged_node = self._find_equivalent_node(state)
            if merged_node is None:
                self.graph.add_node(state.state_id, state=state)
                merged_node = state.state_id
                self._node_counter += 1
            self.graph.add_edge(prev_node, merged_node, action=action)
            prev_node = merged_node
        self.graph.add_edge(prev_node, "END")

    def _tier1_visual_match(self, a: State, b: State) -> bool:
        if abs(a.phash - b.phash) <= self.PHASH_THRESHOLD:
            return True
        img_a = np.array(Image.open(a.screenshot_path).convert("L"))
        img_b = np.array(Image.open(b.screenshot_path).convert("L"))
        if img_a.shape == img_b.shape:
            score, _ = ssim(img_a, img_b, full=True)
            if score >= self.SSIM_THRESHOLD:
                return True
        return False

# dominator_analysis.py — 実装案
import networkx as nx

def extract_essential_states(graph: nx.DiGraph, start: str = "START") -> list[str]:
    idom = nx.immediate_dominators(graph, start)
    terminal_states = [n for n in graph.nodes if graph.out_degree(n) == 0]
    essential = set()
    for terminal in terminal_states:
        current = terminal
        while current != start:
            essential.add(current)
            current = idom.get(current, start)
    return [n for n in nx.topological_sort(graph) if n in essential]

# validate.py — 実装案
def topological_subsequence_match(test_states, essential_states, graph, threshold=1.0):
    matched = []
    ptr = 0
    for test_state in test_states:
        if ptr >= len(essential_states):
            break
        target_node = essential_states[ptr]
        target_state = graph.nodes[target_node].get("state")
        if target_state and _states_match(test_state, target_state, graph):
            matched.append(target_node)
            ptr += 1
    missing = essential_states[ptr:]
    coverage = len(matched) / len(essential_states) if essential_states else 0.0
    passed = coverage >= threshold and len(missing) == 0
    return {
        "passed": passed,
        "coverage": coverage,
        "matched_states": matched,
        "missing_states": missing,
        "failure_reason": None if passed else f"Missing essential states: {missing}",
    }

# .github/workflows/agent-validation.yml — 実装案
name: Agent Behavior Validation
on:
  pull_request:
    branches: [main]
  schedule:
    - cron: "0 2 * * *"

jobs:
  agent-validation:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      DISPLAY: ":99"
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install imagehash scikit-image networkx openai pillow pytest deepeval
      - run: |
          sudo apt-get install -y xvfb
          Xvfb :99 -screen 0 1280x800x24 &
      - run: python scripts/build_pta.py --trace-dir traces/reference --output pta_model.pkl
      - run: |
          for i in 1 2 3 4 5; do
            python agent.py --task search_flow --output traces/ci_run_${i}/ || true
          done
      - run: |
          pytest tests/test_agent_validation.py \
            --pta-model pta_model.pkl \
            --trace-dir traces/ci_run_* \
            --min-pass-rate 0.8 -v
      - if: always()
        uses: actions/upload-artifact@v4
        with:
          name: agent-traces
          path: traces/ci_run_*/

# acceptance_criteria.yml
scenario: "VS Code 検索フロー"
learning:
  trace_dir: "traces/reference"
  num_traces: 5
trials: 5
min_pass: 4
essential_states:
  - id: search_dialog_opened
    description: "検索ダイアログが開いた状態"
  - id: keyword_entered
    description: "検索キーワードが入力された状態"
  - id: search_results_displayed
    description: "検索結果一覧が表示された状態"
side_effects:
  file_reads:
    - pattern: "**/*.{ts,md}"
  api_calls:
    max_count: 1
invariants:
  - "no_file_modification"
  - "no_external_network_call_outside_allowlist"
  - "no_pii_in_response"
audit:
  record_full_trajectory: true
  retention_days: 90

[Stable Release] -> 1. Smoke run (5-10回) -> 2. Trace 正規化
-> 3. PTA 再構築 -> 4. 参照モデルを Git 管理 -> 5. CI ベースライン切替

strategy:
  fail_on_non_essential: true
  pass_threshold: 0.80
  max_trials: 5
  adaptive_sampling: true
  timeout_per_trial: 120s

retry_policy:
  on_llm_timeout: retry_once
  on_essential_miss: no_retry
  on_flaky_optional: ignore

PTA + Dominator        -> essential states 到達 (構造的)
LangSmith trajectory   -> ツール呼び出し列の部分一致
Side-effect monitoring -> DB diff / HTTP egress
Property assertions    -> invariants (PII 非漏洩、削除ツール未呼出等)

Audit Log Layer  -- 全 trajectory を完全保存 (改竄耐性、タイムスタンプ付き)
PTA Layer        -- essential states のみ (CI 検証・表示用)

S+ (成功トレース) -> PTA 構築 -> dominator tree
S- (失敗トレース) -> regression set -> スナップショット比較・root cause 分析