LangGraph StateGraph tutorial: build your first AI workflow in Python

LangGraph StateGraph Tutorial: Build Your First AI Workflow in Python

Welcome! Have you ever wanted to build a smart computer program that can do many steps, remember what it did, and even make decisions? That’s exactly what we’re going to learn today. This LangGraph StateGraph tutorial will show you how to create your very first AI workflow in Python.

Imagine your computer program needs to search for information, then summarize it, and then decide if it needs more details. LangGraph is a fantastic tool that helps you map out these kinds of complex tasks. It’s like drawing a flowchart for your AI!

By the end of this guide, you will understand the core ideas behind LangGraph, how to use StateGraph Python, and build your own practical examples. We’ll break down everything into easy steps, perfect for anyone who wants to create smart, multi-step AI applications. Let’s get started on this exciting journey!

What is LangGraph? Think of it as a Smart Flowchart

LangGraph is a special library built on top of LangChain, which you might have heard of before. LangChain helps you talk to powerful AI models, like the ones that can write stories or answer questions. LangGraph takes this a step further by letting you chain these AI actions together into a proper workflow.

Think of LangGraph like a detailed map or a recipe for your AI helper. Instead of just doing one thing, your AI can follow many steps in a row. It can also choose different paths based on what happens during its work. This makes your AI much more capable and robust.

The main idea is to make your AI process tasks in a structured way, remembering past actions. This way, your AI doesn’t forget important information as it moves from one step to the next. You can learn more about the basics of LangChain by visiting their official documentation here.

Why StateGraph Python is Your AI’s Notepad

At the heart of LangGraph is something called a StateGraph. Imagine your AI has a notepad where it writes down everything important it learns or does. This notepad is the “state” of your graph.

The StateGraph Python lets your AI workflow remember information as it moves between different steps. For example, if one step finds a piece of information, the next step can read that information from the “notepad” and use it. This continuity is super powerful.

Without a StateGraph, each step would be like starting fresh, forgetting everything that happened before. With it, your AI can build on its past actions, making complex decisions and carrying out multi-part tasks. This memory feature is what makes your AI truly intelligent and capable of handling long conversations or projects.

Setting Up Your AI Workbench

Before we can build anything amazing, we need to set up our tools. Don’t worry, it’s quite straightforward! You will need Python installed on your computer, which is usually pre-installed on many systems. If you don’t have it, you can download it from the official Python website.

We also need to install the LangChain and LangGraph libraries. These are like adding new tools to your Python toolkit. We will use a command called pip for this, which helps install Python packages. It’s like going to a store and buying a new game for your console.

You’ll also need an API key for a Large Language Model (LLM), such as OpenAI. This key allows your Python program to connect to powerful AI brains online. Think of it as a secret password that lets your computer talk to the smart AI servers. You can usually get a free or trial key from providers like OpenAI.

H3: Installing LangChain and LangGraph

Open your computer’s terminal or command prompt. This is a window where you can type commands for your computer. Once it’s open, type the following commands one by one and press Enter after each.

pip install langchain langchain-openai langgraph

This command installs LangChain, a special part of LangChain for connecting to OpenAI, and LangGraph itself. You might see a lot of text scroll by, which is normal as the computer installs everything. Once it finishes, you’re ready for the next step.

H3: Getting Your OpenAI API Key

To make our AI smart, we need to use a powerful AI model from OpenAI. You’ll need to create an account on the OpenAI platform. Once you have an account, you can generate an API key.

Look for a section usually labeled “API Keys” or “Developers” in your account settings. It will give you a long string of letters and numbers. This is your secret key, so keep it safe and don’t share it publicly. We will use this key in our Python code to tell the AI models that it’s us asking for help.

When you copy your key, you’ll need to set it up so your Python program can find it. The easiest way is to set it as an environment variable. On most systems, you can do this in your terminal.

export OPENAI_API_KEY="your_openai_api_key_here"

Remember to replace "your_openai_api_key_here" with your actual API key. If you are using an IDE like VS Code or PyCharm, there are often easier ways to manage environment variables specific to your project. This export command works for the current terminal session.

Building Blocks of LangGraph: Nodes and Edges

Every good map has landmarks and roads connecting them. In LangGraph, these are called LangGraph nodes edges. These two concepts are the fundamental parts you’ll use to design any AI workflow. Understanding them is key to mastering this LangGraph StateGraph tutorial.

Think of nodes as individual actions or steps your AI takes. An action could be “read a document,” “ask a question,” or “summarize text.” Each node does one specific job.

Edges are the pathways that connect these nodes. They tell your AI which step to go to next after finishing the current one. Edges decide the flow of your entire workflow, guiding your AI through its tasks.

H3: What are Nodes? Your AI’s Action Steps

Nodes are simply functions or parts of your code that perform a specific task. When your AI workflow reaches a node, it executes the code inside that node. For example, you might have a node that uses an AI model to generate a response.

Another node might be responsible for checking if the generated response is good enough. Or perhaps a node that searches the internet for specific information. Each node is a distinct step in your AI’s journey.

It’s helpful to name your nodes clearly so you know exactly what each part of your workflow does. This makes your AI program easier to understand and fix if something goes wrong.

H3: What are Edges? The Paths Between Actions

Edges are the connections between your nodes, dictating the sequence of operations. After a node finishes its job, an edge tells the graph which node to visit next. This is how your AI moves from one action to another.

There are two main types of edges:

1. Direct Edges: These are simple, straight connections. After Node A, always go to Node B.
2. Conditional Edges: These are smarter connections. After Node A, check something. If the check is true, go to Node B. If false, go to Node C. This allows your AI to make decisions.

We’ll see examples of both direct and conditional edges as we build our first practical StateGraph Python workflow. These connections are what bring your static nodes to life and make your AI dynamic.

Your First StateGraph: A Simple Workflow

Now, let’s put these concepts into practice. We’ll build a very simple AI workflow using StateGraph Python. This example will demonstrate how to define a state, create a few nodes, and connect them with edges. Our first graph will simply process a piece of text through two steps.

This LangGraph StateGraph tutorial is designed to be hands-on, so get ready to type some code! We will start by defining what information our AI needs to remember, then create the actions, and finally link them up.

Remember, the goal here is to understand the basic structure. We’ll make it smarter later. For now, let’s focus on getting a simple flow working.

H3: Defining Your State

First, we need to tell our StateGraph what kind of information it will remember. This is our “notepad.” In LangGraph, we define this using a TypedDict or dict. For this example, our state will just hold a “text” message and a list of “history” messages.

from typing import TypedDict, List
import operator

# Define the state for our graph
# This tells LangGraph what kind of information our AI will remember
class AgentState(TypedDict):
    text: str # This will hold the main text we are working with
    history: List[str] # This will store a list of messages or actions taken
    
# We also need a way to combine changes to our state.
# For example, if two nodes try to update the 'history' list.
# The 'operator.add' here means we'll just add new history items to the existing list.
def combine_history(current_history: List[str], new_history_item: List[str]) -> List[str]:
    return current_history + new_history_item

In this AgentState, text will be the main input we feed into our workflow, and history will track what happens. The combine_history function is important for merging updates to the history list if multiple parts of our graph try to change it. This ensures everything is kept in sync.

H3: Creating Nodes (Actions)

Next, let’s create our “nodes” – the functions that do the actual work. We’ll have two simple nodes:

1. greet_node: This node will simply add a greeting to our history.
2. process_node: This node will take the input text and add a “processed” message to the history.

These nodes will receive the current state of our graph, do something, and then return updates to that state. The updates are usually in the form of a dictionary, where keys match our AgentState.

# Node 1: Greet Node
def greet_node(state: AgentState) -> AgentState:
    print("---GREET NODE---")
    # This node simply adds a greeting to the history
    new_history = [f"Greeting: Hello from LangGraph! Current text: {state['text']}"]
    return {"history": new_history}

# Node 2: Process Node
def process_node(state: AgentState) -> AgentState:
    print("---PROCESS NODE---")
    # This node simulates processing the text and adds a message to history
    current_text = state['text']
    processed_text = f"Processed: '{current_text}' and found it interesting."
    new_history = [processed_text]
    return {"history": new_history, "text": processed_text} # Also updates the text

Notice that each node takes state as an input and returns an AgentState dictionary. Whatever keys are returned will update the overall graph state. If a key is not returned, that part of the state remains unchanged. This is how information flows and is remembered across the workflow.

H3: Connecting Nodes with Edges

Now comes the exciting part: connecting our nodes! We’ll use the StateGraph class from LangGraph to define our workflow. We’ll add our nodes and then specify how they should be linked together using edges.

from langgraph.graph import StateGraph, END

# Create a new StateGraph instance.
# We tell it what our state looks like (AgentState)
workflow = StateGraph(AgentState)

# Add our nodes to the workflow
# We give each node a name (e.g., "greet") and link it to our function
workflow.add_node("greet", greet_node)
workflow.add_node("process", process_node)

# Set the entry point of the graph.
# This is where our workflow will always start.
workflow.set_entry_point("greet")

# Add a direct edge from "greet" to "process"
# This means after "greet" finishes, it will always go to "process"
workflow.add_edge("greet", "process")

# Add a direct edge from "process" to END
# This means after "process" finishes, the workflow stops
workflow.add_edge("process", END)

In this setup, our graph starts at greet. After greet finishes, it automatically moves to process. Finally, after process finishes, the END signal tells the graph to stop. This is a very basic linear flow.

H3: Compiling and Running Your Graph

After defining the state, nodes, and edges, we need to “compile” our graph. This turns our blueprint into a runnable program. Then, we can kick it off with an initial state and watch it work!

