← Back to DevBytes
Developer Tutorial July 18, 2026 9 min read

Building a Data Analysis Agent with LangGraph: Complete Guide

What is a Data Analysis Agent with LangGraph?

A Data Analysis Agent is an AI-powered assistant that can autonomously explore, transform, and visualize data by writing and executing code. Instead of just generating text answers, it interacts with a Python environment—loading CSV files, filtering DataFrames, creating charts with matplotlib, and iteratively refining its analysis based on results. LangGraph, a stateful orchestration framework from the LangChain ecosystem, gives you the control to build these agents as customizable graphs. You define nodes (LLM calls, tool executions) and edges (conditional logic, loops), enabling complex, multi-step reasoning that stays reliable even when tasks require back-and-forth between thinking and acting.

This tutorial walks you through building a complete data analysis agent using LangGraph. You'll learn how to define the agent's state, integrate Python execution tools, construct a ReAct-style loop, and apply best practices to keep your agent safe and effective. By the end, you'll have a working agent that can answer questions like "Load sales.csv, compute monthly revenue, and plot the trend."

Why LangGraph for Data Analysis Agents Matters

Traditional LLM chains are linear: prompt → response. But data analysis is inherently iterative. An agent might need to inspect a dataset, write code, encounter an error, fix the code, re-run, and then produce a final summary. LangGraph provides three critical capabilities that make this natural:

With LangGraph, you're not just piping a single prompt—you're building a robust, stateful workflow that mirrors how a human analyst would work: explore, code, observe, adjust, and conclude.

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

Prerequisites and Setup

Install the required libraries. We'll use LangGraph, LangChain, and the OpenAI integration, plus pandas and matplotlib for data work.

pip install langgraph langchain langchain-openai pandas matplotlib

Set your OpenAI API key as an environment variable. The agent will use gpt-4o (or another model of your choice) for its reasoning. 

import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

Defining the Agent State

LangGraph revolves around a shared state object that flows through the nodes. For our data analysis agent, the state will hold a list of messages (the conversation history). We'll use a TypedDict with an Annotated list to enable additive merging of messages. 

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]

This state ensures every node appends new messages (user inputs, AI responses, tool results) without overwriting previous ones. The graph uses this history to maintain context.

Creating the Tools

We need a Python execution tool that provides a persistent REPL environment. LangChain offers PythonAstREPLTool which runs code in a sandboxed ast-based environment and remembers variables across invocations. We'll also add a simple utility to read local files (like CSV) into the Python workspace. For plotting, we'll instruct the agent to save charts as images and return the file path. 

from langchain_experimental.tools.python.tool import PythonAstREPLTool
from langchain.tools import Tool
import pandas as pd
import sys
from io import StringIO

# Global Python execution environment
python_repl = PythonAstREPLTool(
    globals={"pd": pd, "sys": sys, "plt": None},  # matplotlib imported on demand
    name="python_repl",
    description="Execute Python code to analyze data, perform calculations, or create visualizations. Use pd for pandas, plt for matplotlib. Always save plots with plt.savefig()."
)

# Optional: tool to load a file into the REPL's namespace
def load_csv_to_repl(file_path: str) -> str:
    """Load a CSV file into the Python REPL as DataFrame `df`."""
    try:
        df = pd.read_csv(file_path)
        python_repl.globals["df"] = df
        return f"Loaded {file_path} into variable `df`. Shape: {df.shape}"
    except Exception as e:
        return f"Error loading file: {str(e)}"

load_csv_tool = Tool(
    name="load_csv",
    func=load_csv_to_repl,
    description="Load a CSV file from the given file path into the Python environment as DataFrame `df`."
)

tools = [python_repl, load_csv_tool]

Building the Agent Nodes

Our graph will have two primary nodes: an agent node that calls the LLM with tool bindings, and a tool node that executes any requested tool calls and returns the results as messages.

1. The Agent Node (Call Model with Tools)

We'll bind the tools to a ChatOpenAI model. The system prompt instructs the model to act as a data analyst: first understand the task, write Python code using the provided tools, observe results, and finally produce a natural language answer. 

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage

# System prompt that guides the agent's behavior
SYSTEM_PROMPT = """You are an expert data analyst. Use Python to explore and analyze data.
- If a CSV file path is provided, load it with `load_csv` first.
- Then use `python_repl` to perform calculations, filtering, aggregations, and create visualizations.
- Always save plots using `plt.savefig('plot.png')` and mention the saved file.
- When you have the answer, respond directly to the user with a clear summary.
- Do not hallucinate data; rely on the actual Python outputs."""

