(cls, data: Mapping[str, Any], *, path: str)
| 139 | |
| 140 | @classmethod |
| 141 | def from_dict(cls, data: Mapping[str, Any], *, path: str) -> "GraphDefinition": |
| 142 | mapping = require_mapping(data, path) |
| 143 | graph_id = optional_str(mapping, "id", path) |
| 144 | description = optional_str(mapping, "description", path) |
| 145 | |
| 146 | if "vars" in mapping and mapping["vars"]: |
| 147 | raise ConfigError("vars are only supported at DesignConfig root", extend_path(path, "vars")) |
| 148 | |
| 149 | log_level_raw = mapping.get("log_level", LogLevel.DEBUG.value) |
| 150 | try: |
| 151 | log_level = LogLevel(log_level_raw) |
| 152 | except ValueError as exc: |
| 153 | raise ConfigError( |
| 154 | f"log_level must be one of {[lvl.value for lvl in LogLevel]}", extend_path(path, "log_level") |
| 155 | ) from exc |
| 156 | |
| 157 | is_majority = optional_bool(mapping, "is_majority_voting", path, default=False) |
| 158 | organization = optional_str(mapping, "organization", path) |
| 159 | initial_instruction = optional_str(mapping, "initial_instruction", path) |
| 160 | |
| 161 | nodes_raw = ensure_list(mapping.get("nodes")) |
| 162 | # if not nodes_raw: |
| 163 | # raise ConfigError("graph must define at least one node", extend_path(path, "nodes")) |
| 164 | nodes: List[Node] = [] |
| 165 | for idx, node_dict in enumerate(nodes_raw): |
| 166 | nodes.append(Node.from_dict(node_dict, path=extend_path(path, f"nodes[{idx}]"))) |
| 167 | |
| 168 | edges_raw = ensure_list(mapping.get("edges")) |
| 169 | edges: List[EdgeConfig] = [] |
| 170 | for idx, edge_dict in enumerate(edges_raw): |
| 171 | edges.append(EdgeConfig.from_dict(edge_dict, path=extend_path(path, f"edges[{idx}]"))) |
| 172 | |
| 173 | memory_cfg: List[MemoryStoreConfig] | None = None |
| 174 | if "memory" in mapping and mapping["memory"] is not None: |
| 175 | raw_stores = ensure_list(mapping.get("memory")) |
| 176 | stores: List[MemoryStoreConfig] = [] |
| 177 | seen: set[str] = set() |
| 178 | for idx, item in enumerate(raw_stores): |
| 179 | store = MemoryStoreConfig.from_dict(item, path=extend_path(path, f"memory[{idx}]")) |
| 180 | if store.name in seen: |
| 181 | raise ConfigError( |
| 182 | f"duplicated memory store name '{store.name}'", |
| 183 | extend_path(path, f"memory[{idx}].name"), |
| 184 | ) |
| 185 | seen.add(store.name) |
| 186 | stores.append(store) |
| 187 | memory_cfg = stores |
| 188 | |
| 189 | start_nodes: List[str] = [] |
| 190 | if "start" in mapping and mapping["start"] is not None: |
| 191 | start_value = mapping["start"] |
| 192 | if isinstance(start_value, str): |
| 193 | start_nodes = [start_value] |
| 194 | elif isinstance(start_value, list) and all(isinstance(item, str) for item in start_value): |
| 195 | seen = set() |
| 196 | start_nodes = [] |
| 197 | for item in start_value: |
| 198 | if item not in seen: |
nothing calls this directly
no test coverage detected