Building Data Analysis Agent with AI Agents 2026

Introduction

AI agents that can analyze datasets and generate insights autonomously are transforming how developers work. This guide shows you how to build a production-ready Data Analysis Agent using CrewAI + Pandas Tools.

What We're Building

A Data Analysis Agent that can:

Understand complex requests

Break them into sub-tasks

Execute tasks autonomously

Handle errors and retry

Produce consistent, high-quality output

Architecture

User Request
    ↓
[Data Analysis Agent Orchestrator]
    ↓
[Task Planning] → [Tool Selection] → [Execution]
    ↓                                      ↓
[Validation] ←──────────────────── [Results]
    ↓
Final Output

Implementation with LangGraph

python
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
import json
State definition
class DataAnalysisAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    task: str
    sub_tasks: List[str]
    completed_tasks: List[str]
    final_output: str | None
    iterations: int
Define tools for analyze datasets and generate insights autonomously
@tool
def analyze_task(task: str) -> str:
    """Break down a complex task into sub-tasks."""
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(f"Break this into 3-5 specific, actionable sub-tasks: {task}")
    return response.content
tools = [analyze_task, execute_sub_task, validate_output]
Initialize LLM with tools
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
llm_with_tools = llm.bind_tools(tools)
Agent node
def agent_node(state: DataAnalysisAgentState):
    if state.get("iterations", 0) > 8:
        return {"final_output": "Max iterations reached", "iterations": 9}
    
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "iterations": state.get("iterations", 0) + 1
    }
Tool execution node
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)
def should_continue(state: DataAnalysisAgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
        return "tools"
    return "end"
Build graph
workflow = StateGraph(DataAnalysisAgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
workflow.add_edge("tools", "agent")
agent = workflow.compile()

Using the Agent

python
from langchain_core.messages import HumanMessage
def run_data_analysis_agent(request: str) -> str:
    """Run the Data Analysis Agent on a user request."""
    
    initial_state = {
        "messages": [HumanMessage(content=request)],
        "task": request,
        "sub_tasks": [],
        "completed_tasks": [],
        "final_output": None,
        "iterations": 0
    }
    
    result = agent.invoke(initial_state)
    
    # Extract the final answer
    last_message = result["messages"][-1]
    return last_message.content
Usage
output = run_data_analysis_agent(
    "Create a comprehensive plan to analyze datasets and generate insights autonomously"
)
print(output)

Adding Memory with Persistence

python
from langgraph.checkpoint.sqlite import SqliteSaver
def run_with_memory(request: str, session_id: str) -> str:
    config = {"configurable": {"thread_id": session_id}}
    
    state = {
        "messages": [HumanMessage(content=request)],
        "task": request,
        "sub_tasks": [],
        "completed_tasks": [],
        "final_output": None,
        "iterations": 0
    }
    
    result = agent_with_memory.invoke(state, config=config)
    return result["messages"][-1].content
First interaction
response1 = run_with_memory("Start analyze datasets and generate insights autonomously", session_id="session-001")
Follow-up (agent remembers context)
response2 = run_with_memory("Continue from where we left off", session_id="session-001")

Production: FastAPI Service

python
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio
app = FastAPI(title="Data Analysis Agent Service")
@app.post("/agent/run")
async def run_agent(request: AgentRequest):
    if request.stream:
        async def stream_response():
            async for event in agent.astream_events(
                {"messages": [HumanMessage(content=request.task)], 
                 "task": request.task, "sub_tasks": [], 
                 "completed_tasks": [], "final_output": None, "iterations": 0},
                version="v2"
            ):
                if event["event"] == "on_chat_model_stream":
                    content = event["data"]["chunk"].content
                    if content:
                        yield content
        
        return StreamingResponse(stream_response(), media_type="text/plain")
    
    result = run_data_analysis_agent(request.task)
    return {"result": result, "session_id": request.session_id}
@app.get("/health")
async def health():
    return {"status": "healthy", "agent": "Data Analysis Agent"}

Monitoring Agent Performance

python
from dataclasses import dataclass
from datetime import datetime
import statistics
metrics_store: List[AgentMetrics] = []
def run_with_metrics(task: str) -> tuple[str, AgentMetrics]:
    import time
    start = time.time()
    
    try:
        result = run_data_analysis_agent(task)
        success = True
    except Exception as e:
        result = f"Error: {e}"
        success = False
    
    duration = (time.time() - start) * 1000
    
    # Note: iterations would come from actual state in production
    metrics = AgentMetrics(
        task=task[:50],
        iterations=3,  
        duration_ms=duration,
        success=success,
        output_length=len(result)
    )
    metrics_store.append(metrics)
    return result, metrics
def print_metrics_report():
    if not metrics_store:
        return
    
    successful = [m for m in metrics_store if m.success]
    durations = [m.duration_ms for m in metrics_store]
    
    print(f"Total runs: {len(metrics_store)}")
    print(f"Success rate: {len(successful)/len(metrics_store):.1%}")
    print(f"Avg duration: {statistics.mean(durations):.0f}ms")
    print(f"p95 duration: {sorted(durations)[int(len(durations)*0.95)]:.0f}ms")

Best Practices

Conclusion

Building Data Analysis Agent with AI agents enables autonomous analyze datasets and generate insights autonomously. The LangGraph implementation provides the right balance of control and flexibility for production use.

Start with a simple proof of concept, add persistence, then scale up as confidence grows.

---

*Data Analysis Agent implementation using CrewAI + Pandas Tools | May 2026*