Conditional routing in graphs in LangChain

Introduction

Conditional routing in graphs helps decide which path to take based on certain rules or conditions. It makes the flow smart and flexible.

When you want to choose different actions based on user input in a chatbot.

When processing data flows that depend on specific values or states.

When building workflows that need to branch depending on conditions.

When you want to handle errors or special cases differently in a process.

Syntax

LangChain

from langchain.graphs import Graph

# Create a graph
graph = Graph()

# Add nodes with names
graph.add_node('start')
graph.add_node('option_a')
graph.add_node('option_b')

# Add edges with conditions
# condition is a function that returns True or False

graph.add_edge('start', 'option_a', condition=lambda context: context['choice'] == 'a')
graph.add_edge('start', 'option_b', condition=lambda context: context['choice'] == 'b')

# Run the graph with a context
result = graph.run(start_node='start', context={'choice': 'a'})

The condition is a function that checks the context to decide if the edge should be followed.

The context is a dictionary holding data used to evaluate conditions.

Examples

This edge is taken only if the context's 'choice' is 'a'.

LangChain

graph.add_edge('start', 'option_a', condition=lambda context: context['choice'] == 'a')

This edge is taken only if the context's 'choice' is 'b'.

LangChain

graph.add_edge('start', 'option_b', condition=lambda context: context['choice'] == 'b')

This edge is always taken if no other conditions match, acting as a fallback.

LangChain

graph.add_edge('start', 'default_option', condition=lambda context: True)

If no condition is given, the edge is always available.

LangChain

# Edge with no condition means always true
graph.add_edge('start', 'always_option')

Sample Program

This program creates a graph with conditional edges. It runs the graph with different inputs to show how the path changes based on conditions.

LangChain

from langchain.graphs import Graph

# Create a graph
workflow_graph = Graph()

# Add nodes
workflow_graph.add_node('start')
workflow_graph.add_node('process_a')
workflow_graph.add_node('process_b')
workflow_graph.add_node('end')

# Add edges with conditions
workflow_graph.add_edge('start', 'process_a', condition=lambda context: context.get('input') == 'A')
workflow_graph.add_edge('start', 'process_b', condition=lambda context: context.get('input') == 'B')
workflow_graph.add_edge('process_a', 'end')
workflow_graph.add_edge('process_b', 'end')

# Function to run and print path

def run_and_print_path(graph, start, context):
    path = graph.run(start_node=start, context=context)
    print(f"Context: {context}")
    print(f"Path taken: {path}")

# Run with input 'A'
run_and_print_path(workflow_graph, 'start', {'input': 'A'})

# Run with input 'B'
run_and_print_path(workflow_graph, 'start', {'input': 'B'})

# Run with input 'C' (no matching condition)
run_and_print_path(workflow_graph, 'start', {'input': 'C'})

Important Notes

Time complexity depends on the number of edges checked at each node.

Space complexity depends on the size of the graph and context data.

Common mistake: forgetting to handle cases where no condition matches, which can stop the flow unexpectedly.

Use conditional routing to make flows dynamic and responsive to input or state.

Summary

Conditional routing lets you choose paths based on rules.

Use conditions as simple functions checking context data.

Always consider what happens if no conditions match.