How to intermediate · 3 min read

How to add conditional edges in LangGraph

Quick answer
In LangGraph, add conditional edges by defining edge functions that evaluate conditions and return the next node dynamically. Use the add_edge method with a condition callback to control graph traversal based on runtime data.

PREREQUISITES

  • Python 3.8+
  • pip install langchain>=0.2
  • Basic knowledge of LangChain and LangGraph concepts

Setup

Install the langchain package if you haven't already, and import necessary classes for graph construction.

bash
pip install langchain>=0.2

Step by step

Create a LangGraph instance, add nodes, and define conditional edges by passing a function that returns a boolean or the next node key. This function controls which edge is followed during execution.

python
from langchain.graphs import LangGraph

# Define nodes as simple functions or callables

def start_node():
    print("Start node executed")
    return "data_from_start"

def node_a():
    print("Node A executed")

def node_b():
    print("Node B executed")

# Condition function to decide edge

def condition(data):
    # Example condition based on data
    return data == "data_from_start"

# Create graph
graph = LangGraph()

# Add nodes
graph.add_node("start", start_node)
graph.add_node("A", node_a)
graph.add_node("B", node_b)

# Add conditional edges
# Edge from 'start' to 'A' if condition is True
graph.add_edge("start", "A", condition=condition)
# Edge from 'start' to 'B' if condition is False
graph.add_edge("start", "B", condition=lambda data: not condition(data))

# Execute graph starting from 'start'
result = graph.run("start")

# Output will show which node runs based on condition
output
Start node executed
Node A executed

Common variations

  • Use async functions for nodes and edges to support asynchronous workflows.
  • Replace condition functions with more complex logic or external API calls.
  • Use different LangChain models or tools as node actions.
python
import asyncio
from langchain.graphs import LangGraph

async def async_start_node():
    print("Async start node executed")
    return "async_data"

async def async_node_a():
    print("Async Node A executed")

async def async_node_b():
    print("Async Node B executed")

async def async_condition(data):
    await asyncio.sleep(0.1)  # simulate async check
    return data == "async_data"

async def main():
    graph = LangGraph()
    graph.add_node("start", async_start_node)
    graph.add_node("A", async_node_a)
    graph.add_node("B", async_node_b)
    graph.add_edge("start", "A", condition=async_condition)
    graph.add_edge("start", "B", condition=lambda data: not async_condition(data))

    await graph.run("start")

# To run async main:
# asyncio.run(main())

Troubleshooting

  • If the graph does not follow the expected edge, verify your condition functions return correct boolean values.
  • Ensure node functions return data expected by condition callbacks.
  • For async graphs, confirm you await graph.run() properly.

Key Takeaways

  • Use condition functions in add_edge to create dynamic branching in LangGraph.
  • Node functions can return data used by edge conditions to decide the next node.
  • Async support allows conditional edges to depend on asynchronous computations or API calls.
Verified 2026-04
Verify ↗