Building a Data Analysis Agent with LangGraph: Step-by-Step Tutorial

Introduction: What is LangGraph and Data Analysis Agents?

LangGraph is a framework from the LangChain ecosystem designed for building stateful, multi-actor applications powered by large language models (LLMs). Unlike simple chains that follow a linear sequence, LangGraph lets you define workflows as graphs—nodes represent processing steps, edges represent transitions, and state flows through the graph carrying context from one step to the next. This makes it ideal for building agentic systems that can reason, plan, and interact with tools and data sources iteratively.

A Data Analysis Agent is an LLM-powered application that can autonomously load datasets, explore data, perform statistical computations, generate visualizations, and produce human-readable insights—all from a natural language question. By combining LangGraph with libraries like pandas and matplotlib, you can create a robust analyst that handles real-world CSV files, adapts to user queries, and even asks for human approval before finalizing expensive operations.

In this step-by-step tutorial, you'll learn how to build such an agent from scratch. We'll cover state definition, node logic, conditional routing, tool integration, memory, and best practices. Every code block is ready to run, and by the end you'll have a fully functional data analysis agent you can extend for your own projects.

Why Build a Data Analysis Agent with LangGraph?

Traditional data analysis workflows require manual coding: you write a script to load data, perform transformations, compute summaries, and plot results. An agent automates this by interpreting the user’s question, dynamically deciding what operations to perform, and executing them with safety checks. LangGraph provides key advantages for this:

• Explicit control flow – you design the decision logic as a graph, not a black-box chain.
• Persistent state – intermediate results like DataFrames and plot paths can be stored and reused across nodes.
• Human-in-the-loop – you can pause execution and ask for approval before running code or saving files.
• Tool integration – easily wrap pandas operations, plotting functions, or even a Python REPL as tools.
• Observability – each node's input and output can be logged, making debugging straightforward.

By the end of this tutorial, you'll understand how to leverage these features to build an agent that is reliable, transparent, and ready for production use.

Core Concepts: LangGraph Essentials

Before diving into code, let's review the main building blocks:

State

The State is a shared data object (usually a TypedDict or Pydantic model) that passes through every node. It accumulates information like user input, loaded data, analysis results, and final answers. In LangGraph, state is typically defined with a schema and an optional reducer (e.g., operator.add for appending messages).

Nodes

Nodes are Python functions that receive the current state and return an updated state. Each node performs a specific task: loading a CSV, running a pandas query, generating a chart, etc. Nodes are the "work" units of your graph.

Edges

Edges connect nodes and define the flow. A normal edge always transitions from one node to the next. A conditional edge uses a router function to decide which node to go to next based on the current state (e.g., "if data is loaded, go to analysis; else go to loading").

Graph

The StateGraph object is where you add nodes and edges, compile the graph with a checkpointer (for memory), and then invoke it with an initial state.

Setting Up Your Environment

Create a new Python project and install the required dependencies. We'll use langgraph, langchain-openai (or any chat model), pandas, and matplotlib. You'll also need an OpenAI API key (or swap in a local model later).

pip install langgraph langchain langchain-openai pandas matplotlib python-dotenv

Store your API key in a .env file:

OPENAI_API_KEY=sk-...

Then load it in your script:

from dotenv import load_dotenv
load_dotenv()

Step 1: Define the State Schema

The state will carry all information the agent needs: the user's question, the file path, the loaded DataFrame (as a JSON string or dict to keep it serializable), analysis results, a list of generated plot file paths, and the final answer. We'll use a TypedDict with an operator.add reducer for messages to accumulate chat history.

from typing import TypedDict, List, Optional, Annotated
import operator

class AnalystState(TypedDict):
    # User input and conversation history
    messages: Annotated[List[str], operator.add]
    # Path to the data file
    file_path: str
    # Loaded data as a serializable dict (column -> list of values) or None
    data_dict: Optional[dict]
    # Analysis results as a string (e.g., statistical summary)
    analysis_result: Optional[str]
    # Paths to generated visualizations
    plot_paths: List[str]
    # Final answer to return to the user
    final_answer: Optional[str]
    # Flag for human approval if needed
    human_approved: bool

The Annotated[List[str], operator.add] ensures that when a node returns a partial state with new messages, they are appended to the existing list, preserving conversation history.

Step 2: Build the Node Functions

Each node is a function that takes the state and returns a dictionary with updates. We'll implement five core nodes: loading data, analyzing, visualizing, formatting the final answer, and a human-in-the-loop approval node. We'll also create a router node that decides the next step based on the current state.

2.1 Data Loading Node

This node reads the CSV file specified in file_path, converts it to a dictionary, and stores it in the state. If data is already loaded, it skips.

import pandas as pd

