fix(evaluation): 统一审计口径并补强提示词恢复

Author: guocfu

examples/optimization/eval_optimize_loop/artifact_writer.py

@@ -24,6 +24,8 @@
 from .schemas import ArtifactIndex, ArtifactReference, FailureReport
 from .schemas import OptimizationReport, ReportPhase
 from .schemas import TraceCandidateProposal
+from .sensitive_config import SensitiveConfigError
+from .sensitive_config import validate_persisted_sensitive_values
 
 
 ArtifactType: TypeAlias = Literal[
@@ -36,46 +38,6 @@
 ]
 
 _INPUT_COPY_DISABLED = "artifacts.copy_input_files=false"
-_SENSITIVE_CONFIG_KEYS = {
-    "accesstoken",
-    "apikey",
-    "auth",
-    "authorization",
-    "authtoken",
-    "baseurl",
-    "bearertoken",
-    "clientsecret",
-    "credential",
-    "credentials",
-    "password",
-    "passwd",
-    "privatekey",
-    "secret",
-    "secretkey",
-    "token",
-    "xapikey",
-}
-_SENSITIVE_CONFIG_KEY_SUFFIXES = {
-    "accesstoken",
-    "apikey",
-    "authtoken",
-    "baseurl",
-    "bearertoken",
-    "clientsecret",
-    "credential",
-    "credentials",
-    "endpointurl",
-    "password",
-    "passwd",
-    "privatekey",
-    "secretkey",
-}
-_APPROVED_SENSITIVE_VALUES = {
-    "",
-    "${TRPC_AGENT_API_KEY}",
-    "${TRPC_AGENT_BASE_URL}",
-    "fake-not-used-in-offline-mode",
-}
 _AT_FDCWD = -100
 _RENAME_NOREPLACE = 1
 _RENAME_EXCL = 0x4
@@ -179,52 +141,17 @@ def _json_text(model: BaseModel) -> str:
     return model.model_dump_json(by_alias=False, indent=2) + "\n"
 
 
-def _normalized_sensitive_key(key: str) -> str:
-    return key.replace("_", "").replace("-", "").casefold()
-
-
-def _is_sensitive_config_key(key: str) -> bool:
-    normalized = _normalized_sensitive_key(key)
-    return normalized in _SENSITIVE_CONFIG_KEYS or any(
-        normalized.endswith(suffix)
-        for suffix in _SENSITIVE_CONFIG_KEY_SUFFIXES
-    )
-
-
-def _validate_sensitive_config_values(value: object, *, path: str = "$") -> None:
-    if isinstance(value, str):
-        if value.strip().casefold().startswith(("http://", "https://")):
-            raise ArtifactWriteError(
-                "sensitive optimizer config value is not an approved "
-                f"placeholder: {path}"
-            )
-        return
-    if isinstance(value, list):
-        for index, item in enumerate(value):
-            _validate_sensitive_config_values(item, path=f"{path}[{index}]")
-        return
-    if not isinstance(value, dict):
-        return
-    for key, item in value.items():
-        item_path = f"{path}.{key}"
-        if _is_sensitive_config_key(key):
-            if not isinstance(item, str) or item not in _APPROVED_SENSITIVE_VALUES:
-                raise ArtifactWriteError(
-                    "sensitive optimizer config value is not an approved "
-                    f"placeholder: {item_path}"
-                )
-        else:
-            _validate_sensitive_config_values(item, path=item_path)
-
-
 def _validate_optimizer_config_for_copy(path: Path) -> None:
     try:
         payload = json.loads(path.read_text(encoding="utf-8"))
     except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
         raise ArtifactWriteError(
             f"failed to parse optimizer config snapshot: {path}: {exc}"
         ) from exc
-    _validate_sensitive_config_values(payload)
+    try:
+        validate_persisted_sensitive_values(payload)
+    except SensitiveConfigError as exc:
+        raise ArtifactWriteError(str(exc)) from exc
 
 
 def _rename_directory_no_replace(source: Path, target: Path) -> None:
