🕸️ 技術調査 - エージェントワークフローの構造カバレッジテスト

# expected_graph.yaml — 型付き調整グラフ宣言 (実装案/例)
agents:
  - id: triage_agent
    description: "ユーザー入力を分類し適切なエージェントに委譲"
  - id: sales_agent
    description: "製品・購入に関する問い合わせを処理"
  - id: refund_agent
    description: "返金・クレームを処理"

# C2: 許可ツールエッジ
allowed_tool_edges:
  - { agent: triage_agent, tool: get_user_info }
  - { agent: sales_agent,  tool: get_product_catalog }
  - { agent: sales_agent,  tool: create_order }
  - { agent: refund_agent, tool: issue_refund }
  - { agent: refund_agent, tool: get_order_history }

# C3: 制限ツールエッジ (通ってはいけない組み合わせ)
restricted_tool_edges:
  - { agent: triage_agent, tool: issue_refund, reason: "triage は返金処理権限なし" }
  - { agent: sales_agent,  tool: issue_refund, reason: "sales は返金権限なし" }

# C4: 委譲エッジ (handoff)
handoff_edges:
  - { from: triage_agent, to: sales_agent }
  - { from: triage_agent, to: refund_agent }

# obligations.py — YAML から C1-C4 義務を生成 (実装案/例)
from dataclasses import dataclass
from typing import Literal
import yaml

ObligationType = Literal["C1_agent", "C2_allowed_tool", "C3_restricted_tool", "C4_handoff"]

@dataclass
class Obligation:
    type: ObligationType
    agent: str
    target: str  # tool 名 or 委譲先 agent 名
    reason: str = ""

def load_obligations(graph_yaml_path: str) -> list[Obligation]:
    with open(graph_yaml_path) as f:
        g = yaml.safe_load(f)
    obligations: list[Obligation] = []
    for a in g.get("agents", []):                      # C1
        obligations.append(Obligation("C1_agent", a["id"], a["id"]))
    for e in g.get("allowed_tool_edges", []):          # C2
        obligations.append(Obligation("C2_allowed_tool", e["agent"], e["tool"]))
    for e in g.get("restricted_tool_edges", []):       # C3
        obligations.append(Obligation("C3_restricted_tool", e["agent"], e["tool"], e.get("reason", "")))
    for e in g.get("handoff_edges", []):               # C4
        obligations.append(Obligation("C4_handoff", e["from"], e["to"]))
    return obligations

# instrumentation.py — OTel GenAI span を emit する薄いラッパー (実装案/例)
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import SpanKind

exporter = InMemorySpanExporter()  # 本番では OTLPSpanExporter に差し替え
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agent-coverage")

def invoke_agent_span(agent_name: str, conversation_id: str):
    """invoke_agent span。remote=CLIENT / local framework 実行=INTERNAL"""
    return tracer.start_as_current_span(
        f"invoke_agent {agent_name}", kind=SpanKind.CLIENT,
        attributes={
            "gen_ai.operation.name": "invoke_agent",
            "gen_ai.agent.name": agent_name,
            "gen_ai.conversation.id": conversation_id,
        },
    )

def execute_tool_span(agent_name: str, tool_name: str, call_args: str = ""):
    """execute_tool span の SpanKind は仕様上 INTERNAL"""
    return tracer.start_as_current_span(
        f"execute_tool {tool_name}", kind=SpanKind.INTERNAL,
        attributes={
            "gen_ai.operation.name": "execute_tool",
            "gen_ai.agent.name": agent_name,
            "gen_ai.tool.name": tool_name,
            "gen_ai.tool.call.arguments": call_args,
        },
    )

[期待グラフ YAML] + [テストシナリオ] -> [エージェント実行] -> [OTel span 収集]
                                                          -> [カバレッジ集計]
                                                          -> [カバレッジ率 / 制限違反 / 未到達一覧]

# coverage_report.py — span 集合 × 期待グラフ → カバレッジ算出 (実装案/例)
from dataclasses import dataclass
from obligations import load_obligations, Obligation

@dataclass
class CoverageResult:
    c1_covered: list; c1_missing: list
    c2_covered: list; c2_missing: list
    c3_violations: list   # 出てはいけないのに出た
    c3_confirmed: list    # 出てはいけなくて出なかった (正常)
    c4_covered: list; c4_missing: list