# Compile the graph
# This prepares our workflow to be run
app = workflow.compile()

# Now, let's run our graph!
# We give it an initial state with some text
initial_state = {"text": "Hello, LangGraph!", "history": []}
result = app.invoke(initial_state)

# Print the final state of the graph after it has run
print("\n---FINAL STATE---")
print(result)

# Let's see the history separately
print("\n---HISTORY OF ACTIONS---")
for item in result['history']:
    print(f"- {item}")

When you run this code, you will see output from the print statements inside greet_node and process_node. The final state will show how the text was updated and all the messages added to the history list. This clearly demonstrates how information is passed and modified across your StateGraph Python workflow. You’ve just built and run your first AI workflow!

Making Your StateGraph Smarter: Conditional Edges

Our first graph was a straight line. But what if your AI needs to make a choice? For example, “If the answer is short, I’ll summarize it. If it’s long, I’ll ask for more details.” This is where conditional edges come in. They allow your StateGraph to branch its path based on the current state.

Conditional edges are a powerful feature that lets your AI workflow adapt to different situations. This is a crucial part of making intelligent agents, as discussed in many articles on LangChain agent graph designs. Instead of always going the same way, your graph can follow different paths based on a decision.

Let’s expand our example to include a conditional choice. We’ll add a new node and a function to decide the next step.

H3: Introducing a Decision Node

First, we need a new node that will perform some check. This node won’t necessarily update the state but will return a string that tells the conditional edge where to go next.

Let’s imagine our AI needs to check if the input text contains a specific keyword. If it does, we’ll take one path; otherwise, we’ll take another.

# Let's add an actual LLM for smarter decision making
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# Initialize our LLM (Large Language Model)
# Make sure your OPENAI_API_KEY environment variable is set!
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# Node 3: Decide Node (uses LLM)
def decide_node(state: AgentState) -> AgentState:
    print("---DECIDE NODE---")
    current_text = state['text']
    
    # Use the LLM to make a decision
    # We're asking it to categorize the text
    response = llm.invoke([HumanMessage(content=f"Is the following text a question or a statement? Text: '{current_text}'. Respond with 'question' or 'statement'.")])
    decision = response.content.strip().lower()

    print(f"LLM decided: {decision}")
    
    # Update history with the decision
    new_history = [f"Decision: LLM categorized text as '{decision}'"]
    
    # We're returning the decision as a special key,
    # which will be used by our conditional edge.
    # We also update the history.
    return {"decision": decision, "history": new_history}

# Node 4: Question Handler
def question_handler_node(state: AgentState) -> AgentState:
    print("---QUESTION HANDLER NODE---")
    new_history = [f"Handled as a question: '{state['text']}'"]
    return {"history": new_history, "text": "Question handled: Please provide more details."}

# Node 5: Statement Handler
def statement_handler_node(state: AgentState) -> AgentState:
    print("---STATEMENT HANDLER NODE---")
    new_history = [f"Handled as a statement: '{state['text']}'"]
    return {"history": new_history, "text": "Statement acknowledged: Interesting point."}

Here, decide_node uses an LLM to categorize the text. It returns {"decision": "question"} or {"decision": "statement"}. This decision key in the state will be crucial for our conditional edge.

H3: Building the Conditional Logic

Now we need a special function that our conditional edge will call. This function looks at the current state and tells the graph which node to go to next. It should return the name of the next node as a string.

# This function defines the conditional logic
# It looks at the 'decision' key in the state and returns the next node name
def route_decision(state: AgentState) -> str:
    print("---ROUTE DECISION---")
    if "decision" in state and state["decision"] == "question":
        return "handle_question"
    elif "decision" in state and state["decision"] == "statement":
        return "handle_statement"
    else:
        # Default or error path
        return "error_node" # Let's add an error node for completeness

# Node 6: Error Node
def error_node(state: AgentState) -> AgentState:
    print("---ERROR NODE---")
    new_history = [f"Error: Could not categorize text: '{state['text']}'"]
    return {"history": new_history, "text": "Error: Uncategorized input."}

The route_decision function is the brain of our conditional edge. It checks the decision that the LLM put into the state and directs the flow accordingly.

H3: Redefining the Graph with Conditional Edges

We’ll clear our old workflow and build a new one with our LLM, decision node, and conditional logic. Remember, our AgentState needs to be updated to include the decision key.

# Let's redefine AgentState to include the 'decision' key
class AgentState(TypedDict):
    text: str
    history: List[str]
    decision: str # Added for conditional routing

# We also need to define how 'decision' updates.
# For simplicity, we'll just overwrite it with the new decision.
def combine_decision(current_decision: str, new_decision: str) -> str:
    return new_decision

# Re-create the workflow with updated state definition
workflow_conditional = StateGraph(AgentState)

# Add all our new nodes
workflow_conditional.add_node("decide", decide_node)
workflow_conditional.add_node("handle_question", question_handler_node)
workflow_conditional.add_node("handle_statement", statement_handler_node)
workflow_conditional.add_node("error_node", error_node) # Add the error node

# Set the entry point
workflow_conditional.set_entry_point("decide")

# Add the conditional edge!
# From "decide" node, call route_decision to determine the next step
workflow_conditional.add_conditional_edges(
    "decide",     # The node where the decision is made
    route_decision, # The function that makes the routing decision
    {               # A dictionary mapping decision outcomes to target nodes
        "question": "handle_question",
        "statement": "handle_statement",
        "error_node": "error_node"
    }
)

# After handling, both paths should go to END
workflow_conditional.add_edge("handle_question", END)
workflow_conditional.add_edge("handle_statement", END)
workflow_conditional.add_edge("error_node", END) # The error path also ends

Here, add_conditional_edges is the key. It takes the source node ("decide"), the function to call (route_decision), and a dictionary mapping the function’s return values to target nodes. This makes your StateGraph Python workflow truly dynamic!

H3: Running the Conditional Graph

Now, let’s test our smart graph with different inputs to see how it branches.

# Compile the conditional graph
app_conditional = workflow_conditional.compile()

print("\n---RUNNING CONDITIONAL GRAPH (QUESTION)---")
initial_state_q = {"text": "What is the capital of France?", "history": [], "decision": ""}
result_q = app_conditional.invoke(initial_state_q)
print("\n---FINAL STATE (QUESTION)---")
print(result_q)
print("\n---HISTORY OF ACTIONS (QUESTION)---")
for item in result_q['history']:
    print(f"- {item}")


print("\n---RUNNING CONDITIONAL GRAPH (STATEMENT)---")
initial_state_s = {"text": "The sky is blue today.", "history": [], "decision": ""}
result_s = app_conditional.invoke(initial_state_s)
print("\n---FINAL STATE (STATEMENT)---")
print(result_s)
print("\n---HISTORY OF ACTIONS (STATEMENT)---")
for item in result_s['history']:
    print(f"- {item}")

You should see that for the question, the graph goes through decide and then handle_question. For the statement, it goes through decide and then handle_statement. This clearly shows the power of conditional routing in your LangGraph StateGraph tutorial example.

Advanced Concepts with LangGraph

Now that you’ve got the basics down, let’s explore some more advanced features of LangGraph. These will make your AI workflows even more powerful and robust. We’ll look at looping, human interaction, and how StateGraph integrates with LangChain agent graph structures.

These features move beyond simple linear or branching paths, allowing for more sophisticated interactions. You can create complex decision trees, iterative processes, and even workflows that pause to ask for your input. This is where your AI truly starts to shine.

Mastering these advanced concepts will allow you to build AI applications that can handle real-world challenges. It opens up possibilities for automated assistants, research tools, and much more.

H3: Looping in LangGraph: Doing Things Again and Again

Sometimes, your AI needs to repeat a set of actions until a certain condition is met. This is called looping. LangGraph makes it easy to create loops in your StateGraph Python workflows.

Imagine an AI that tries to find information, and if it doesn’t find enough, it tries again with a different search query. Or a game AI that makes moves until the game ends. Loops are essential for these kinds of iterative tasks.

To create a loop, you simply create a conditional edge that, under certain circumstances, routes back to an earlier node in the graph. This creates a cycle.

# Let's extend our AgentState for a loop example
class LoopingAgentState(TypedDict):
    search_query: str
    search_results: List[str]
    max_attempts: int
    current_attempt: int
    found_info: bool
    history: List[str]

# Node for searching (dummy function for now)
def search_node(state: LoopingAgentState) -> LoopingAgentState:
    print(f"---SEARCH NODE (Attempt {state['current_attempt']})---")
    query = state['search_query']
    current_attempt = state['current_attempt']
    
    # Simulate searching
    if current_attempt < state['max_attempts'] and "important" not in query:
        # Simulate not finding info on first attempt, so we change the query
        new_results = [f"Result for '{query}' (attempt {current_attempt}): No specific info found."]
        new_query = f"{query} important" # Modify query to try again
        found = False
    else:
        # Simulate finding info or max attempts reached
        new_results = [f"Result for '{query}' (attempt {current_attempt}): Found important details!"]
        new_query = query # Keep query same
        found = True

    new_history = [f"Search: '{query}' - found: {found}"]
    return {
        "search_results": new_results,
        "current_attempt": state['current_attempt'] + 1,
        "found_info": found,
        "search_query": new_query, # Update query for next attempt
        "history": new_history
    }

# Node for analyzing search results
def analyze_node(state: LoopingAgentState) -> LoopingAgentState:
    print("---ANALYZE NODE---")
    results = state['search_results']
    analysis = f"Analysis: Found {len(results)} results. Info found status: {state['found_info']}"
    new_history = [analysis]
    return {"history": new_history}

