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#

Install the exact sibling versions from the tested package family before using the source-checkout run command:

python -m pip install "design-research-problems==0.4.0" \
    "design-research-agents==0.6.0" \
    "design-research-analysis==0.3.1"

  1. Import those 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    )
57    conditions = drex.build_design(study)
58    run_results = drex.run_study(
59        study,
60        conditions=conditions,
61        show_progress=False,
62    )
63    exported_paths = drex.export_analysis_tables(
64        study,
65        conditions=conditions,
66        run_results=run_results,
67        output_dir=study.output_dir / "analysis",
68        validate_with_analysis_package=True,
69    )
70    report = analysis_module.validate_experiment_events(exported_paths["events.csv"])
71    primary_metric_rows = analysis_module.build_condition_metric_table_from_artifacts(
72        exported_paths["events.csv"],
73        metric="primary_outcome",
74        condition_column="agent_id",
75    )
76    run_result = run_results[0]
77
78    print("Problem ID:", packaged_problem.metadata.problem_id)
79    print("Problem kind:", packaged_problem.metadata.kind.value)
80    print("Problem package:", problems_module.__name__)
81    print("Agent package:", agents_module.__name__)
82    print("Agent:", study.agent_specs[0])
83    print("Completed runs:", len(run_results))
84    print("Run status:", run_result.status.value)
85    print("Output keys:", ", ".join(sorted(run_result.outputs)))
86    print("Primary outcome:", run_result.metrics.get("primary_outcome"))
87    print("Metric rows:", len(primary_metric_rows))
88    print("Event rows valid:", report.is_valid, f"(rows={report.n_rows})")
89    print("Exported artifacts:", ", ".join(path.name for path in exported_paths.values()))
90
91
92if __name__ == "__main__":
93    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.