🎯 What You'll Learn Today
LangGraph Tutorial: Conversation Flow Control - Unit 2.2 Exercise 8
This tutorial is also available in Google Colab here or for download here
Joint Initiative: This tutorial is part of a collaboration between AI Product Engineer and the Nebius Academy.
This tutorial demonstrates how to implement effective flow control in LangGraph conversations, managing transitions between states and end conditions.
Key Concepts Covered
- Flow State Management
- End Condition Detection
- Step Transition Logic
- State-based Routing
from typing import Annotated, Literal, TypedDict
!pip install langchain-core
!pip install langgraph
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langgraph.graph.message import add_messagesStep 1: State Definition for Flow Control
Define state structure for managing conversation flow.
Why This Matters
Flow control state is crucial because
- Tracks conversation progress
- Maintains execution context
- Enables decision making
Debug Tips
- State Structure Issues:
- Verify message initialization
- Check tool_outputs structure
- Monitor state consistency
class State(TypedDict):
"""State container for flow control.
Notes:
- messages tracks conversation history
- tool_outputs needed for completion check
- Both fields must be properly initialized
Attributes:
messages: Conversation history with proper typing
tool_outputs: List of tool execution results
"""
messages: Annotated[list[BaseMessage], add_messages]
tool_outputs: list[str]Step 2: Flow Control Implementation
Implement logic to determine next conversation steps.
Why This Matters
Flow control is critical because
- Manages conversation progression
- Handles termination conditions
- Ensures proper tool execution
Debug Tips
- Flow Control Issues:
- Check message type handling
- Verify end conditions
- Monitor state transitions
def get_next_step(state: State) -> Literal["tool_selector", "end"]:
"""Determine next conversation step based on state.
Notes:
- Checks conversation state
- Detects termination conditions
- Routes to appropriate next step
Args:
state: Current conversation state
Returns:
Literal["tool_selector", "end"]: Next step identifier
"""
# Safety check for empty state
if not state.get("messages"):
return "end"
# Get latest message for analysis
last_message = state["messages"][-1]
# Check termination conditions
# 1. User indicates completion
if isinstance(last_message, HumanMessage):
if any(word in last_message.content.lower() for word in ["thanks", "bye"]):
return "end"
# 2. Tool execution completed
if isinstance(last_message, AIMessage) and state.get("tool_outputs"):
return "end"
# Continue with tool selection if no end conditions met
return "tool_selector"Step 3: System Demonstration
Test the flow control system with various scenarios.
Why This Matters
Testing verifies
- Proper state transitions
- End condition detection
- Flow control logic
def demonstrate_flow_control():
"""Demonstrate flow control with various test cases."""
# Test cases
test_cases = [
{
"name": "Empty State",
"state": {"messages": [], "tool_outputs": []},
"description": "Testing empty state handling",
},
{
"name": "User Goodbye",
"state": {
"messages": [HumanMessage(content="thanks for your help!")],
"tool_outputs": [],
},
"description": "Testing user termination",
},
{
"name": "Tool Completion",
"state": {
"messages": [AIMessage(content="Here's the result")],
"tool_outputs": ["calculation complete"],
},
"description": "Testing tool completion",
},
{
"name": "Continue Conversation",
"state": {
"messages": [HumanMessage(content="what's the weather?")],
"tool_outputs": [],
},
"description": "Testing continuation",
},
]
print("Flow Control Demonstration")
print("=========================")
for case in test_cases:
print(f"\nTest: {case['name']}")
print(f"Description: {case['description']}")
# Process flow
next_step = get_next_step(case["state"])
# Print results
print("State:")
print(f"- Messages: {len(case['state']['messages'])} messages")
print(f"- Tool Outputs: {len(case['state']['tool_outputs'])} outputs")
print(f"Next Step: {next_step}")
print("-" * 50)Common Pitfalls
- Not handling empty states
- Missing message type checks
- Incomplete end conditions
- Poor state validation
Key Takeaways
- Always verify state integrity
- Check message types explicitly
- Define clear end conditions
- Maintain consistent routing
Expected Output
Flow Control Demonstration
Test: Empty State
Description: Testing empty state handling
## State
- Messages: 0 messages
- Tool Outputs: 0 outputs
## Next Step: end
Test: User Goodbye
Description: Testing user termination
## State
- Messages: 1 messages
- Tool Outputs: 0 outputs
## Next Step: end
Test: Tool Completion
Description: Testing tool completion
## State
- Messages: 1 messages
- Tool Outputs: 1 outputs
## Next Step: end
Test: Continue Conversation
Description: Testing continuation
## State
- Messages: 1 messages
- Tool Outputs: 0 outputs
## Next Step: tool_selector
if __name__ == "__main__":
demonstrate_flow_control()
🇬🇧 Chapter