How to pass along a variable and then combine with multiple outputs

[brandonwilde]

Is it possible to create a chain that (1) produces multiple named outputs and (2) also uses a RunnablePassthrough to pass its input unchanged to the subsequent Runnable/chain?

I'm running into a problem where the output of my (sub)chain eventually has to be assigned to a single variable name, which I would like to avoid for the sake of being able to handle variables separately. Below are two approaches I've found for implementing variable passthrough, but both methods also force me to store my classify_chain's output (a dictionary with multiple keys and values) as a single variable.

RunnableParallel(
        input=RunnablePassthrough(),
        output=classify_chain
    )

    {'output': classify_chain, 'input': itemgetter('input')}

The result in both cases is something like:

{
  'input': input_value,
  'output': {
    'output_a': output_a_value,
    'output_b': output_b_value'
  }
}

I would like to be get it in this form, however:

{
  'input': input_value,
  'output_a': output_a_value,
  'output_b': output_b_value'
}

If I pipe the output of classify_chain directly into something else (i.e. I neglect the passthrough), then the multiple outputs of classify_chain are treated as separate input variables to the next Runnable, as I'd like. It's only when I need to combine it with a passthrough that I become constrained.

Hopefully there's something simple I'm missing. I'd love to hear if anyone knows a solution to this!

[dimaelzein]

@brandonwilde Did you manage to do this? I am looking for a similar solution too..

[brandonwilde]

Hey @dimaelzein, I ended up adding an extra function to my chain like this. You can make it more specific to your use case, but the result is that it's pulling out the contents of any dictionaries (here it will find a dictionary of the form {"output": {...}}) and then add those contents to the top-level dictionary.

from langchain.schema.runnable import RunnablePassthrough

chain = (
    RunnablePassthrough.assign(
        output=classify_chain  # pass along existing variables, add "output" variable
    )
    | flatten
)

def flatten(*vars):
    '''
    Flatten a dictionary by removing unnecessary mid-level keys.
    Returns a Runnable (chainable) function.
    '''
    flat = {}
    for var in vars:
        keys = [k for k in var]
        for key in keys:
            if isinstance(var[key], dict):
                flat.update(var[key])
            else:
                flat[key] = var[key]
    return

[rob-err]

Thanks for this. Minor issue with flatten, missing value, flat, in the return statement for anyone else