Adds a node to the workflow.

        :param node_id: A unique identifier for the new node.
        :param task: The task to run.
        :param inputs: A mapping of inputs from workflow inputs, or outputs from other nodes. The format should be
            `{input_name: (source, value), ...}` where `input_name` is the parameter name for the task input, source is
            "input" or "dependency" and `value` is either the workflow input name (if source is "input") or a 2-tuple
            with a node id and an output parameter name from that node's task to map to the input.

def add_node(self, node_id, task, inputs):
        """
        Adds a node to the workflow.

        :param node_id: A unique identifier for the new node.
        :param task: The task to run.
        :param inputs: A mapping of inputs from workflow inputs, or outputs from other nodes. The format should be
            `{input_name: (source, value), ...}` where `input_name` is the parameter name for the task input, source is
            "input" or "dependency" and `value` is either the workflow input name (if source is "input") or a 2-tuple
            with a node id and an output parameter name from that node's task to map to the input.
        """

        if node_id in self.nodes_by_id:
            raise ValueError('The node {0} already exists in this workflow.'.format(node_id))

        node = WorkflowNode(node_id, task, inputs)
        self.nodes_by_id[node_id] = node

        for source, value in six.itervalues(inputs):
            if source == 'dependency':
                dependents = self.dependents_by_node_id.get(value[0], set())
                dependents.add(node_id)
                self.dependents_by_node_id[value[0]] = dependents