04. LLM chain routing (RunnableLambda, RunnableBranch)

RunnableBranch Is a powerful tool that allows you to dynamically route logic based on input. Through this, the developer Flexible definition of various processing paths based on the characteristics of the input data You can.

RunnableBranch Helps to implement complex decision trees in a simple and intuitive way. This greatly improves the readability and maintainability of the code, and promotes the modularity and reusability of the logic.

Also, RunnableBranch At runtime, you can dynamically evaluate the branch conditions and choose the appropriate processing routine, which increases the adaptability and scalability of the system.

Due to these features, RunnableBranch can be utilized in a variety of domains, especially useful for developing applications with high diversity and volatility of input data. RunnableBranch Effectively utilizing it can reduce the complexity of your code and improve the flexibility and performance of your system.

Dynamic logic routing by input

LangChain Expression Language covers how to perform routing.

Routing allows the output of the previous step to create an indeterminate chain that defines the next step. Routing helps to provide structure and consistency to interaction with LLM.

There are two ways to do routing.

1. RunnableLambda Returns a conditionally viable object from (recommended)

2. RunnableBranch use

Both methods ask input questions in the first step 수학 , 과학 or 기타 I'll explain it using a two-step sequence that categorizes it as for and then routes it to that prompt chain.

Copy

# Configuration file for managing API keys as environment variables
from dotenv import load_dotenv

# Load API key information
load_dotenv()

Copy

 True 

Copy

# LangSmith set up tracking. https://smith.langchain.com
# !pip install langchain-teddynote
from langchain_teddynote import logging

# Enter a project name.
logging.langsmith("LCEL-Advanced")

Copy

 Start tracking LangSmith. 
[Project name] 
LCEL-Advanced 

Simple example

First, the incoming question math , science , or etc I'll create a Chain that categorizes it as one of them.

Copy

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(
    """Please categorize the given user question as either `Math`, `Science`, or `Other`. Do not answer with more than one word.

<question>
{question}
</question>

Classification:"""
)

# Create a chain.
chain = (
    prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()  # Use a string output parser.

Classify questions using the chain you created.

Copy

# Call the chain by entering a question.
chain.invoke({"question": "2+2 Whwat is?"})

Copy

 'Mathematics' 

Copy

# Call the chain by entering a question.
chain.invoke({"question": "What is the law of action and reaction?"})

Copy

 'science' 

Copy

# Call the chain by entering a question.
chain.invoke({"question": "What kind of company is Google?"})

Copy

 'Other' 

Now let's create three subchains.

Copy

math_chain = (
    PromptTemplate.from_template(
        """You are an expert in math. \
Always answer questions starting with "As Mr. Kebong said,..". \
Respond to the following question:

Question: {question}
Answer:"""
    )
    # I use OpenAI's LLM.
    | ChatOpenAI(model="gpt-4o-mini")
)

science_chain = (
    PromptTemplate.from_template(
        """You are an expert in science. \
Always answer questions starting with "Sir Isaac Newton said,..". \
Respond to the following question:

Question: {question}
Answer:"""
    )
    # I use OpenAI's LLM.
    | ChatOpenAI(model="gpt-4o-mini")
)

general_chain = (
    PromptTemplate.from_template(
        """Respond to the following question concisely:

Question: {question}
Answer:"""
    )
    # I use OpenAI's LLM.
    | ChatOpenAI(model="gpt-4o-mini")
)

Using custom functions

This is the method recommended by the LangChain official dock, for routing between different outputs Custom function for RunnableLambda You can also use it by wrapping it with.

Copy

def route(info):
    # If the topic includes “Mathematics”
    if "수학" in info["topic"].lower():
        # return datascience_chain
        return math_chain
    # If the topic includes “science”
    elif "과학" in info["topic"].lower():
        # Returns art_chain
        return science_chain
    # In other cases
    else:
        # Return general_chain
        return general_chain

Copy

from operator import itemgetter
from langchain_core.runnables import RunnableLambda

full_chain = (
    {"topic": chain, "question": itemgetter("question")}
    | RunnableLambda(
        # Pass a function specifying the path as an argument.
        route
    )
    | StrOutputParser()
)

Copy

# Call up the chain by entering a math-related question.
full_chain.invoke({"question": "Tell me about the concept of calculus."})

Copy

 'Speaking from the teacher, Calculus is a field of mathematics, consisting of two main concepts: differential and integral. Differentials deal with the rate of change of the function, and the process of finding the slope of the function at a specific point. This is useful for analyzing the speed of an object, the marginal cost of economics, etc. On the other hand, the integral is to get the area of the function, which is used to calculate the area under the function in a given interval. Calculus is applied in a variety of fields, including science, engineering, and economics, and is an important tool for understanding successive changes and the resulting totals.' 

Copy

# Call up the chain by typing a science-related question.
full_chain.invoke({"question": "How does gravity work?"})

Copy

 'Speaking from Isaac Newton, gravity is a force proportional to the mass between objects and inversely proportional to the square of the distance. This is the force that the two objects attract each other, and the power decreases as the square of the distance increases. This principle is explained by Newton's law of manpower, indicating that all objects have properties that attract each other. Thus, large-massed objects like Earth will pull the surrounding small objects strongly, which is also why we are attached to Earth.' 

Copy

# Call the chain by entering other questions.
full_chain.invoke({"question": "RAG(Retrieval Augmented Generation)What is?"})

Copy

 'RAG (Creative Enhancement Search) is a model that combines information search with natural language generation, and after searching for relevant documents for a given question, it is a way to generate responses based on these documents. This allows you to provide a more accurate and information-based answer.' 

RunnableBranch

RunnableBranch A special type that can define the conditions to run and the Runnable according to the input value Runnable is.

However, we do not provide any functionality that cannot be implemented with the custom functions described above, so we recommend using custom functions.

grammar

• RunnableBranch Is initialized with a (condition, Runnable) pair of list and basic Runnable.

• Select the branch by passing the input value passed at the time of call to each condition.

• Select the first condition that is evaluated as True, and run the Runnable corresponding to that condition with the input value.

• If nothing matches the conditions provided basic Runnable Run.

Copy

from operator import itemgetter
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    # Check if the topic contains "math", and if so, run math_chain.
    (lambda x: "수학" in x["topic"].lower(), math_chain),
    # Check if the topic contains "science", and if so, run science_chain.
    (lambda x: "과학" in x["topic"].lower(), science_chain),
    # If the above conditions are not met, run general_chain.
    general_chain,
)
# Defines an entire chain that takes a topic and a question and executes a branch.
full_chain = (
    {"topic": chain, "question": itemgetter("question")} | branch | StrOutputParser()
)