model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)

def agent_node(state: AgentState):
    # Prepend system message only once at the start of the conversation
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}

The returned AIMessage may contain tool_calls if the model decides to use a tool, or just a text response if it's ready to answer.

2. The Tool Execution Node

This node iterates over any parallel tool calls, executes them, and returns a ToolMessage for each, capturing the output (or an error). 

from langgraph.prebuilt import ToolNode

# LangGraph provides a prebuilt ToolNode that handles execution
tool_node = ToolNode(tools)

ToolNode automatically extracts tool calls from the last AI message, runs them, and returns the corresponding ToolMessage objects. If a tool fails, it still returns a message with the error, allowing the agent to self-correct.

Assembling the Graph

Now we create the state graph, add our nodes, and wire them with conditional edges that implement the ReAct loop: after the agent node, if there are tool calls, go to the tool node; otherwise, end. 

from langgraph.graph import StateGraph, END
from langchain_core.messages import AIMessage

# Initialize graph with our state schema
graph = StateGraph(AgentState)

# Add nodes
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

# Set entry point: the first node to run
graph.set_entry_point("agent")

# Conditional edge: after agent, decide whether to execute tools or stop
def route_after_agent(state: AgentState):
    last_message = state["messages"][-1]
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return "tools"
    return END

graph.add_conditional_edges("agent", route_after_agent, {"tools": "tools", END: END})

# After tools, always go back to the agent to process results
graph.add_edge("tools", "agent")

# Compile the graph
app = graph.compile()

The graph now looks like: START → agent → (tools ↔ agent) → END. The agent repeatedly invokes tools and reflects on results until it decides to respond with a final answer.

Running the Agent

Invoke the compiled app with an initial user message. The agent will load the CSV, perform analysis, generate a plot, and return a summary. 

user_query = "Load sales_data.csv, calculate total revenue per product category, and create a bar chart."
initial_state = {"messages": [HumanMessage(content=user_query)]}

# Stream the steps to see the agent's thinking
for event in app.stream(initial_state):
    for key, value in event.items():
        if key == "agent":
            last_msg = value["messages"][-1]
            if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
                print(f"🔧 Calling tools: {[tc['name'] for tc in last_msg.tool_calls]}")
            else:
                print(f"🤖 Agent: {last_msg.content[:200]}...")
        elif key == "tools":
            print(f"📊 Tool results received.")

# Final state after execution
final_state = app.invoke(initial_state)
final_answer = final_state["messages"][-1]
print(final_answer.content)

Example output (simplified): the agent first calls load_csv, then uses python_repl to group by category and sum revenue, then calls python_repl again to create a bar chart with matplotlib and save it as plot.png, and finally responds: "Total revenue per category: Electronics $50k, Furniture $30k. Bar chart saved as plot.png."

Adding Human-in-the-Loop (Optional)

For data modification tasks (e.g., dropping rows, overwriting files), you may want to pause execution and require human approval. LangGraph supports interrupting the graph before a specific node. 

# Compile with interrupt before the 'tools' node
app_with_approval = graph.compile(interrupt_before=["tools"])

# Run until the interrupt
state = app_with_approval.invoke(initial_state)
# Graph pauses, showing the pending tool calls
pending_calls = state["messages"][-1].tool_calls
print("Pending actions:", pending_calls)

# Human approves (simulate with input)
approval = input("Approve execution? (yes/no): ")
if approval.lower() == "yes":
    # Resume execution, providing no extra input
    state = app_with_approval.invoke(None, config={"configurable": {"thread_id": "1"}})
else:
    # You could modify state to reject the actions
    print("Execution rejected.")

This pattern is invaluable for sensitive operations. You can extend it to ask for confirmation only when certain dangerous functions (like os.remove) are called.

Best Practices for Data Analysis Agents

Conclusion

You've built a fully functional data analysis agent using LangGraph that can load data, write and execute Python, generate visualizations, and interactively refine its answers. LangGraph's stateful graph architecture gives you fine-grained control over the agent's reasoning loop, making it far more reliable than a single-shot prompt. By combining LLM reasoning with a safe Python execution environment, you unlock powerful use cases: automated report generation, exploratory data analysis, on-the-fly dashboard creation, and more. The patterns you learned—defining state, wiring conditional edges, and integrating human oversight—form the foundation for building even more sophisticated agents that can handle multi-step analytical workflows with confidence.

— Ad —

Google AdSense will appear here after approval

← Back to all articles