Introduction to Data Analysis Agents
Data analysis agents represent a paradigm shift in how we interact with data. Instead of manually writing SQL queries, Python scripts, or spreadsheet formulas, an AI-powered agent can autonomously reason about your data, formulate analytical questions, execute computations, and iteratively refine its approach based on results. LangGraph, built on top of LangChain, provides the ideal framework for constructing these agents by enabling stateful, multi-step reasoning workflows with precise control flow.
What is LangGraph?
LangGraph is a library for building stateful, multi-actor applications with LLMs. It extends LangChain's capabilities by providing a graph-based orchestration framework where each node represents a computation step and edges define the flow of data and control between steps. Unlike a linear chain, a graph can have cycles, conditional branches, and persistent state — all essential for building an agent that needs to iteratively explore data, reflect on intermediate results, and backtrack when necessary.
The core abstractions in LangGraph include:
- State: A typed dictionary that persists across all nodes in the graph, carrying context like the user's question, intermediate results, and memory
- Nodes: Python functions that read the current state, perform work (like calling an LLM or executing code), and return updates to the state
- Edges: Connections between nodes that can be static (always follow the same path) or conditional (route based on the current state)
- Graph: The overall structure that compiles nodes and edges into a runnable application with built-in streaming, checkpointing, and debugging support
Why LangGraph for Data Analysis?
Data analysis is inherently iterative and non-linear. You might start with a broad question, inspect the data's shape, realize you need to clean outliers, reformulate your hypothesis, run a statistical test, visualize the distribution, and then decide whether to drill down further or present conclusions. This cyclical process maps naturally to a graph-based agent architecture. LangGraph specifically excels here because:
- Persistent State: The agent carries forward the dataframe metadata, previous queries, computed results, and error history across multiple reasoning cycles
- Conditional Routing: You can route to different nodes based on whether code execution succeeded or failed, whether the analysis is complete, or whether the user needs to clarify their question
- Human-in-the-Loop: LangGraph supports interrupt points where the agent pauses for human approval before executing potentially destructive operations
- Streaming and Observability: Every step can be streamed to the UI, giving users real-time visibility into the agent's reasoning process
Architecture Overview
A well-designed data analysis agent typically follows a ReAct-style (Reasoning + Acting) loop augmented with tool access for code execution. The graph structure we'll build contains the following nodes:
- Reasoner Node: The LLM analyzes the current state and decides the next analytical step — whether to write code, ask a clarifying question, or present final results
- Executor Node: Safely executes the generated code in a sandboxed environment, captures output and errors, and appends results to state
- Reflector Node: Reviews execution results, determines if the analysis is complete or if more work is needed, and updates the plan accordingly
- Finalizer Node: Synthesizes all findings into a coherent narrative response for the user
The edges form a loop: Reasoner → Executor → Reflector → back to Reasoner if more work is needed, or Reasoner → Finalizer when done. Conditional edges handle error recovery, timeouts, and user interruption gracefully.
Step-by-Step Implementation
1. Setting Up the Environment
First, install the required dependencies. We'll use LangGraph for orchestration, LangChain for LLM interactions, and a sandboxed execution environment for safety.
pip install langgraph langchain langchain-openai pandas numpy matplotlib seaborn
pip install duckdb-engine # For SQL querying capabilities
pip install pydantic # For typed state definitions2. Defining the Agent State
The state is the backbone of the agent. It must be comprehensive enough to carry all context between cycles while remaining structured and type-safe. Here's a production-grade state definition using Pydantic:
from typing import TypedDict, Annotated, List, Dict, Any, Optional
from langchain_core.messages import BaseMessage
import operator
class DataAnalysisState(TypedDict):
"""State schema for the data analysis agent."""
# Conversation history
messages: Annotated[List[BaseMessage], operator.add]
# The user's original question
user_question: str
# DataFrame metadata (never send raw data to LLM)
dataframe_columns: List[str]
dataframe_dtypes: Dict[str, str]
dataframe_shape: tuple
dataframe_sample: str # First 5 rows as string
# Analysis tracking
analysis_plan: List[str] # Steps the agent plans to take
completed_steps: List[str] # Steps already completed
# Code execution results
last_code: Optional[str]
last_output: Optional[str]
last_error: Optional[str]
# Accumulated results
computed_results: Dict[str, Any] # Named results from previous steps
visualizations_generated: List[str] # Paths to saved charts
# Control flags
analysis_complete: bool
needs_clarification: bool
clarification_question: Optional[str]
# Iteration tracking
iteration_count: int
max_iterations: int3. Loading and Preparing Data
Before the agent runs, we need to load the user's data and extract metadata. This metadata — not the raw data itself — is what gets passed to the LLM for context, preserving privacy and keeping token usage manageable.
import pandas as pd
import json
def prepare_data_context(file_path: str) -> dict:
"""
Load a CSV file and extract metadata for the agent's context.
Returns a dictionary with column info, dtypes, shape, and sample.
"""
df = pd.read_csv(file_path)
# Handle large datasets by sampling
if len(df) > 100_000:
df_sample = df.sample(n=10_000, random_state=42)
else:
df_sample = df
context = {
"dataframe_columns": list(df.columns),
"dataframe_dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
"dataframe_shape": df.shape,
"dataframe_sample": df.head(5).to_string(index=False),
"null_counts": df.isnull().sum().to_dict(),
"numeric_summary": df.describe().to_string() if len(df.select_dtypes(include='number').columns) > 0 else None,
}
# Store the full dataframe in a global or session-level variable
# that the executor can access (NOT passed to the LLM)
return context, df
# Example usage
context_dict, raw_df = prepare_data_context("sales_data.csv")4. Building the Reasoner Node
The reasoner is the brain of the agent. It receives the current state — including conversation history, data metadata, previous results, and the analysis plan — and decides the next action. We use LangChain's structured output parsing to force the LLM to return a well-defined decision object.
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from pydantic import BaseModel, Field
from typing import Literal
class ReasonerDecision(BaseModel):
"""Structured decision from the reasoner node."""
action: Literal["write_code", "ask_clarification", "finalize"] = Field(
description="The action to take next"
)
reasoning: str = Field(
description="Step-by-step reasoning for this decision"
)
code_to_execute: Optional[str] = Field(
default=None,
description="Python code to execute if action is 'write_code'. "
"Use pandas, matplotlib, etc. Access the dataframe via 'df' variable."
)
expected_output_type: Optional[str] = Field(
default=None,
description="What kind of result this code should produce (numeric, table, chart, etc.)"
)
clarification_question: Optional[str] = Field(
default=None,
description="Question to ask the user if action is 'ask_clarification'"
)
final_summary: Optional[str] = Field(
default=None,
description="Summary of findings if action is 'finalize'"
)
SYSTEM_PROMPT = """You are an expert data analyst agent. You have access to a pandas DataFrame
stored in the variable `df`. You cannot see the full data, but you have metadata.
Current DataFrame metadata:
- Columns: {dataframe_columns}
- Dtypes: {dataframe_dtypes}
- Shape: {dataframe_shape}
- Sample (first 5 rows):
{dataframe_sample}
Analysis plan so far:
{analysis_plan}
Completed steps:
{completed_steps}
Previous code output (if any):
{last_output}
Previous error (if any):
{last_error}
Computed results from previous steps:
{computed_results}
Available libraries: pandas, numpy, matplotlib, seaborn, scipy.stats, duckdb
Guidelines:
1. Write small, focused code snippets — one analytical step at a time
2. Always assign meaningful variable names to results (e.g., 'avg_sales_by_region')
3. Use print() to display results and df.head() to show tables
4. For visualizations, use plt.savefig('chart_n.png') and print the filename
5. If you encounter an error, try a different approach or ask for clarification
6. When the analysis is complete, summarize findings clearly
7. NEVER write destructive code (no df.to_csv overwrite, no os.system, no subprocess)
8. Keep code under 50 lines per step
"""
def build_reasoner_node(llm: ChatOpenAI):
"""Build the reasoner node function."""
prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_PROMPT),
MessagesPlaceholder(variable="messages"),
("human", "Based on the current state, what should be the next analytical step? "
"Respond with a structured decision.")
])
# Use structured output with the LLM
structured_llm = llm.with_structured_output(ReasonerDecision)
chain = prompt | structured_llm
def reasoner_node(state: DataAnalysisState):
# Prepare the prompt variables from state
prompt_vars = {
"dataframe_columns": state["dataframe_columns"],
"dataframe_dtypes": state["dataframe_dtypes"],
"dataframe_shape": state["dataframe_shape"],
"dataframe_sample": state["dataframe_sample"],
"analysis_plan": state.get("analysis_plan", []),
"completed_steps": state.get("completed_steps", []),
"last_output": state.get("last_output", "None"),
"last_error": state.get("last_error", "None"),
"computed_results": state.get("computed_results", {}),
}
decision: ReasonerDecision = chain.invoke({
"messages": state["messages"],
**prompt_vars
})
# Update state based on decision
updates = {
"iteration_count": state["iteration_count"] + 1,
}
if decision.action == "write_code":
updates["last_code"] = decision.code_to_execute
updates["analysis_plan"] = state.get("analysis_plan", []) + [decision.reasoning]
# Route to executor
updates["next_node"] = "executor"
elif decision.action == "ask_clarification":
updates["needs_clarification"] = True
updates["clarification_question"] = decision.clarification_question
updates["next_node"] = "finalizer" # Present question to user
elif decision.action == "finalize":
updates["analysis_complete"] = True
updates["next_node"] = "finalizer"
updates["final_summary"] = decision.final_summary
return updates
return reasoner_node5. Building the Executor Node
The executor node runs the generated code in a restricted environment. Security is paramount — we use a combination of namespace restriction, AST parsing for dangerous patterns, and timeout controls. The executor captures stdout, stderr, and any exceptions, then appends structured results back to the state.
import sys
import io
import traceback
import ast
import signal
from contextlib import contextmanager
import matplotlib.pyplot as plt
# Global reference to the dataframe (set per session)
CURRENT_DATAFRAME = None
def set_dataframe(df):
global CURRENT_DATAFRAME
CURRENT_DATAFRAME = df
class CodeSafetyChecker:
"""Checks Python code for potentially dangerous operations."""
FORBIDDEN_MODULES = {"os", "subprocess", "sys", "shutil", "socket", "requests", "urllib"}
FORBIDDEN_FUNCTIONS = {"exec", "eval", "compile", "__import__", "open"}
FORBIDDEN_METHODS = {"to_csv", "to_excel", "to_pickle"} # Prevent overwrites
@staticmethod
def is_safe(code: str) -> tuple[bool, str]:
"""Check if code is safe to execute. Returns (is_safe, reason)."""
try:
tree = ast.parse(code)
except SyntaxError as e:
return False, f"Syntax error: {str(e)}"
for node in ast.walk(tree):
# Check for forbidden imports
if isinstance(node, ast.Import):
for alias in node.names:
if alias.name.split('.')[0] in CodeSafetyChecker.FORBIDDEN_MODULES:
return False, f"Forbidden import: {alias.name}"
if isinstance(node, ast.ImportFrom):
if node.module and node.module.split('.')[0] in CodeSafetyChecker.FORBIDDEN_MODULES:
return False, f"Forbidden import from: {node.module}"
# Check for forbidden function calls
if isinstance(node, ast.Call):
if isinstance(node.func, ast.Name) and node.func.id in CodeSafetyChecker.FORBIDDEN_FUNCTIONS:
return False, f"Forbidden function: {node.func.id}"
# Check method calls like df.to_csv()
if isinstance(node.func, ast.Attribute):
if node.func.attr in CodeSafetyChecker.FORBIDDEN_METHODS:
return False, f"Forbidden method: {node.func.attr}"
return True, "Code is safe"
@contextmanager
def timeout_context(seconds: int):
"""Context manager for execution timeout."""
def timeout_handler(signum, frame):
raise TimeoutError(f"Code execution exceeded {seconds} seconds")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
def execute_code_safely(code: str, state: DataAnalysisState) -> dict:
"""
Execute the provided code in a restricted namespace.
Returns a dictionary with output, error, and any captured variables.
"""
# Safety check
is_safe, reason = CodeSafetyChecker.is_safe(code)
if not is_safe:
return {
"output": None,
"error": f"CODE REJECTED: {reason}",
"captured_results": {}
}
# Prepare execution namespace with safe builtins
safe_builtins = {
"print": print,
"len": len,
"range": range,
"enumerate": enumerate,
"zip": zip,
"list": list,
"dict": dict,
"set": set,
"tuple": tuple,
"int": int,
"float": float,
"str": str,
"bool": bool,
"abs": abs,
"round": round,
"min": min,
"max": max,
"sum": sum,
"sorted": sorted,
"isinstance": isinstance,
"type": type,
}
namespace = {
"__builtins__": safe_builtins,
"df": CURRENT_DATAFRAME,
"pd": pd,
"np": __import__('numpy'),
"plt": plt,
"sns": __import__('seaborn'),
"scipy_stats": __import__('scipy.stats'),
}
# Capture stdout
old_stdout = sys.stdout
sys.stdout = io.StringIO()
try:
with timeout_context(30): # 30-second timeout
exec(code, namespace)
output = sys.stdout.getvalue()
error = None
# Capture any variables that look like results
captured = {}
for var_name, value in namespace.items():
# Skip modules, functions, and internal variables
if var_name.startswith('_') or callable(value) or isinstance(value, type):
continue
if var_name in {"df", "pd", "np", "plt", "sns", "scipy_stats"}:
continue
# Capture DataFrames, Series, numbers, strings, lists, dicts
if isinstance(value, (pd.DataFrame, pd.Series)):
captured[var_name] = {
"type": "dataframe",
"shape": value.shape,
"preview": value.head(100).to_string()
}
elif isinstance(value, (int, float, str, list, dict, tuple)):
captured[var_name] = value
return {
"output": output,
"error": error,
"captured_results": captured
}
except TimeoutError as e:
return {
"output": sys.stdout.getvalue(),
"error": f"TIMEOUT: {str(e)}",
"captured_results": {}
}
except Exception as e:
return {
"output": sys.stdout.getvalue(),
"error": f"{type(e).__name__}: {str(e)}\n{traceback.format_exc()}",
"captured_results": {}
}
finally:
sys.stdout = old_stdout
def build_executor_node():
"""Build the executor node function."""
def executor_node(state: DataAnalysisState):
code = state.get("last_code")
if not code:
return {"last_error": "No code to execute", "next_node": "reasoner"}
result = execute_code_safely(code, state)
updates = {
"last_output": result["output"],
"last_error": result["error"],
}
# Merge captured results into computed_results
if result["captured_results"]:
current_results = state.get("computed_results", {})
current_results.update(result["captured_results"])
updates["computed_results"] = current_results
if result["error"]:
updates["next_node"] = "reflector" # Reflect on error
else:
updates["next_node"] = "reflector" # Reflect on success
return updates
return executor_node6. Building the Reflector Node
The reflector node reviews the output of code execution and updates the agent's understanding. It determines whether the analysis is on track, if errors need addressing, or if the overall goal has been achieved. This node prevents the agent from looping endlessly on the same mistake.
def build_reflector_node(llm: ChatOpenAI):
"""Build the reflector node that reviews execution results."""
REFLECTOR_PROMPT = """You are a reflective analyst reviewing the results of a code execution.
User's original question: {user_question}
Analysis plan: {analysis_plan}
Completed steps: {completed_steps}
Last code executed:
{last_code}
Output received:
{last_output}
Error (if any):
{last_error}
Current computed results: {computed_results}
Your task: Determine if the analysis is progressing correctly.
Consider these questions:
1. Did the code execute successfully? If not, what went wrong?
2. Does the output meaningfully advance the analysis?
3. Are we repeating a step that was already completed?
4. Is the analysis complete, or do we need more steps?
5. If there's an error, should we retry with a fix, or try a different approach?
Respond with a structured reflection."""
class ReflectionOutput(BaseModel):
status: Literal["success", "error_recoverable", "error_fatal", "complete", "stuck"] = Field(
description="Status of the current step"
)
reflection: str = Field(description="Detailed reflection on what happened")
suggestion: str = Field(description="What to do next")
should_continue: bool = Field(description="True if more analysis steps are needed")
mark_step_complete: bool = Field(description="True if the current step achieved its goal")
prompt = ChatPromptTemplate.from_messages([
("system", REFLECTOR_PROMPT),
])
structured_llm = llm.with_structured_output(ReflectionOutput)
chain = prompt | structured_llm
def reflector_node(state: DataAnalysisState):
reflection: ReflectionOutput = chain.invoke({
"user_question": state["user_question"],
"analysis_plan": state.get("analysis_plan", []),
"completed_steps": state.get("completed_steps", []),
"last_code": state.get("last_code", "None"),
"last_output": state.get("last_output", "None"),
"last_error": state.get("last_error", "None"),
"computed_results": str(state.get("computed_results", {}))[:2000], # Truncate
})
updates = {}
# If step was successful, mark it complete
if reflection.mark_step_complete and state.get("analysis_plan"):
current_plan = state["analysis_plan"]
if current_plan:
completed = state.get("completed_steps", []) + [current_plan[-1]]
updates["completed_steps"] = completed
# Handle different statuses
if reflection.status == "complete":
updates["analysis_complete"] = True
updates["next_node"] = "finalizer"
elif reflection.status == "error_fatal":
updates["analysis_complete"] = True
updates["last_error"] = f"Fatal: {reflection.reflection}"
updates["next_node"] = "finalizer"
elif reflection.status == "stuck":
updates["analysis_complete"] = True
updates["last_error"] = f"Stuck: {reflection.suggestion}"
updates["next_node"] = "finalizer"
else:
# Continue the loop — go back to reasoner
updates["next_node"] = "reasoner"
updates["last_error"] = state.get("last_error") # Preserve for next iteration
return updates
return reflector_node7. Building the Finalizer Node
The finalizer synthesizes all findings into a comprehensive natural language response. It aggregates computed results, describes visualizations, and presents a coherent analytical narrative to the user.
def build_finalizer_node(llm: ChatOpenAI):
"""Build the finalizer node that synthesizes final results."""
FINALIZER_PROMPT = """You are an expert data analyst presenting final results to a user.
User's original question: {user_question}
Analysis plan executed: {analysis_plan}
Completed steps: {completed_steps}
Computed results: {computed_results}
Visualizations generated: {visualizations_generated}
Errors encountered: {last_error}
Your task: Synthesize all findings into a clear, comprehensive response.
Guidelines:
1. Start with a direct answer to the user's question
2. Present key findings with specific numbers and insights
3. Mention any visualizations created
4. Note any limitations or caveats
5. If there were errors, explain what couldn't be completed
6. Use clear section headers and bullet points for readability
7. Be concise but thorough — aim for quality over quantity
"""
prompt = ChatPromptTemplate.from_messages([
("system", FINALIZER_PROMPT),
])
def finalizer_node(state: DataAnalysisState):
# Check if we need clarification instead
if state.get("needs_clarification") and state.get("clarification_question"):
return {
"final_response": f"I need some clarification before I can proceed:\n\n"
f"{state['clarification_question']}\n\n"
f"Please provide more details so I can continue the analysis.",
"analysis_complete": False, # Wait for user response
}
final_response = llm.invoke(FINALIZER_PROMPT.format(
user_question=state["user_question"],
analysis_plan=state.get("analysis_plan", []),
completed_steps=state.get("completed_steps", []),
computed_results=str(state.get("computed_results", {}))[:3000],
visualizations_generated=state.get("visualizations_generated", []),
last_error=state.get("last_error", "None"),
)).content
return {
"final_response": final_response,
"analysis_complete": True,
}
return finalizer_node8. Assembling the Graph
With all nodes built, we now assemble them into a LangGraph StateGraph. This is where we define the topology — nodes, edges, and conditional routing logic.
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver
from typing import Literal
def build_analysis_graph(llm: ChatOpenAI) -> StateGraph:
"""Assemble the complete data analysis agent graph."""
# Create the graph with our state type
workflow = StateGraph(DataAnalysisState)
# Build all nodes
reasoner = build_reasoner_node(llm)
executor = build_executor_node()
reflector = build_reflector_node(llm)
finalizer = build_finalizer_node(llm)
# Add nodes to the graph
workflow.add_node("reasoner", reasoner)
workflow.add_node("executor", executor)
workflow.add_node("reflector", reflector)
workflow.add_node("finalizer", finalizer)
# Set entry point — start at reasoner
workflow.set_entry_point("reasoner")
# Define routing function for conditional edges
def route_after_reasoner(state: DataAnalysisState) -> Literal["executor", "finalizer"]:
next_node = state.get("next_node", "executor")
if next_node == "finalizer":
return "finalizer"
return "executor"
def route_after_executor(state: DataAnalysisState) -> Literal["reflector"]:
# Always go to reflector after execution
return "reflector"
def route_after_reflector(state: DataAnalysisState) -> Literal["reasoner", "finalizer"]:
next_node = state.get("next_node", "reasoner")
if next_node == "finalizer":
return "finalizer"
# Check max iterations
if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
return "finalizer"
return "reasoner"
# Add edges
workflow.add_conditional_edges(
"reasoner",
route_after_reasoner,
{
"executor": "executor",
"finalizer": "finalizer",
}
)
workflow.add_conditional_edges(
"executor",
route_after_executor,
{"reflector": "reflector"}
)
workflow.add_conditional_edges(
"reflector",
route_after_reflector,
{
"reasoner": "reasoner",
"finalizer": "finalizer",
}
)
# Finalizer goes to END
workflow.add_edge("finalizer", END)
# Add checkpointing for state persistence
memory = MemorySaver()
return workflow.compile(checkpointer=memory)
# Build the agent
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.1)
agent = build_analysis_graph(llm)9. Running the Agent
Here's how to invoke the agent on actual data. We set up the initial state with the user's question and data metadata, then stream through the graph steps for real-time visibility.
import json
from langchain_core.messages import HumanMessage, AIMessage
def run_analysis(
user_question: str,
dataframe_path: str,
max_iterations: int = 10,
thread_id: str = "session_1"
) -> str:
"""
Run the data analysis agent on a CSV file.
Returns the final response string.
"""
# Load and prepare data
context_dict, raw_df = prepare_data_context(dataframe_path)
set_dataframe(raw_df) # Make available to executor
# Build initial state
initial_state: DataAnalysisState = {
"messages": [HumanMessage(content=user_question)],
"user_question": user_question,
"dataframe_columns": context_dict["dataframe_columns"],
"dataframe_dtypes": context_dict["dataframe_dtypes"],
"dataframe_shape": context_dict["dataframe_shape"],
"dataframe_sample": context_dict["dataframe_sample"],
"analysis_plan": [],
"completed_steps": [],
"last_code": None,
"last_output": None,
"last_error": None,
"computed_results": {},
"visualizations_generated": [],
"analysis_complete": False,
"needs_clarification": False,
"clarification_question": None,
"iteration_count": 0,
"max_iterations": max_iterations,
}
# Configure for streaming
config = {"configurable": {"thread_id": thread_id}, "recursion_limit": 50}
# Run the graph with streaming
print(f"Starting analysis for: {user_question}\n")
print("=" * 60)
step_count = 0
for event in agent.stream(initial_state, config=config):
step_count += 1
node_name = list(event.keys())[0]
node_output = event[node_name]
print(f"\n--- Step {step_count}: Node '{node_name}' ---")
if node_name == "reasoner":
if "analysis_plan" in node_output:
print(f" New plan step: {node_output.get('analysis_plan', [])[-1] if node_output.get('analysis_plan') else 'N/A'}")
if "last_code" in node_output:
print(f" Code to execute:\n{node_output['last_code'][:300]}...")
elif node_name == "executor":
if node_output.get("last_error"):
print(f" ERROR: {node_output['last_error'][:200]}")
else:
print(f" Output: {node_output.get('last_output', '')[:300]}")
elif node_name == "reflector":
print(f" Status: Analyzing results...")
elif node_name == "finalizer":
print(f" Final response ready!")
# Get final state
final_state = agent.get_state(config)
state_values = final_state.values
if state_values.get("final_response"):
return state_values["final_response"]
else:
return "Analysis could not be completed. Please check the data and try again."
# Example usage
result = run_analysis(
user_question="What are the top 5 products by revenue, and how does their performance vary by region?",
dataframe_path="sales_data.csv",
max_iterations=10
)
print("\n" + "=" * 60)
print("FINAL RESULT:")
print(result)Advanced Features and Enhancements
Human-in-the-Loop Approval
For sensitive operations or when the agent proposes expensive computations, you can add interrupt points where the graph pauses and waits for human approval before proceeding.
# Add an interrupt before code execution
workflow.add_node("human_approval", human_approval_node)
workflow.add_conditional_edges(
"reasoner",
route_with_approval_check,
{
"human_approval": "human_approval",
"executor": "executor",
"finalizer": "finalizer",
}
)
# When compiling, specify interrupt points
agent = workflow.compile(
checkpointer=memory,
interrupt_before=["executor"] # Pause before executing code
)
# To resume after approval:
# agent.invoke(None, config=config) # None means "proceed"Adding SQL Querying Capabilities
For larger datasets, you can extend the executor to support DuckDB SQL queries alongside Python code. The agent can choose between pandas operations and SQL depending on the task.
def execute_sql_query(query: str) -> dict:
"""Execute a DuckDB SQL query against the dataframe."""
import duckdb
conn = duckdb.connect()
conn.register('df', CURRENT_DATAFRAME)
try:
result_df = conn.execute(query).fetchdf()
return {
"output": result_df.to_string(),
"error": None,
"captured_results": {"sql_result": result_df}
}
except Exception as e:
return {
"output": None,
"error": str(e),
"captured_results": {}
}
finally:
conn.close()
# Modify the executor to detect SQL vs Python
def detect_code_type(code: str) -> Literal["python", "sql"]:
"""Detect if code is SQL or Python."""
code_stripped = code.strip()
# SQL keywords pattern
sql_patterns = ["SELECT", "WITH", "CREATE", "INSERT", "UPDATE", "DELETE"]
if any(code_stripped.upper().startswith(p) for p in sql_patterns):
return "sql"
return "python"Memory and Conversation Continuity
LangGraph's checkpointing system allows the agent to remember previous analyses across sessions. By using the same thread_id, the agent can recall prior findings and build upon them incrementally.
# Continue a previous analysis session
config = {"configurable": {"thread_id": "user_123_session"}}
agent.invoke(
{"messages": [HumanMessage(content="Now let's look at monthly trends for those top products")]},
config=config
)
# The agent will have access to all previous state from that threadBest Practices for Production Data Analysis Agents
- Sandbox Execution Rigorously: Never execute raw LLM-generated code in an unsandboxed environment