🔧 技術調査 - 自動ハーネス適応で小型モデルを96%安く動かす

# 実装案: 業務タスク用の成功判定関数（migration-analysis の task_evals パターンを簡略化）
from pathlib import Path
import json


def run_single_instance_eval(workspace_dir: str, example: dict) -> dict:
    """エージェントの作業ディレクトリと期待値を比較し score/feedback を返す。"""
    workspace = Path(workspace_dir)
    expected = json.loads(Path(example["expected_path"]).read_text())
    actual_state = json.loads((workspace / "state.json").read_text())

    checkpoints: list[tuple[str, bool]] = [
        ("required_contacts_reached", _contacted_all(actual_state, expected["contacts"])),
        ("required_docs_read", _docs_accessed(actual_state, expected["doc_paths"])),
        ("final_decision_correct", actual_state.get("decision") == expected["decision"]),
    ]
    # 最終チェックポイントが通れば満点、それ以外は通過数の加重平均
    if checkpoints[-1][1]:
        score = 1.0
    else:
        score = sum(ok for _, ok in checkpoints) / len(checkpoints)

    feedback = "; ".join(f"{name}={ok}" for name, ok in checkpoints)
    return {"workspace_dir": workspace_dir, "score": score, "feedback": feedback}

# 実装案: モデルカタログ (configs/models.yaml 相当、環境変数はダミー)
models:
  - name: gemma-4-26b-a4b          # SLM 候補
    model: openrouter/google/gemma-4-26b-a4b
    api_key: ${OPENROUTER_API_KEY}
    input_cost_per_token: 0.0000001
    output_cost_per_token: 0.0000003
  - name: gemini-3.1-pro-preview   # フロンティア baseline / メタエージェントの分析用 LM
    model: vertex_ai/gemini-3.1-pro-preview
    vertex_credentials: ${GOOGLE_APPLICATION_CREDENTIALS}

# 実装案: モデル辞書のロードと実行用ラッパーへの変換
import dspy, yaml, os

def load_lmdict(yaml_path: str) -> dict:
    config = yaml.safe_load(open(yaml_path))
    lm_dict = {}
    for m in config["models"]:
        kwargs = {k: (os.getenv(v[2:-1]) if isinstance(v, str) and v.startswith("${") else v)
                  for k, v in m.items() if k not in ("name", "model")}
        lm_dict[m["name"]] = dspy.LM(m["model"], **kwargs)
    return lm_dict

LM_DICT = load_lmdict("configs/models.yaml")
task_lm = LM_DICT["gemma-4-26b-a4b"]               # SLM: タスク実行本体
reflection_lm = LM_DICT["gemini-3.1-pro-preview"]  # メタエージェントの分析用

# 実装案: 探索メモリと初期ハーネス（seed candidate）の初期化
from dataclasses import dataclass, asdict
import json, os


@dataclass
class CandidateRecord:
    iteration: int
    code_hash: str
    agent_code: str            # ハーネス実装そのもの (build_agent 関数)
    parent_hash: str | None
    status: str | None = None  # "seed" | "accepted_on_subsample" | "rejected_on_subsample"
    val_score: float | None = None


class HarnessMemory:
    def __init__(self, run_dir: str):
        self.path = os.path.join(run_dir, "memory.json")
        self.records: list[CandidateRecord] = []

    def add(self, record: CandidateRecord) -> None:
        self.records.append(record)
        json.dump([asdict(r) for r in self.records], open(self.path, "w"), indent=2)

    def pareto_frontier(self) -> list[CandidateRecord]:
        # 検証済み (val_score が入っている) 候補のみを対象にスコア降順
        scored = [r for r in self.records if r.val_score is not None]
        return sorted(scored, key=lambda r: r.val_score, reverse=True)


SEED_AGENT_CODE = '''
def build_agent(base_dir: str, llm):
    # 最小構成のベースラインハーネス。ここから GEPA が改訂を重ねる。
    from openhands.sdk import Agent
    return Agent(llm=llm, system_prompt_filename="system_prompt.md")
'''

memory = HarnessMemory(run_dir="results/budget_approval/gemma-4-26b-a4b/seed0")
memory.add(CandidateRecord(iteration=0, code_hash="seed", agent_code=SEED_AGENT_CODE,
                            parent_hash=None, status="seed"))

# 実装案: 予算超過で停止する Stopper（migration-analysis の CostBudgetStopper を簡略化）
class CostBudgetStopper:
    """Stop optimization when total accumulated cost exceeds a budget in USD."""

    def __init__(self, max_cost: float, cost_tracker: dict):
        self.max_cost = max_cost
        self._cost_tracker = cost_tracker

    def __call__(self, gepa_state) -> bool:
        total = sum(bucket["accumulated_cost"] for bucket in self._cost_tracker.values())
        return total >= self.max_cost

# 実装案: GEPA スタイル探索ループの骨格
from gepa import optimize

adapter = AgentOptimizationAdapter(  # gepa.GEPAAdapter を実装した自作クラス（③のメモリ・②の LM を内包）
    task_id="budget_approval",
    task_lm=task_lm,
    reflection_lm=reflection_lm,
    eval_fn=run_single_instance_eval,   # ①の成功判定関数
    memory=memory,                      # ③の探索メモリ
)
stopper = CostBudgetStopper(max_cost=20.0, cost_tracker=adapter.cost_tracker)  # ペア予算 $20

result = optimize(
    seed_candidate={"agent_code": SEED_AGENT_CODE},
    trainset=trainset,
    valset=valset,
    adapter=adapter,
    stopper=stopper,
)