# Conditional router for the loop
def check_loop_condition(state: LoopingAgentState) -> str:
    print("---CHECK LOOP CONDITION---")
    # If info is found OR max attempts reached, we END the loop
    if state['found_info'] or state['current_attempt'] > state['max_attempts']:
        return "end_loop"
    else:
        # Otherwise, loop back to search
        return "search"

# Define the looping workflow
workflow_loop = StateGraph(LoopingAgentState)
workflow_loop.add_node("search", search_node)
workflow_loop.add_node("analyze", analyze_node)

workflow_loop.set_entry_point("search")

# From search, always go to analyze
workflow_loop.add_edge("search", "analyze")

# From analyze, conditionally loop back to search or end
workflow_loop.add_conditional_edges(
    "analyze",
    check_loop_condition,
    {
        "search": "search", # Loop back to search
        "end_loop": END     # Exit loop
    }
)

app_loop = workflow_loop.compile()

print("\n---RUNNING LOOPING GRAPH---")
initial_loop_state = {
    "search_query": "LangGraph tutorial",
    "search_results": [],
    "max_attempts": 3,
    "current_attempt": 1,
    "found_info": False,
    "history": []
}
result_loop = app_loop.invoke(initial_loop_state)

print("\n---FINAL STATE (LOOP)---")
print(result_loop)
print("\n---HISTORY OF ACTIONS (LOOP)---")
for item in result_loop['history']:
    print(f"- {item}")

In this example, the check_loop_condition function decides if the AI should try searching again ("search") or if it’s done (END). This creates a dynamic loop that repeats actions based on the state. You can read more about designing complex loops in the official LangGraph documentation.

H3: Integrating with LangChain Agent Graph

LangGraph is built to work seamlessly with LangChain. This means you can use powerful LangChain agents as nodes within your StateGraph. A LangChain agent graph is essentially a more advanced form of AI program that can choose which tools to use and when to use them.

Imagine you have a LangChain agent that can search the web, calculate math, or summarize articles. You can make this entire agent a single node in your StateGraph. Your StateGraph can then decide when to call this agent and what to do with its results.

This combination is incredibly powerful. Your StateGraph provides the overall structure and flow, while the LangChain agent provides sophisticated, tool-using intelligence within individual steps.

Here’s a simple idea of how an agent could fit in (without full implementation, as a complete agent is quite detailed):

# This is a conceptual example of using a LangChain Agent as a node

# For this, you would typically define a LangChain AgentExecutor
# from langchain.agents import AgentExecutor, create_react_agent, Tool
# from langchain import hub
# from langchain_community.tools import WikipediaQueryRun
# from langchain_community.utilities import WikipediaAPIWrapper

# Let's use a dummy agent for demonstration purposes to avoid complex setup
class DummyAgentExecutor:
    def __init__(self, name):
        self.name = name
    def invoke(self, inputs):
        print(f"---Dummy Agent '{self.name}' Invoked---")
        # In a real scenario, this would involve LLM calls and tool usage
        response = f"Agent '{self.name}' processed: '{inputs['input']}'. Result: Some useful information!"
        return {"output": response}

# Our state for this graph
class AgentGraphState(TypedDict):
    query: str
    agent_response: str
    history: List[str]

# A node that calls our dummy agent
def call_agent_node(state: AgentGraphState) -> AgentGraphState:
    print("---CALLING AGENT NODE---")
    query = state['query']
    
    # In a real setup, this would be your configured LangChain Agent
    # For instance: agent_executor = create_agent_executor(llm, tools, prompt)
    dummy_agent = DummyAgentExecutor("ResearchAgent") 
    
    agent_output = dummy_agent.invoke({"input": query})
    
    new_history = [f"Agent called with query: '{query}'", f"Agent responded: {agent_output['output']}"]
    return {"agent_response": agent_output['output'], "history": new_history}

# A node to process the agent's response
def process_agent_response_node(state: AgentGraphState) -> AgentGraphState:
    print("---PROCESS AGENT RESPONSE NODE---")
    response = state['agent_response']
    processed = f"Processed agent response: {response}. Extracted key info."
    new_history = [f"Processed agent's output: {processed}"]
    return {"history": new_history, "agent_response": processed} # Update the agent_response with processed info

# Build the graph
workflow_agent = StateGraph(AgentGraphState)
workflow_agent.add_node("call_agent", call_agent_node)
workflow_agent.add_node("process_agent_response", process_agent_response_node)

workflow_agent.set_entry_point("call_agent")
workflow_agent.add_edge("call_agent", "process_agent_response")
workflow_agent.add_edge("process_agent_response", END)

app_agent = workflow_agent.compile()

print("\n---RUNNING AGENT INTEGRATION GRAPH---")
initial_agent_state = {"query": "Tell me about large language models.", "agent_response": "", "history": []}
result_agent = app_agent.invoke(initial_agent_state)

