Conditional Workflows in LangGraph

Conditional workflows enable dynamic, non-linear execution paths within an agentic graph. Unlike sequential graphs that execute tasks in a fixed pipeline or parallel graphs that execute multiple branches simultaneously, conditional graphs evaluate runtime state to dynamically route control flow to specific nodes.

1. Conceptual Foundation: Sequential vs. Parallel vs. Conditional Workflows

In agentic system design with LangGraph, control flow falls into three primary topologies:

Structure & Concept Flow

Sequential Workflow:

Task 1 -> Task 2 -> Task 3 -> END

Parallel Workflow:

+--> Task 2A --+

Task 1 --| |--> Task 3 -> END

+--> Task 2B --+

Conditional Workflow:

+-- [Condition A] --> Task 2A --+

Task 1 --| |--> END

+-- [Condition B] --> Task 2B --+

Comparative Analysis of Workflow Topologies

Topology	Execution Pattern	Branching Behavior	Primary Use Case
Sequential	Linear ( $A \to B \to C$ )	Zero branching; single deterministic path.	Fixed pipelines, ETL, simple prompt chaining.
Parallel	Concurrent ( $A \to [B \parallel C] \to D$ )	All active branches execute concurrently; results aggregate downstream.	Multi-agent research, simultaneous tool execution, multi-perspective evaluations.
Conditional	Selective ( $A \to B \text{ OR } C$ )	Single branch selected at runtime based on predicate evaluation.	Dynamic routing, classification-based handling, guardrails, triage systems.

Real-Life Analogy

The Triage Desk: Think of a hospital emergency room. In a sequential workflow, every patient would see the receptionist, then the triage nurse, then the exact same general doctor in a single queue. In a parallel workflow, a single patient would be examined by a cardiologist and an orthopedic surgeon simultaneously. In a conditional workflow, the triage nurse checks the patient's vitals (state); if chest pain is detected, the patient is routed exclusively to cardiology, whereas if a bone fracture is detected, they are routed exclusively to radiology.

2. The LangGraph Routing Mechanism

In LangGraph, conditional routing separates state modification (performed by Nodes) from routing decisions (governed by Routing Functions and Conditional Edges).

Structure & Concept Flow

State (Input)

[Source Node]

v (Updates State)

[Routing Function]

|-- Evaluates state values

|-- Returns exact string identifier of target node

[add_conditional_edges]

+---> Target Node A

+---> Target Node B

+---> Target Node C

Core Components

Source Node: The node where the conditional branch originates.
Routing Function (Predicate): A standard Python function that takes the current State as input and returns a str matching the name of the next destination node.
graph.add_conditional_edges(): The LangGraph API method that wires the source node to downstream candidates via the routing function:
PYTHON
```
graph.add_conditional_edges(
    source_node="node_name",
    path=routing_function
)
```

Note: A newer alternative in LangGraph utilizes the Command(goto="...") control primitive to perform dynamic routing directly inside a node function. However, the standard add_conditional_edges method provides an explicit separation of node logic and routing topology.

3. Deep Dive 1: Deterministic Math Engine (Quadratic Equation Solver)

To understand conditional routing without LLM non-determinism, consider solving a standard quadratic equation:

ax^2 + bx + c = 0

Mathematical Logic

The roots depend entirely on the discriminant ( $D$ ):

D = b^2 - 4ac

The downstream mathematical execution branches into three mutually exclusive pathways:

Structure & Concept Flow

Discriminant Evaluation (D)

+-----------------------+-----------------------+

| (D > 0) | (D == 0) | (D < 0)

v v v

Two Distinct Real Roots One Repeated Root No Real Roots (Complex)

x = (-b ± √D) / (2a) x = -b / (2a) (Result: "No Real Roots")

System Architecture Flow

Structure & Concept Flow

START

show_equation (Formats and logs equation string)

calculate_discriminant (Computes D = b^2 - 4ac)

[check_condition router]

+-- (D > 0) --> real_roots node --------+

| |

+-- (D == 0) --> repeated_roots node -----+---> END

| |

+-- (D < 0) --> no_real_roots node ------+

4. Step-by-Step Code Architecture: Quadratic Equation Graph

