Efficient RunnableLambda Chaining

2025-02-12 11:50 UTC gpt-4o Open in ChatGPT ↗

The cleanest and most efficient way to procedurally chain a list of RunnableLambda instances in a Langchain RunnableSequence is to use Python’s unpacking operator (*) within the sequence composition.

Steps:

Define your RunnableLambda functions.
Store them in a list.
Unpack them into a RunnableSequence.

Example:

from langchain.schema.runnable import RunnableLambda, RunnableSequence

# Define lambda functions
def step1(input):
    return f"Step 1 processed: {input}"

def step2(input):
    return f"Step 2 processed: {input}"

def step3(input):
    return f"Step 3 processed: {input}"

# Create RunnableLambdas
runnables = [
    RunnableLambda(step1),
    RunnableLambda(step2),
    RunnableLambda(step3)
]

# Chain them efficiently
chain = RunnableSequence(*runnables)

# Execute the chain
output = chain.invoke("Start")
print(output)

Why is this the cleanest?

Concise & Readable: Using *runnables in RunnableSequence directly passes all items as arguments.
Scalable: Easily extendable by adding more functions to the list.
Procedural & Declarative: Keeps function definitions and chaining logic separate for clarity.

This approach ensures that all functions are executed in order while keeping the implementation minimal and maintainable. 🚀

You’re right! The RunnableSequence executes the chain in a left-to-right manner, meaning each function’s output is passed as input to the next one. So the output structure you’re seeing is expected.

If you want the results to be accumulated in a different way (e.g., a list of results instead of a nested string), you might want to modify how each function processes the input.

Alternative Approaches

1. Collect Intermediate Outputs in a List

Modify the lambdas to accumulate the results instead of overriding them:

from langchain.schema.runnable import RunnableLambda, RunnableSequence

# Define processing steps that accumulate results
def step1(input):
    return input + ["Step 1 processed"]

def step2(input):
    return input + ["Step 2 processed"]

def step3(input):
    return input + ["Step 3 processed"]

# Create RunnableLambdas
runnables = [
    RunnableLambda(lambda x: step1(x) if isinstance(x, list) else step1([x])),
    RunnableLambda(step2),
    RunnableLambda(step3)
]

# Chain them efficiently
chain = RunnableSequence(*runnables)

# Execute the chain
output = chain.invoke([])
print(output)  # ['Step 1 processed', 'Step 2 processed', 'Step 3 processed']

2. Store Outputs as a Dictionary

If you want structured output, a dictionary can be used:

def step1(input):
    input["step1"] = "Step 1 processed"
    return input

def step2(input):
    input["step2"] = "Step 2 processed"
    return input

def step3(input):
    input["step3"] = "Step 3 processed"
    return input

runnables = [
    RunnableLambda(lambda x: step1(x) if isinstance(x, dict) else step1({"input": x})),
    RunnableLambda(step2),
    RunnableLambda(step3)
]

chain = RunnableSequence(*runnables)

output = chain.invoke({})
print(output)  # {'input': {}, 'step1': 'Step 1 processed', 'step2': 'Step 2 processed', 'step3': 'Step 3 processed'}

Why is the Original Output Nested?

The reason the original output was formatted like:

"Step 3 processed: Step 2 processed: Step 1 processed: Start"

is because each function in the chain takes the previous function’s output as its input and returns a single string.

If you want a sequential execution without nesting, using a list or dictionary as shown above is the best approach.

Let me know what output format you prefer! 🚀

You’re absolutely right to question this! Let’s break it down clearly.

Understanding `RunnableSequence` Execution Order

RunnableSequence executes left to right, meaning:

The first RunnableLambda gets the initial input.
Its output is passed as input to the next RunnableLambda.
This continues until the final result.

However, the issue arises due to how the functions are structured.

Why Does the Output Look Reversed?

In the original example:

def step1(input):
    return f"Step 1 processed: {input}"

def step2(input):
    return f"Step 2 processed: {input}"

def step3(input):
    return f"Step 3 processed: {input}"

Each function prepends its message to the previous output. Since function execution is left to right, but string concatenation happens as “new text + previous result”, the final output appears reversed.

Fix: Ensure Left-to-Right Concatenation

If you want step1 → step2 → step3 to execute in that order and reflect properly in the output, modify the functions to append instead of prepend:

from langchain.schema.runnable import RunnableLambda, RunnableSequence

# Define processing steps that append instead of prepend
def step1(input):
    return f"{input} -> Step 1 processed"

def step2(input):
    return f"{input} -> Step 2 processed"

def step3(input):
    return f"{input} -> Step 3 processed"

# Create RunnableLambdas
runnables = [
    RunnableLambda(step1),
    RunnableLambda(step2),
    RunnableLambda(step3)
]

# Chain them efficiently
chain = RunnableSequence(*runnables)

# Execute the chain
output = chain.invoke("Start")
print(output)  # Expected: "Start -> Step 1 processed -> Step 2 processed -> Step 3 processed"

Why Does This Work?

Each step appends to the existing string instead of prepending.
Since execution is left to right, the final result maintains the correct order.

Alternative: Using a List

If you want a structured output that avoids string manipulation issues:

def step1(input):
    return input + ["Step 1 processed"]

def step2(input):
    return input + ["Step 2 processed"]

def step3(input):
    return input + ["Step 3 processed"]

runnables = [
    RunnableLambda(lambda x: step1(x) if isinstance(x, list) else step1([x])),
    RunnableLambda(step2),
    RunnableLambda(step3)
]

chain = RunnableSequence(*runnables)
output = chain.invoke([])
print(output)  # ['Step 1 processed', 'Step 2 processed', 'Step 3 processed']

Key Takeaways

✅ RunnableSequence executes left to right.
✅ If functions prepend their output, it appears reversed.
✅ The correct approach is to append results to preserve execution order.

This should now behave exactly as you expect! 🚀 Let me know if you need further tweaks.

Great question! By default, RunnableSequence passes the output of one step directly as the input to the next step. However, there are ways to introduce steps that don’t feed into the next step’s input while keeping everything procedural. Here are some approaches:

1️⃣ Using Side Effects (Parallel Processing)

If a step should run but not modify the main data flow, you can use side effects like logging, database updates, or function calls that don’t alter the pipeline.

def step1(input):
    print("Logging: Step 1 executed")  # Side effect
    return f"Step 1 processed: {input}"  # Only passes this forward

def step2(input):
    print("Logging: Step 2 executed")  # Another side effect
    return f"Step 2 processed: {input}"

def step3(input):
    return f"Step 3 processed: {input}"

runnables = [
    RunnableLambda(step1),
    RunnableLambda(step2),
    RunnableLambda(step3),
]

chain = RunnableSequence(*runnables)

output = chain.invoke("Start")
print(output)

Here, print() statements run but don’t modify the flow.
If you need to store data instead, replace print() with a database.save() or logging.info().

2️⃣ Using `RunnableParallel` for Branching

Langchain provides RunnableParallel, which allows for multiple independent execution paths.

from langchain.schema.runnable import RunnableParallel

def step1(input):
    return f"Step 1 processed: {input}"

def step2(input):
    return f"Step 2 processed independently"

def step3(input):
    return f"Step 3 processed: {input}"

# Run step1 → (step2 and step3 run in parallel)
chain = RunnableSequence(
    RunnableLambda(step1),
    RunnableParallel({
        "main": RunnableLambda(step3),  # Continues the sequence
        "side_task": RunnableLambda(step2),  # Runs independently
    })
)

output = chain.invoke("Start")
print(output)

Expected Output

{
  "main": "Step 3 processed: Step 1 processed: Start",
  "side_task": "Step 2 processed independently"
}

Step 1 → feeds into Step 3
Step 2 runs independently (not affecting the main flow)

3️⃣ Using a Custom Wrapper to Skip Inputs

Another approach is to create a function that takes input but discards it, ensuring it runs but doesn’t alter the chain.

def step1(input):
    return f"Step 1 processed: {input}"

def independent_step(_):  # Discards input
    print("This is an independent step that doesn't modify flow.")

def step3(input):
    return f"Step 3 processed: {input}"

runnables = [
    RunnableLambda(step1),
    RunnableLambda(independent_step),  # Runs but discards input
    RunnableLambda(step3),
]

chain = RunnableSequence(*runnables)
output = chain.invoke("Start")
print(output)

Step 2 (independent_step) runs but doesn’t affect the input.
Step 3 gets input directly from Step 1.

✅ Summary

