🧪 業務エージェント評価を 4 シグナル採点で組み直す: Claw-Eval-Live を実務に落とす

# 実装案: task_snapshot_v1.yaml
version: "1"
snapshot_date: "2026-05-05"

tasks:
  - task_id: "crm-001"
    family: "service_assistance"
    execution_surface: "multi_system"
    description: "CRM の顧客ステータスを更新し、メールで担当者に通知する"
    expected_state_changes:
      - system: "crm"
        entity: "customer"
        field: "status"
        before: "lead"
        after: "active"
      - system: "email"
        entity: "outbox"
        field: "sent_count"
        delta: 1
    scoring_pattern: "operation_verification"
    weight:
      data_retrieval: 0.15      # 15-20% (論文 §3.3)
      data_accuracy: 0.60       # 40-60%
      action_verification: 0.25 # 10-20%
    pass_threshold: 0.80        # τ=0.80 (論文 §2.3)
    artifact_required: true
    hitl_required: false

  - task_id: "ws-repair-001"
    family: "workspace_repair"
    execution_surface: "terminal"
    description: "壊れた Python 仮想環境を診断して再構築する"
    expected_state_changes:
      - system: "filesystem"
        entity: "venv"
        field: "health_check_exit_code"
        after: 0
    scoring_pattern: "script_first"
    weight:
      data_retrieval: 0.10
      data_accuracy: 0.30
      action_verification: 0.60
    pass_threshold: 0.80
    artifact_required: true

# 実装案: otel_agent_tracer.py
# 参考: https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-agent-spans/

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
import json, time

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("claw-eval-agent", "0.1.0")


def run_task_with_signals(task_id: str, agent_fn, task_input: dict) -> dict:
    """エージェント実行を trace でラップし 4 シグナルを収集する。"""
    with tracer.start_as_current_span("agent.task") as span:
        span.set_attribute("gen_ai.operation.name", "agent.execute")
        span.set_attribute("gen_ai.agent.id", task_id)

        state_before = capture_state_snapshot(task_input["expected_state_changes"])
        t_start = time.time()
        result = agent_fn(task_input)
        elapsed = time.time() - t_start
        state_after = capture_state_snapshot(task_input["expected_state_changes"])

        span.set_attribute("eval.state_before", json.dumps(state_before))
        span.set_attribute("eval.state_after", json.dumps(state_after))
        span.set_attribute("eval.elapsed_sec", elapsed)
        span.set_attribute("eval.tool_calls", json.dumps(result.get("tool_calls", [])))
        artifact_path = save_artifact(task_id, result.get("artifact"))
        span.set_attribute("eval.artifact_path", str(artifact_path))

        return {
            "state_before": state_before,
            "state_after": state_after,
            "tool_calls": result.get("tool_calls", []),
            "artifact_path": artifact_path,
            "elapsed_sec": elapsed,
        }

# 実装案: langfuse_eval_tracer.py
from langfuse import Langfuse
import uuid

lf = Langfuse()


def trace_task_with_langfuse(task_id: str, agent_fn, task_input: dict) -> dict:
    session_id = str(uuid.uuid4())
    tr = lf.trace(
        name=f"eval-task-{task_id}",
        session_id=session_id,
        tags=["claw-eval", task_input["family"], task_input["scoring_pattern"]],
        metadata={"task_id": task_id, "snapshot_date": task_input.get("snapshot_date")},
    )

    state_span = tr.span(name="state-capture-before")
    state_before = capture_state_snapshot(task_input["expected_state_changes"])
    state_span.end(output=state_before)

    agent_span = tr.span(name="agent-execution")
    result = agent_fn(task_input)
    agent_span.end(output={
        "tool_calls": result.get("tool_calls", []),
        "artifact_path": result.get("artifact_path"),
    })

    state_after_span = tr.span(name="state-capture-after")
    state_after = capture_state_snapshot(task_input["expected_state_changes"])
    state_after_span.end(output=state_after)

    return {"trace_id": tr.id, "session_id": session_id}

# 実装案: scoring_engine.py
from dataclasses import dataclass

@dataclass
class TaskScore:
    task_id: str
    data_retrieval: float
    data_accuracy: float
    action_verification: float
    overall: float
    passed: bool


