LangGraph Tutorial: Stateful Multi-Agent AI Workflows

The landscape of Artificial Intelligence is rapidly evolving, moving beyond simple prompt-response interactions to more sophisticated, collaborative systems. While Large Language Models (LLMs) have demonstrated incredible capabilities, real-world applications often require a series of interdependent steps, dynamic decision-making, and the ability to maintain context across multiple turns or agents. This is where LangGraph steps in, offering a robust framework for building stateful, multi-agent AI workflow applications.

LangGraph, an extension of the popular LangChain library, provides a powerful and intuitive way to construct complex AI systems as graphs. Think of it as a state machine for your AI agents, enabling you to define nodes (where processing happens) and edges (how control flows between nodes), all while maintaining a persistent state throughout the execution. This capability is crucial for applications that need to remember past interactions, make decisions based on accumulating information, or involve multiple specialized AI agents working together towards a common goal.

The Challenge of Stateful AI Workflows

Developing AI applications that mimic human-like reasoning and collaboration presents several architectural challenges. Traditional approaches often struggle with:

• Maintaining Context: Simple LLM calls are stateless. Each interaction is a fresh start, making it difficult to build applications that remember previous turns or decisions without explicit, cumbersome context passing.
• Complex Logic: Real-world problems rarely fit into a single, linear chain of operations. They require conditional branching, loops, and dynamic routing based on intermediate results.
• Multi-Agent Coordination: Imagine an AI system that needs to research a topic, then summarize it, then draft an email, and finally review the email. This involves multiple specialized “agents” or modules, each with its own task. Orchestrating their collaboration and ensuring smooth handoffs can be incredibly complex.
• Error Handling and Recovery: In a multi-step process, if one step fails, how do you gracefully recover or retry? Without a clear workflow definition, managing these scenarios becomes a nightmare.

These challenges highlight the need for a framework that can not only define the steps of an AI application but also manage the state as it progresses through those steps. LangGraph addresses these pain points by providing a clear, graph-based abstraction for designing such systems.

Understanding LangGraph’s Core Concepts

At its heart, LangGraph is about defining a directed graph where nodes represent computational steps and edges represent transitions between these steps. The magic, however, lies in its ability to manage a shared, mutable state that evolves as the graph executes. Let’s break down the fundamental concepts:

Graph State

The Graph State is the central piece of information that all nodes can read from and write to. It’s typically a dictionary-like object that holds all the relevant data for your workflow. When a node executes, it receives the current state, performs its operation, and then returns updates to that state. These updates are then merged into the overall graph state for the next node to use.

Key Idea: The state is the shared memory of your AI workflow. It allows agents to communicate and build upon each other’s work without explicit message passing.

Nodes

Nodes are the individual units of computation within your LangGraph workflow. They can be:

• LLM Invocations: Calling a Large Language Model to generate text, classify input, or extract information.
• Tool Calls: Executing external functions or APIs (e.g., searching the web, querying a database, sending an email).
• Custom Functions: Any Python function that performs a specific task, such as data processing, validation, or formatting.

Each node takes the current graph state as input and returns a partial update to that state. LangGraph then merges these updates.

Edges

Edges define the flow of execution between nodes. There are two primary types of edges:

• Direct Edges: These are straightforward transitions from one node to another. After Node A completes, Node B always executes next.
• Conditional Edges: These are powerful for dynamic decision-making. A conditional edge specifies a “router” function that inspects the current graph state and decides which node to execute next based on certain conditions. This allows for branching logic in your workflows.

Checkpoints

LangGraph can persist the state of your graph at various points, often after each step. This feature, called checkpointing, is invaluable for several reasons:

• Fault Tolerance: If your application crashes, you can restart from the last saved state.
• Human-in-the-Loop: You can pause the workflow, allow a human to review or modify the state, and then resume.
• Debugging and Observability: You can inspect the state at any point in the workflow’s history.

Setting Up Your Environment

Before diving into code, ensure you have the necessary libraries installed. We’ll need langchain, langgraph, and an LLM provider (like OpenAI or Anthropic). For this tutorial, we’ll assume you have an OpenAI API key.

# Install necessary librariespip install langchain langchain-openai langgraph

You’ll also need to set up your API key as an environment variable:

import osos.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

Building a Simple Stateful Agent Workflow

Let’s construct a basic LangGraph application. We’ll create a simple workflow where an AI agent receives a user query, processes it, and then potentially updates a “history” in its state.

1. Define the Graph State

First, we define the structure of our shared state using a TypedDict. This makes the state explicit and easier to manage.

from typing import TypedDict, Listfrom langchain_core.messages import BaseMessage# Define the state for our graphclass AgentState(TypedDict):    """    Represents the state of our graph.    Attributes:        messages: A list of messages in the current conversation turn.        tool_output: Any output from a tool call (optional).    """    messages: List[BaseMessage]    tool_output: str # Could be a more complex type, but str for simplicity

2. Define Nodes

Next, we’ll create a node that uses an LLM to respond to messages.

from langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, AIMessage# Initialize the LLMllm = ChatOpenAI(model="gpt-4o", temperature=0)# Define a node function that interacts with the LLMdef call_llm(state: AgentState):    """    Node to call the LLM and update the state with its response.    """    messages = state["messages"]    response = llm.invoke(messages)    return {"messages": [response]} # Return a list containing the AI's response message

For a more advanced example, let’s add a “tool” node that simulates a data lookup.

from langchain_core.tools import tool# Define a dummy tool@tooldef search_data(query: str) -> str:    """    Simulates searching an internal knowledge base for the query.    """    if "weather" in query.lower():        return "The weather in London is currently 15°C and cloudy."    elif "population of new york" in query.lower():        return "The population of New York City is approximately 8.4 million."    else:        return "Could not find specific data for that query."# Define a node function to use the tooldef call_tool(state: AgentState):    """    Node to call a tool based on the LLM's decision (if any).    For simplicity, this example directly calls a dummy tool.    In a real scenario, the LLM would decide *which* tool to call.    """    messages = state["messages"]    last_message = messages[-1]    # In a real agent, you'd parse tool_calls from the LLM's response.    # For this simple example, we'll just check if the query implies a search.    if "search" in last_message.content.lower() or "?" in last_message.content:        print("---Calling Search Tool---")        tool_result = search_data.invoke({"query": last_message.content})        return {"tool_output": tool_result, "messages": [AIMessage(content=f"Tool output: {tool_result}")]}    else:        return {"tool_output": "No tool called."}

3. Define the Graph and Edges

Now, we’ll assemble our nodes into a graph. We’ll use a StateGraph and define the entry point, nodes, and edges.

from langgraph.graph import StateGraph, END# Build the workflowworkflow = StateGraph(AgentState)# Add nodes to the graphworkflow.add_node("llm_node", call_llm)workflow.add_node("tool_node", call_tool)# Set the entry pointworkflow.set_entry_point("llm_node")# Define conditional logic for the next step after the LLM nodedef decide_what_to_do(state: AgentState):    """    Router function to decide whether to call a tool or end the conversation.    """    last_message = state["messages"][-1]    # For simplicity, if the LLM's response contains 'tool_code', it implies a tool call.    # In a real LangChain agent, this would parse actual tool_calls from the LLM's output.    if "tool_code" in last_message.content: # Placeholder for real tool calling intent        print("---DECISION: CALL TOOL---")        return "tool_node"    else:        print("---DECISION: GENERATE RESPONSE---")        return END # End the workflow if no tool is needed# Add edges# From llm_node, decide whether to go to tool_node or endworkflow.add_conditional_edges(    "llm_node",    decide_what_to_do,    {        "tool_node": "tool_node",        END: END    })# After tool_node, always go back to llm_node to process tool outputworkflow.add_edge("tool_node", "llm_node")

4. Compile and Run the Graph

Finally, compile the graph and invoke it. We’ll add a persistent checkpointing mechanism for better state management.

