MCPcopy Create free account
hub / github.com/ScaleML/AgentSPEX / WorkflowIR

Class WorkflowIR

verifier/AgentVerifier/WorkflowToLean.py:199–240  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

197
198@dataclass
199class WorkflowIR:
200 name: str
201 goal: str
202 parameters: list[TypedVar]
203 nodes: list[NodeIR]
204 edges: list[EdgeIR]
205 entry: int
206 exits: list[int]
207 submodule_map: Optional[dict[str, SubmoduleRef]] = None
208
209 def to_dict(self) -> dict:
210 d: dict = {
211 "name": self.name, "goal": self.goal,
212 "parameters": [p.to_dict() for p in self.parameters],
213 "nodes": [n.to_dict() for n in self.nodes],
214 "edges": [e.to_dict() for e in self.edges],
215 "entry": self.entry, "exits": self.exits,
216 }
217 if self.submodule_map:
218 d["submodule_map"] = {k: v.to_dict() for k, v in self.submodule_map.items()}
219 return d
220
221 @staticmethod
222 def from_dict(d: dict) -> "WorkflowIR":
223 return WorkflowIR(
224 name=d["name"], goal=d["goal"],
225 parameters=[TypedVar.from_dict(p) for p in d["parameters"]],
226 nodes=[NodeIR.from_dict(n) for n in d["nodes"]],
227 edges=[EdgeIR.from_dict(e) for e in d["edges"]],
228 entry=d["entry"], exits=d["exits"],
229 submodule_map={k: SubmoduleRef.from_dict(v) for k, v in d["submodule_map"].items()}
230 if d.get("submodule_map") else None,
231 )
232
233 def save_json(self, path: str, indent: int = 2) -> None:
234 with open(path, "w", encoding="utf-8") as f:
235 json.dump(self.to_dict(), f, indent=indent, ensure_ascii=False)
236
237 @staticmethod
238 def load_json(path: str) -> "WorkflowIR":
239 with open(path, "r", encoding="utf-8") as f:
240 return WorkflowIR.from_dict(json.load(f))
241
242
243# ============================================================================

Callers 2

from_dictMethod · 0.85
parse_task_jsonFunction · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected