Conditional edges in LangGraph
Quick answer
In
LangGraph, conditional edges are implemented by defining nodes that evaluate conditions and return different states accordingly. You create functions that return different outputs based on input state, then add edges from these nodes to other nodes or the END state conditionally.PREREQUISITES
Python 3.8+pip install langgraphBasic knowledge of Python functions and typing
Setup
Install the langgraph package via pip and import necessary classes. Define a TypedDict for your state to ensure type safety.
pip install langgraph output
Collecting langgraph Downloading langgraph-0.1.0-py3-none-any.whl (10 kB) Installing collected packages: langgraph Successfully installed langgraph-0.1.0
Step by step
Define a state type and nodes as functions that return different states based on conditions. Add conditional edges by connecting nodes and specifying the flow based on the returned state.
from langgraph.graph import StateGraph, END
from typing import TypedDict
# Define the state type
class State(TypedDict):
count: int
messages: list[str]
# Node that increments count and decides next node based on count
def increment_node(state: State) -> State:
new_count = state["count"] + 1
new_messages = state["messages"] + [f"Count is {new_count}"]
return {"count": new_count, "messages": new_messages}
# Node that checks count and decides to continue or end
def check_node(state: State) -> State:
if state["count"] < 3:
# Continue looping
return state
else:
# End the graph
return {"count": state["count"], "messages": state["messages"] + ["Done"]}
# Create graph and add nodes
graph = StateGraph(State)
graph.add_node("increment", increment_node)
graph.add_node("check", check_node)
graph.set_entry_point("increment")
# Add edges conditionally by connecting nodes
# From increment to check
graph.add_edge("increment", "check")
# From check back to increment if count < 3
# From check to END if count >= 3
def edge_condition(state: State) -> str:
return "increment" if state["count"] < 3 else END
graph.add_edge("check", edge_condition)
# Compile and invoke
app = graph.compile()
initial_state = {"count": 0, "messages": []}
result = app.invoke(initial_state)
print("Messages:")
for msg in result["messages"]:
print(msg) output
Messages: Count is 1 Count is 2 Count is 3 Done
Common variations
- Use async functions for nodes if your logic involves I/O or async calls.
- Return different types of states or use enums to represent multiple conditional branches.
- Combine multiple conditions in edge functions for complex branching.
from langgraph.graph import StateGraph, END
from typing import TypedDict
import asyncio
class AsyncState(TypedDict):
flag: bool
messages: list[str]
async def async_node(state: AsyncState) -> AsyncState:
await asyncio.sleep(0.1) # simulate async work
new_messages = state["messages"] + ["Async step done"]
return {"flag": not state["flag"], "messages": new_messages}
async def async_edge_condition(state: AsyncState) -> str:
return "async_node" if state["flag"] else END
graph = StateGraph(AsyncState)
graph.add_node("async_node", async_node)
graph.set_entry_point("async_node")
graph.add_edge("async_node", async_edge_condition)
app = graph.compile()
import asyncio
result = asyncio.run(app.invoke({"flag": True, "messages": []}))
print(result["messages"]) output
['Async step done', 'Async step done', ... depending on loop or termination]
Troubleshooting
- If your graph does not terminate, ensure your conditional edges eventually return
END. - Type errors often arise if your
TypedDictstate does not match returned dictionaries; keep state consistent. - Use
printstatements inside nodes and edge functions to debug flow.
Key Takeaways
- Define nodes as functions returning state to implement conditional logic in LangGraph.
- Use edge functions returning node names or
ENDto create conditional edges. - TypedDict state typing ensures safe and clear state transitions.
- Async nodes and edges support I/O-bound or asynchronous workflows.
- Always ensure conditional edges lead to termination to avoid infinite loops.