def load_data_node(state: AnalystState) -> dict:
    # Skip if data already loaded
    if state.get("data_dict") is not None:
        return {}
    file_path = state["file_path"]
    df = pd.read_csv(file_path)
    # Convert DataFrame to dict of lists (serializable)
    data_dict = df.to_dict(orient='list')
    return {"data_dict": data_dict}

2.2 Analysis Node

The analysis node reconstructs the DataFrame, performs a basic statistical summary using pandas describe(), and stores the result as a string. In a more advanced agent, you'd use an LLM to generate pandas code dynamically; here we keep it simple for clarity.

def analyze_node(state: AnalystState) -> dict:
    data_dict = state.get("data_dict")
    if not data_dict:
        return {"analysis_result": "No data loaded."}
    df = pd.DataFrame(data_dict)
    # Generate descriptive statistics
    stats = df.describe(include='all').to_string()
    return {"analysis_result": stats}

2.3 Visualization Node

This node creates a histogram for the first numeric column and saves it as a PNG. It updates the plot_paths list. We'll use matplotlib and a fixed output directory.

import matplotlib.pyplot as plt
import os

def visualize_node(state: AnalystState) -> dict:
    data_dict = state.get("data_dict")
    if not data_dict:
        return {}
    df = pd.DataFrame(data_dict)
    # Pick first numeric column
    numeric_cols = df.select_dtypes(include='number').columns
    if len(numeric_cols) == 0:
        return {}
    col = numeric_cols[0]
    plt.figure()
    df[col].hist(bins=20)
    plt.title(f"Distribution of {col}")
    os.makedirs("plots", exist_ok=True)
    path = f"plots/{col}_hist.png"
    plt.savefig(path)
    plt.close()
    new_paths = state.get("plot_paths", []) + [path]
    return {"plot_paths": new_paths}

2.4 Final Answer Node

This node composes a final answer string that summarizes the analysis and references the generated plots. In a production agent, you'd pass the state to an LLM to generate a narrative. We'll do a basic version.

def final_answer_node(state: AnalystState) -> dict:
    analysis = state.get("analysis_result", "No analysis performed.")
    plots = state.get("plot_paths", [])
    plot_str = "\n".join(plots) if plots else "No visualizations generated."
    answer = f"Analysis Result:\n{analysis}\n\nVisualizations:\n{plot_str}"
    return {"final_answer": answer}

2.5 Human Approval Node (for demonstration)

A simple node that sets human_approved to True after a simulated check. In a real system, you'd integrate a front-end interrupt.

def human_approval_node(state: AnalystState) -> dict:
    # Simulate human approval (always approved in this example)
    return {"human_approved": True}

Step 3: Create the Graph and Connect Nodes

Now we instantiate a StateGraph with our state schema, add nodes, and wire them with normal edges. We'll also include a router node that uses a conditional edge to decide whether to go to analysis or directly to final answer based on data availability.

3.1 Router Node Logic

The router is a regular node that returns a dictionary with a special key next_step (or we can use a separate function for conditional edges). We'll implement a conditional edge function that inspects the state.

def route_after_load(state: AnalystState) -> str:
    if state.get("data_dict") is None:
        # Data loading failed or not yet done; but we just loaded it.
        # For safety, go to final answer with error
        return "final_answer"
    return "analyze"

We'll also route after analysis to either visualization (if numeric columns exist) or final answer.

def route_after_analysis(state: AnalystState) -> str:
    if state.get("data_dict"):
        df = pd.DataFrame(state["data_dict"])
        if df.select_dtypes(include='number').columns.size > 0:
            return "visualize"
    return "final_answer"

3.2 Building the Graph

from langgraph.graph import StateGraph

# Create graph builder
builder = StateGraph(AnalystState)

# Add nodes
builder.add_node("load_data", load_data_node)
builder.add_node("analyze", analyze_node)
builder.add_node("visualize", visualize_node)
builder.add_node("final_answer", final_answer_node)
builder.add_node("human_approval", human_approval_node)

# Set entry point: start at load_data
builder.set_entry_point("load_data")

# After load, decide next step with conditional edge
builder.add_conditional_edges(
    "load_data",
    route_after_load,
    {
        "analyze": "analyze",
        "final_answer": "final_answer"
    }
)

# After analyze, decide whether to visualize or finalize
builder.add_conditional_edges(
    "analyze",
    route_after_analysis,
    {
        "visualize": "visualize",
        "final_answer": "final_answer"
    }
)

# After visualization, go to final_answer
builder.add_edge("visualize", "final_answer")

# After final_answer, optionally go to human_approval (if not already approved)
# We'll use a conditional edge that loops back if not approved
def route_after_final(state: AnalystState) -> str:
    if state.get("human_approved", False):
        return "end"
    return "human_approval"

