feat(examples): add counterfactual trace optimization loop

Author: Leoluis0705

examples/optimization/eval_optimize_loop/DESIGN.md

@@ -0,0 +1,16 @@
+# 方案设计
+
+本方案将评测优化闭环的核心从“根据失败文本猜测原因”改为“通过同一评测器验证最小行为反事实”。系统先检查训练集、验证集、提示词和预算契约，并将样例标记为 trusted、suspect 或 invalid。只有 trusted 且具有修复证据的失败能够驱动优化。
+
+对于失败轨迹，系统深拷贝实际对话，分别替换最终回复、工具名称、工具参数或受限组合，保持预期对话不变，再调用 `AgentEvaluator.get_executer()` 评分。单一干预修复失败时形成强归因；只有组合有效时判定为复合失败。数据、评测器和基础设施异常不进入优化摘要。若局部替换与原工具响应不一致，证据会降低置信度。
+
+归因结果映射到 router、skill 和 system 三类 `TargetPrompt`。真实优化入口固定使用 `AgentOptimizer.optimize(update_source=False)`，候选必须在完整验证集重新评测，新增退化再次执行反事实诊断。Gate 采用 all-must-pass，检查验证增益、可信子集、hard fail、关键样例、严重度、成本、耗时和证据。只有 gate 接受且显式传入 `--apply` 时才调用 `TargetPrompt.write_all()`，并记录写前写后哈希。报告保存轨迹证据、候选差异、拒绝原因、随机种子及复现命令。
+
+## 技术附录
+
+- 反事实归因：结论来自真实 metric delta，不依赖 case ID、failure reason 或人工标签。
+- Prompt actionability：只有 agent behavior failure 能够选择优化表面。
+- Gate：关键检查必须全部通过，证据不足时以 `NEEDS_REVIEW` 拒绝。
+- 防过拟合：训练失败驱动优化，完整 validation 决策，新退化独立诊断。
+- 审计：记录输入和提示词哈希、seed、耗时、成本、候选及每项 gate 证据。
+- 限制：局部轨迹替换可能形成现实中不可执行的状态，因此显式记录一致性并降低置信度。

examples/optimization/eval_optimize_loop/README.md

@@ -0,0 +1,33 @@
+# Trust-Aware Counterfactual Trace Diagnosis Loop
+
+This example closes evaluation and prompt optimization with evidence from trace interventions. Unlike loops that classify a failure from reason text or sample metadata, it deep-copies the actual trace, changes one execution surface, and sends the counterfactual `EvalCase` through the same public `AgentEvaluator` metrics. Evaluation-data, evaluator, and infrastructure failures are excluded from prompt optimization.
+
+## Quick start
+
+```bash
+python examples/optimization/eval_optimize_loop/run_counterfactual_probe.py
+python examples/optimization/eval_optimize_loop/run_pipeline.py --mode fake
+python examples/optimization/eval_optimize_loop/run_pipeline.py --mode trace
+python examples/optimization/eval_optimize_loop/run_pipeline.py --mode fake --candidate-profile accepted
+python examples/optimization/eval_optimize_loop/run_pipeline.py --mode fake --candidate-profile ineffective
+```
+
+Both fake and trace modes need no API key. Trace mode replays `actual_conversation`; `conversation` remains the expected trace. The fake optimizer is prompt-sensitive and introduces an intentionally broad billing rule without reading case IDs.
+
+## Real optimizer
+
+Fake and trace modes are fully runnable and verified end to end. `pipeline.optimizer.run_real_optimizer()` wires the public `AgentOptimizer.optimize(..., update_source=False)` API to selected `TargetPrompt` files and is verified with a mock/spy, including actionable-case filtering and unified optimization fields. It has not been run against a real model in this example. Production use requires a business `call_agent`, model credentials, and a trace-capture adapter for candidate regression diagnosis.
+
+## Write-back
+
+`--apply` is necessary but not sufficient. Files are written only after every gate check passes. The implementation records baseline hashes, calls `TargetPrompt.write_all()` once, verifies changed hashes, and restores the baseline on failure. The bundled candidate is rejected, so running with `--apply` leaves all source prompts unchanged.
+
+## Outputs
+
+- `prototype_output/counterfactual_probe.{json,md}`: feasibility evidence.
+- `sample_output/optimizer_failure_digest.json`: actionable and excluded failures.
+- `sample_output/optimization_report.{json,md}`: baseline, reliability, attribution, candidate deltas, regression diagnosis, gate, cost, duration, hashes, and reproduction command.
+
+## Known limits
+
+Counterfactual traces can create states that a real agent could not produce, especially when a tool name changes but its response does not. The loop therefore restricts combinations, validates trace shape, exposes invalid interventions, and rejects insufficient evidence. Exact metrics are deterministic; LLM judges require repeated sampling before a case can be trusted. Windows absolute evalset paths are converted to relative paths because the current SDK interprets a drive-letter colon as a case selector.

