Executable root graph for compiled Dify workflows.
| 6 | |
| 7 | |
| 8 | class DifyRootGraph(RootGraph): |
| 9 | """Executable root graph for compiled Dify workflows.""" |
| 10 | |
| 11 | def invoke(self, input: dict, attributes: dict[str, object] | None = None): |
| 12 | if attributes: |
| 13 | self._attributes_store = {**self._attributes_store, **attributes} |
| 14 | self._attributes_store.update(input) |
| 15 | metadata = self._attributes_store.get("compatibility") |
| 16 | if isinstance(metadata, dict): |
| 17 | init_dify_state(metadata, self._attributes_store).attach(self._attributes_store) |
| 18 | output, attrs = super().invoke(input, attributes) |
| 19 | state = DifyRuntimeState.from_store(self._attributes_store) |
| 20 | if state.workflow_result is not None: |
| 21 | output = state.workflow_result |
| 22 | self._output = output |
| 23 | return output, attrs |
no outgoing calls
no test coverage detected