def compute_coverage(obligations, observed_agents, observed_tool_edges, observed_handoffs):
    r = CoverageResult([], [], [], [], [], [], [], [])
    for ob in obligations:
        if ob.type == "C1_agent":
            (r.c1_covered if ob.agent in observed_agents else r.c1_missing).append(ob.agent)
        elif ob.type == "C2_allowed_tool":
            pair = (ob.agent, ob.target)
            (r.c2_covered if pair in observed_tool_edges else r.c2_missing).append(pair)
        elif ob.type == "C3_restricted_tool":
            pair = (ob.agent, ob.target)
            (r.c3_violations if pair in observed_tool_edges else r.c3_confirmed).append(pair)
        elif ob.type == "C4_handoff":
            pair = (ob.agent, ob.target)
            (r.c4_covered if pair in observed_handoffs else r.c4_missing).append(pair)
    return r

def extract_observations_from_spans(spans):
    """FinishedSpan リストから観測セットを返す。親子スパン関係から handoff を近似検出"""
    observed_agents, observed_tool_edges, observed_handoffs = set(), set(), set()
    span_agent: dict = {}
    for span in spans:
        attrs = dict(span.attributes or {})
        op, agent = attrs.get("gen_ai.operation.name", ""), attrs.get("gen_ai.agent.name", "")
        if op == "invoke_agent" and agent:
            observed_agents.add(agent)
            span_agent[span.context.span_id] = agent
            parent_id = span.parent.span_id if span.parent else None
            if parent_id and parent_id in span_agent:
                observed_handoffs.add((span_agent[parent_id], agent))
        elif op == "execute_tool":
            tool = attrs.get("gen_ai.tool.name", "")
            if agent and tool:
                observed_tool_edges.add((agent, tool))
    return observed_agents, observed_tool_edges, observed_handoffs

# scenario_generator.py — DSPy でカバレッジ駆動シナリオ生成 (実装案/例)
import dspy
from obligations import Obligation

class ScenarioFromObligation(dspy.Signature):
    """未カバー構造義務を満たすテストシナリオ (ユーザー入力) を生成する。"""
    obligation_type: str = dspy.InputField(desc="C1/C2/C3/C4 のいずれか")
    agent_name: str = dspy.InputField(desc="対象エージェント名")
    target: str = dspy.InputField(desc="ツール名 or 委譲先エージェント名")
    workflow_description: str = dspy.InputField(desc="ワークフロー全体の説明")
    scenario: str = dspy.OutputField(desc="対象エージェント起動 ツール実行を誘発する入力テキスト")

class CoverageDrivenScenarioGenerator(dspy.Module):
    def __init__(self):
        self.generate = dspy.ChainOfThought(ScenarioFromObligation)
    def forward(self, obligation: Obligation, workflow_desc: str) -> str:
        pred = self.generate(
            obligation_type=obligation.type, agent_name=obligation.agent,
            target=obligation.target, workflow_description=workflow_desc)
        return pred.scenario

# temperature=0 推奨だが完全な決定性は保証されない (後述のベストプラクティス参照)
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini", temperature=0))

# test_workflow_coverage.py — pytest + OTel カバレッジ検証 (実装案/例)
import pytest
from instrumentation import exporter
from obligations import load_obligations
from coverage_report import compute_coverage, extract_observations_from_spans

GRAPH_YAML = "expected_graph.yaml"

@pytest.fixture(autouse=True)
def clear_spans():
    exporter.clear(); yield; exporter.clear()

@pytest.mark.parametrize("scenario", [
    "商品Aの在庫状況を教えてください",
    "先週の注文を返金してください",
    "プランのアップグレードを検討しています",
])
def test_no_c3_violation(scenario):
    run_workflow(scenario)  # OpenAI Agents SDK の runner.run() 等を呼ぶ
    obligations = load_obligations(GRAPH_YAML)
    obs = extract_observations_from_spans(exporter.get_finished_spans())
    result = compute_coverage(obligations, *obs)
    assert result.c3_violations == [], f"C3 制限違反: {result.c3_violations}"

# promptfooconfig.yaml — 構造カバレッジ検査の assert 例 (実装案/例)
description: "エージェントワークフロー 構造カバレッジテスト"
providers:
  - id: openai:gpt-4o-mini
    config: { temperature: 0 }
prompts: ["{{scenario}}"]
tests:
  - description: "triage_agent が issue_refund を呼ばない C3 制限"
    vars: { scenario: "すぐに返金してください triage に直接送る" }
    assert:
      - type: not-contains
        value: "execute_tool:triage_agent:issue_refund"

// Claude .claude/settings.json — deny ルール (settings.json は JSON 形式)
{
  "permissions": {
    "deny": [
      "Bash(rm -rf *)",
      "mcp__payment__charge"
    ]
  }
}

Phase 1 観測: trace を Phoenix/Langfuse に流して coverage/violation をダッシュボード表示。閾値なし
Phase 2 ソフト: CI で coverage < 60% / 制限違反 > 0 を warning 表示。fail はしない
Phase 3 ハード: ゲーミングが発生しないことを確認してから exit 1 を有効化