38 lines
2 KiB
Python
38 lines
2 KiB
Python
from __future__ import annotations
|
|
|
|
from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec
|
|
|
|
|
|
def progeny_replay_experiment_graph_v1() -> ExecutionGraphSpec:
|
|
spec = ExecutionGraphSpec(
|
|
name="progeny_replay_experiment",
|
|
version=1,
|
|
graph_type="PROGENY_REPLAY_EXPERIMENT",
|
|
entry="prepare_experiment",
|
|
nodes={
|
|
"prepare_experiment": GraphNodeSpec("prepare_experiment", "replay_prepare"),
|
|
"select_cases": GraphNodeSpec("select_cases", "replay_select_cases"),
|
|
"validate_variants": GraphNodeSpec("validate_variants", "replay_validate_variants"),
|
|
"run_champion": GraphNodeSpec("run_champion", "replay_run_champion"),
|
|
"run_challenger": GraphNodeSpec("run_challenger", "replay_run_challenger"),
|
|
"compare": GraphNodeSpec("compare", "replay_compare"),
|
|
"experiment_judge": GraphNodeSpec("experiment_judge", "replay_experiment_judge"),
|
|
"await_human_decision": GraphNodeSpec("await_human_decision", "replay_human_decision"),
|
|
"complete": GraphNodeSpec("complete", "complete", {"terminal": True}),
|
|
},
|
|
edges=[
|
|
GraphEdgeSpec("prepare_experiment", "select_cases", "success"),
|
|
GraphEdgeSpec("select_cases", "validate_variants", "success"),
|
|
GraphEdgeSpec("validate_variants", "run_champion", "success"),
|
|
GraphEdgeSpec("run_champion", "run_challenger", "success"),
|
|
GraphEdgeSpec("run_challenger", "compare", "success"),
|
|
GraphEdgeSpec("compare", "experiment_judge", "success"),
|
|
GraphEdgeSpec("experiment_judge", "await_human_decision", "success"),
|
|
GraphEdgeSpec("await_human_decision", "complete", "approved"),
|
|
GraphEdgeSpec("await_human_decision", "complete", "rejected"),
|
|
],
|
|
terminal_nodes=["complete"],
|
|
metadata={"description": "Replay Arena experiment workflow with human promotion gate."},
|
|
)
|
|
spec.validate()
|
|
return spec
|