feat(examples): add auditable AgentOptimizer evaluation loop

.gitignore

@@ -21,6 +21,9 @@ coverage.xml
 test-ngtest-ut-trpc-agent-py.xml
 .pytest_cache
 
+# Runtime artefacts for Issue #91; stable reviewed reports live in sample_output/.
+examples/optimization/eval_optimize_loop/runs/
+
 node_modules
 package-lock.json
 pyrightconfig.json

examples/optimization/eval_optimize_loop/DESIGN.md

@@ -0,0 +1,13 @@
+# 方案设计说明
+
+本示例在公开的 `AgentEvaluator` 与 `AgentOptimizer` 之上增加独立编排层，形成“基线评测、失败归因、候选生成、独立回归、Gate 决策、审计落盘”的完整闭环，不修改 SDK 公共 API。失败归因采用规则优先策略：先根据执行异常、实际与期望工具选择、参数差异、工具响应、JSON 格式和判定原因分类；trace 模式优先使用真实中间轨迹，期望会话只作为参照。只有规则不能解释时才保留可扩展的 judge 入口；每个结论都带有类型、证据和置信度。
+
+## Why rule-first instead of counterfactual-first
+
+Issue #170 style counterfactual attribution can be stronger because it can rerun controlled variants and distinguish prompt, tool, data, and environment causes more directly. This example deliberately starts with rule-first attribution instead: the checked-in fake and trace paths need to be stable, credential-free, low cost, and reproducible in CI without an LLM judge. The rule layer also makes every failure reason auditable from the stored eval result, rather than from optimizer self-reporting. Counterfactual probes remain a future extension point behind the existing attribution interface, where they can add stronger diagnosis without changing the Gate contract.
+
+接受策略由 Gate 独立掌握，优化器的 round 状态或摘要不能直接决定通过。候选必须具有完整提示词映射，并在 `PromptSandbox` 中重新跑完整训练集和验证集。Gate 会拒绝评测不完整、新增 hard fail、关键回归、超出允许范围的回归、指标低于下限、验证集无增益、成本或耗时超预算以及违反平局策略的候选；未知成本会记录为警告。赢家按新增 hard fail、关键回归数、验证集通过率和得分、成本、耗时与稳定候选 ID 排序。
+
+防过拟合依靠训练集与验证集的独立全量回归，而不是读取优化器轮次指标。若训练提升而验证下降，或泛化差距超过配置阈值，Gate 会直接拒绝。优化期间固定 `update_source=False`，候选只作用于临时目标提示词目录；默认不回写，显式允许回写时还要校验基线和候选摘要。
+
+每次运行都会生成可由 Pydantic 读取的 `OptimizationReport`，并保存脱敏后的输入、环境、原始、归一化、候选与 Gate 审计产物。fake 和 trace 路径不需要 API Key、网络或 LLM judge，且输出稳定可复现；live 模式缺少必要环境变量会在启动优化器前安全退出。

examples/optimization/eval_optimize_loop/README.md