from langgraph.checkpoint.sqlite import SqliteSaverfrom langchain_core.messages import SystemMessage# Initialize checkpointingmemory = SqliteSaver.from_conn_string(":memory:")# Compile the graphapp = workflow.compile(checkpointer=memory)# Example conversationsprint("---Conversation 1---")config = {"configurable": {"thread_id": "1"}} # Unique ID for this conversation threadinitial_message = HumanMessage(content="What is the weather like in London?")# The first message goes into the state.# The graph will start at the entry point ("llm_node").for s in app.stream({"messages": [initial_message]}, config=config):    if "__end__" not in s:        print(s)        print("---")print("\n---Conversation 2 (Continuing Thread 1)---")# Continuing the same thread, the state will be loaded from checkpointnext_message = HumanMessage(content="And what about the population of New York?")for s in app.stream({"messages": [next_message]}, config=config):    if "__end__" not in s:        print(s)        print("---")# You can inspect the final state for a threadfinal_state = app.get_state(config)print("\nFinal state for Thread 1:")print(final_state.values)

In this example, the decide_what_to_do function acts as a simple router. In a production scenario, the LLM itself would output a structured tool call, and the router would parse that to decide whether to invoke a tool or simply respond. The key takeaway is how the state (messages and tool_output) is passed and updated across nodes and turns.

Advanced LangGraph Patterns: Multi-Agent Collaboration

LangGraph truly shines when orchestrating multiple specialized agents. Let’s outline a more complex scenario: a research assistant that can search for information and then summarize it. This involves two conceptual agents: a “Researcher” (which uses a search tool) and a “Summarizer” (which uses an LLM to condense information).

Designing a Research and Summarize Workflow

1. User Query: The user asks a question.
2. Researcher Agent:
   • Receives the query.
   • Decides if a search tool is needed.
   • If yes, calls the search tool.
   • Updates the state with search results.
3. Summarizer Agent:
   • Receives the search results (from the state).
   • Uses an LLM to summarize the results.
   • Updates the state with the summary.
4. Final Response: The main LLM node provides a final answer incorporating the summary.

For brevity, we’ll focus on the architectural concepts and the conditional routing, rather than full detailed code for every sub-agent, but the structure remains the same as our simple example. Each “agent” would essentially be a node or a sub-graph within the larger graph.

Implementing Conditional Routing for Agent Handoffs

A crucial part of multi-agent systems is the ability to intelligently hand off tasks. This is where conditional edges become indispensable. Imagine our Researcher node outputs a flag indicating if it found relevant information or if it needs further clarification. This flag can then dictate whether the workflow proceeds to the Summarizer or goes back to the Researcher for more input.

# Simplified representation of a multi-agent stateclass MultiAgentState(TypedDict):    query: str    research_results: str    summary: str    next_action: str # e.g., "research", "summarize", "respond", "clarify"    messages: List[BaseMessage]# Node for Research Agentdef research_node(state: MultiAgentState):    print("---RESEARCHING---")    query = state["query"]    # Simulate tool call based on query    if "latest tech trends" in query.lower():        results = "AI advancements, quantum computing, cybersecurity threats."        next_action = "summarize"    else:        results = "No specific research results found."        next_action = "respond" # Or "clarify"    return {"research_results": results, "next_action": next_action, "messages": [AIMessage(content=f"Research complete. Results: {results}")]}# Node for Summarizer Agentdef summarize_node(state: MultiAgentState):    print("---SUMMARIZING---")    research_results = state["research_results"]    # LLM call to summarize    summary = llm.invoke(f"Summarize the following research results: {research_results}").content    return {"summary": summary, "next_action": "respond", "messages": [AIMessage(content=f"Summary: {summary}")]}# Node for Final Response Agentdef final_response_node(state: MultiAgentState):    print("---GENERATING FINAL RESPONSE---")    query = state["query"]    summary = state.get("summary", state.get("research_results", "No information found."))    final_answer = llm.invoke(f"Based on the query '{query}' and summary '{summary}', provide a concise answer.").content    return {"messages": [AIMessage(content=final_answer)], "next_action": "end"}# Router function for multi-agent workflowdef router(state: MultiAgentState):    if state["next_action"] == "research":        return "research_node"    elif state["next_action"] == "summarize":        return "summarize_node"    elif state["next_action"] == "respond":        return "final_response_node"    else: # Default to end or error handling        return END# Build the multi-agent workflow graphmulti_agent_workflow = StateGraph(MultiAgentState)multi_agent_workflow.add_node("research_node", research_node)multi_agent_workflow.add_node("summarize_node", summarize_node)multi_agent_workflow.add_node("final_response_node", final_response_node)multi_agent_workflow.set_entry_point("research_node") # Start with research# Conditional edges based on the 'next_action' in statemulti_agent_workflow.add_conditional_edges(    "research_node",    router,    {        "summarize_node": "summarize_node",        "final_response_node": "final_response_node",        END: END    })multi_agent_workflow.add_conditional_edges(    "summarize_node",    router,    {        "final_response_node": "final_response_node",        END: END    })multi_agent_workflow.add_conditional_edges(    "final_response_node",    router,    {        END: END # Final response always ends    })multi_agent_app = multi_agent_workflow.compile(checkpointer=memory)print("\n---Multi-Agent Conversation Example---")config_multi = {"configurable": {"thread_id": "2"}}initial_query_multi = HumanMessage(content="Tell me about the latest tech trends.")for s in multi_agent_app.stream({"query": initial_query_multi.content, "next_action": "research", "messages": [initial_query_multi]}, config=config_multi):    if "__end__" not in s:        print(s)        print("---")final_multi_state = multi_agent_app.get_state(config_multi)print("\nFinal state for Multi-Agent Thread 2:")print(final_multi_state.values)