Approach	Method	Effect
Side Effects	Logging, DB updates inside steps	Runs code without modifying flow
RunnableParallel	Parallel execution	Allows independent processing paths
Custom Wrapper	Step that ignores input	Runs but doesn’t interfere

🚀 Now you have full control over which steps feed into each other and which run independently! Let me know if you need more customization. 😃

I am having an issue, I have two chain elements following this format

prompt_template = prompts[step[“prompt”]][“template”] prompt = PromptTemplate.from_template(prompt_template)

        def store_result_and_return(response, _output=None):
            """Stores LLM-generated outputs into `generated_inputs` and returns them."""

            # 🛑 Debug: Check if step_output is set correctly
            if _output is None:
                raise ValueError("⚠️ ERROR: `step_output` is missing in store_result_and_return()!")

            # ✅ Store response in generated_inputs dictionary
            generated_inputs[_output] = response

            # ✅ Logging Execution Step
            log_entry = {
                "timestamp": str(datetime.datetime.now()),
                "step": step_name,
                "type": "LLM",
                "input": get_step_inputs(step_inputs),
                "output": response
            }
            execution_log.append(log_entry)
            print(f"\n✅ LLM Step Executed: {step_name}\n🔹 Input: {get_step_inputs(step_inputs)}\n📝 Output: {response}\n")

            # 📝 Debugging Log
            print(f"\n📝 Stored Result -> `{_output}`:\n{response}\n")

            return {_output: response}

        def execute_llm_call(_inputs, _output, _prompt):
            """Executes the LLM call and stores the result."""
            required_input = get_step_inputs(_inputs)
            print(f"{step_name} ------> ")
            print(required_input)
            formatted_prompt = prompt.format(**required_input)
            print(formatted_prompt)
            response = llm.invoke(formatted_prompt)
            return store_result_and_return(response, _output=_output)

        current_step = RunnableLambda(
            lambda _: execute_llm_call(_inputs=step_inputs, _output=step_output, _prompt=prompt),  # ✅ Calls function with correct argument
            name=f"LLM_{step_name}"
        )

i am adding the current step to a list. When i pass this list to a runnablesequence i get the last element twice

It looks like you’re encountering a late binding issue due to how Python handles closures in loops. This means that step_inputs, step_output, and prompt might be referencing the last value set in the loop, instead of capturing the correct value at the moment of function definition.

✅ Solution: Capture Loop Variables Using `functools.partial` or `lambda` Default Arguments

Instead of relying on a closure that captures mutable loop variables, explicitly bind the arguments when creating RunnableLambda.

🔹 Fix Using `functools.partial` (Recommended)

Modify your RunnableLambda definition to capture the values properly:

from functools import partial
from langchain.schema.runnable import RunnableLambda

# Inside your loop where you define `current_step`
current_step = RunnableLambda(
    partial(
        execute_llm_call,
        _inputs=step_inputs,  # Capturing values immediately
        _output=step_output,
        _prompt=prompt
    ),
    name=f"LLM_{step_name}"
)

✅ Why does this work?

partial() freezes the values of _inputs, _output, and _prompt at the time of function creation.
Prevents the function from using the last values that exist after the loop ends.

🔹 Alternative Fix Using `lambda` Default Arguments

If you don’t want to use functools.partial, another approach is to explicitly pass values in the lambda using default arguments:

current_step = RunnableLambda(
    lambda _: execute_llm_call(
        _inputs=step_inputs,
        _output=step_output,
        _prompt=prompt
    ),
    name=f"LLM_{step_name}"
)

Becomes:

current_step = RunnableLambda(
    lambda _, _inputs=step_inputs, _output=step_output, _prompt=prompt:
        execute_llm_call(_inputs, _output, _prompt),
    name=f"LLM_{step_name}"
)

✅ Why does this work?

lambda _, _inputs=step_inputs, _output=step_output, _prompt=prompt: ...
ensures that _inputs, _output, and _prompt are captured as default arguments.
Prevents Python from using the last values after the loop completes.

Final Thoughts

Approach	Pros	Cons
`functools.partial` ✅	Cleaner, avoids lambda nesting, best practice	Slightly more verbose
`lambda` with default args ✅	More inline, avoids `partial` import	Can look cluttered if many parameters

I recommend using partial, but both solutions will work. This should fix the issue of duplicate last elements in your RunnableSequence. 🚀 Let me know if you need further clarification!