print("\n---FINAL STATE (AGENT)---")
print(result_agent)
print("\n---HISTORY OF ACTIONS (AGENT)---")
for item in result_agent['history']:
    print(f"- {item}")

This conceptual example shows how a node could encapsulate an entire LangChain agent. This approach allows you to build very complex and capable AI systems by combining the best features of both LangGraph and LangChain agents. For more on building LangChain agent graph workflows, you might find this external resource helpful: Building AI Agents with LangChain.

H3: Human in the Loop

Sometimes, your AI needs to ask you a question or get your approval before moving forward. This is called “Human in the Loop.” LangGraph supports this by allowing your graph to pause and wait for external input.

When a node needs human input, it can return a special value that tells the graph to pause. You can then provide the input, and the graph will resume from where it left off. This is great for sensitive tasks or when the AI needs clarification.

This makes your AI workflows more collaborative and trustworthy. It ensures that critical decisions or actions always have a human review step.

from langgraph.checkpoint.memory import MemoryCheckpointManager

# Let's redefine AgentState for Human-in-the-Loop
class HumanLoopState(TypedDict):
    query: str
    review_needed: bool
    human_input: str
    final_answer: str
    history: List[str]

# Node: Initial query processing
def initial_process_node(state: HumanLoopState) -> HumanLoopState:
    print("---INITIAL PROCESS NODE---")
    new_history = [f"Initial query: {state['query']}. Checking if review is needed."]
    # Simulate a condition where human review is always needed for a specific query
    needs_review = "urgent" in state['query'].lower()
    return {"review_needed": needs_review, "history": new_history}

# Node: Prompt for human review
def prompt_for_review_node(state: HumanLoopState) -> HumanLoopState:
    print("---PROMPT FOR REVIEW NODE---")
    new_history = [f"Prompting for human review for query: {state['query']}"]
    # This node doesn't *directly* update human_input, that comes from outside.
    # It just signals that review is needed.
    return {"history": new_history}

# Node: Final answer generation
def generate_final_answer_node(state: HumanLoopState) -> HumanLoopState:
    print("---GENERATE FINAL ANSWER NODE---")
    if state['review_needed'] and state['human_input']:
        final_ans = f"Final Answer (with human input): {state['human_input']} based on query: {state['query']}"
    else:
        final_ans = f"Final Answer (no human input): Automatically processed query: {state['query']}"
    new_history = [final_ans]
    return {"final_answer": final_ans, "history": new_history}

# Conditional logic for human review
def check_review_status(state: HumanLoopState) -> str:
    print("---CHECK REVIEW STATUS---")
    if state['review_needed'] and not state.get('human_input'):
        return "human_review" # Go to human review
    else:
        return "generate_answer" # Generate answer (either reviewed or not needed)

# Define the workflow for Human in the Loop
workflow_human = StateGraph(HumanLoopState)
workflow_human.add_node("initial_process", initial_process_node)
workflow_human.add_node("prompt_for_review", prompt_for_review_node)
workflow_human.add_node("generate_answer", generate_final_answer_node)

workflow_human.set_entry_point("initial_process")

# After initial process, check if review is needed
workflow_human.add_conditional_edges(
    "initial_process",
    check_review_status,
    {
        "human_review": "prompt_for_review",
        "generate_answer": "generate_answer"
    }
)

# After prompting for review, it means we expect human input,
# which will then push it to generate_answer.
# For simplicity in this example, we assume human_input will trigger the next step.
# In a real app, you'd use check_review_status again after external update
workflow_human.add_edge("prompt_for_review", "generate_answer") # This is simplified for this example
workflow_human.add_edge("generate_answer", END)

# Compile with a checkpoint manager for persistence (crucial for human-in-the-loop)
# This allows the graph to save its state and resume later.
app_human = workflow_human.compile(checkpointer=MemoryCheckpointManager())

# --- Running the Human-in-the-Loop graph ---

print("\n---RUNNING HUMAN-IN-THE-LOOP GRAPH (NO REVIEW)---")
# This run will not require human input
thread_id_no_review = {"configurable": {"thread_id": "1"}} # Unique ID for this conversation
initial_human_state_no_review = {"query": "What is AI?", "review_needed": False, "human_input": "", "final_answer": "", "history": []}
result_no_review = app_human.invoke(initial_human_state_no_review, thread_id_no_review)
print("\n---FINAL STATE (NO REVIEW)---")
print(result_no_review)
print("\n---HISTORY OF ACTIONS (NO REVIEW)---")
for item in result_no_review['history']:
    print(f"- {item}")

print("\n\n---RUNNING HUMAN-IN-THE-LOOP GRAPH (WITH REVIEW)---")
# This run *will* require human input, and we will simulate providing it
thread_id_review = {"configurable": {"thread_id": "2"}} # Another unique ID

