5. Chains Branching

Dynamic routing and conditional branching using RunnableBranch

In this section, we will explore dynamic workflow routing by branching execution paths based on conditions using RunnableBranch.

Objectives

1. Implement conditional branching using RunnableBranch in LCEL.
2. Build a sentiment classification chain to automatically categorize text inputs.
3. Route input dynamically to dedicated specialized response chains depending on classification results.

---

Branching Chains Plan

Goal

Classify the sentiment of user feedback (positive, negative, neutral, or escalate) and dynamically route execution to the appropriate support responder chain.

Sample Input

{"feedback": "The product is terrible. It broke after just one use and the quality is very poor."}

Sample Output

An automated customer service response addressing the negative sentiment of the feedback.

Plan

1. Create specialized prompt templates for positive, negative, neutral, and escalation feedback.
2. Construct a classification chain that analyzes feedback and returns a category string (“positive”, “negative”, “neutral”, “escalate”).
3. Set up a RunnableBranch that contains conditional checks mapped to each responder chain, plus a default fallback.
4. Compose the final pipeline by connecting the classification chain to the branching router: chain = classification_chain | branches.
5. Invoke the pipeline with user feedback.

---

Step-by-Step Implementation

Step 1: Define Specialized Responder Chains

First, we create prompt templates and chains tailored to specific sentiment reactions: positive thank-yous, negative issue handling, neutral detail gathering, and an escalation fallback.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser

model = ChatOpenAI(model="gpt-4o")

# Positive responder
positive_feedback_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "Generate a thank you note for this positive feedback: {feedback}."),
])
positive_chain = positive_feedback_template | model | StrOutputParser()

# Negative responder
negative_feedback_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "Generate a response addressing this negative feedback: {feedback}."),
])
negative_chain = negative_feedback_template | model | StrOutputParser()

# Neutral responder
neutral_feedback_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "Generate a request for more details for this neutral feedback: {feedback}."),
])
neutral_chain = neutral_feedback_template | model | StrOutputParser()

# Escalation fallback responder
escalate_feedback_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "Generate a message to escalate this feedback to a human agent: {feedback}."),
])
escalate_chain = escalate_feedback_template | model | StrOutputParser()

Step 2: Define the Sentiment Classifier Chain

We create a classifier chain that acts as the entry node. It prompts the model to classify the user’s feedback into one of the four categories: positive, negative, neutral, or escalate.

classification_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "Classify the sentiment of this feedback as positive, negative, neutral, or escalate: {feedback}."),
])

classification_chain = classification_template | model | StrOutputParser()

Step 3: Configure routing conditions via RunnableBranch

Now, we define conditional checks mapped to the corresponding chains we set up in Step 1. RunnableBranch takes pairs of (condition_callable, runnable_chain) and a final fallback chain.

from langchain.schema.runnable import RunnableBranch

branches = RunnableBranch(
    (lambda x: "positive" in x, positive_chain),
    (lambda x: "negative" in x, negative_chain),
    (lambda x: "neutral" in x, neutral_chain),
    escalate_chain  # Default fallback
)

Step 4: Compose Classifier and Router

Finally, we connect the classifier chain directly to the router branches using the pipe operator. The classification result is forwarded directly to the branches logic to select the correct execution path.

# The output text of classification_chain flows into the branches routing logic
chain = classification_chain | branches

---

Complete Combined Code

Below is the complete, consolidated Python script uniting all of the steps above:

from dotenv import load_dotenv
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableBranch
from langchain_openai import ChatOpenAI

# Load environment variables from .env
load_dotenv()

# Create a ChatOpenAI model
model = ChatOpenAI(model="gpt-4o")

# Define prompt templates for different feedback types
positive_feedback_template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human",
         "Generate a thank you note for this positive feedback: {feedback}."),
    ]
)

negative_feedback_template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human",
         "Generate a response addressing this negative feedback: {feedback}."),
    ]
)

neutral_feedback_template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        (
            "human",
            "Generate a request for more details for this neutral feedback: {feedback}.",
        ),
    ]
)

escalate_feedback_template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        (
            "human",
            "Generate a message to escalate this feedback to a human agent: {feedback}.",
        ),
    ]
)

# Define the feedback classification template
classification_template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human",
         "Classify the sentiment of this feedback as positive, negative, neutral, or escalate: {feedback}."),
    ]
)

# Define the runnable branches for handling feedback
branches = RunnableBranch(
    (
        lambda x: "positive" in x,
        positive_feedback_template | model | StrOutputParser()  # Positive feedback chain
    ),
    (
        lambda x: "negative" in x,
        negative_feedback_template | model | StrOutputParser()  # Negative feedback chain
    ),
    (
        lambda x: "neutral" in x,
        neutral_feedback_template | model | StrOutputParser()  # Neutral feedback chain
    ),
    escalate_feedback_template | model | StrOutputParser()
)

# Create the classification chain
classification_chain = classification_template | model | StrOutputParser()

# Combine classification and response generation into one chain
chain = classification_chain | branches

# Run the chain with an example review
review = "The product is terrible. It broke after just one use and the quality is very poor."
result = chain.invoke({"feedback": review})

# Output the result
print(result)

---

Practice & Exercises

To reinforce what you’ve learned in this section, practice with the interactive notebook:

Practice & Exercises

Practice setting up conditional logic nodes, routing user inputs to custom domains, and implementing fallback handlers.

💻 VS Code | 🚀 Colab | 📥 Download