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:
- Start Node: The node that just finished executing.
- 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 Rootsnode to calculate two distinct solutions. - If D=0: Route to the
Repeated Rootsnode for a single repeating solution. - If D<0: Route to the
No Real Rootsnode.
Graph Construction:
- Nodes:
Show Equation→Calculate Discriminant→ [Conditional Split]. - Routing: A
check_conditionfunction 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
# -----------------------------
# 1. Define State
# -----------------------------
class QuadState(TypedDict):
a: int
b: int
c: int
equation: str
discriminant: float
result: str
# -----------------------------
# 2. Show Equation
# -----------------------------
def show_equation(state: QuadState):
equation = f'{state["a"]}x² + {state["b"]}x + {state["c"]}'
return {
'equation': equation
}
# -----------------------------
# 3. Calculate Discriminant
# -----------------------------
def calculate_discriminant(state: QuadState):
discriminant = state["b"]**2 - (4 * state["a"] * state["c"])
return {
'discriminant': discriminant
}
# -----------------------------
# 4. Real Roots
# -----------------------------
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
}
# -----------------------------
# 5. Repeated Root
# -----------------------------
def repeated_roots(state: QuadState):
root = (-state["b"]) / (2 * state["a"])
result = f'Only repeating root is {root}'
return {
'result': result
}
# -----------------------------
# 6. No Real Roots
# -----------------------------
def no_real_roots(state: QuadState):
result = 'No real roots'
return {
'result': result
}
# -----------------------------
# 7. Conditional Router
# -----------------------------
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"
# -----------------------------
# 8. Create Graph
# -----------------------------
graph = StateGraph(QuadState)
# -----------------------------
# 9. Add Nodes
# -----------------------------
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
)
# -----------------------------
# 10. Add Edges
# -----------------------------
graph.add_edge(
START,
'show_equation'
)
graph.add_edge(
'show_equation',
'calculate_discriminant'
)
# -----------------------------
# 11. Conditional Edge
# -----------------------------
graph.add_conditional_edges(
'calculate_discriminant',
check_condition
)
# -----------------------------
# 12. Connect Branches to END
# -----------------------------
graph.add_edge(
'real_roots',
END
)
graph.add_edge(
'repeated_roots',
END
)
graph.add_edge(
'no_real_roots',
END
)
# -----------------------------
# 13. Compile Graph
# -----------------------------
workflow = graph.compile()
# -----------------------------
# 14. Initial State
# -----------------------------
initial_state = {
'a': 2,
'b': 4,
'c': 2
}
# -----------------------------
# 15. Run Workflow
# -----------------------------
result = workflow.invoke(initial_state)
print(result)Other without LLM use example for customer support chatbot
from typing import TypedDict
from langgraph.graph import StateGraph,START,END
class State(TypedDict):
query: str
category: str
response: str
def classify_query(state:State):
query=state["query"].lower()
if "refund" in query:
category = "faq"
elif "order" in query:
category = "order"
elif "cancel" in query:
category = "cancel"
else:
category = "unknown"
return {
"category": category
}
def faq_node(state:State):
return {
"response": "Here is our refund policy..."
}
def order_node(state:State):
return{
"response": "Let me check your order status..."
}
def cancel_node(state: State):
return {
"response": "Your cancellation request has been received."
}
def unknown_node(state: State):
return {
"response": "Sorry, I don't understand your request."
}
graph=StateGraph(State)
graph.add_node("classify",classify_query)
graph.add_node("faq",faq_node)
graph.add_node("order",order_node)
graph.add_node("cancel",cancel_node)
graph.add_node("unknown",unknown_node)
graph.add_edge(START,"classify")
graph.add_conditional_edges("classify",lambda state : state["category"],
{
"faq": "faq",
"order": "order",
"cancel": "cancel",
"unknown": "unknown"
})
graph.add_edge("faq",END)
graph.add_edge("order",END)
graph.add_edge("cancel",END)
graph.add_edge("unknown",END)
app=graph.compile()
result = app.invoke({
"query": "I want a refund",
"category": "",
"response": ""
})
print(result) 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 Responsenode 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
# ============================================================
from typing import TypedDict, Literal
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
# ============================================================
# 2. Define Structured Output Schemas
# ============================================================
# Schema used by the LLM to classify the sentiment.
# The LLM is restricted to either "Positive" or "Negative".
class SentimentSchema(BaseModel):
sentiment: Literal["Positive", "Negative"] = Field(
description="Sentiment of the review"
)
# Schema used by the LLM to diagnose a negative 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"
]
# ============================================================
# 3. Define the Global Graph State
# ============================================================
# This state is shared between all nodes in the graph.
#
# Each node can:
# - Read information from the state
# - Return updates to the state
class ReviewState(TypedDict):
review: str
sentiment: str
diagnosis: dict
response: str
# ============================================================
# 4. Initialize the LLM
# ============================================================
model = ChatOpenAI(
model="gpt-4o-mini"
)
# ============================================================
# 5. Create Structured-Output Versions of the LLM
# ============================================================
# This model will return a SentimentSchema object
# instead of free-form text.
structured_model_sentiment = model.with_structured_output(
SentimentSchema
)
# This model will return a DiagnosisSchema object.
structured_model_diagnosis = model.with_structured_output(
DiagnosisSchema
)
# ============================================================
# 6. Define Graph Nodes
# ============================================================
# ------------------------------------------------------------
# Node 1: Find Sentiment
# ------------------------------------------------------------
#
# The LLM analyzes the review and determines whether
# the sentiment is Positive or Negative.
def find_sentiment(state: ReviewState):
prompt = f"""
What is the sentiment of the following review?
Review:
{state['review']}
"""
# Invoke the structured-output model.
result = structured_model_sentiment.invoke(prompt)
# Update the "sentiment" field in the graph state.
return {
"sentiment": result.sentiment
}
# ------------------------------------------------------------
# Node 2: Positive Response
# ------------------------------------------------------------
#
# This node runs when the review is positive.
def positive_response(state: ReviewState):
prompt = f"""
Write a warm and professional thank-you message
for this customer review:
{state['review']}
"""
response = model.invoke(prompt)
# Store the LLM response in the global state.
return {
"response": response.content
}
# ------------------------------------------------------------
# Node 3: Run Diagnosis
# ------------------------------------------------------------
#
# This node runs only when the review is negative.
#
# It identifies:
# - Type of issue
# - User's emotional tone
# - Urgency
def run_diagnosis(state: ReviewState):
prompt = f"""
Diagnose this negative customer review.
Review:
{state['review']}
Identify:
- Issue type
- Emotional tone
- Urgency
"""
result = structured_model_diagnosis.invoke(prompt)
# Convert the Pydantic object into a normal dictionary
# so it can be stored in the graph state.
return {
"diagnosis": result.model_dump()
}
# ------------------------------------------------------------
# Node 4: Negative Response
# ------------------------------------------------------------
#
# This node uses the diagnosis generated by the previous node
# to create a personalized response.
def negative_response(state: ReviewState):
# Get diagnosis information from the graph state.
diag = state["diagnosis"]
prompt = f"""
The user had a {diag['issue_type']} issue
with a {diag['tone']} tone.
Urgency:
{diag['urgency']}
Write an empathetic and helpful resolution.
"""
response = model.invoke(prompt)
# Store the generated response in the state.
return {
"response": response.content
}
# ============================================================
# 7. Define the Routing Function
# ============================================================
# This function decides which node should run next
# after the sentiment has been detected.
#
# It returns the NAME of the next node.
def check_sentiment(state: ReviewState):
if state["sentiment"] == "Positive":
return "positive_response"
else:
return "run_diagnosis"
# ============================================================
# 8. Create the StateGraph
# ============================================================
builder = StateGraph(ReviewState)
# ============================================================
# 9. Add 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
)
# ============================================================
# 10. Define the Graph Flow
# ============================================================
# START → Find Sentiment
builder.add_edge(
START,
"find_sentiment"
)
# ============================================================
# 11. Add Conditional Routing
# ============================================================
# After "find_sentiment" finishes,
# check_sentiment() decides which node should run.
#
# Possible routes:
#
# Positive → positive_response
# Negative → run_diagnosis
#
# Because check_sentiment() directly returns the
# destination node name, no mapping dictionary is required.
builder.add_conditional_edges(
"find_sentiment",
check_sentiment
)
# ============================================================
# 12. Connect the Remaining Nodes
# ============================================================
# Positive path:
#
# find_sentiment
# ↓
# positive_response
# ↓
# END
builder.add_edge(
"positive_response",
END
)
# Negative path:
#
# find_sentiment
# ↓
# run_diagnosis
# ↓
# negative_response
builder.add_edge(
"run_diagnosis",
"negative_response"
)
# After generating the negative response,
# terminate the workflow.
builder.add_edge(
"negative_response",
END
)
# ============================================================
# 13. Compile the Graph
# ============================================================
workflow = builder.compile()
# ============================================================
# 14. Execute the Workflow
# ============================================================
initial_state = {
"review": (
"The app keeps freezing on the login screen. "
"This is unacceptable!"
)
}
result = workflow.invoke(initial_state)
# ============================================================
# 15. Display the Result
# ============================================================
print("Sentiment:", result["sentiment"])
print("Diagnosis:", result.get("diagnosis"))
print("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.
Q.1 What is the primary difference between a parallel workflow and a conditional workflow in LangGraph?
Parallel workflows require an LLM, whereas conditional workflows are only used for mathematical computations.
In a parallel workflow, multiple branches execute simultaneously, while in a conditional workflow, only one branch is selected based on specific logic.
Parallel workflows are linear, while conditional workflows are non-linear.
Conditional workflows must always end at a single node, while parallel workflows can have multiple end points.
Explanation
Parallel workflows execute multiple branches at the same time, whereas conditional workflows evaluate a condition and follow only one appropriate execution path.
Q.2 Why is a routing function considered one of the most important components in a conditional LangGraph workflow?
It increases the speed of LLM inference.
It determines the execution path by analysing the current state and selecting the appropriate next node.
It automatically generates Python functions for new nodes.
It replaces the need for edges between nodes.
Explanation
The routing function acts as the decision-maker of a conditional workflow. It inspects the current state, evaluates conditions, and returns the next node to execute, enabling dynamic, intelligent workflow execution instead of a fixed sequence.
Q.3 When defining a conditional edge using add_conditional_edges(), what is the purpose of the routing function?
To compile the graph into an executable workflow.
To examine the current state and return the name of the next node to execute.
To force every branch to execute sequentially.
To reset the workflow state before moving to the next node.
Explanation
The routing function analyses the current state and returns the appropriate node name, allowing LangGraph to dynamically determine the next execution path.
Q.4 In the customer support workflow, if a review is classified as negative, what is the immediate next step?
Generate a thank-you response.
Immediately terminate the workflow.
Run a diagnosis to determine the issue type, tone, and urgency.
Ask the customer to submit another review.
Explanation
Negative reviews require further analysis. The diagnosis node extracts structured information such as issue type, customer tone, and urgency before generating the final response.
Q.5 Why are conditional workflows especially useful in AI applications?
They allow the workflow to make decisions dynamically based on the current state.
They always execute faster than parallel workflows.
They eliminate the need for state management.
They prevent the use of LLMs.
Explanation
Conditional workflows enable dynamic decision-making by allowing the graph to choose different execution paths depending on the current state or LLM output.
Q.6 Which LangGraph feature ensures an LLM returns fields like Sentiment or Issue Type in a structured format?
add_conditional_edges().
.with_structured_output() together with a Pydantic BaseModel.
Python's split() function.
The internal StateGraph dictionary.
Explanation
The with_structured_output() method combined with a Pydantic BaseModel forces the LLM to return predictable structured fields instead of unstructured text.
Q.7 During the Diagnosis phase for negative reviews, which three fields are extracted?
Root 1, Root 2, and Discriminant.
Username, Password, and Login Time.
Sentiment, Polarity, and Subjectivity.
Issue Type, Tone, and Urgency.
Explanation
The diagnosis stage extracts the Issue Type, Tone, and Urgency, allowing the workflow to generate an appropriate customer support response.
Q.8 What happens if a routing function returns a node name that does not exist in the graph?
The graph raises an execution error because the destination node cannot be found.
The graph automatically skips to the END node.
LangGraph creates the missing node automatically.
The graph defaults to the first node that was added.
Explanation
Every node referenced by a routing function must exist in the graph. Otherwise, execution fails because LangGraph cannot determine where to continue.
Q.9 In LangGraph visualizations, what does a dotted-line arrow typically represent?
A conditional edge whose destination is determined at runtime.
A broken connection.
A parallel execution path.
A Human-in-the-Loop approval step.
Explanation
A dotted edge represents conditional routing. The destination is selected dynamically during execution based on the routing function’s output rather than being fixed beforehand.
Q.12 Which advantage of conditional workflows makes them especially useful for real-world AI applications such as customer support and document processing?
Every branch executes regardless of the input.
The workflow can dynamically choose different execution paths based on the current state or LLM output.
The workflow completely eliminates the need for Large Language Models.
Conditional workflows always execute faster than parallel workflows.
Explanation
Conditional workflows allow LangGraph to make intelligent decisions at runtime. Based on the current state or an LLM’s output, the workflow can follow different execution paths, making it ideal for applications like customer support, document routing, fraud detection, and recommendation systems where different inputs require different actions.
