Handing the Pipeline to an Agent
In data analysis workflows, you often find yourself repeating the same cycle: load a CSV, run some statistics, generate charts, and write up a report. What if you could hand that entire pipeline over to an AI agent that handles each step autonomously?
We'll pair Gemini 3 Pro's reasoning with LangGraph's stateful, graph-based workflow engine to build an agent that runs end-to-end data analysis without manual intervention.
What you'll learn:
- Designing agent workflows using LangGraph's StateGraph
- Integrating Gemini 3 Pro API with LangGraph
- Building a pipeline: CSV parsing → statistical analysis → visualization → report generation
- Implementing error handling and retry strategies
- Best practices for deploying to production
Prerequisites: Familiarity with Python and the Gemini API basics
What Is LangGraph?
LangGraph is a framework built by the LangChain team for creating stateful, multi-actor applications. Unlike traditional LangChain Chains (linear pipelines), LangGraph lets you define workflows as directed graphs, making it ideal for complex agent architectures.
Here's why LangGraph works well for AI agent development:
- State management: Each step can access results from previous steps through a shared state object
- Conditional routing: The agent can dynamically choose its next action based on intermediate results
- Loop support: If an analysis step produces unsatisfactory results, the agent can automatically retry
- Human-in-the-loop: You can insert approval checkpoints for critical decisions
Setting Up Your Environment
Installing Dependencies
pip install google-genai langgraph langchain-google-genai pandas matplotlib seabornConfiguring Your API Key
Grab an API key from Google AI Studio and set it as an environment variable.
import os
os.environ["GOOGLE_API_KEY"] = "your-api-key-here"Quick Verification
Let's confirm the Gemini 3 Pro API is working correctly.
from google import genai
client = genai.Client()
response = client.models.generate_content(
model="gemini-3-pro",
contents="What are 3 best practices for data analysis in Python?"
)
print(response.text)Expected output:
1. Prioritize data cleaning — Handle missing values and outliers before diving into analysis...
2. Never skip exploratory data analysis (EDA) — Use visualizations to understand the big picture...
3. Ensure reproducibility — Fix random seeds, version-control your notebooks...
Agent Architecture Design
Our data analysis agent is built as a state graph with five nodes:
- Data Loading Node — Reads the CSV and captures basic metadata
- Analysis Planning Node — Gemini 3 Pro examines the data and creates an execution plan
- Statistical Analysis Node — Executes statistical computations based on the plan
- Visualization Node — Generates charts from the analysis results
- Report Generation Node — Gemini 3 Pro writes a human-readable report
Defining the State
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
import pandas as pd
class AnalysisState(TypedDict):
"""Shared state across the entire agent workflow"""
file_path: str # Path to the input CSV
dataframe: pd.DataFrame | None # Loaded data
data_summary: str # Text summary of the dataset
analysis_plan: list[str] # List of planned analysis steps
statistics: dict # Statistical analysis results
chart_paths: list[str] # File paths of generated charts
report: str # Final report text
error: str | None # Error information
retry_count: int # Number of retry attemptsImplementing Each Node
Node 1: Data Loading
import pandas as pd
import json
def load_data_node(state: AnalysisState) -> AnalysisState:
"""Load a CSV file and generate a basic data summary"""
try:
df = pd.read_csv(state["file_path"])
# Build a text summary of the data
summary_parts = [
f"Rows: {len(df)}, Columns: {len(df.columns)}",
f"Columns: {', '.join(df.columns.tolist())}",
f"Data types:\n{df.dtypes.to_string()}",
f"Missing values:\n{df.isnull().sum().to_string()}",
f"First 5 rows:\n{df.head().to_string()}",
f"Descriptive statistics:\n{df.describe().to_string()}"
]
data_summary = "\n\n".join(summary_parts)
return {
**state,
"dataframe": df,
"data_summary": data_summary,
"error": None
}
except Exception as e:
return {**state, "error": f"Data loading error: {str(e)}"}Node 2: Analysis Planning with Gemini 3 Pro
This is where the agent's intelligence shines. We pass the data summary to Gemini 3 Pro and ask it to design an optimal analysis strategy.
from google import genai
client = genai.Client()
def plan_analysis_node(state: AnalysisState) -> AnalysisState:
"""Have Gemini 3 Pro analyze the data and create an execution plan"""
if state.get("error"):
return state
prompt = f"""You are a data analysis expert.
Review the following dataset summary and create an analysis plan in JSON format.
## Dataset Summary
{state['data_summary']}
## Output Format
Respond with the following JSON structure:
{{
"analysis_steps": [
{{
"step": "Description of the step",
"method": "Statistical method to use",
"columns": ["target columns"]
}}
],
"recommended_charts": [
{{
"type": "Chart type (histogram, scatter, heatmap, bar, line)",
"columns": ["target columns"],
"title": "Chart title"
}}
]
}}
"""
response = client.models.generate_content(
model="gemini-3-pro",
contents=prompt,
config={
"response_mime_type": "application/json"
}
)
try:
plan = json.loads(response.text)
analysis_steps = [step["step"] for step in plan["analysis_steps"]]
return {
**state,
"analysis_plan": analysis_steps,
"_full_plan": plan # Keep the full plan for internal use
}
except json.JSONDecodeError:
return {**state, "error": "Failed to parse the analysis plan"}Node 3: Running Statistical Analysis
import numpy as np
def run_statistics_node(state: AnalysisState) -> AnalysisState:
"""Execute statistical computations based on the analysis plan"""
if state.get("error"):
return state
df = state["dataframe"]
stats = {}
# Auto-detect column types
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
categorical_cols = df.select_dtypes(include=["object", "category"]).columns.tolist()
# Descriptive statistics
stats["basic"] = {
col: {
"mean": float(df[col].mean()),
"median": float(df[col].median()),
"std": float(df[col].std()),
"min": float(df[col].min()),
"max": float(df[col].max()),
"skew": float(df[col].skew()),
"kurtosis": float(df[col].kurtosis())
}
for col in numeric_cols
}
# Correlation analysis (when 2+ numeric columns exist)
if len(numeric_cols) >= 2:
corr_matrix = df[numeric_cols].corr()
stats["correlation"] = corr_matrix.to_dict()
# Detect strong correlations (|r| > 0.7)
strong_correlations = []
for i in range(len(numeric_cols)):
for j in range(i + 1, len(numeric_cols)):
r = corr_matrix.iloc[i, j]
if abs(r) > 0.7:
strong_correlations.append({
"col1": numeric_cols[i],
"col2": numeric_cols[j],
"correlation": round(r, 4)
})
stats["strong_correlations"] = strong_correlations
# Categorical data aggregation
if categorical_cols:
stats["categorical"] = {
col: df[col].value_counts().head(10).to_dict()
for col in categorical_cols
}
return {**state, "statistics": stats}Node 4: Generating Visualizations
import matplotlib.pyplot as plt
import matplotlib
import seaborn as sns
import os
matplotlib.rcParams['font.family'] = 'DejaVu Sans'
def create_visualizations_node(state: AnalysisState) -> AnalysisState:
"""Generate charts based on the statistical results"""
if state.get("error"):
return state
df = state["dataframe"]
chart_paths = []
output_dir = "analysis_output"
os.makedirs(output_dir, exist_ok=True)
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
# 1. Histograms for numeric column distributions
if numeric_cols:
fig, axes = plt.subplots(
1, min(len(numeric_cols), 4),
figsize=(5 * min(len(numeric_cols), 4), 4)
)
if len(numeric_cols) == 1:
axes = [axes]
for idx, col in enumerate(numeric_cols[:4]):
axes[idx].hist(df[col].dropna(), bins=30, edgecolor='black', alpha=0.7)
axes[idx].set_title(f'Distribution: {col}')
axes[idx].set_xlabel(col)
axes[idx].set_ylabel('Frequency')
plt.tight_layout()
path = f"{output_dir}/histograms.png"
plt.savefig(path, dpi=150, bbox_inches='tight')
plt.close()
chart_paths.append(path)
# 2. Correlation heatmap
if len(numeric_cols) >= 2:
fig, ax = plt.subplots(figsize=(10, 8))
corr = df[numeric_cols].corr()
sns.heatmap(corr, annot=True, fmt=".2f", cmap="coolwarm",
center=0, ax=ax, square=True)
ax.set_title("Correlation Heatmap")
plt.tight_layout()
path = f"{output_dir}/correlation_heatmap.png"
plt.savefig(path, dpi=150, bbox_inches='tight')
plt.close()
chart_paths.append(path)
# 3. Scatter plots for strongly correlated pairs
strong_corrs = state["statistics"].get("strong_correlations", [])
for i, corr_info in enumerate(strong_corrs[:3]):
fig, ax = plt.subplots(figsize=(8, 6))
ax.scatter(
df[corr_info["col1"]],
df[corr_info["col2"]],
alpha=0.5, edgecolors='black', linewidth=0.5
)
ax.set_xlabel(corr_info["col1"])
ax.set_ylabel(corr_info["col2"])
ax.set_title(
f'{corr_info["col1"]} vs {corr_info["col2"]} '
f'(r={corr_info["correlation"]})'
)
plt.tight_layout()
path = f"{output_dir}/scatter_{i+1}.png"
plt.savefig(path, dpi=150, bbox_inches='tight')
plt.close()
chart_paths.append(path)
return {**state, "chart_paths": chart_paths}Node 5: Report Generation with Gemini 3 Pro
def generate_report_node(state: AnalysisState) -> AnalysisState:
"""Have Gemini 3 Pro write a human-readable report from the statistics"""
if state.get("error"):
return state
stats_summary = json.dumps(state["statistics"], ensure_ascii=False, indent=2)
prompt = f"""You are a data analyst. Based on the following statistical results,
write a report that business stakeholders can easily understand.
## Dataset Overview
{state['data_summary'][:2000]}
## Statistical Results
{stats_summary[:3000]}
## Generated Charts
{', '.join(state['chart_paths'])}
## Report Requirements
1. Executive summary (3 lines max)
2. Key findings (bullet points)
3. Data trends and patterns
4. Interpretation of correlations (include business implications)
5. Recommended actions
6. Caveats and data limitations
Avoid jargon and reference specific numbers from the analysis.
"""
response = client.models.generate_content(
model="gemini-3-pro",
contents=prompt
)
return {**state, "report": response.text}Assembling the Workflow with LangGraph
Now we wire everything together as a LangGraph state graph.
from langgraph.graph import StateGraph, END
def should_retry(state: AnalysisState) -> str:
"""Determine whether to retry on error"""
if state.get("error") and state.get("retry_count", 0) < 3:
return "retry"
elif state.get("error"):
return "fail"
return "continue"
# Build the graph
workflow = StateGraph(AnalysisState)
# Add nodes
workflow.add_node("load_data", load_data_node)
workflow.add_node("plan_analysis", plan_analysis_node)
workflow.add_node("run_statistics", run_statistics_node)
workflow.add_node("create_visualizations", create_visualizations_node)
workflow.add_node("generate_report", generate_report_node)
# Define edges
workflow.set_entry_point("load_data")
workflow.add_conditional_edges(
"load_data",
should_retry,
{
"continue": "plan_analysis",
"retry": "load_data",
"fail": END
}
)
workflow.add_conditional_edges(
"plan_analysis",
should_retry,
{
"continue": "run_statistics",
"retry": "plan_analysis",
"fail": END
}
)
workflow.add_edge("run_statistics", "create_visualizations")
workflow.add_edge("create_visualizations", "generate_report")
workflow.add_edge("generate_report", END)
# Compile the graph
app = workflow.compile()Running the Agent
# Set up initial state and run
initial_state = {
"file_path": "sales_data.csv",
"dataframe": None,
"data_summary": "",
"analysis_plan": [],
"statistics": {},
"chart_paths": [],
"report": "",
"error": None,
"retry_count": 0
}
# Execute the agent
result = app.invoke(initial_state)
# Print the report
if result.get("error"):
print(f"An error occurred: {result['error']}")
else:
print("=" * 60)
print("Data Analysis Report")
print("=" * 60)
print(result["report"])
print(f"\nGenerated charts: {result['chart_paths']}")Expected output:
============================================================
Data Analysis Report
============================================================
## Executive Summary
Analysis of the sales dataset reveals 23% year-over-year growth in Q3.
The online channel accounts for 62% of total revenue and is the primary
growth driver.
## Key Findings
- Average monthly revenue: $124,500 (std dev: $23,400)
- Ad spend and revenue show a strong positive correlation (r = 0.87)
...
Generated charts: ['analysis_output/histograms.png', 'analysis_output/correlation_heatmap.png']
Taking It to Production
Streaming Execution for Progress Tracking
async def run_with_streaming(file_path: str):
"""Display real-time progress as each step completes"""
initial_state = {
"file_path": file_path,
"dataframe": None,
"data_summary": "",
"analysis_plan": [],
"statistics": {},
"chart_paths": [],
"report": "",
"error": None,
"retry_count": 0
}
node_labels = {
"load_data": "Loading data...",
"plan_analysis": "Planning analysis...",
"run_statistics": "Running statistics...",
"create_visualizations": "Generating charts...",
"generate_report": "Writing report..."
}
async for event in app.astream(initial_state):
for node_name, node_state in event.items():
label = node_labels.get(node_name, node_name)
if node_state.get("error"):
print(f" ✗ {label} Error: {node_state['error']}")
else:
print(f" ✓ {label} Done")
# Run it
import asyncio
asyncio.run(run_with_streaming("sales_data.csv"))Adding Logging and Monitoring
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("data_analysis_agent")
def with_logging(node_func):
"""Decorator that adds logging to any node function"""
def wrapper(state: AnalysisState) -> AnalysisState:
node_name = node_func.__name__
start_time = datetime.now()
logger.info(f"[{node_name}] Started")
result = node_func(state)
elapsed = (datetime.now() - start_time).total_seconds()
if result.get("error"):
logger.error(f"[{node_name}] Error ({elapsed:.1f}s): {result['error']}")
else:
logger.info(f"[{node_name}] Completed ({elapsed:.1f}s)")
return result
return wrapper
# Apply logging to each node
load_data_node = with_logging(load_data_node)
plan_analysis_node = with_logging(plan_analysis_node)
run_statistics_node = with_logging(run_statistics_node)
create_visualizations_node = with_logging(create_visualizations_node)
generate_report_node = with_logging(generate_report_node)Wrapping Up
The agent is complete: Gemini 3 Pro's reasoning driving LangGraph's stateful workflow engine.
The key takeaways: LangGraph's StateGraph makes it intuitive to design complex analysis workflows as nodes and edges. Gemini 3 Pro's structured output (JSON mode) enables high-quality automated planning and reporting. Built-in conditional routing and retry logic make the agent robust enough for production use.
For your next steps, consider integrating [Gemini API Function Calling]((/articles/gemini-api/function-calling-guide) to give the agent access to external data sources, leveraging [Gemini Structured Output]((/articles/gemini-api/gemini-structured-output-advanced) for stricter type safety, or exploring [multi-agent architectures with Google ADK]((/articles/gemini-advanced/google-adk-typescript-multi-agent-guide) for even more complex workflows.