🎯 What You'll Learn Today
LangGraph Tutorial: Conversation Flow Control - Unit 2.2 Exercise 8
Try It Yourself
📢 Joint Initiative
This tutorial is part of a collaboration between AI Product Engineer and 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()