Runtime context for a workflow graph (state + business logic). Differences from ``GraphConfig``: - ``GraphConfig`` is immutable configuration data. - ``GraphContext`` is mutable runtime state with dynamic execution data. Attributes: config: Graph configuration
| 13 | |
| 14 | |
| 15 | class GraphContext: |
| 16 | """Runtime context for a workflow graph (state + business logic). |
| 17 | |
| 18 | Differences from ``GraphConfig``: |
| 19 | - ``GraphConfig`` is immutable configuration data. |
| 20 | - ``GraphContext`` is mutable runtime state with dynamic execution data. |
| 21 | |
| 22 | Attributes: |
| 23 | config: Graph configuration |
| 24 | nodes: Mapping of ``node_id`` to ``Node`` |
| 25 | edges: List of edges |
| 26 | layers: Topological layer layout |
| 27 | outputs: Node outputs captured during execution |
| 28 | topology: Topological ordering list |
| 29 | subgraphs: Mapping of ``node_id`` to nested ``GraphContext`` |
| 30 | has_cycles: Whether the graph contains cycles |
| 31 | cycle_execution_order: Execution order for cycles |
| 32 | directory: Output directory for artifacts |
| 33 | depth: Graph depth |
| 34 | """ |
| 35 | |
| 36 | def __init__(self, config: GraphConfig) -> None: |
| 37 | """Initialize the graph context. |
| 38 | |
| 39 | Args: |
| 40 | config: Graph configuration |
| 41 | """ |
| 42 | self.config = config |
| 43 | self.vars: Dict[str, Any] = dict(config.vars) |
| 44 | |
| 45 | # Graph structure |
| 46 | self.nodes: Dict[str, Node] = {} |
| 47 | self.edges: List[Dict[str, Any]] = [] |
| 48 | self.layers: List[List[str]] = [] |
| 49 | self.topology: List[str] = [] |
| 50 | self.depth: int = 0 |
| 51 | self.start_nodes: List[str] = [] |
| 52 | self.explicit_start_nodes: List[str] = [] |
| 53 | |
| 54 | # Runtime state |
| 55 | self.outputs: Dict[str, str] = {} |
| 56 | self.subgraphs: Dict[str, "GraphContext"] = {} |
| 57 | |
| 58 | # Cycle support |
| 59 | self.has_cycles: bool = False |
| 60 | self.cycle_execution_order: List[Dict[str, Any]] = [] |
| 61 | |
| 62 | # Output directory |
| 63 | timestamp = datetime.now().strftime("%Y%m%d%H%M%S") |
| 64 | fixed_output_dir = bool(config.metadata.get("fixed_output_dir")) |
| 65 | if fixed_output_dir or "session_" in config.name: |
| 66 | self.directory = config.output_root / config.name |
| 67 | else: |
| 68 | self.directory = config.output_root / f"{config.name}_{timestamp}" |
| 69 | self.directory.mkdir(parents=True, exist_ok=True) |
| 70 | # Voting mode flag |
| 71 | self.is_majority_voting: bool = config.is_majority_voting |
| 72 |
no outgoing calls
no test coverage detected