examples/optimization/eval_optimize_loop/__init__.py

@@ -0,0 +1 @@
+"""Trust-aware counterfactual trace optimization example."""

examples/optimization/eval_optimize_loop/fake/__init__.py

@@ -0,0 +1 @@
+"""Deterministic semantic fake components."""

examples/optimization/eval_optimize_loop/fake/model.py

@@ -0,0 +1,64 @@
+"""Prompt-sensitive fake behavior generator; never reads eval IDs."""
+
+from __future__ import annotations
+
+import re
+
+from trpc_agent_sdk.evaluation import EvalCase, Invocation
+
+
+def user_text(case: EvalCase) -> str:
+    return " ".join(part.text or "" for part in case.conversation[0].user_content.parts).lower()
+
+
+def generate_trace(case: EvalCase, prompts: dict[str, str]) -> EvalCase:
+    """Generate behavior from user semantics and prompt rules."""
+    changed = case.model_copy(deep=True)
+    text = user_text(case)
+    router = prompts.get("router_prompt", "")
+    skill = prompts.get("skill_prompt", "")
+    system = prompts.get("system_prompt", "")
+    tool_name = None
+    tool_args = None
+    if "shipping" in text or "shipment" in text:
+        tracking = re.search(r"t-\d+", text).group(0).upper()
+        tool_name, tool_args = "track_shipment", {"tracking_id": tracking}
+        response = f'{{"tracking_id":"{tracking}","status":"in_transit"}}'
+    elif "billing" in text or "invoice" in text:
+        invoice = re.search(r"i-\d+", text).group(0).upper()
+        tool_name = "create_refund" if "BILLING_TO_REFUND=ON" in router else "get_invoice"
+        tool_args = {"invoice_id": invoice}
+        response = f'{{"invoice_id":"{invoice}","status":"open"}}'
+    else:
+        order_match = re.search(r"o-\d+", text)
+        if order_match is None:
+            response = '{"status":"unsupported"}'
+            raw = {
+                "user_content": case.conversation[0].user_content.model_dump(mode="json"),
+                "final_response": {"role": "model", "parts": [{"text": response}]},
+            }
+            changed.actual_conversation = [Invocation.model_validate(raw)]
+            return changed
+        order = order_match.group(0).upper()
+        if "summarize" in text:
+            response = (
+                f'{{"status":"pending","order_id":"{order}"}}'
+                if "JSON_STATUS=ALWAYS_REQUIRED" in system
+                else f'{{"order_id":"{order}"}}'
+            )
+        else:
+            reason = "duplicate" if "duplicate" in text else "damaged"
+            strict = "REFUND_ROUTE=STRICT" in router
+            tool_name = "create_refund" if strict or reason == "duplicate" else "get_invoice"
+            tool_args = {"order_id": order}
+            if "REFUND_REASON=REQUIRED" in skill or reason == "damaged":
+                tool_args["reason"] = reason
+            response = f'{{"status":"submitted","order_id":"{order}"}}'
+    raw = {
+        "user_content": case.conversation[0].user_content.model_dump(mode="json"),
+        "final_response": {"role": "model", "parts": [{"text": response}]},
+    }
+    if tool_name:
+        raw["intermediate_data"] = {"tool_uses": [{"name": tool_name, "args": tool_args}]}
+    changed.actual_conversation = [Invocation.model_validate(raw)]
+    return changed

examples/optimization/eval_optimize_loop/fake/optimizer.py

@@ -0,0 +1,56 @@
+"""Fake optimizer with the same candidate contract as the real adapter."""
+
+from __future__ import annotations
+
+
+async def optimize(prompts: dict[str, str], targets: list[str], profile: str = "overfit") -> dict:
+    candidate = dict(prompts)
+    if profile == "ineffective":
+        return {
+            "total_rounds": 1,
+            "rounds": [
+                {
+                    "round": 1,
+                    "profile": profile,
+                    "targets": [],
+                    "candidate_prompts": candidate,
+                    "cost": 0.0,
+                    "tokens": 0,
+                    "duration_seconds": 0.0,
+                }
+            ],
+            "best_prompts": candidate,
+            "cost": 0.0,
+            "tokens": 0,
+            "seed": 42,
+            "candidate_profile": profile,
+        }
+    if profile not in ("accepted", "overfit"):
+        raise ValueError(f"unknown candidate profile: {profile}")
+    if "router_prompt" in targets:
+        candidate["router_prompt"] += "\nREFUND_ROUTE=STRICT\n"
+        if profile == "overfit":
+            candidate["router_prompt"] += "BILLING_TO_REFUND=ON\n"
+    if "skill_prompt" in targets:
+        candidate["skill_prompt"] += "\nREFUND_REASON=REQUIRED\n"
+    if "system_prompt" in targets:
+        candidate["system_prompt"] += "\nJSON_STATUS=ALWAYS_REQUIRED\n"
+    return {
+        "total_rounds": 1,
+        "rounds": [
+            {
+                "round": 1,
+                "profile": profile,
+                "targets": list(targets),
+                "candidate_prompts": candidate,
+                "cost": 0.0,
+                "tokens": 0,
+                "duration_seconds": 0.0,
+            }
+        ],
+        "best_prompts": candidate,
+        "cost": 0.0,
+        "tokens": 0,
+        "seed": 42,
+        "candidate_profile": profile,
+    }