@@ -0,0 +1,33 @@
+# Issue #91: Auditable evaluation and prompt optimization loop
+
+This example demonstrates an evaluation-first prompt-optimization loop around the public `AgentOptimizer` API.  It keeps the checked-in fake and trace paths deterministic and credential-free, while live mode is an opt-in adapter with independent acceptance checks.
+
+The loop uses a Champion-Challenger architecture. The source prompt is the Champion; each `AgentOptimizer` or fixture candidate is a Challenger. The Gate trusts only independent train and validation re-evaluation in `PromptSandbox`, not optimizer self-scores or round summaries.
+
+## Run
+
+From the repository root:
+
+```text
+python examples/optimization/eval_optimize_loop/run_pipeline.py --mode fake --output-dir examples/optimization/eval_optimize_loop/sample_output
+python examples/optimization/eval_optimize_loop/run_pipeline.py --mode trace --output-dir .tmp/issue-91-trace
+python examples/optimization/eval_optimize_loop/run_pipeline.py --mode live --output-dir .tmp/issue-91-live
+```
+
+`fake` uses checked-in candidates and a local rubric; it needs no key and makes no network calls. `trace` evaluates the checked-in recorded conversation only. `live` validates `TRPC_AGENT_API_KEY`, `TRPC_AGENT_BASE_URL`, and `TRPC_AGENT_MODEL_NAME` before constructing an optimizer. It exits with code `2` if any are absent, and does not start network work in that case.
+
+Live mode writes candidates to a temporary `TargetPrompt` workspace, passes only the SDK `evaluate`/`optimize` configuration to `AgentOptimizer.optimize`, and always uses `update_source=False`. Source write-back is disabled by default; it is considered only after a Gate-approved winner and guarded by prompt-digest checks.
+
+## CI collection
+
+The regression tests live under `tests/examples/optimization/eval_optimize_loop`. The repository-level pytest configuration in `pyproject.toml` already uses `testpaths = ["tests"]`, so these tests are collected by the existing `python -m pytest` CI path without a custom test runner.
+
+## Outputs
+
+Each run writes `optimization_report.json`, `optimization_report.md`, and secret-free audit evidence:
+
+- `input.snapshot.json` and `environment.snapshot.json`
+- `audit/raw_reports.json` and `audit/normalized_reports.json`
+- `audit/candidate_reports.json` and `audit/gate_decisions.json`
+
+The JSON report is Pydantic-readable as `OptimizationReport` and is the source of truth for the selected candidate.

examples/optimization/eval_optimize_loop/__init__.py

@@ -0,0 +1 @@
+"""Deterministic no-key evaluation and optimization loop example."""

examples/optimization/eval_optimize_loop/agent/__init__.py

@@ -0,0 +1 @@
+"""Prompt files for the deterministic fake support agent."""

examples/optimization/eval_optimize_loop/agent/agent.py

@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+from ..fake.fake_agent import FakeSupportAgent
+
+__all__ = ["FakeSupportAgent"]

examples/optimization/eval_optimize_loop/agent/prompts/router.md

@@ -0,0 +1 @@
+ROUTER_V1

examples/optimization/eval_optimize_loop/agent/prompts/system.md

@@ -0,0 +1 @@
+BASELINE

examples/optimization/eval_optimize_loop/fake/__init__.py

@@ -0,0 +1 @@
+"""No-key deterministic dependencies for the fake pipeline."""

examples/optimization/eval_optimize_loop/fake/candidates.json

@@ -0,0 +1,5 @@
+[
+  {"candidate_id": "candidate_general_fix", "prompts": {"system_prompt": "GENERAL_FIX\n", "router_prompt": "ROUTER_V1\n"}, "round_index": 1},
+  {"candidate_id": "candidate_noop", "prompts": {"system_prompt": "BASELINE\n", "router_prompt": "ROUTER_V1\n"}, "round_index": 2},
+  {"candidate_id": "candidate_overfit", "prompts": {"system_prompt": "OVERFIT\n", "router_prompt": "ROUTER_V1\n"}, "round_index": 3}
+]

examples/optimization/eval_optimize_loop/fake/fake_agent.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import json
+
+from trpc_agent_sdk.evaluation import TargetPrompt
+
+
+def _json(route: str, tool: str, arguments: dict[str, str], answer: str) -> str:
+    return json.dumps({"route": route, "tool": tool, "arguments": arguments, "answer": answer}, ensure_ascii=False, sort_keys=True)
+
+
+class FakeSupportAgent:
+    """Prompt-backed support router with no model or network dependency."""
+
+    def __init__(self, target_prompt: TargetPrompt) -> None:
+        self._target_prompt = target_prompt
+
+    async def call_agent(self, query: str) -> str:
+        prompts = await self._target_prompt.read_all()
+        mode = prompts["system_prompt"].strip()
+        general = mode == "GENERAL_FIX"
+        overfit = mode == "OVERFIT"
+        if query == "生成订单 A100 的 JSON 工单摘要":
+            return _json("order_lookup", "lookup_order", {"order_id": "A100"}, "正在查询订单 A100。") if general or overfit else "订单 A100 正在查询"
+        if query == "查询订单 A100":
+            return _json("order_lookup", "lookup_order", {"order_id": "A100"}, "正在查询订单 A100。") if general or overfit else _json("order_lookup", "lookup_order", {}, "正在查询订单。")
+        if query == "特殊退款政策是什么":
+            return _json("knowledge_gap", "none", {}, "未提供该政策，不能猜测。") if overfit else _json("knowledge_gap", "none", {}, "我不确定。")
+        if query == "生成订单 B200 的 JSON 工单摘要":
+            return _json("order_lookup", "lookup_order", {"order_id": "B200"}, "正在查询订单 B200。") if general or overfit else "订单 B200 正在查询"
+        if query == "退款订单 R900，金额 12 USD":
+            if overfit:
+                return _json("order_lookup", "lookup_order", {"order_id": "R900"}, "正在查询订单 R900。")
+            return _json("refund", "refund_order", {"order_id": "R900", "currency": "USD", "amount": "12"}, "正在退款订单 R900。")
+        if query == "如何查看订单状态":
+            return _json("faq", "none", {}, "在订单详情页可查看订单状态。")
+        raise ValueError(f"unknown fake query: {query}")

