Serialize a node (BaseNode subclasses like FunctionNode, AgentNode, etc.).
(node: Any)
| 72 | |
| 73 | |
| 74 | def serialize_node(node: Any) -> dict[str, Any]: |
| 75 | """Serialize a node (BaseNode subclasses like FunctionNode, AgentNode, etc.).""" |
| 76 | class_name = type(node).__name__ |
| 77 | node_name = _get_node_field(node, "name") |
| 78 | |
| 79 | # Handle START node |
| 80 | if node_name == "__START__": |
| 81 | return { |
| 82 | "name": "__START__", |
| 83 | "type": "start", |
| 84 | "rerun_on_resume": _get_node_field(node, "rerun_on_resume"), |
| 85 | } |
| 86 | |
| 87 | if hasattr(node, "model_fields"): |
| 88 | result = serialize_agent(node) |
| 89 | if "type" not in result: |
| 90 | if getattr(node, "graph", None) is not None: |
| 91 | result["type"] = "workflow" |
| 92 | else: |
| 93 | result["type"] = NODE_TYPE_MAP.get( |
| 94 | class_name, "agent" if "Agent" in class_name else "node" |
| 95 | ) |
| 96 | return result |
| 97 | |
| 98 | # Get node type from mapping or default to 'node' |
| 99 | node_type = NODE_TYPE_MAP.get(class_name, "node") |
| 100 | |
| 101 | return { |
| 102 | "name": node_name, |
| 103 | "type": node_type, |
| 104 | "rerun_on_resume": _get_node_field(node, "rerun_on_resume"), |
| 105 | } |
| 106 | |
| 107 | |
| 108 | def serialize_agent(agent: BaseAgent) -> dict[str, Any]: |
no test coverage detected