def score_task(task_def: dict, signals: dict) -> TaskScore:
    pattern = task_def["scoring_pattern"]

    if pattern == "script_first":
        action_score = score_state_diff_deterministic(
            task_def["expected_state_changes"],
            signals["state_before"], signals["state_after"])
        retrieval_score = score_tool_calls_deterministic(task_def, signals["tool_calls"])
        accuracy_score = retrieval_score
    elif pattern == "operation_verification":
        retrieval_score = score_tool_calls_deterministic(task_def, signals["tool_calls"])
        action_score = score_state_diff_deterministic(
            task_def["expected_state_changes"],
            signals["state_before"], signals["state_after"])
        accuracy_score = (retrieval_score + action_score) / 2
    else:  # analytic
        retrieval_score = score_tool_calls_deterministic(task_def, signals["tool_calls"])
        accuracy_det = score_numeric_entities(task_def, signals["state_after"])
        accuracy_llm = llm_judge_semantic(task_def, signals)
        accuracy_score = 0.7 * accuracy_det + 0.3 * accuracy_llm
        action_score = score_state_diff_deterministic(
            task_def["expected_state_changes"],
            signals["state_before"], signals["state_after"])

    w = task_def["weight"]
    overall = (w["data_retrieval"] * retrieval_score
               + w["data_accuracy"] * accuracy_score
               + w["action_verification"] * action_score)
    tau = task_def.get("pass_threshold", 0.80)
    return TaskScore(
        task_id=task_def["task_id"],
        data_retrieval=retrieval_score,
        data_accuracy=accuracy_score,
        action_verification=action_score,
        overall=overall,
        passed=overall >= tau,
    )


def score_state_diff_deterministic(expected, before: dict, after: dict) -> float:
    """期待する状態変更がすべて発生したか 0/1 で検証し平均を返す。"""
    if not expected:
        return 1.0
    scores = []
    for change in expected:
        key = f"{change['system']}.{change['entity']}.{change['field']}"
        after_val = after.get(key)
        if "after" in change:
            scores.append(1.0 if after_val == change["after"] else 0.0)
        elif "delta" in change:
            before_val = before.get(key, 0)
            scores.append(1.0 if (after_val - before_val) == change["delta"] else 0.0)
    return sum(scores) / len(scores)

# 実装案: Dockerfile.eval
FROM python:3.12-slim

RUN pip install langfuse opentelemetry-sdk opentelemetry-exporter-otlp \
    fakeredis faketime httpretty

ENV LD_PRELOAD=/usr/lib/x86_64-linux-gnu/faketime/libfaketime.so.1
ENV FAKETIME="2026-05-05 00:00:00"

COPY eval_mocks/ /app/eval_mocks/
COPY tasks/ /app/tasks/
COPY scoring_engine.py /app/

WORKDIR /app
CMD ["python", "-m", "pytest", "eval_suite/", "-v", "--tb=short"]

# 実装案: docker-compose.eval.yaml
services:
  eval-runner:
    build: { context: ., dockerfile: Dockerfile.eval }
    environment:
      EVAL_SEED: "42"
      FAKETIME: "2026-05-05 00:00:00"
      LANGFUSE_HOST: "http://langfuse:3000"
    depends_on: [langfuse, wiremock]
  langfuse:
    image: langfuse/langfuse:latest
    ports: ["3000:3000"]
  wiremock:
    image: wiremock/wiremock:latest
    volumes: ["./eval_mocks:/home/wiremock/mappings"]
    ports: ["8080:8080"]

# 実装案: refreshable_signal_layer.py
from dataclasses import dataclass
import yaml, pathlib, uuid, datetime

@dataclass
class DemandSignal:
    skill_name: str
    frequency_score: float   # 過去 30 日の出現頻度を正規化
    recency_score: float     # 直近 7 日の出現率
    salience: float          # 0.4 * frequency + 0.6 * recency (設計案)


class DemandSourceAdapter:
    """Slack / Jira / ServiceNow など、実装は差し替え可能。"""
    def get_top_n_skills(self, n: int = 50) -> list[DemandSignal]:
        raise NotImplementedError


