Experiment Artifacts Handoff#

Source: examples/experiment_artifacts_handoff.py

Introduction#

Start from a study-output directory shaped like a design-research-experiments export and run standard analysis workflows without manually joining runs.csv, conditions.csv, events.csv, or evaluations.csv.

Technical Implementation#

  1. Write a tiny deterministic artifact bundle that stands in for an exported experiment.

  2. Validate the bundle through top-level artifact helpers.

  3. Run condition comparisons, Markov-chain comparisons, and regression directly from the artifact directory.

  1from __future__ import annotations
  2
  3import csv
  4import json
  5from pathlib import Path
  6from tempfile import TemporaryDirectory
  7
  8import design_research_analysis as dran
  9
 10
 11def _write_csv(path: Path, rows: list[dict[str, object]]) -> None:
 12    with path.open("w", encoding="utf-8", newline="") as handle:
 13        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
 14        writer.writeheader()
 15        writer.writerows(rows)
 16
 17
 18def _write_artifacts(output_dir: Path) -> None:
 19    output_dir.mkdir(parents=True, exist_ok=True)
 20    (output_dir / "manifest.json").write_text(
 21        json.dumps({"schema_version": "0.1.0", "study_id": "artifact-demo"}),
 22        encoding="utf-8",
 23    )
 24    conditions = [
 25        {
 26            "study_id": "artifact-demo",
 27            "condition_id": "cond-baseline-small",
 28            "agent_treatment": "baseline",
 29            "model_size_b": 7,
 30            "task_family": "mechanical",
 31            "admissible": True,
 32            "constraint_messages": "[]",
 33            "assignment_meta_json": "{}",
 34        },
 35        {
 36            "study_id": "artifact-demo",
 37            "condition_id": "cond-baseline-large",
 38            "agent_treatment": "baseline",
 39            "model_size_b": 13,
 40            "task_family": "thermal",
 41            "admissible": True,
 42            "constraint_messages": "[]",
 43            "assignment_meta_json": "{}",
 44        },
 45        {
 46            "study_id": "artifact-demo",
 47            "condition_id": "cond-planner-small",
 48            "agent_treatment": "planner",
 49            "model_size_b": 7,
 50            "task_family": "thermal",
 51            "admissible": True,
 52            "constraint_messages": "[]",
 53            "assignment_meta_json": "{}",
 54        },
 55        {
 56            "study_id": "artifact-demo",
 57            "condition_id": "cond-planner-large",
 58            "agent_treatment": "planner",
 59            "model_size_b": 13,
 60            "task_family": "mechanical",
 61            "admissible": True,
 62            "constraint_messages": "[]",
 63            "assignment_meta_json": "{}",
 64        },
 65    ]
 66    _write_csv(output_dir / "conditions.csv", conditions)
 67
 68    scores = [0.42, 0.55, 0.78, 0.93]
 69    runs: list[dict[str, object]] = []
 70    for index, condition in enumerate(conditions, start=1):
 71        runs.append(
 72            {
 73                "study_id": "artifact-demo",
 74                "condition_id": condition["condition_id"],
 75                "run_id": f"run-{index}",
 76                "problem_id": f"problem-{condition['task_family']}",
 77                "problem_family": condition["task_family"],
 78                "agent_id": condition["agent_treatment"],
 79                "agent_kind": "scripted",
 80                "pattern_name": "ideation",
 81                "model_name": f"model-{condition['model_size_b']}b",
 82                "seed": index,
 83                "replicate": 1,
 84                "status": "success",
 85                "start_time": "2026-01-01T00:00:00Z",
 86                "end_time": "2026-01-01T00:00:03Z",
 87                "latency_s": 3.0,
 88                "input_tokens": 10,
 89                "output_tokens": 20,
 90                "cost_usd": 0.0,
 91                "primary_outcome": scores[index - 1],
 92                "trace_path": "",
 93                "manifest_path": "manifest.json",
 94            }
 95        )
 96    _write_csv(output_dir / "runs.csv", runs)
 97
 98    events: list[dict[str, object]] = []
 99    for index, run in enumerate(runs, start=1):
100        middle_event = "simulate" if run["agent_id"] == "planner" else "revise"
101        for step, event_type in enumerate(("inspect", middle_event, "submit")):
102            events.append(
103                {
104                    "timestamp": f"2026-01-01T00:00:{index}{step}Z",
105                    "record_id": f"evt-{index}-{step}",
106                    "text": event_type,
107                    "session_id": run["run_id"],
108                    "actor_id": "agent",
109                    "event_type": event_type,
110                    "meta_json": "{}",
111                    "run_id": run["run_id"],
112                }
113            )
114    _write_csv(output_dir / "events.csv", events)
115
116    _write_csv(
117        output_dir / "evaluations.csv",
118        [
119            {
120                "run_id": f"run-{index}",
121                "evaluator_id": "rubric",
122                "metric_name": "quality_score",
123                "metric_value": score,
124                "metric_unit": "unitless",
125                "aggregation_level": "run",
126                "notes_json": "{}",
127            }
128            for index, score in enumerate(scores, start=1)
129        ],
130    )
131
132
133def main() -> None:
134    """Run artifact-first analysis workflows over one tiny export."""
135    with TemporaryDirectory() as tmp:
136        output_dir = Path(tmp) / "study-output"
137        _write_artifacts(output_dir)
138
139        validation = dran.validate_experiment_events(output_dir / "events.csv")
140        joined_events = dran.build_event_table_from_artifacts(output_dir)
141        metric_rows = dran.build_condition_metric_table_from_artifacts(
142            output_dir,
143            metric="quality_score",
144            condition_column="agent_treatment",
145        )
146        report = dran.compare_condition_pairs_from_artifacts(
147            output_dir,
148            metric="quality_score",
149            condition_column="agent_treatment",
150            condition_pairs=[("planner", "baseline")],
151            alternative="greater",
152            seed=17,
153        )
154        chains = dran.fit_markov_chains_from_artifacts(
155            output_dir,
156            condition_column="agent_treatment",
157        )
158        chain_delta = dran.compare_markov_chains_from_artifacts(
159            output_dir,
160            condition_column="agent_treatment",
161            left_condition="planner",
162            right_condition="baseline",
163        )
164        run_metrics = dran.build_run_metric_table_from_artifacts(
165            output_dir,
166            metrics="quality_score",
167        )
168        regression = dran.fit_regression_from_artifacts(
169            output_dir,
170            outcome="quality_score",
171            predictors=("model_size_b", "task_family"),
172            categorical_predictors=("task_family",),
173        )
174
175        print("Events valid:", validation.is_valid, f"rows={validation.n_rows}")
176        print("Joined event rows:", len(joined_events))
177        print("Metric rows:", len(metric_rows), f"run rows={len(run_metrics)}")
178        print("Condition comparisons:", len(report.comparisons))
179        print("Markov chains:", ", ".join(sorted(chains)))
180        print("Transition delta:", f"{chain_delta.estimate:.4f}")
181        print("Regression coefficients:", regression.coefficients)
182
183
184if __name__ == "__main__":
185    main()

Expected Results#

Run Command

PYTHONPATH=src python examples/experiment_artifacts_handoff.py

Prints validation status, derived table sizes, condition-comparison count, Markov-chain labels, transition-comparison estimate, and regression coefficients.

References#

  • docs/experiments_handoff.rst