LangChain's Runnable objects provide a modular and flexible approach to designing workflows by enabling the chaining, parallel execution, and transformation of data. These utilities allow for efficient handling of structured inputs and outputs, with minimal code overhead.
Key Components is:
RunnableLambda: A lightweight utility that enables the application of custom logic through lambda functions, ideal for dynamic and quick data transformations.
RunnablePassthrough: Designed to pass input data unchanged or augment it with additional attributes when paired with the .assign() method.
itemgetter: A Python operator module utility for efficiently extracting specific keys or indices from structured data such as dictionaries or tuples.
These tools can be combined to build powerful workflows, such as:
Extracting and processing specific data elements using itemgetter.
Performing custom transformations with RunnableLambda.
Creating end-to-end data pipelines with Runnable chains.
By leveraging these components, users can design scalable and reusable pipelines for machine learning and data processing workflows.
Table of Contents
References
Environment Setup
[Note]
langchain-opentutorial is a package that provides a set of easy-to-use environment setup, useful functions and utilities for tutorials.
You can also load the OPEN_API_KEY from the .env file.
from dotenv import load_dotenv
load_dotenv(override=True)
True
# Set local environment variables
from langchain_opentutorial import set_env
set_env(
{
"LANGCHAIN_TRACING_V2": "true",
"LANGCHAIN_ENDPOINT": "https://api.smith.langchain.com",
"LANGCHAIN_PROJECT": "05-Runnable",
}
)
Environment variables have been set successfully.
Efficient Data Handling with RunnablePassthrough
RunnablePassthrough is a utility designed to streamline data processing workflows by either passing input data unchanged or enhancing it with additional attributes. Its flexibility makes it a valuable tool for handling data in pipelines where minimal transformation or selective augmentation is required.
Simple Data Forwarding
Suitable for scenarios where no transformation is required, such as logging raw data or passing it to downstream systems.
Dynamic Data Augmentation
Enables the addition of metadata or context to input data for use in machine learning pipelines or analytics systems.
RunnablePassthrough can either pass the input unchanged or append additional keys to it.
When RunnablePassthrough() is called on its own, it simply takes the input and passes it as is.
When called using RunnablePassthrough.assign(...), it takes the input and adds additional arguments provided to the assign function.
RunnablePassthrough
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
# Create the prompt and llm
prompt = PromptTemplate.from_template("What is 10 times {num}?")
llm = ChatOpenAI(temperature=0)
# Create the chain
chain = prompt | llm
When invoking the chain with invoke(), the input data must be of type dictionary.
# Execute the chain : input dtype as 'dictionary'
chain.invoke({"num": 5})
Here is an example using RunnablePassthrough.RunnablePassthrough is a runnable object with the following characteristics:
Basic Operation
Performs a simple pass-through function that forwards input values directly to output
Can be executed independently using the invoke() method
Use Cases
Useful when you need to pass data through chain steps without modification
Can be combined with other components to build complex data pipelines
Particularly helpful when you need to preserve original input while adding new fields
Input Handling
Accepts dictionary-type inputs
Can handle single values as well
Maintains data structure throughout the chain
from langchain_core.runnables import RunnablePassthrough
# Runnable
RunnablePassthrough().invoke({"num": 10})
{'num': 10}
Here is an example of creating a chain using RunnablePassthrough.
runnable_chain = {"num": RunnablePassthrough()} | prompt | ChatOpenAI()
# The dict value has been updated with RunnablePassthrough().
runnable_chain.invoke(10)
Efficient Parallel Execution with RunnableParallel
RunnableParallel is a utility designed to execute multiple Runnable objects concurrently, streamlining workflows that require parallel processing. It distributes input data across different components, collects their results, and combines them into a unified output. This functionality makes it a powerful tool for optimizing workflows where tasks can be performed independently and simultaneously.
Concurrent Execution
Executes multiple Runnable objects simultaneously, reducing the time required for tasks that can be parallelized.
Unified Output Management
Combines the results from all parallel executions into a single, cohesive output, simplifying downstream processing.
Flexibility
Can handle diverse input types and support complex workflows by distributing the workload efficiently.
from langchain_core.runnables import RunnableParallel
# Create an instance of RunnableParallel. This instance allows multiple Runnable objects to be executed in parallel.
runnable = RunnableParallel(
# Pass a RunnablePassthrough instance as the 'passed' keyword argument. This simply passes the input data through without modification.
passed=RunnablePassthrough(),
# Use RunnablePassthrough.assign as the 'extra' keyword argument to assign a lambda function 'mult'.
# This function multiplies the value associated with the 'num' key in the input dictionary by 3.
extra=RunnablePassthrough.assign(mult=lambda x: x["num"] * 3),
# Pass a lambda function as the 'modified' keyword argument.
# This function adds 1 to the value associated with the 'num' key in the input dictionary.
modified=lambda x: x["num"] + 1,
)
# Call the invoke method on the runnable instance, passing a dictionary {'num': 1} as input.
runnable.invoke({"num": 1})
chain1 = (
{"country": RunnablePassthrough()}
| PromptTemplate.from_template("What is the capital of {country}?")
| ChatOpenAI()
)
chain2 = (
{"country": RunnablePassthrough()}
| PromptTemplate.from_template("What is the area of {country}?")
| ChatOpenAI()
)
combined_chain = RunnableParallel(capital=chain1, area=chain2)
combined_chain.invoke("United States of America")
{'capital': AIMessage(content='The capital of the United States of America is Washington, D.C.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 15, 'prompt_tokens': 17, 'total_tokens': 32, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-29437a26-8661-4f15-a655-3b3ca6aa0e8c-0', usage_metadata={'input_tokens': 17, 'output_tokens': 15, 'total_tokens': 32, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}),
'area': AIMessage(content='The total land area of the United States of America is approximately 3.8 million square miles.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 21, 'prompt_tokens': 17, 'total_tokens': 38, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-5004e08c-dd66-4c7c-bc3f-60821fecc403-0', usage_metadata={'input_tokens': 17, 'output_tokens': 21, 'total_tokens': 38, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})}
Dynamic Processing with RunnableLambda
RunnableLambda is a flexible utility that allows developers to define custom data transformation logic using lambda functions. By enabling quick and easy implementation of custom processing workflows, RunnableLambda simplifies the creation of tailored data pipelines while ensuring minimal setup overhead.
Customizable Data Transformation
Allows users to define custom logic for transforming input data using lambda functions, offering unparalleled flexibility.
Lightweight and Simple
Provides a straightforward way to implement ad-hoc processing without the need for extensive class or function definitions.
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from datetime import datetime
def get_today(a):
# Get today's date
return datetime.today().strftime("%b-%d")
# Print today's date
get_today(None)
'Jan-04'
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
# Create the prompt and llm
prompt = PromptTemplate.from_template(
"List {n} famous people whose birthday is on {today}. Include their date of birth."
)
llm = ChatOpenAI(temperature=0, model_name="gpt-4o")
# Create the chain
chain = (
{"today": RunnableLambda(get_today), "n": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# Output
print(chain.invoke(3))
Here are three famous people born on January 4:
1. **Isaac Newton** - Born on January 4, 1643 (according to the Gregorian calendar; December 25, 1642, in the Julian calendar), he was an English mathematician, physicist, astronomer, and author who is widely recognized as one of the most influential scientists of all time.
2. **Louis Braille** - Born on January 4, 1809, he was a French educator and inventor of a system of reading and writing for use by the blind or visually impaired, known as Braille.
3. **Michael Stipe** - Born on January 4, 1960, he is an American singer-songwriter and the lead vocalist of the alternative rock band R.E.M.
Extracting Specific Keys Using itemgetter
itemgetter is a utility function from Python's operator module with the following features and benefits:
Core Functionality
Efficiently extracts values from specific keys or indices in dictionaries, tuples, and lists
Capable of extracting multiple keys or indices simultaneously
Supports functional programming style
Performance Optimization
More efficient than regular indexing for repetitive key access operations
Optimized memory usage
Performance advantages when processing large datasets
Usage in LangChain
Data filtering in chain compositions
Selective extraction from complex input structures
Combines with other Runnable objects for data preprocessing
from operator import itemgetter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
# Function that returns the length of a sentence.
def length_function(text):
return len(text)
# Function that returns the product of the lengths of two sentences.
def _multiple_length_function(text1, text2):
return len(text1) * len(text2)
# Function that uses _multiple_length_function to return the product of the lengths of two sentences.
def multiple_length_function(_dict):
return _multiple_length_function(_dict["text1"], _dict["text2"])
prompt = ChatPromptTemplate.from_template("What is {a} + {b}?")
model = ChatOpenAI()
chain1 = prompt | model
chain = (
{
"a": itemgetter("word1") | RunnableLambda(length_function),
"b": {"text1": itemgetter("word1"), "text2": itemgetter("word2")}
| RunnableLambda(multiple_length_function),
}
| prompt
| model
)