class TaskRefresher:
    def __init__(self, adapter: DemandSourceAdapter, template_dir: str):
        self.adapter = adapter
        self.template_dir = pathlib.Path(template_dir)

    def refresh(self, dry_run: bool = False) -> list[dict]:
        signals = self.adapter.get_top_n_skills(50)
        new_tasks = []
        for sig in signals:
            if sig.salience < 0.30:
                continue
            template = self._load_template(sig.skill_name)
            if template is None:
                continue
            task = dict(template)
            task["task_id"] = f"{sig.skill_name}-{uuid.uuid4().hex[:6]}"
            task["generated_date"] = str(datetime.date.today())
            task["demand_salience"] = sig.salience
            new_tasks.append(task)
            if not dry_run:
                self._save(task)
        return new_tasks

    def _load_template(self, name: str):
        path = self.template_dir / f"{name}.yaml"
        return yaml.safe_load(path.read_text()) if path.exists() else None

    def _save(self, task: dict):
        path = self.template_dir / "generated" / f"{task['task_id']}.yaml"
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(yaml.dump(task, allow_unicode=True))

# 実装案: eval CLI
python -m claw_eval run \
  --task-dir ./tasks/ \
  --agent-module myagent.run \
  --output-dir ./eval_results/ \
  --seed 42

# 出力例
# task_id          retrieval  accuracy  action  overall  passed
# crm-001          0.90       0.85      1.00    0.884    PASS
# ws-repair-001    0.80       0.75      0.50    0.668    FAIL
# TOTAL (n=105)                                          66.7%
# Pass rate (τ=0.80): 66.7%  Overall: 83.6

# 実装案: .github/workflows/agent-eval.yml
name: Agent Eval - 4-Signal Scoring

on:
  pull_request:
    branches: [main]
  schedule:
    - cron: "0 2 * * 1"