# 実装案: 軌跡の保存（SLM 軌跡と教師軌跡を分離）
import json, os

def save_trajectory(run_dir: str, example_id: str, steps: list, is_teacher: bool = False):
    subdir = "teacher_trajectories" if is_teacher else "trajectories"
    dest = os.path.join(run_dir, "memory", "current", subdir)
    os.makedirs(dest, exist_ok=True)
    json.dump(steps, open(os.path.join(dest, f"{example_id}.json"), "w"), indent=2)

- project/agent.py                      # 現行ハーネス実装
- memory/current/trajectories/          # タスク軌跡（SLM 自身）
- memory/current/teacher_trajectories/  # タスク軌跡（教師、任意）
- memory/scoreboard.md, past_agents.md  # 探索メモリ（履歴サマリ・過去候補）
- docs/adaptation.md, docs/sdk_reference.md  # 設計空間ドキュメント（失敗モード→適応 / 実装 API）

# 実装案: メタエージェント呼び出しの骨格
PROPOSER_SYSTEM_PROMPT = """You are an agent optimization expert.
You improve an AI agent's task performance by analyzing execution trajectories
and modifying the agent's configuration code.

## Workspace
- project/agent.py            # 現行ハーネス実装
- memory/current/trajectories/ # タスク軌跡
- memory/scoreboard.md         # 探索メモリ (履歴サマリ)
- docs/adaptation.md           # 設計空間ドキュメント (失敗モード→適応)

## Constraints
- MUST define `build_agent(base_dir, llm) -> Agent`.
- Code must be valid, self-contained Python with explicit imports.
"""

def propose_new_harness(workspace_dir: str, reflection_lm) -> str:
    """4種 context を読み込んだコーディングエージェントに改訂 agent.py を書かせる。"""
    conversation = build_coding_agent(system_prompt=PROPOSER_SYSTEM_PROMPT,
                                       workspace=workspace_dir, llm=reflection_lm)
    conversation.run("軌跡と探索メモリを分析し、adaptation.md を参照して agent.py を改訂してください。")
    return open(os.path.join(workspace_dir, "project", "agent.py")).read()

# 実装案: タスク設定 (gepa_optimize.yaml 相当、抜粋)
task_id: "budget_approval_s2l"
model_name: "ministral-3-8b"
reflection_lm: "gemini-3.1-pro-preview"
max_cost: 10.0        # ペア予算 (paper 全体では平均 $20)
max_examples: 20
train_ratio: 0.5
use_adaptation_guide: true

# 実装案: 性能×コストの Pareto フロント抽出
def pareto_frontier(candidates: list) -> list:
    """candidates: [{"val_score": float, "cost_ratio": float, ...}, ...]
    性能は高いほど・コストは低いほど良い、の 2 目的 Pareto。"""
    frontier = []
    for c in candidates:
        dominated = any(
            o["val_score"] >= c["val_score"] and o["cost_ratio"] <= c["cost_ratio"]
            and (o["val_score"] > c["val_score"] or o["cost_ratio"] < c["cost_ratio"])
            for o in candidates
        )
        if not dominated:
            frontier.append(c)
    return sorted(frontier, key=lambda c: c["cost_ratio"])

# harness-registry/attendance-auditing_gemma-4-26b-a4b.yaml
task: attendance-auditing
model: gemma-4-26b-a4b
version: v3
baseline_llm_score: 1.00          # gemini-3.1-pro-preview 基準
score: 0.89                       # LLM 比 89%
cost_ratio: 0.04                  # LLM 比 4%
latency_delta: -0.25              # 25% 削減
adaptation_strategies:
  - type: context_addition
    detail: "勤怠規則の暗黙知を context として外在化"
  - type: tool_creation
    detail: "複雑な承認パターンを単純 IF に変換するツール"
failure_modes_addressed:
  - instruction-following
  - knowledge
exploration_cost_usd: 20
promoted: true

def should_retrigger_optimization(pair_metrics, baseline):
    """性能比が基準線から一定閾値を下回ったら再最適化を起票する"""
    perf_drop = baseline.perf_ratio - pair_metrics.perf_ratio
    diversity_up = pair_metrics.input_diversity > baseline.input_diversity * 1.3

    if perf_drop > 0.10:          # 例: 89% -> 79% 未満
        return "perf_degradation"
    if diversity_up:              # 論文: 89.1% -> 68.0% の多様性減衰パターン
        return "input_diversity_shift"
    if pair_metrics.model_version != baseline.model_version:
        return "model_migration"  # SLM 差し替えは一律再学習が必要
    return None

def breakeven_runs(exploration_cost_usd=20, per_run_saving_usd=None,
                    llm_cost_per_run=None, slm_cost_per_run=None):
    """探索コストの回収に必要な実行回数を試算する"""
    if per_run_saving_usd is None:
        per_run_saving_usd = llm_cost_per_run - slm_cost_per_run
    return exploration_cost_usd / per_run_saving_usd

# 論文の実測平均: breakeven_runs ≈ 13 回（$20 / 平均節約額）

def pick_adaptation_strategy(failure_mode):
    """成功実績の高い順に適応戦略を提案する"""
    priority = {
        "instruction-following": ["context_addition", "tool_creation", "tool_management"],
        "knowledge": ["context_addition", "tool_creation", "tool_management"],
        "tool-use": ["tool_management", "tool_creation", "context_addition"],
        "long-context": ["context_compression", "context_addition"],
        "planning-reasoning": ["multi_agent_orchestration", "deterministic_check"],
    }
    return priority.get(failure_mode, ["context_addition"])