builder.add_conditional_edges(
    "final_answer",
    route_after_final,
    {
        "human_approval": "human_approval",
        "end": "__end__"
    }
)

# After human approval, go back to final_answer to regenerate or just end
builder.add_edge("human_approval", "final_answer")

Note the special node name "__end__" is the graph's built-in terminal node. The human approval loop demonstrates a simple pattern: after the final answer is drafted, we check if a human approved it; if not, we go to the approval node, then back to final_answer (which could regenerate the answer with an approval flag).

Step 4: Compile and Run the Agent

Compile the graph with a checkpointer if you want to persist state across invocations. For a simple run without persistence, we can compile without a checkpointer. Then invoke with an initial state.

# Compile graph (no persistence for this basic example)
graph = builder.compile()

# Initial state
initial_state = {
    "messages": ["User asked for a data analysis report"],
    "file_path": "sales_data.csv",  # replace with your CSV file
    "data_dict": None,
    "analysis_result": None,
    "plot_paths": [],
    "final_answer": None,
    "human_approved": False
}

# Run the agent
final_state = graph.invoke(initial_state)

# Display the final answer
print(final_state["final_answer"])
print("Plots generated:", final_state["plot_paths"])

If you want to use a checkpointer for memory (e.g., to resume a paused execution), you can pass a MemorySaver or a persistent checkpointer like SqliteSaver:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph_with_memory = builder.compile(checkpointer=checkpointer)
# Then invoke with a thread_id for the session
final_state = graph_with_memory.invoke(
    initial_state,
    config={"configurable": {"thread_id": "session-1"}}
)

Step 5: Adding an LLM-Powered Reasoning Node (Advanced)

The above agent follows a fixed path. To make it truly "agentic," you can replace the fixed analysis and routing with an LLM that decides what to do next. LangGraph integrates seamlessly with LangChain tools. Here's a sketch of how to add an LLM node that uses a python_repl tool to run arbitrary pandas code.

5.1 Define Tools

from langchain_core.tools import tool

@tool
def python_repl(code: str) -> str:
    """Execute Python code and return the result. Use for pandas operations."""
    import pandas as pd
    # We need access to the DataFrame from state, so we'll pass it via globals in the node.
    # This is a simplified stub; real implementation would capture output.
    exec(code, globals())
    return "Code executed successfully."

5.2 LLM Node

You would create a node that calls an LLM with the conversation history and the tools. The LLM can request tool calls, and LangGraph's ToolNode can execute them and return results. The state would then be updated with new messages and analysis results. This pattern is known as the ReAct agent.

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [python_repl]
tool_node = ToolNode(tools)

def llm_node(state: AnalystState):
    # Convert messages to LangChain message format
    from langchain_core.messages import HumanMessage, AIMessage
    msgs = [HumanMessage(content=m) for m in state["messages"]]
    # Bind tools to LLM
    llm_with_tools = llm.bind_tools(tools)
    response = llm_with_tools.invoke(msgs)
    # Return updated messages (AIMessage includes tool_calls)
    return {"messages": [response.content]}  # simplified

The full implementation is beyond our simple tutorial, but LangGraph's documentation provides excellent examples of ReAct agents. For data analysis, you'd let the LLM decide when to load data, run code, and generate charts, looping until a final answer is ready.

Best Practices for LangGraph Data Analysis Agents

• Keep state serializable – avoid storing complex objects like DataFrames directly; use JSON-serializable dicts or base64-encoded Parquet buffers for large data.
• Use checkpoints for reliability – enable graph persistence so if a node fails, you can resume from the last successful step without re-executing expensive operations.
• Add human-in-the-loop for sensitive actions – before writing files, executing code that modifies data, or sending results, pause for approval. Use interrupt before critical nodes.
• Instrument nodes with logging – log state snapshots at each node to quickly identify where errors occur.
• Limit tool permissions – if using a Python REPL tool, sandbox it (e.g., restrict imports, network access, and filesystem writes to a tmp directory).
• Iterate with small, testable graphs – start with a simple linear flow, then add conditional branching and tools incrementally.
• Use typed state and validation – Pydantic models can automatically validate state updates, catching bugs early.

Conclusion

You've now built a complete data analysis agent using LangGraph. We started with a fixed pipeline that loads a CSV, computes statistics, generates a histogram, and compiles a final report. Then we added conditional routing and a human-in-the-loop pattern to simulate approval. Finally, we glimpsed how to integrate LLM-powered reasoning for more dynamic, agentic behavior.

The graph-based architecture gives you full control over every step, making your agent transparent, debuggable, and extensible. You can easily swap in your own analysis logic, connect to databases, or add new visualization types. LangGraph's state management and checkpointing also make it production-ready for long-running and interruptible workflows.

Now it's your turn: grab a dataset, tweak the nodes, and start building agents that answer real-world data questions with confidence.