Real Stack Interoperability#

Source: examples/real_stack_interoperability.py

Introduction#

Run one packaged problem from design-research-problems through a public design-research-agents baseline and validate the exported events.csv contract with design-research-analysis’s artifact-first integration API.

Technical Implementation#

  1. Bootstrap sibling src/ directories from the local workspace when present.

  2. Execute a one-run study that uses a packaged optimization problem together with SeededRandomBaselineAgent.

  3. Export canonical artifacts and validate the event table through the analysis package’s artifact-first integration contract.

  1from __future__ import annotations
  2
  3import importlib
  4import sys
  5from pathlib import Path
  6
  7import design_research_experiments as drex
  8
  9
 10def _bootstrap_sibling_sources() -> None:
 11    """Add sibling package `src/` directories when the repos are checked out locally."""
 12    workspace_root = Path(__file__).resolve().parents[2]
 13    for repo_name in (
 14        "design-research-problems",
 15        "design-research-agents",
 16        "design-research-analysis",
 17    ):
 18        src_path = workspace_root / repo_name / "src"
 19        src_path_text = str(src_path)
 20        if src_path.exists() and src_path_text not in sys.path:
 21            sys.path.insert(0, src_path_text)
 22
 23
 24def _load_stack_modules() -> dict[str, object] | None:
 25    """Import sibling stack packages when available from the local workspace."""
 26    _bootstrap_sibling_sources()
 27    try:
 28        problems_module = importlib.import_module("design_research_problems")
 29        importlib.import_module("design_research_agents")
 30    except ImportError as exc:
 31        print(f"Real stack example skipped: {exc}")
 32        return None
 33
 34    try:
 35        analysis_module = importlib.import_module("design_research_analysis")
 36        analysis_integration = importlib.import_module("design_research_analysis.integration")
 37    except ImportError as exc:
 38        print(f"Real stack example skipped: {exc}")
 39        return None
 40
 41    return {
 42        "problems": problems_module,
 43        "analysis": analysis_module,
 44        "analysis_integration": analysis_integration,
 45    }
 46
 47
 48def main() -> None:
 49    """Run one real interoperability path across the sibling libraries."""
 50    modules = _load_stack_modules()
 51    if modules is None:
 52        return
 53
 54    problems_module = modules["problems"]
 55    analysis_module = modules["analysis"]
 56    analysis_integration = modules["analysis_integration"]
 57    problem_id = "gmpb_default_dynamic_min"
 58    packaged_problem = problems_module.get_problem(problem_id)
 59    problem_packet = drex.resolve_problem(problem_id)
 60
 61    study = drex.Study(
 62        study_id="real-stack-interoperability",
 63        title="Real stack interoperability",
 64        description="Packaged problem + public agent + validated analysis handoff.",
 65        output_dir=Path("artifacts") / "real-stack-interoperability",
 66        problem_ids=(problem_id,),
 67        agent_specs=("SeededRandomBaselineAgent",),
 68        outcomes=(
 69            drex.OutcomeSpec(
 70                name="primary_outcome",
 71                source_table="runs",
 72                column="primary_outcome",
 73                aggregation="mean",
 74                primary=True,
 75            ),
 76        ),
 77        run_budget=drex.RunBudget(replicates=1, parallelism=1, max_runs=1),
 78        primary_outcomes=("primary_outcome",),
 79    )
 80    conditions = drex.build_design(study)
 81    run_results = drex.run_study(
 82        study,
 83        conditions=conditions,
 84        problem_registry={problem_id: problem_packet},
 85        show_progress=False,
 86    )
 87    exported_paths = drex.export_analysis_tables(
 88        study,
 89        conditions=conditions,
 90        run_results=run_results,
 91        output_dir=study.output_dir / "analysis",
 92        validate_with_analysis_package=True,
 93    )
 94    loaded_artifacts = analysis_integration.load_experiment_artifacts(exported_paths["events.csv"])
 95    report = analysis_integration.validate_experiment_events(exported_paths["events.csv"])
 96    primary_metric_rows = analysis_module.build_condition_metric_table(
 97        loaded_artifacts["runs.csv"],
 98        metric="primary_outcome",
 99        condition_column="agent_id",
100        conditions=loaded_artifacts["conditions.csv"],
101        evaluations=loaded_artifacts["evaluations.csv"],
102    )
103    run_result = run_results[0]
104
105    print("Problem ID:", packaged_problem.metadata.problem_id)
106    print("Problem family:", packaged_problem.metadata.kind.value)
107    print("Agent:", study.agent_specs[0])
108    print("Completed runs:", len(run_results))
109    print("Run status:", run_result.status.value)
110    print("Output keys:", ", ".join(sorted(run_result.outputs)))
111    print("Primary outcome:", run_result.metrics.get("primary_outcome"))
112    print("Metric rows:", len(primary_metric_rows))
113    print("Event rows valid:", report.is_valid, f"(rows={report.n_rows})")
114    print("Exported artifacts:", ", ".join(path.name for path in exported_paths.values()))
115
116
117if __name__ == "__main__":
118    main()

Expected Results#

Run Command

PYTHONPATH=src python examples/real_stack_interoperability.py

The script prints the packaged problem identity, one successful run result, and the exported artifact filenames after the event table passes validation.