jobs:
  eval:
    runs-on: ubuntu-latest
    timeout-minutes: 60
    services:
      wiremock:
        image: wiremock/wiremock:latest
        ports: ["8080:8080"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r requirements-eval.txt
      - name: Run evaluation suite
        env:
          EVAL_SEED: "42"
          FAKETIME: "2026-05-05 00:00:00"
        run: |
          python -m claw_eval run \
            --task-dir ./tasks/ \
            --agent-module myagent.run \
            --output-dir ./eval_results/ \
            --report-json ./eval_results/summary.json
      - name: Check pass threshold
        run: |
          python - <<'EOF'
          import json, sys
          summary = json.load(open("eval_results/summary.json"))
          tau = 0.80
          if summary["pass_rate"] < tau:
              print(f"FAIL: pass_rate {summary['pass_rate']:.3f} < τ={tau}")
              sys.exit(1)
          print("PASS")
          EOF
      - if: always()
        uses: actions/upload-artifact@v4
        with:
          name: eval-results-${{ github.sha }}
          path: eval_results/

# 実装案: hitl_router.py
from dataclasses import dataclass
from enum import Enum

class ReviewDecision(Enum):
    AUTO_APPROVE = "auto_approve"
    SAMPLE_REVIEW = "sample_review"
    FULL_HITL = "full_hitl"


@dataclass
class RiskContext:
    task_family: str
    overall_score: float
    involves_financial: bool
    financial_amount: float
    multi_system: bool


def route_to_hitl(ctx: RiskContext) -> tuple[ReviewDecision, float]:
    """
    3 シグナル合議: rule 30% + history 55% + llm_confidence 15%
    """
    HIGH_RISK_FAMILIES = {"hr_people", "mgmt"}
    rule_score = 0.0
    if ctx.task_family in HIGH_RISK_FAMILIES:
        rule_score = 1.0
    if ctx.involves_financial and ctx.financial_amount >= 5000:
        rule_score = max(rule_score, 1.0)

    history_score = 1.0 - ctx.overall_score
    llm_risk = 1.0 - ctx.overall_score
    composite_risk = 0.30 * rule_score + 0.55 * history_score + 0.15 * llm_risk

    if rule_score >= 1.0 or composite_risk >= 0.70:
        return ReviewDecision.FULL_HITL, 1.0
    if ctx.overall_score >= 0.85:
        return ReviewDecision.AUTO_APPROVE, 0.0
    if ctx.overall_score <= 0.30:
        return ReviewDecision.FULL_HITL, 1.0

    sampling_rate = 0.05 + 0.15 * (1.0 - ctx.overall_score)
    return ReviewDecision.SAMPLE_REVIEW, min(sampling_rate, 0.20)

# 実装案: golden_trace_regression.py
import json, pathlib, hashlib
from dataclasses import dataclass

GOLDEN_DIR = pathlib.Path("golden_traces/")

@dataclass
class GoldenTrace:
    task_id: str
    tool_calls: list[dict]
    state_after: dict
    artifact_hash: str


def save_golden(trace: GoldenTrace) -> None:
    GOLDEN_DIR.mkdir(parents=True, exist_ok=True)
    path = GOLDEN_DIR / f"{trace.task_id}.json"
    path.write_text(json.dumps({
        "task_id": trace.task_id,
        "tool_calls": trace.tool_calls,
        "state_after": trace.state_after,
        "artifact_hash": trace.artifact_hash,
    }, ensure_ascii=False, indent=2))


def diff_against_golden(task_id: str, new_signals: dict) -> dict:
    path = GOLDEN_DIR / f"{task_id}.json"
    if not path.exists():
        return {"status": "no_golden", "task_id": task_id}
    golden = json.loads(path.read_text())
    diffs = {}
    if golden["tool_calls"] != new_signals.get("tool_calls", []):
        diffs["tool_calls"] = {"golden": golden["tool_calls"],
                               "current": new_signals.get("tool_calls", [])}
    state_diffs = {}
    for k, v in golden["state_after"].items():
        actual = new_signals.get("state_after", {}).get(k)
        if actual != v:
            state_diffs[k] = {"golden": v, "current": actual}
    if state_diffs:
        diffs["state_after"] = state_diffs
    return {"status": "diff" if diffs else "ok", "task_id": task_id, "diffs": diffs}

git tag -a snapshot/2026-05 -m "tasks: 105 (+12 refreshed, HR/MGMT 更新)"
git push origin snapshot/2026-05

git checkout snapshot/2026-04
eval --config eval.yaml --task-dir tasks/ --output results/2026-04/

docker run --rm \
  --network none \
  --read-only \
  --mount type=bind,src=./tasks,dst=/tasks,readonly \
  --mount type=tmpfs,dst=/workspace \
  claw-eval-sandbox:snapshot-2026-05 \
  eval-runner --task-id $TASK_ID --output /workspace/result.json

judge:
  primary: claude-opus-4-6
  secondary: gpt-5-4
  temperature: 0
  seed: 42
  cross_check_threshold: 0.10
  deterministic_fields:
    - state_change_verified
    - artifact_hash_match
    - audit_log_entry_count

aws s3api put-object \
  --bucket eval-audit-logs \
  --key "2026/05/05/${TASK_ID}/${RUN_ID}.jsonl.gz" \
  --body result.jsonl.gz \
  --object-lock-mode COMPLIANCE \
  --object-lock-retain-until-date "2029-05-05T00:00:00Z"

ALWAYS_SAMPLE = ["ecommerce-monthly-adjust", "first-response-audit", "multi-doc-merge"]
MONTHLY_ONLY = [t for t in ALL_TASKS if t.pass_rate < 0.05 or t.pass_rate > 0.95]

def should_record_full(task_id: str) -> bool:
    if task_id in ALWAYS_SAMPLE:
        return True
    if task_id in MONTHLY_ONLY and not is_monthly_run():
        return False
    return True

# Grafana アラート例
- alert: HRFamilyDegradation
  expr: |
    avg_over_time(eval_pass_rate{family="HR"}[7d]) < 0.18
  for: 1d
  labels:
    severity: warning
  annotations:
    summary: "HR family pass 率が 18% を下回りました"

eval-compare \
  --baseline snapshot/2026-04 \
  --current snapshot/2026-05 \
  --output diff-report/2026-04-vs-05.md

import asyncio

async def run_step(step_fn, timeout_sec=10, step_name="unknown"):
    try:
        return await asyncio.wait_for(step_fn(), timeout=timeout_sec)
    except asyncio.TimeoutError:
        logger.warning(f"Step {step_name} timed out after {timeout_sec}s")
        raise StepTimeoutError(step_name)