examples/optimization/eval_optimize_loop/gate.json

@@ -0,0 +1 @@
+{"min_validation_delta":0.01,"max_cost":1.0,"max_latency_seconds":180,"max_counterfactual_evaluations_per_case":7,"protected_cases":["val_billing"]}

examples/optimization/eval_optimize_loop/optimizer.json

@@ -0,0 +1 @@
+{"evaluate":{"metrics":[{"metric_name":"tool_trajectory_avg_score","threshold":1.0},{"metric_name":"final_response_avg_score","threshold":1.0}],"num_runs":1},"optimize":{"algorithm":{"name":"gepa_reflective","seed":42,"reflection_lm":{"model_name":"${TRPC_AGENT_MODEL_NAME}","base_url":"${TRPC_AGENT_BASE_URL}","api_key":"${TRPC_AGENT_API_KEY}"},"max_metric_calls":20}}}

examples/optimization/eval_optimize_loop/pipeline/__init__.py

@@ -0,0 +1 @@
+"""Pipeline components for the counterfactual trace optimization loop."""

examples/optimization/eval_optimize_loop/pipeline/counterfactual.py

@@ -0,0 +1,70 @@
+"""Budgeted counterfactual evaluation using the official evaluator."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from trpc_agent_sdk.evaluation import EvalCase
+
+from .diagnosis import attribute_from_evidence
+from .interventions import InterventionKind, build_counterfactual
+from .models import CounterfactualEvidence, FailureAttribution
+from .probe import evaluate_trace_cases
+
+
+async def diagnose_trace_case(case: EvalCase, workspace: Path, max_evaluations: int = 7) -> FailureAttribution:
+    baseline = (await evaluate_trace_cases([case], workspace))[case.eval_id]
+    evidence = []
+    singles = [
+        InterventionKind.REPLACE_FINAL_RESPONSE,
+        InterventionKind.REPLACE_TOOL_NAME,
+        InterventionKind.REPLACE_TOOL_ARGUMENTS,
+        InterventionKind.NORMALIZE_FORMAT,
+    ]
+    combinations = [
+        InterventionKind.REPLACE_TOOL_NAME_AND_ARGUMENTS,
+        InterventionKind.REPLACE_TOOL_NAME_AND_FINAL_RESPONSE,
+        InterventionKind.REPLACE_TOOL_ARGUMENTS_AND_FINAL_RESPONSE,
+    ]
+    for kind in singles + combinations:
+        if len(evidence) >= max_evaluations:
+            break
+        built = build_counterfactual(case, kind)
+        if not built.valid or built.eval_case is None:
+            evidence.append(
+                CounterfactualEvidence(
+                    kind.value,
+                    False,
+                    built.status,
+                    False,
+                    [],
+                    [],
+                    baseline,
+                    {},
+                    structurally_valid=built.structurally_valid,
+                    semantically_coherent=built.semantically_coherent,
+                    coherence_warnings=list(built.coherence_warnings),
+                )
+            )
+            continue
+        after = (await evaluate_trace_cases([built.eval_case], workspace))[built.eval_case.eval_id]
+        repaired = sorted(k for k, v in baseline.items() if v < 1 and after.get(k, v) >= 1)
+        unchanged = sorted(k for k, v in baseline.items() if after.get(k) == v)
+        evidence.append(
+            CounterfactualEvidence(
+                kind.value,
+                True,
+                built.status,
+                all(v >= 1 for v in after.values()),
+                repaired,
+                unchanged,
+                baseline,
+                after,
+                structurally_valid=built.structurally_valid,
+                semantically_coherent=built.semantically_coherent,
+                coherence_warnings=list(built.coherence_warnings),
+            )
+        )
+        if evidence[-1].changed_fail_to_pass and kind in singles:
+            break
+    return attribute_from_evidence(case.eval_id, evidence, evaluations_used=len(evidence), budget=max_evaluations)