Conditional Workflows in LangGraph

Welcome to the seventh installment of our Agentic AI with LangGraph series. Having covered sequential and parallel workflows, we now move to the most critical logic gate in agentic design: Conditional Workflows. This documentation explores how to build intelligent agents that can branch into different paths based on real-time data and reasoning.

1. Understanding Conditional Workflows #

A Conditional Workflow is a branching architecture where the system chooses a specific path based on a predefined condition or LLM reasoning. While it looks visually similar to a parallel workflow, there is a fundamental difference:

• Parallel Workflows: You enter all branches simultaneously to execute multiple tasks at once.
• Conditional Workflows: You enter only one branch based on a condition.

Think of it as the “if-else” statement of AI workflows. This capability is essential for building complex agents, as nearly every professional-grade system requires the ability to choose different actions depending on the situation.

---

2. The Core Mechanism: Routing and Conditional Edges #

To implement conditional logic in LangGraph, you need two primary components:

The Routing Function

Unlike standard nodes, a routing function (often called a “condition checker”) is a standalone function that takes the current State as input. Instead of returning a state update, it returns the name of the next node the graph should execute.

add_conditional_edges

Instead of using the standard add_edge function, you use add_conditional_edges. This function requires two arguments:

1. Start Node: The node that just finished executing.
2. Routing Function: The logic that determines which node comes next.

---

3. Non-LLM Example: Quadratic Equation Solver #

To understand the logic without LLM complexity, consider a workflow that solves a quadratic equation (ax2+bx+c=0).

The Decision Logic: The workflow calculates the discriminant (D=b2−4ac). The path taken depends entirely on the value of D:

• If D>0: Route to the Real Roots node to calculate two distinct solutions.
• If D=0: Route to the Repeated Roots node for a single repeating solution.
• If D<0: Route to the No Real Roots node.

Graph Construction:

1. Nodes: Show Equation → Calculate Discriminant → [Conditional Split].
2. Routing: A check_condition function reads the discriminant from the state and returns the string name of the appropriate root-calculation node.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Literal
class QuadState(TypedDict):

    a: int
    b: int
    c: int

    equation: str
    discriminant: float
    result: str

def show_equation(state: QuadState):

    equation = f'{state["a"]}x2{state["b"]}x{state["c"]}'

    return {'equation':equation}

def calculate_discriminant(state: QuadState):

    discriminant = state["b"]**2 - (4*state["a"]*state["c"])

    return {'discriminant': discriminant}

def real_roots(state: QuadState):

    root1 = (-state["b"] + state["discriminant"]**0.5)/(2*state["a"])
    root2 = (-state["b"] - state["discriminant"]**0.5)/(2*state["a"])

    result = f'The roots are {root1} and {root2}'

    return {'result': result}

def repeated_roots(state: QuadState):

    root = (-state["b"])/(2*state["a"])

    result = f'Only repeating root is {root}'

    return {'result': result}

def no_real_roots(state: QuadState):

    result = f'No real roots'

    return {'result': result}

def check_condition(state: QuadState) -> Literal["real_roots", "repeated_roots", "no_real_roots"]:

    if state['discriminant'] > 0:
        return "real_roots"
    elif state['discriminant'] == 0:
        return "repeated_roots"
    else:
        return "no_real_roots"

graph = StateGraph(QuadState)

graph.add_node('show_equation', show_equation)
graph.add_node('calculate_discriminant', calculate_discriminant)
graph.add_node('real_roots', real_roots)
graph.add_node('repeated_roots', repeated_roots)
graph.add_node('no_real_roots', no_real_roots)


graph.add_edge(START, 'show_equation')
graph.add_edge('show_equation', 'calculate_discriminant')

graph.add_conditional_edges('calculate_discriminant', check_condition)
graph.add_edge('real_roots', END)
graph.add_edge('repeated_roots', END)
graph.add_edge('no_real_roots', END)

workflow = graph.compile()

initial_state = {
    'a': 2, 
    'b': 4,
    'c': 2
}

workflow.invoke(initial_state)

---

4. LLM-Based Example: Smart Customer Support Agent #

In a production scenario, we use the LLM to drive the branching logic. Consider a system that processes customer reviews and generates a tailored response.

Step 1: Sentiment Extraction

The agent first uses an LLM with a structured output schema to classify the review as “Positive” or “Negative”.

Step 2: The Conditional Split

A routing function checks the “Sentiment” key in the state.

