LangGraph Conditional Workflows: Branching Without Prompt Spaghetti

This walkthrough rebuilds an operable path for: Conditional Workflows in LangGraph: The If-Else of Agentic AI. Focus on contracts, checks, and code you can drop into a repo without guessing intent. For Overview, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

What Even Is a Conditional Workflow?

For What Even Is a Conditional Workflow?, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Bound tool schemas tightly. Wide free-text args invite injection and make audits expensive.

Workflow #1: Solving a Quadratic Equation

For Workflow #1: Solving a Quadratic Equation, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Bound tool schemas tightly. Wide free-text args invite injection and make audits expensive.

The Workflow Structure

For The Workflow Structure, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and cost next to functional results. Visibility early prevents surprise bills when the path moves from demo to shared environments. Bound tool schemas tightly. Wide free-text args invite injection and make audits expensive. For The Workflow Structure, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

Building the State

For Building the State, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Checkpoint after expensive model calls so a retry does not re-bill the same work.

class QuadState(TypedDict):
    a: float
    b: float
    c: float
    equation: str
    discriminant: float
    result: str

The Nodes

For The Nodes, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Checkpoint after expensive model calls so a retry does not re-bill the same work.

def show_equation(state: QuadState):
    equation = f"{state['a']}x² + {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):
    r1 = (-state['b'] + state['discriminant'] ** 0.5) / (2 * state['a'])
    r2 = (-state['b'] - state['discriminant'] ** 0.5) / (2 * state['a'])
    result = f"The roots are {r1} and {r2}"
    return {"result": result}

def repeated_root(state: QuadState):
    root = -state['b'] / (2 * state['a'])
    result = f"Only repeating root is {root}"
    return {"result": result}

def no_real_roots(state: QuadState):
    return {"result": "No real roots"}

The Routing Function — This Is the Key Part

For The Routing Function — This Is the Key Part, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and cost next to functional results. Visibility early prevents surprise bills when the path moves from demo to shared environments. Checkpoint after expensive model calls so a retry does not re-bill the same work. For The Routing Function — This Is the Key Part, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

def check_condition(state: QuadState):
    if state['discriminant'] > 0:
        return "real_roots"
    elif state['discriminant'] == 0:
        return "repeated_root"
    else:
        return "no_real_roots"

Wiring It Together

For Wiring It Together, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Separate planning from tool execution. The planner proposes; the executor mutates; the verifier checks outcomes against the goal.

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_root", repeated_root)
graph.add_node("no_real_roots", no_real_roots)

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

# This is the magic line
graph.add_conditional_edges("calculate_discriminant", check_condition)

graph.add_edge("real_roots", END)
graph.add_edge("repeated_root", END)
graph.add_edge("no_real_roots", END)

workflow = graph.compile()

Workflow #2: Customer Review Handler (LLM-Based)

For Workflow #2: Customer Review Handler (LLM-Based), define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Separate planning from tool execution. The planner proposes; the executor mutates; the verifier checks outcomes against the goal.

The Scenario

For The Scenario, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and cost next to functional results. Visibility early prevents surprise bills when the path moves from demo to shared environments. Separate planning from tool execution. The planner proposes; the executor mutates; the verifier checks outcomes against the goal. For The Scenario, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

Structured Output Schemas

For Structured Output Schemas, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Bound tool schemas tightly. Wide free-text args invite injection and make audits expensive.

from pydantic import BaseModel
from typing import Literal

class SentimentSchema(BaseModel):
    sentiment: Literal["positive", "negative"]

class DiagnosisSchema(BaseModel):
    issue_type: Literal["UI", "performance", "bug", "support", "other"]
    tone: Literal["frustrated", "angry", "disappointed", "neutral"]
    urgency: Literal["low", "medium", "high"]

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")
structured_model_sentiment = model.with_structured_output(SentimentSchema)
structured_model_diagnosis = model.with_structured_output(DiagnosisSchema)

The State

For The State, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Bound tool schemas tightly. Wide free-text args invite injection and make audits expensive.

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

The Nodes

For The Nodes, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and cost next to functional results. Visibility early prevents surprise bills when the path moves from demo to shared environments. Bound tool schemas tightly. Wide free-text args invite injection and make audits expensive. For The Nodes, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

def find_sentiment(state: ReviewState):
    prompt = f"For the following review, find out the sentiment:\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 in response to this review:
    {state['review']}
    Also kindly ask the user to leave feedback on our website."""

    result = model.invoke(prompt)
    return {"response": result.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)
    return {"diagnosis": result.model_dump()}

def negative_response(state: ReviewState):
    diagnosis = state['diagnosis']
    prompt = f"""You are a support assistant. The user had a {diagnosis['issue_type']} issue,
    sounded {diagnosis['tone']}, and marked urgency as {diagnosis['urgency']}.
    Write an empathetic, helpful resolution message."""

    result = model.invoke(prompt)
    return {"response": result.content}

def check_sentiment(state: ReviewState):
    if state['sentiment'] == "positive":
        return "positive_response"
    else:
        return "run_diagnosis"

The Graph

For The Graph, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Checkpoint after expensive model calls so a retry does not re-bill the same work.

graph = StateGraph(ReviewState)

graph.add_node("find_sentiment", find_sentiment)
graph.add_node("positive_response", positive_response)
graph.add_node("run_diagnosis", run_diagnosis)
graph.add_node("negative_response", negative_response)

graph.add_edge(START, "find_sentiment")
graph.add_conditional_edges("find_sentiment", check_sentiment)
graph.add_edge("positive_response", END)
graph.add_edge("run_diagnosis", "negative_response")
graph.add_edge("negative_response", END)

workflow = graph.compile()

Testing It

For Testing It, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Checkpoint after expensive model calls so a retry does not re-bill the same work.

The Core Mental Model

For The Core Mental Model, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and cost next to functional results. Visibility early prevents surprise bills when the path moves from demo to shared environments. Checkpoint after expensive model calls so a retry does not re-bill the same work. For The Core Mental Model, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

What’s Next

For What’s Next, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Separate planning from tool execution. The planner proposes; the executor mutates; the verifier checks outcomes against the goal.

Structured Learning Beyond Blogs

Separate planning from tool execution. The planner proposes; the executor mutates; the verifier checks outcomes against the goal.

Courses

Separate planning from tool execution. The planner proposes; the executor mutates; the verifier checks outcomes against the goal.

Operational checklist

Checkpoint after expensive model calls so a retry does not re-bill the same work.

Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

Write a short runbook: how to rotate keys, how to drain the queue, how to roll back the last change.