PYTHON

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# Step 1: Define the Graph State
class QuadState(TypedDict):
    a: float
    b: float
    c: float
    equation: str
    discriminant: float
    result: str

# Step 2: Define Node Functions
def show_equation(state: QuadState) -> dict:
    eq_str = f"{state['a']}x^2 + {state['b']}x + {state['c']} = 0"
    return {"equation": eq_str}

def calculate_discriminant(state: QuadState) -> dict:
    d = (state["b"] ** 2) - (4 * state["a"] * state["c"])
    return {"discriminant": d}

def real_roots(state: QuadState) -> dict:
    d = state["discriminant"]
    a = state["a"]
    b = state["b"]
    
    root1 = (-b + (d ** 0.5)) / (2 * a)
    root2 = (-b - (d ** 0.5)) / (2 * a)
    return {"result": f"The two distinct real roots are: {root1} and {root2}"}

def repeated_roots(state: QuadState) -> dict:
    a = state["a"]
    b = state["b"]
    root = -b / (2 * a)
    return {"result": f"The single repeated root is: {root}"}

def no_real_roots(state: QuadState) -> dict:
    return {"result": "No real roots exist (discriminant < 0)."}

# Step 3: Define Routing Predicate
def check_condition(state: QuadState) -> str:
    d = state["discriminant"]
    if d > 0:
        return "real_roots"
    elif d == 0:
        return "repeated_roots"
    else:
        return "no_real_roots"

# Step 4: Construct the Graph
builder = StateGraph(QuadState)

# Add Nodes
builder.add_node("show_equation", show_equation)
builder.add_node("calculate_discriminant", calculate_discriminant)
builder.add_node("real_roots", real_roots)
builder.add_node("repeated_roots", repeated_roots)
builder.add_node("no_real_roots", no_real_roots)

# Add Standard Edges
builder.add_edge(START, "show_equation")
builder.add_edge("show_equation", "calculate_discriminant")

# Add Conditional Edge Routing
builder.add_conditional_edges(
    source="calculate_discriminant",
    path=check_condition
)

# Connect branch terminations to END
builder.add_edge("real_roots", END)
builder.add_edge("repeated_roots", END)
builder.add_edge("no_real_roots", END)

# Compile Graph
quad_app = builder.compile()

Execution Traces Across Distinct Conditions

PYTHON

# Case 1: D > 0 (Two Real Roots)
input_state_1 = {"a": 1.0, "b": -5.0, "c": 6.0}
output_1 = quad_app.invoke(input_state_1)
print(output_1["result"])
# Output: "The two distinct real roots are: 3.0 and 2.0"

# Case 2: D == 0 (One Repeated Root)
input_state_2 = {"a": 1.0, "b": -4.0, "c": 4.0}
output_2 = quad_app.invoke(input_state_2)
print(output_2["result"])
# Output: "The single repeated root is: 2.0"

# Case 3: D < 0 (No Real Roots)
input_state_3 = {"a": 2.0, "b": 1.0, "c": 3.0}
output_3 = quad_app.invoke(input_state_3)
print(output_3["result"])
# Output: "No real roots exist (discriminant < 0)."

5. Deep Dive 2: LLM-Powered Agentic Customer Support Workflow

Real-world AI systems route unstructured textual inputs into specialized downstream LLM sub-pipelines based on intent and sentiment.

Business Logic & Routing Strategy

Structure & Concept Flow

Review Text Ingested

[find_sentiment Node]

(Structured LLM Classification: "positive" | "negative")

[check_sentiment Router]

+-- "positive" --> [positive_response Node] ---------------------+

| (Generates warm thank you & feedback request) |

| |

+-- "negative" --> [run_diagnosis Node] |

(Extracts Issue Type, Tone, Urgency) |

| |

v |

[negative_response Node] |

(Generates tailored resolution message) v

| END

+--------------------------------------+

Decision Breakdown

Positive Branch:
- Single-step response generation.
- Objective: Express gratitude and request community/web review.
Negative Branch:
- Multi-step diagnostic and remediation pipeline.
- Step 1 (run_diagnosis): Structural feature extraction (Issue Type, Emotional Tone, Urgency Level).
- Step 2 (negative_response): Highly contextualized, empathetic recovery message addressing the specific operational friction point.