examples/optimization/eval_optimize_loop/fake/fake_judge.py

@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+import json
+import statistics
+from dataclasses import dataclass
+from typing import Optional
+
+from trpc_agent_sdk.evaluation._eval_case import Invocation
+from trpc_agent_sdk.evaluation._eval_metrics import EvalMetric, EvalStatus
+from trpc_agent_sdk.evaluation._eval_result import EvaluationResult, PerInvocationResult
+from trpc_agent_sdk.evaluation._evaluator_base import Evaluator
+from trpc_agent_sdk.evaluation._evaluator_registry import EVALUATOR_REGISTRY
+
+
+_FAKE_RUBRIC_REGISTERED = False
+
+
+@dataclass(frozen=True)
+class FakeRubricResult:
+    score: float
+    reason: str
+
+
+def _response_text(invocation: Invocation) -> str:
+    response = invocation.final_response
+    if response is None or not response.parts:
+        return ""
+    return "".join(part.text or "" for part in response.parts)
+
+
+def _parse_response(response: str) -> dict[str, object] | None:
+    try:
+        payload = json.loads(response)
+    except (TypeError, json.JSONDecodeError):
+        return None
+    return payload if isinstance(payload, dict) else None
+
+
+def evaluate_fake_response(response: str, expected_response: str) -> FakeRubricResult:
+    """Score the local fake-response rubric without a model or external state."""
+    actual = _parse_response(response)
+    if actual is None:
+        return FakeRubricResult(0.0, "invalid JSON response")
+
+    expected = _parse_response(expected_response)
+    if expected is None:
+        raise ValueError("fake rubric expected response must be a JSON object")
+
+    score = 0.25
+    reasons: list[str] = []
+    required_fields = ("route", "tool", "arguments", "answer")
+    missing_fields = [field for field in required_fields if field not in actual]
+    if missing_fields:
+        return FakeRubricResult(0.25, "missing required fields: " + ", ".join(missing_fields))
+
+    if actual.get("route") == expected.get("route") and actual.get("tool") == expected.get("tool"):
+        score += 0.25
+    else:
+        reasons.append("route or tool does not match expected response")
+
+    expected_arguments = expected.get("arguments")
+    actual_arguments = actual.get("arguments")
+    if isinstance(expected_arguments, dict) and isinstance(actual_arguments, dict) and all(
+        actual_arguments.get(name) == value for name, value in expected_arguments.items()
+    ):
+        score += 0.25
+    else:
+        reasons.append("required tool arguments do not match expected response")
+
+    unknown_knowledge = expected.get("route") == "knowledge_gap" and expected.get("tool") == "none"
+    answer = actual.get("answer")
+    explicit_refusal = isinstance(answer, str) and any(
+        phrase in answer.lower() for phrase in ("不能猜测", "cannot guess", "do not guess", "don't guess")
+    )
+    if not unknown_knowledge or explicit_refusal:
+        score += 0.25
+    else:
+        reasons.append("unknown knowledge must explicitly refuse to guess")
+
+    return FakeRubricResult(score, "; ".join(reasons) if reasons else "all fake rubric checks passed")
+
+
+def score_fake_response(response: str, expected_response: str) -> float:
+    """Return the deterministic fake-rubric score for a response pair."""
+    return evaluate_fake_response(response, expected_response).score
+
+
+def fake_rubric_score(response: str) -> float:
+    """Backward-compatible structural score for callers without a reference response."""
+    payload = _parse_response(response)
+    return 1.0 if payload and all(key in payload for key in ("route", "tool", "arguments", "answer")) else 0.0
+
+
+class FakeRubricEvaluator(Evaluator):
+    """A local deterministic evaluator registered only by the fake example."""
+
+    requires_reference = True
+
+    def __init__(self, threshold: Optional[float] = None, eval_metric: Optional[EvalMetric] = None) -> None:
+        if threshold is not None and eval_metric is not None:
+            raise ValueError("Either eval_metric or threshold may be specified, not both")
+        self._threshold = eval_metric.threshold if eval_metric is not None else threshold
+        if self._threshold is None:
+            self._threshold = 0.75
+
+    def evaluate_invocations(
+        self,
+        actual_invocations: list[Invocation],
+        expected_invocations: Optional[list[Invocation]],
+    ) -> EvaluationResult:
+        if expected_invocations is None:
+            raise ValueError("expected_invocations is required for fake_rubric_score")
+        if len(actual_invocations) != len(expected_invocations):
+            raise ValueError("actual and expected invocations must contain the same number of invocations")
+
+        per_invocation_results: list[PerInvocationResult] = []
+        for actual, expected in zip(actual_invocations, expected_invocations):
+            result = evaluate_fake_response(_response_text(actual), _response_text(expected))
+            per_invocation_results.append(
+                PerInvocationResult(
+                    actual_invocation=actual,
+                    expected_invocation=expected,
+                    score=result.score,
+                    eval_status=EvalStatus.PASSED if result.score >= self._threshold else EvalStatus.FAILED,
+                    reason=result.reason,
+                )
+            )
+        if not per_invocation_results:
+            return EvaluationResult()
+
+        overall_score = statistics.mean(result.score for result in per_invocation_results)
+        return EvaluationResult(
+            overall_score=overall_score,
+            overall_eval_status=EvalStatus.PASSED if overall_score >= self._threshold else EvalStatus.FAILED,
+            per_invocation_results=per_invocation_results,
+        )
+
+
+def register_fake_rubric_evaluator() -> None:
+    """Register the example-only evaluator with the SDK's existing registry."""
+    global _FAKE_RUBRIC_REGISTERED
+    if _FAKE_RUBRIC_REGISTERED:
+        return
+    EVALUATOR_REGISTRY.register("fake_rubric_score", FakeRubricEvaluator)
+    _FAKE_RUBRIC_REGISTERED = True