Copy

# Enter a question to run the entire chain.
full_chain.invoke({"question": "Tell me about the concept of calculus."})

Copy

 'Speaking from the teacher, Calculus is a field of mathematics, a study of function changes and the amount that follows. Calculus can be largely divided into two main concepts: derivative and integral.\n\n derivative is a method of measuring the instantaneous rate of change of a function, the process of calculating the slope at a specific point. For example, when measuring the speed of a car, analyzing the time change for a certain moment's position change is a derivative application. On the other hand, integration is the process of obtaining the total area or sum of functions. This can be understood as a differential inverse process, and is used to calculate the area under the function in a given interval. For example, you use integrals when you want to know the distance the object has moved or the area under the curve.\n\n Calculus is widely utilized in a variety of fields such as physics, engineering, economics, etc., and plays an important role in understanding and predicting changing systems. To do.' 

Copy

# Enter a question to run the entire chain.
full_chain.invoke({"question": "How do you calculate acceleration due to gravity?"})

Copy

 'Speaking from Isaac Newton, gravitational acceleration means the acceleration that an object receives by the gravity of an object like Earth. Gravity acceleration on Earth is known as about 9.81 m/s². To calculate this, we use the laws of manpower. According to the law of manpower, the gravity between two objects is represented by the formula: \n\n\[F = \\frac{G \\cdot m_1 \\cdot m_2}{r^2} \\]\n\n where \\ (F\) is gravity, For Earth, if you place the mass of an object \\(m\), the mass of the Earth \\(M\\), and the radius of the Earth as \\(R\\), the gravitational acceleration \\(g\\) can be expressed as: \n\n\[g =\\\* 

Copy

# Enter a question to run the entire chain.
full_chain.invoke({"question": "RAG(Retrieval Augmented Generation)은 무엇인가요?"})

Copy

 'RAG (Litrigal Augmentation Generation) is an approach that combines information retrieval and generation models, a way to retrieve relevant information from external databases and utilize it to generate more accurate and rich responses to generate answers to a given question.'