6. Step-by-Step Code Architecture: Sentiment Routing & Diagnostic Pipeline

PYTHON

import os
from typing import TypedDict, Literal, Optional
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

# Initialize Base LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ==========================================
# 1. Pydantic Schemas for Structured Output
# ==========================================

class SentimentSchema(BaseModel):
    sentiment: Literal["positive", "negative"] = Field(
        description="Sentiment classification of the customer review."
    )

class DiagnosisSchema(BaseModel):
    issue_type: Literal["UX", "Performance", "Bug", "Support", "Other"] = Field(
        description="The category of the issue identified in the review."
    )
    tone: Literal["Frustrated", "Angry", "Disappointed", "Neutral"] = Field(
        description="The emotional tone conveyed by the user."
    )
    urgency: Literal["Low", "Medium", "High"] = Field(
        description="The operational criticality or urgency of the issue."
    )

# Instantiate Structured LLM Clients
structured_sentiment_llm = llm.with_structured_output(SentimentSchema)
structured_diagnosis_llm = llm.with_structured_output(DiagnosisSchema)

# ==========================================
# 2. Graph State Schema
# ==========================================

class ReviewState(TypedDict):
    review: str
    sentiment: Optional[Literal["positive", "negative"]]
    diagnosis: Optional[dict]
    response: Optional[str]

# ==========================================
# 3. Node Functions
# ==========================================

def find_sentiment(state: ReviewState) -> dict:
    prompt = f"Analyze the following review and determine its sentiment:\n\n{state['review']}"
    result: SentimentSchema = structured_sentiment_llm.invoke(prompt)
    return {"sentiment": result.sentiment}

def positive_response(state: ReviewState) -> dict:
    prompt = (
        f"Write a warm, professional thank-you response to this positive review:\n\n"
        f"\"{state['review']}\"\n\n"
        f"Encourage the customer to leave feedback on our public website."
    )
    response = llm.invoke(prompt)
    return {"response": response.content}

def run_diagnosis(state: ReviewState) -> dict:
    prompt = (
        f"Diagnose this negative customer review by extracting issue type, tone, and urgency:\n\n"
        f"\"{state['review']}\""
    )
    diagnosis_result: DiagnosisSchema = structured_diagnosis_llm.invoke(prompt)
    return {"diagnosis": diagnosis_result.model_dump()}

def negative_response(state: ReviewState) -> dict:
    diagnosis = state["diagnosis"]
    prompt = (
        f"You are a Senior Customer Escalation Specialist.\n"
        f"A customer submitted this negative review: \"{state['review']}\"\n\n"
        f"Diagnostic Context:\n"
        f"- Issue Category: {diagnosis['issue_type']}\n"
        f"- Customer Tone: {diagnosis['tone']}\n"
        f"- Urgency Level: {diagnosis['urgency']}\n\n"
        f"Draft an empathetic, highly actionable resolution response addressing their exact pain point."
    )
    response = llm.invoke(prompt)
    return {"response": response.content}

# ==========================================
# 4. Routing Predicate
# ==========================================

def check_sentiment(state: ReviewState) -> str:
    if state["sentiment"] == "positive":
        return "positive_response"
    return "run_diagnosis"

# ==========================================
# 5. Graph Assembly & Compilation
# ==========================================

review_builder = StateGraph(ReviewState)

# Register Nodes
review_builder.add_node("find_sentiment", find_sentiment)
review_builder.add_node("positive_response", positive_response)
review_builder.add_node("run_diagnosis", run_diagnosis)
review_builder.add_node("negative_response", negative_response)

# Connect Base Ingress Edge
review_builder.add_edge(START, "find_sentiment")

# Connect Conditional Branching
review_builder.add_conditional_edges(
    source="find_sentiment",
    path=check_sentiment
)

# Connect Linear Paths and Terminations
review_builder.add_edge("positive_response", END)
review_builder.add_edge("run_diagnosis", "negative_response")
review_builder.add_edge("negative_response", END)