Key Benefits of LangGraph

Adopting LangGraph for your AI applications brings a host of advantages, especially for complex, stateful interactions:

• Explicit State Management: The graph state provides a clear, centralized repository for all relevant information, making it easy to understand and debug your application’s memory.
• Modularity and Reusability: Nodes are independent units of logic. You can easily swap out LLMs, add new tools, or reuse nodes across different workflows, promoting a cleaner codebase.
• Complex Control Flow: Conditional edges allow for sophisticated branching, looping, and dynamic routing, which are essential for building intelligent agents that can adapt to varying inputs and scenarios.
• Observability: With features like checkpointing and the clear graph structure, it’s significantly easier to trace the execution path of your agents, inspect the state at each step, and identify where issues might occur.
• Human-in-the-Loop Capabilities: Checkpointing and the explicit state enable seamless integration of human intervention, allowing for review, correction, and resumption of AI workflows.
• Scalability: By breaking down complex tasks into smaller, manageable nodes, LangGraph makes it easier to distribute and scale different parts of your AI system.

Best Practices for LangGraph Development

To get the most out of LangGraph and build robust, maintainable applications, consider these best practices:

• Define a Clear State Schema: Start by meticulously defining your TypedDict for the graph state. This acts as the contract for all your nodes and helps prevent errors.
• Keep Nodes Atomic: Each node should ideally perform a single, well-defined task. This improves readability, testability, and reusability. Avoid cramming too much logic into one node.
• Use Descriptive Node and Edge Names: Clear naming conventions make your graph easier to understand, especially as it grows in complexity.
• Implement Robust Error Handling: Consider how your nodes will handle failures. LangGraph allows for error handling within nodes and can be configured to retry or transition to specific error-handling nodes.
• Leverage Checkpointing: Always use a checkpointer, especially for long-running or multi-turn conversations. This provides resilience and enables powerful features like human review.
• Test Thoroughly: Test individual nodes in isolation, and then test the entire graph flow with various inputs and edge cases.
• Visualize Your Graph: For complex graphs, consider using LangGraph’s built-in visualization tools (often requiring graphviz) to get a clear picture of your workflow.

Conclusion

LangGraph represents a significant leap forward in building sophisticated AI applications. By providing a structured, stateful, and modular framework, it empowers developers to move beyond linear chains and create truly dynamic, multi-agent workflows. Whether you’re building a complex RAG system, an autonomous research assistant, or an interactive conversational agent, LangGraph offers the tools to design, implement, and manage the intricate logic required. Embracing its graph-based approach will not only simplify your development process but also unlock new possibilities for creating intelligent, resilient, and highly capable AI systems. The future of AI is collaborative and stateful, and LangGraph is an essential tool for navigating that future.