initial_human_state_review = {"query": "Summarize urgent report on climate change.", "review_needed": False, "human_input": "", "final_answer": "", "history": []}

# First invocation: it will pause at prompt_for_review because review_needed becomes True
print("\n---First invocation (will pause for human review)---")
# The current LangGraph method for human-in-the-loop involves calling app.stream
# or managing through an API endpoint. For a simple script, we simulate it
# by manually updating the state and reinvoking.

# First, run up to the point it needs review.
# We modify the check_review_status logic slightly for this manual simulation
# (In a real app, this is handled by LangGraph's internal streaming/checkpointing)
current_state_before_human = app_human.invoke(initial_human_state_review, thread_id_review)
print("Graph state after initial process (waiting for human):", current_state_before_human)
print("Graph should now be in 'prompt_for_review' state, waiting for human_input...")

# Simulate human input by calling app.invoke again with updated state
human_provided_input = "Human says: Ensure to highlight the key environmental impacts."
print(f"\n---Simulating human input: '{human_provided_input}'---")

# We provide the human input as a *partial* state update
# This is how you would typically inject human input when the graph is paused
final_result_review = app_human.invoke({"human_input": human_provided_input}, thread_id_review)

print("\n---FINAL STATE (AFTER REVIEW)---")
print(final_result_review)
print("\n---HISTORY OF ACTIONS (AFTER REVIEW)---")
for item in final_result_review['history']:
    print(f"- {item}")

This example shows how your StateGraph can handle human intervention. The MemoryCheckpointManager is crucial for saving and reloading the state, allowing your graph to pick up exactly where it left off after you provide input. This makes your AI a true assistant that can work with you.

Real-World Applications of LangGraph StateGraph

The power of LangGraph StateGraph isn’t just for theoretical examples; it’s for building real, useful AI applications. Once you understand LangGraph nodes edges and conditional logic, a whole world of possibilities opens up. You can design complex systems that are far beyond what a single AI prompt can do.

Think about how many real-world tasks involve multiple steps, decisions, and remembering information. LangGraph provides the perfect framework for automating these processes with AI. This LangGraph StateGraph tutorial has just scratched the surface of what you can build.

Let’s explore some practical examples where LangGraph can make a huge difference.

H3: Customer Service Bots

Imagine a customer service bot that doesn’t just answer simple questions but can guide a user through troubleshooting steps. If the first solution doesn’t work, it tries another. If it needs user information, it asks for it and remembers the response.

• Nodes: “Identify Issue,” “Suggest Solution A,” “Ask for More Details,” “Suggest Solution B,” “Escalate to Human Agent.”
• Edges: Conditional edges based on user’s response (“Did Solution A work?”), leading to different nodes or loops back to trying new solutions.
• State: Stores user’s issue, attempted solutions, user’s replies, and whether escalation is needed.

This kind of multi-turn, adaptive conversation is a perfect fit for LangGraph. It ensures a consistent and intelligent customer experience.

H3: Research Assistants

An AI research assistant built with LangGraph could be incredibly powerful. It could take a complex topic, break it down, search for information from multiple sources, summarize findings, and even identify gaps in knowledge.

• Nodes: “Break Down Topic,” “Search Wikipedia,” “Search Academic Papers,” “Summarize Findings,” “Identify Knowledge Gaps,” “Generate Report.”
• Edges: Loop back to “Search Academic Papers” if knowledge gaps are found, conditional edges for prioritizing search sources.
• State: Stores the research topic, sub-topics, search queries, raw search results, summarized information, and a list of unanswered questions.

Such an assistant could significantly speed up research tasks, providing comprehensive and well-structured information. You could even incorporate LangChain agent graph nodes to delegate specific research tasks.

H3: Automated Content Generation

For content creators, LangGraph can automate parts of the writing process. An AI could brainstorm ideas, create outlines, write drafts, and even revise text based on feedback.

• Nodes: “Brainstorm Topics,” “Generate Outline,” “Write Draft Section A,” “Write Draft Section B,” “Review and Edit,” “Check for Plagiarism,” “Publish.”
• Edges: Loop back to “Write Draft Section X” for revisions, conditional edges to “Review and Edit” if initial quality is low.
• State: Holds the content topic, generated outline, text drafts, revision notes, and publishing status.

This allows for efficient creation of blogs, articles, or marketing copy, ensuring consistency and quality throughout.

H3: Task Automation and Orchestration

Beyond text-based tasks, LangGraph can orchestrate complex automation workflows. Imagine managing project tasks: checking statuses, updating databases, sending notifications, and triggering further actions based on specific criteria.

• Nodes: “Check Task Status,” “Update Database,” “Send Email Notification,” “Create Follow-up Task,” “Generate Summary Report.”
• Edges: Conditional edges based on task completion status, leading to different actions (e.g., if complete, update database; if stuck, create follow-up).
• State: Stores project ID, task lists, current task status, team members involved, and communication logs.