# Compile Application
review_app = review_builder.compile()

Complete End-to-End Execution Demonstrations

Execution Case A: Positive Customer Review

PYTHON

positive_input = {
    "review": "I have been using this app for a month. The user interface is incredibly clean, intuitive, and works seamlessly without tutorials. Great job to the team!"
}

result_pos = review_app.invoke(positive_input)
print("Detected Sentiment:", result_pos["sentiment"])
print("Generated Response:\n", result_pos["response"])

Execution Path: START $\to$ find_sentiment $\to$ positive_response $\to$ END

Execution Case B: Negative Critical Review

PYTHON

negative_input = {
    "review": "I have been trying to log in for over an hour now. The app freezes immediately on the authentication screen. I reinstalled it twice with no luck. Unacceptable for a production app!"
}

result_neg = review_app.invoke(negative_input)
print("Detected Sentiment:", result_neg["sentiment"])
print("Structured Diagnosis:", result_neg["diagnosis"])
print("Tailored Response:\n", result_neg["response"])

Execution Path: START $\to$ find_sentiment $\to$ run_diagnosis $\to$ negative_response $\to$ END

7. Edge Cases, Routing Mechanics & Best Practices

Structure & Concept Flow

add_conditional_edges Validation Flow

+----------------------+----------------------+

| |

v v

Dynamic String Match Explicit Mapping Dictionary

Returns exact target node name Maps arbitrary return values to nodes

e.g., return "positive_response" e.g., {"pos": "positive_response"}

Critical Operational Considerations

Exact String Matching: The return value of the routing function must match the string registered in add_node("node_name", ...) exactly. A typo will raise a runtime graph execution error.
Deterministic Schemas with Pydantic: When building LLM-based routers, never rely on raw text LLM responses for conditional routing (e.g., asking an LLM to "say 'yes' or 'no'"). Always use .with_structured_output(Schema) with Literal[...] fields to guarantee strict schema conformance.
Partial State Updates: In LangGraph, node functions only need to return a dictionary containing the specific state keys they update. The graph engine merges these partial updates into the global state automatically.
Complex Subgraphs: Branch targets do not have to be terminal nodes. A conditional edge can route to an entire multi-node linear or parallel subgraph (as shown in the run_diagnosis $\to$ negative_response pipeline).

Summary and Key Takeaways

Conditional Branching Primitives: LangGraph implements conditional logic via graph.add_conditional_edges(source_node, routing_function), routing control flow based on runtime state evaluations.
Node-Router Separation of Concerns: Nodes are pure execution units that transform state; routing functions are lightweight inspection units that return target node names without mutating state.
Structured LLM Enforcements: When routing depends on LLM decisions, use Pydantic models paired with .with_structured_output() to prevent hallucinations and maintain strict structural typing.
Graph Topology Flexibility: Branches in a conditional workflow can have varying lengths and complexities (e.g., a simple 1-step response on one branch vs. a multi-stage diagnostic pipeline on another).
Safe Graph Convergence: Multiple disjoint conditional paths can safely terminate at common synchronization points or route directly to the terminal END state.

Conditional Workflows in LangGraph | Agentic AI using LangGraph | Video 7 | CampusX

Conditional Workflows in LangGraph

1. Conceptual Foundation: Sequential vs. Parallel vs. Conditional Workflows

Comparative Analysis of Workflow Topologies

Real-Life Analogy

2. The LangGraph Routing Mechanism

Core Components

3. Deep Dive 1: Deterministic Math Engine (Quadratic Equation Solver)

Mathematical Logic

System Architecture Flow

4. Step-by-Step Code Architecture: Quadratic Equation Graph

Execution Traces Across Distinct Conditions

5. Deep Dive 2: LLM-Powered Agentic Customer Support Workflow

Business Logic & Routing Strategy

Decision Breakdown

6. Step-by-Step Code Architecture: Sentiment Routing & Diagnostic Pipeline

Complete End-to-End Execution Demonstrations

Execution Case A: Positive Customer Review

Execution Case B: Negative Critical Review

7. Edge Cases, Routing Mechanics & Best Practices

Critical Operational Considerations

Summary and Key Takeaways