@@ -621,6 +548,8 @@ def publish_report_bundle(
 
         for relative in native_paths:
             native_path = root / relative
+            if native_path.name == "optimizer.runtime.json":
+                _validate_optimizer_config_for_copy(native_path)
             references.append(
                 _available_reference(
                     root,

examples/optimization/eval_optimize_loop/candidate_provider.py

@@ -24,6 +24,7 @@
 from .schemas import CandidateScenario
 from .schemas import OptimizerCandidateProposal
 from .schemas import OptimizerRuntimeParameters
+from .sensitive_config import replace_persisted_sensitive_values
 
 
 class CandidateProviderError(RuntimeError):
@@ -96,26 +97,7 @@ def __init__(self, call_agent: CallAgent) -> None:
     @staticmethod
     def _replace_persisted_connection_values(value: object) -> object:
         """递归将可能被 SDK 复制到产物的连接值替换为环境占位符。"""
-        if isinstance(value, list):
-            return [
-                AgentOptimizerCandidateProvider._replace_persisted_connection_values(item)
-                for item in value
-            ]
-        if not isinstance(value, dict):
-            return value
-        placeholders = {
-            "api_key": "${TRPC_AGENT_API_KEY}",
-            "apiKey": "${TRPC_AGENT_API_KEY}",
-            "base_url": "${TRPC_AGENT_BASE_URL}",
-            "baseUrl": "${TRPC_AGENT_BASE_URL}",
-        }
-        return {
-            key: placeholders.get(
-                key,
-                AgentOptimizerCandidateProvider._replace_persisted_connection_values(item),
-            )
-            for key, item in value.items()
-        }
+        return replace_persisted_sensitive_values(value)
 
     @staticmethod
     def _prepare_runtime_config(request: CandidateRequest) -> Path:

examples/optimization/eval_optimize_loop/pipeline.py

@@ -24,7 +24,6 @@
 from trpc_agent_sdk.evaluation import CallAgent
 from trpc_agent_sdk.evaluation import EvalCaseResult
 from trpc_agent_sdk.evaluation import EvalSet
-from trpc_agent_sdk.evaluation import EvalStatus
 from trpc_agent_sdk.evaluation import OptimizeConfigFile
 from trpc_agent_sdk.evaluation import TargetPrompt
 from trpc_agent_sdk.evaluation import load_optimize_config
@@ -41,6 +40,7 @@
 from .config import PipelineConfig
 from .config import load_pipeline_config
 from .evaluation_adapter import EvaluationAnalysisError
+from .evaluation_adapter import standardize_snapshot
 from .fake import DeterministicFakeModel
 from .gate import evaluate_gate
 from .gate import GateEvaluationError
@@ -473,22 +473,6 @@ def prepare_run(pipeline_config_path: str | Path, *, run_id: str | None = None)
         raise
 
 
-def _summarize_results(
-    eval_results_by_eval_id: dict[str, list[EvalCaseResult]],
-) -> tuple[int, int, float | None]:
-    passed_cases = 0
-    scores: list[float] = []
-    for runs in eval_results_by_eval_id.values():
-        if runs and all(getattr(run, "final_eval_status", None) == EvalStatus.PASSED for run in runs):
-            passed_cases += 1
-        for run in runs:
-            for metric in getattr(run, "overall_eval_metric_results", []):
-                if metric.score is not None:
-                    scores.append(float(metric.score))
-    average_score = sum(scores) / len(scores) if scores else None
-    return passed_cases, len(eval_results_by_eval_id), average_score
-
-
 def _validate_results(
     *,
     eval_set: EvalSet,
@@ -547,18 +531,34 @@ async def _evaluate_split(
         phase=phase,
         split=split,
     )
-    passed_cases, total_cases, average_score = _summarize_results(eval_results_by_eval_id)
-    return EvaluationSnapshot(
+    snapshot = EvaluationSnapshot(
         phase=phase,
         split=split,
         eval_set_id=eval_set.eval_set_id,
         failed_summary=failed_summary,
         details_lines=details_lines,
         result_lines=result_lines,
         eval_results_by_eval_id=eval_results_by_eval_id,
-        passed_case_count=passed_cases,
-        total_case_count=total_cases,
-        average_score=average_score,
+        passed_case_count=0,
+        total_case_count=len(eval_results_by_eval_id),
+        average_score=None,
+    )
+    try:
+        standardized = standardize_snapshot(snapshot)
+    except EvaluationAnalysisError as exc:
+        raise PipelineStageExecutionError(
+            f"{phase} {split} evaluation result standardization failed: {exc}"
+        ) from exc
+    return snapshot.model_copy(
+        update={
+            "passed_case_count": standardized.passed_case_count,
+            "total_case_count": len(standardized.cases),
+            "average_score": (
+                standardized.average_score.value
+                if standardized.average_score.status == "available"
+                else None
+            ),
+        }
     )
 
 
@@ -567,14 +567,34 @@ async def _restore_working_baseline(
     baseline_prompts: dict[str, str],
 ) -> bool:
     """Restore optimizer leftovers and prove the isolated baseline is present."""
+    initial_read_error: Exception | None = None
+    was_modified = True
     try:
         current = await prepared.working_target.read_all()
+    except Exception as exc:
+        initial_read_error = exc
+    else:
         was_modified = current != baseline_prompts
-        if was_modified:
+
+    if was_modified:
+        try:
             await prepared.working_target.write_all(baseline_prompts)
+        except Exception as exc:
+            if initial_read_error is not None:
+                raise PipelineStageExecutionError(
+                    "failed to restore optimizer working prompts after initial "
+                    f"read failed ({initial_read_error}): {exc}"
+                ) from exc
+            raise PipelineStageExecutionError(
+                f"failed to restore optimizer working prompts: {exc}"
+            ) from exc
+
+    try:
         restored = await prepared.working_target.read_all()
     except Exception as exc:
-        raise PipelineStageExecutionError(f"failed to restore optimizer working prompts: {exc}") from exc
+        raise PipelineStageExecutionError(
+            f"failed to verify restored optimizer working prompts: {exc}"
+        ) from exc
     if restored != baseline_prompts:
         raise PipelineStageExecutionError("optimizer working prompts did not match baseline after restoration")
     return was_modified
@@ -592,7 +612,6 @@ async def _execute_offline_stage(
     the isolated working target on success or candidate-evaluation failure so
     the run can be inspected later.
     """
-    progress.enter("baseline_train")
     if prepared.config.execution.mode != "offline":
         raise PipelineStageExecutionError(
             "run_offline_stage requires execution.mode='offline', got "
@@ -756,7 +775,6 @@ async def _execute_real_stage(
     progress: _MutableReportProgress,
 ) -> RealStageResult:
     """Generate a real optimizer candidate and run the full guarded regression."""
-    progress.enter("baseline_train")
     if prepared.config.execution.mode != "real":
         raise PipelineStageExecutionError(
             f"run_real_stage requires execution.mode='real', got {prepared.config.execution.mode!r}"