luigi_multi_in_out.py

class TaskA1(luigi.Task):
    def requires(self):
        ....
    def output(self):
        return { 'output1' : luigi.LocalTarget( self.input().path + ".output1" ),
                   'output2' : luigi.LocalTarget( self.input().path + ".output2" ) }
    def run(self):
        ....

class TaskA2(luigi.Task):
    def requires(self):
        ....
    def output(self):
        return { 'output1' : luigi.LocalTarget( self.input().path + ".output1" ),
                   'output2' : luigi.LocalTarget( self.input().path + ".output2" ) }
    def run(self):
        ....

class TaskB(luigi.Task):
    def requires(self):
        return { 'input1' : TaskA1(),
                   'input2' : TaskA2() }

    def output(self):
        return { 'output1' : self.input()['input1']['output1'].path + ".output1",
                   'output2' : self.input()['input1']['output1'].path + ".output2" }

    def run(self):
        with self.input()['input11']['output1'].open("r") as infile, self.output['output1'].open("w") as outfile:
            for line in infile:
                line = do_something(line)
                outfile.write(line)