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 helpers.

Technical Implementation#

  1. Import the installed sibling libraries through their package-level APIs.

  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 helpers.

 1from __future__ import annotations
 2
 3import importlib
 4from pathlib import Path
 5from typing import Any
 6
 7import design_research_experiments as drex
 8
 9
10def _load_stack_modules() -> dict[str, Any] | None:
11    """Import sibling stack packages when the coordinated libraries are installed."""
12    try:
13        problems_module = importlib.import_module("design_research_problems")
14        agents_module = importlib.import_module("design_research_agents")
15        analysis_module = importlib.import_module("design_research_analysis")
16    except ImportError as exc:
17        print(f"Real stack example skipped: {exc}")
18        return None
19
20    return {
21        "problems": problems_module,
22        "agents": agents_module,
23        "analysis": analysis_module,
24    }
25
26
27def main() -> None:
28    """Run one real interoperability path across the sibling libraries."""
29    modules = _load_stack_modules()
30    if modules is None:
31        return
32
33    problems_module = modules["problems"]
34    agents_module = modules["agents"]
35    analysis_module = modules["analysis"]
36    problem_id = "gmpb_default_dynamic_min"
37    packaged_problem = problems_module.get_problem(problem_id)
38
39    study = drex.Study(
40        study_id="real-stack-interoperability",
41        title="Real stack interoperability",
42        description="Packaged problem + public agent + validated analysis handoff.",
43        output_dir=Path("artifacts") / "real-stack-interoperability",
44        problem_ids=(problem_id,),
45        agent_specs=("SeededRandomBaselineAgent",),
46        outcomes=(
47            drex.OutcomeSpec(
48                name="primary_outcome",
49                source_table="runs",
50                column="primary_outcome",
51                aggregation="mean",
52                primary=True,
53            ),
54        ),
55        run_budget=drex.RunBudget(replicates=1, parallelism=1, max_runs=1),
56        primary_outcomes=("primary_outcome",),
57    )
58    conditions = drex.build_design(study)
59    run_results = drex.run_study(
60        study,
61        conditions=conditions,
62        show_progress=False,
63    )
64    exported_paths = drex.export_analysis_tables(
65        study,
66        conditions=conditions,
67        run_results=run_results,
68        output_dir=study.output_dir / "analysis",
69        validate_with_analysis_package=True,
70    )
71    report = analysis_module.validate_experiment_events(exported_paths["events.csv"])
72    primary_metric_rows = analysis_module.build_condition_metric_table_from_artifacts(
73        exported_paths["events.csv"],
74        metric="primary_outcome",
75        condition_column="agent_id",
76    )
77    run_result = run_results[0]
78
79    print("Problem ID:", packaged_problem.metadata.problem_id)
80    print("Problem family:", packaged_problem.metadata.kind.value)
81    print("Problem package:", problems_module.__name__)
82    print("Agent package:", agents_module.__name__)
83    print("Agent:", study.agent_specs[0])
84    print("Completed runs:", len(run_results))
85    print("Run status:", run_result.status.value)
86    print("Output keys:", ", ".join(sorted(run_result.outputs)))
87    print("Primary outcome:", run_result.metrics.get("primary_outcome"))
88    print("Metric rows:", len(primary_metric_rows))
89    print("Event rows valid:", report.is_valid, f"(rows={report.n_rows})")
90    print("Exported artifacts:", ", ".join(path.name for path in exported_paths.values()))
91
92
93if __name__ == "__main__":
94    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.