examples/optimization/eval_optimize_loop/fake/fixture_optimizer.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from ..pipeline.models import CandidateRecord
+
+
+class FixtureOptimizerBackend:
+    def __init__(self, candidates_path: Path) -> None:
+        self._candidates_path = candidates_path
+
+    def load_candidates(self) -> list[CandidateRecord]:
+        payload = json.loads(self._candidates_path.read_text(encoding="utf-8"))
+        return [CandidateRecord.model_validate(item) for item in payload]

examples/optimization/eval_optimize_loop/optimizer.json

@@ -0,0 +1,28 @@
+{
+  "evaluate": {
+    "metrics": [
+      {"metric_name": "final_response_avg_score", "threshold": 1.0, "criterion": {"final_response": {"text": {"match": "exact"}}}},
+      {"metric_name": "fake_rubric_score", "threshold": 0.75}
+    ],
+    "num_runs": 1
+  },
+  "optimize": {
+    "eval_case_parallelism": 1,
+    "stop": {"required_metrics": "all"},
+    "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": 10
+    }
+  },
+  "pipeline": {
+    "reproducibility": {"seed": 42},
+    "write_back_when_accepted": false,
+    "gate": {"min_validation_score_delta": 0.05, "critical_case_ids": ["val_refund_critical"]}
+  }
+}

examples/optimization/eval_optimize_loop/pipeline/__init__.py

@@ -0,0 +1 @@
+"""Pipeline components for the Issue #91 example."""