def add_node(self, node_id, task, inputs):
if node_id in self.nodes_by_id:<EOL><INDENT>raise ValueError('<STR_LIT>'.format(node_id))<EOL><DEDENT>node = WorkflowNode(node_id, task, inputs)<EOL>self.nodes_by_id[node_id] = node<EOL>for source, value in six.itervalues(inputs):<EOL><INDENT>if source == '<STR_LIT>':<EOL><INDENT>dependents = self.dependents_by_node_id.get(value[<NUM_LIT:0>], set())<EOL>dependents.add(node_id)<EOL>self.dependents_by_node_id[value[<NUM_LIT:0>]] = dependents<EOL><DEDENT><DEDENT>
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.