LangGraph provides a robust way to ensure that multi-step processes are executed reliably and intelligently, making it an excellent tool for many business applications.

Tips for Building Great AI Workflows

Building effective AI workflows with LangGraph is a skill that improves with practice. Here are some tips to help you on your journey after this LangGraph StateGraph tutorial. These best practices will ensure your StateGraph Python applications are robust, easy to understand, and perform well.

Remember, even the most complex AI systems start with simple building blocks. Don’t be afraid to experiment and iterate on your designs.

H3: Start Simple

It’s tempting to jump into complex designs right away, but the best approach is to start with a very simple workflow. Get a basic graph with a few nodes and direct edges working first. This helps you understand the core mechanics without getting overwhelmed.

Once your simple graph is running, gradually add more complexity: introduce conditional edges, then loops, and finally integrate with LangChain agent graph components. This incremental approach makes debugging easier and helps you grasp each concept firmly. Think of it as learning to walk before you run.

H3: Test Often

Just like any other software, your LangGraph workflows need to be tested regularly. Run your graph with different initial states to ensure all paths work as expected. Pay close attention to your conditional edges to make sure they’re routing correctly.

When you’re building a loop, test the conditions for both entering and exiting the loop. Regular testing helps catch bugs early and ensures your AI behaves predictably in various scenarios. Debugging a small, tested segment is much easier than a large, untested one.

H3: Draw Your Graph First

Before you write any code, it’s incredibly helpful to draw your workflow on paper or using a diagram tool. Sketch out your LangGraph nodes edges, showing where the information flows and where decisions are made. This visual representation helps you clarify your logic and identify potential issues before coding.

A clear diagram serves as a blueprint for your StateGraph Python implementation. It also makes it easier to explain your workflow to others and to track changes as your project evolves. This simple step can save you a lot of time and frustration later on.

Troubleshooting Common Issues

Even with careful planning, you might encounter issues when building your LangGraph workflows. Don’t get discouraged! Troubleshooting is a normal part of the development process. Here are some common problems and how to approach them within your LangGraph StateGraph tutorial journey.

Understanding common pitfalls will help you quickly identify and fix problems, making your development process smoother.

H3: Graph Not Moving or Getting Stuck

If your graph starts but then doesn’t seem to progress, or just stops unexpectedly without reaching END, here are things to check:

• Missing END Node: Did you forget to add an END node? Every valid path in your graph should eventually lead to END.
• Incorrect Edges: Double-check your add_edge and add_conditional_edges calls. Is the source node spelled correctly? Is the target node spelled correctly?
• Conditional Edge Logic: If you’re using conditional edges, ensure your routing function (route_decision in our example) returns one of the keys you’ve mapped. If it returns something unexpected, the graph won’t know where to go.
• Infinite Loops: If your graph keeps running indefinitely, you might have created an unintentional loop without an exit condition. Review your loop’s conditional logic to ensure it eventually leads to an END state.

Look at the print statements within your nodes. If a node isn’t printing, it means the graph isn’t reaching that node. This helps pinpoint where the flow breaks down.

H3: State Not Updating as Expected

If your AI isn’t remembering information correctly, or a node isn’t seeing the right data from previous steps, check your StateGraph Python definitions:

• TypedDict Keys: Ensure that the keys you are returning from your nodes exactly match the keys defined in your AgentState (e.g., AgentState(TypedDict)). A typo here will mean the state isn’t updated.
• Return Values from Nodes: Each node function should return an AgentState dictionary. If a node returns None or an empty dictionary when it should be updating the state, the state won’t change.
• State Accumulation: Remember that LangGraph combines state updates. If you’re using a complex object like a list, ensure your combine_ function (like combine_history) correctly merges the new data with the existing state.
• Print the State: Add print(state) at the beginning of each node to see exactly what information that node is receiving. This is a powerful debugging technique.

Ensuring correct state management is crucial for the memory and continuity of your AI workflow.

Conclusion

Congratulations! You’ve reached the end of this LangGraph StateGraph tutorial. You’ve taken your first significant steps into building sophisticated AI workflows in Python. You now understand what LangGraph is, how StateGraph Python provides memory to your AI, and how to create dynamic paths using LangGraph nodes edges.

We covered setting up your environment, building a simple linear graph, and then making it smarter with conditional logic. We also touched upon advanced concepts like looping, integrating with a LangChain agent graph, and adding human interaction. You’ve seen practical examples and gained insights into real-world applications.

The world of AI is rapidly evolving, and tools like LangGraph empower you to build intelligent systems that go beyond single-turn interactions. Keep practicing, experiment with new ideas, and don’t hesitate to refer back to the official LangGraph documentation for more detailed information. Your journey into building complex AI workflows has just begun. What amazing AI applications will you create next?