• Path A (Positive): The agent moves to a Positive Response node to draft a warm thank-you message.
• Path B (Negative): The agent branches into a deeper Diagnosis path.

Step 3: Deep Diagnosis (Negative Branch Only)

If the sentiment is negative, the agent runs a diagnosis to extract:

• Issue Type: (e.g., UI, Performance, Bug, or Support).
• Tone: (e.g., Frustrated, Angry).
• Urgency: (e.g., Low, Medium, High).

The final response for negative reviews is generated by considering all these diagnosed factors, ensuring a much more empathetic and helpful resolution than a generic template.

---

5. Implementation Framework: add_conditional_edges #

1. Setup and State Definition
First, define the schemas for the LLM's structured responses and the global graph state
.
from typing import TypedDict, Literal
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

# Define Schemas for Structured Output
class SentimentSchema(BaseModel):
    sentiment: Literal["Positive", "Negative"] = Field(description="Sentiment of the review")

class DiagnosisSchema(BaseModel):
    issue_type: Literal["UX", "Performance", "Bug", "Support", "Other"]
    tone: str = Field(description="Emotional tone expressed by the user")
    urgency: Literal["Low", "Medium", "High"]

# Define the Global State
class ReviewState(TypedDict):
    review: str
    sentiment: str
    diagnosis: dict
    response: str

# Initialize Models
model = ChatOpenAI(model="gpt-4o-mini")
structured_model_sentiment = model.with_structured_output(SentimentSchema)
structured_model_diagnosis = model.with_structured_output(DiagnosisSchema)
2. Defining the Nodes (Functions)
Nodes are the "muscles" of the graph. Each function performs a specific task and updates the state
.
def find_sentiment(state: ReviewState):
    prompt = f"What is the sentiment of the following review? \n\n {state['review']}"
    result = structured_model_sentiment.invoke(prompt)
    return {"sentiment": result.sentiment}

def positive_response(state: ReviewState):
    prompt = f"Write a warm thank-you message for this review: {state['review']}"
    response = model.invoke(prompt)
    return {"response": response.content}

def run_diagnosis(state: ReviewState):
    prompt = f"Diagnose this negative review: {state['review']}. Return issue type, tone, and urgency."
    result = structured_model_diagnosis.invoke(prompt)
    # Convert Pydantic object to dictionary for the state
    return {"diagnosis": result.model_dump()}

def negative_response(state: ReviewState):
    diag = state['diagnosis']
    prompt = (f"The user had a {diag['issue_type']} issue with a {diag['tone']} tone. "
              f"Urgency is {diag['urgency']}. Write an empathetic resolution.")
    response = model.invoke(prompt)
    return {"response": response.content}
3. The Routing Logic
The Routing Function acts as the decision-maker, determining which node to visit next based on the state
.
def check_sentiment(state: ReviewState):
    if state["sentiment"] == "Positive":
        return "positive_response"
    else:
        return "run_diagnosis"
4. Building and Compiling the Graph
This stage connects the nodes using standard and conditional edges
.
builder = StateGraph(ReviewState)

# Add all nodes to the graph
builder.add_node("find_sentiment", find_sentiment)
builder.add_node("positive_response", positive_response)
builder.add_node("run_diagnosis", run_diagnosis)
builder.add_node("negative_response", negative_response)

# Define the flow
builder.add_edge(START, "find_sentiment")

# Apply the Conditional Logic
builder.add_conditional_edges(
    "find_sentiment",
    check_sentiment
)

# Connect remaining paths to END
builder.add_edge("positive_response", END)
builder.add_edge("run_diagnosis", "negative_response")
builder.add_edge("negative_response", END)

# Compile the workflow
workflow = builder.compile()
5. Execution Example
You can now invoke the workflow by passing a customer review in the initial state
.
initial_state = {"review": "The app keeps freezing on the login screen. This is unacceptable!"}
result = workflow.invoke(initial_state)

print(f"Sentiment: {result['sentiment']}")
print(f"Diagnosis: {result['diagnosis']}")
print(f"Agent Response: {result['response']}")

---

Summary Table: Workflow Comparison #

Feature	Sequential	Parallel	Conditional
Logic	Linear (A → B)	Simultaneous (A + B)	Selection (A OR B)
Execution	One node at a time	Multiple nodes at once	One branch per run
Routing	Static Edges	Static Edges	Dynamic Functions
API	add_edge	add_edge	add_conditional_edges

By mastering conditional workflows, you transition from building simple scripts to creating reasoning agents that can adapt their behavior based on the specific content and context of a user’s request.