A Natural-Language Data Analyst That Queries SQL, Plots Charts, and Sanitizes Its Own Output
Chapter 10: Practical Project 1 — Intelligent Data Analysis Agent
The distance from theory to practice can only be shortened by doing. Starting with this chapter, we will use three complete practical projects to integrate the knowledge learned previously. The first project is an Intelligent Data Analysis Agent — a user asks questions in natural language, and the Agent automatically queries data, performs analysis, generates charts, and produces reports.
10.1 Requirements Analysis: From Natural Language to SQL and Chart Generation
Project Goals
Build a data analysis Agent with the following core capabilities:
| Capability | User Input Example | Agent Output |
|---|---|---|
| Data Query | "What were the sales last month?" | SQL query + results table |
| Trend Analysis | "What is the sales growth trend?" | Trend chart + textual analysis |
| Anomaly Detection | "Are there any abnormal orders?" | Abnormal data list + cause speculation |
| Report Generation | "Generate this month's sales report" | Complete analysis report (including charts) |
Architecture Design
User Natural Language Question
|
[Intent Recognition] ──> Query / Analysis / Report
|
[Schema Understanding] ──> Read database metadata
|
[SQL Generation] ──> Generate and execute SQL
|
[Result Processing] ──> Format / Generate chart / Textual interpretation
|
[Security Check] ──> Confirm no sensitive data leakage
|
Return Result
Technology Selection
| Component | Choice | Reason |
|---|---|---|
| LLM | GPT-4o | SQL generation and data analysis require strong reasoning ability |
| Database | SQLite (demo) / PostgreSQL (production) | Lightweight and compatible |
| Chart Library | matplotlib + Plotly | Static charts + interactive charts |
| Framework | LangChain | Agent orchestration + tool management |
| Frontend | Streamlit | Rapidly build interactive interfaces |
10.2 Data Preprocessing and Schema Mapping Strategy
Sample Data Preparation
import sqlite3
import pandas as pd
def create_sample_database(db_path: str = "sales.db"):
"""Create a sample sales database"""
conn = sqlite3.connect(db_path)
# Orders table
orders_data = {
"order_id": [f"ORD{i:06d}" for i in range(1, 1001)],
"customer_id": [f"C{np.random.randint(1, 200):04d}" for _ in range(1000)],
"product_id": [f"P{np.random.randint(1, 50):04d}" for _ in range(1000)],
"order_date": pd.date_range("2024-01-01", periods=1000, freq="8H"),
"quantity": [np.random.randint(1, 20) for _ in range(1000)],
"unit_price": [round(np.random.uniform(10, 500), 2) for _ in range(1000)],
}
df_orders = pd.DataFrame(orders_data)
df_orders["total_amount"] = df_orders["quantity"] * df_orders["unit_price"]
df_orders.to_sql("orders", conn, if_exists="replace", index=False)
# Products table
products_data = {
"product_id": [f"P{i:04d}" for i in range(1, 51)],
"product_name": [f"Product {i}" for i in range(1, 51)],
"category": [np.random.choice(["Electronics", "Clothing", "Food", "Home"]) for _ in range(50)],
}
pd.DataFrame(products_data).to_sql("products", conn, if_exists="replace", index=False)
conn.close()
create_sample_database()
Schema Mapping: Letting the LLM Understand Database Structure
The LLM needs to know the database's table structure to generate correct SQL. The key strategy is to describe the Schema concisely but completely:
def get_db_schema(db_path: str) -> str:
"""Extract database schema description"""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
schema_description = ""
for (table_name,) in tables:
cursor.execute(f"PRAGMA table_info({table_name})")
columns = cursor.fetchall()
schema_description += f"\nTable {table_name}:\n"
for col in columns:
col_name, col_type = col[1], col[2]
schema_description += f" - {col_name} ({col_type})"
# Add sample data (3 rows)
cursor.execute(f"SELECT * FROM {table_name} LIMIT 3")
rows = cursor.fetchall()
col_names = [desc[0] for desc in cursor.description]
sample_df = pd.DataFrame(rows, columns=col_names)
schema_description += f"\n Sample data:\n{sample_df.to_string()}\n"
conn.close()
return schema_description
Schema Description Optimization
Stuffing the full Schema directly into the LLM wastes tokens. Optimization strategy:
def get_condensed_schema(db_path: str, relevant_tables: list[str] = None) -> str:
"""Condensed schema description, only including key information"""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
all_tables = [t[0] for t in cursor.fetchall()]
target_tables = relevant_tables or all_tables
schema_lines = []
for table_name in target_tables:
if table_name not in all_tables:
continue
cursor.execute(f"PRAGMA table_info({table_name})")
columns = cursor.fetchall()
col_descriptions = ", ".join([f"{c[1]} {c[2]}" for c in columns])
schema_lines.append(f"{table_name}({col_descriptions})")
conn.close()
return "\n".join(schema_lines)
# Output example:
# orders(order_id TEXT, customer_id TEXT, product_id TEXT, order_date DATETIME, quantity INTEGER, unit_price REAL, total_amount REAL)
# products(product_id TEXT, product_name TEXT, category TEXT)
10.3 Code Generation and Execution: Core Implementation Logic of PandasAI
SQL Generation and Execution Tool
from langchain_core.tools import tool
import sqlite3
@tool
def execute_sql(query: str, db_path: str = "sales.db") -> str:
"""Execute SQL query and return results.
Args:
query: SQL query statement (only SELECT supported)
db_path: Database file path
"""
# Security check
forbidden_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]
if any(kw in query.upper() for kw in forbidden_keywords):
return "Security restriction: Only SELECT queries are allowed"
try:
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute(query)
rows = cursor.fetchall()
col_names = [desc[0] for desc in cursor.description]
df = pd.DataFrame(rows, columns=col_names)
conn.close()
if len(df) > 50:
return f"Query returned {len(df)} rows of data, first 50 rows:\n{df.head(50).to_string()}"
return df.to_string()
except Exception as e:
return f"SQL execution error: {e}"
@tool
def generate_chart(data_description: str, chart_type: str = "bar") -> str:
"""Generate a chart based on data description.
Args:
data_description: Data description, format "col1,col2,...\\nval1,val2,...\\n..."
chart_type: Chart type, options: bar/line/pie/scatter
"""
import matplotlib.pyplot as plt
lines = data_description.strip().split("\n")
headers = lines[0].split(",")
data = [line.split(",") for line in lines[1:]]
df = pd.DataFrame(data, columns=headers)
for col in df.columns[1:]:
df[col] = pd.to_numeric(df[col], errors="coerce")
fig, ax = plt.subplots(figsize=(10, 6))
if chart_type == "bar":
df.plot(x=headers[0], y=headers[1:], kind="bar", ax=ax)
elif chart_type == "line":
df.plot(x=headers[0], y=headers[1:], kind="line", ax=ax)
elif chart_type == "pie":
df.plot(y=headers[1], labels=df[headers[0]], kind="pie", ax=ax)
chart_path = f"/tmp/chart_{int(time.time())}.png"
plt.savefig(chart_path, dpi=150, bbox_inches="tight")
plt.close()
return f"Chart generated: {chart_path}"
Building the Data Analysis Agent
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate
db_schema = get_condensed_schema("sales.db")
system_prompt = f"""You are a professional data analysis assistant.
## Database Structure
{db_schema}
## Workflow
1. Analyze the user's question, understand data requirements
2. Generate the correct SQL query statement
3. Use the execute_sql tool to execute the query
4. Analyze the query results
5. If a chart is needed, use the generate_chart tool to generate it
6. Interpret the analysis results in natural language
## Notes
- Only generate SELECT statements, do not modify data
- Date format: YYYY-MM-DD
- Use total_amount for monetary fields
- Provide specific numbers in analysis, avoid vague descriptions
"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [execute_sql, generate_chart]
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=5)
Usage Example
result = agent_executor.invoke({
"input": "What is the sales ranking by product category? Please generate a bar chart"
})
print(result["output"])
10.4 Security Fencing: Preventing Malicious Code Execution and Data Leakage
Three-Layer Security Protection
Layer 1: Prompt Layer - Inform Agent of security rules
Layer 2: Tool Layer - Perform security checks inside tools
Layer 3: Output Layer - Check if returned content contains sensitive data
Output Sanitization
class OutputSanitizer:
"""Output sanitization processor"""
SENSITIVE_PATTERNS = {
"phone": (r"1[3-9]\d{9}", lambda m: m.group()[:3] + "****" + m.group()[-4:]),
"email": (r"[\w.-]+@[\w.-]+\.\w+", lambda m: m.group()[0] + "***@" + m.group().split("@")[1]),
"id_card": (r"\d{17}[\dXx]", lambda m: m.group()[:6] + "********" + m.group()[-4:]),
}
def sanitize(self, text: str) -> str:
for pattern_name, (pattern, replace_fn) in self.SENSITIVE_PATTERNS.items():
text = re.sub(pattern, replace_fn, text)
return text
def check_sensitive_columns(self, df: pd.DataFrame) -> pd.DataFrame:
"""Check and sanitize sensitive columns in DataFrame"""
sensitive_keywords = ["phone", "email", "id_card", "password", "secret"]
for col in df.columns:
if any(kw in col.lower() for kw in sensitive_keywords):
df[col] = "[REDACTED]"
return df
Query Auditing
class QueryAuditor:
"""Query audit log"""
def __init__(self, log_path: str = "query_audit.jsonl"):
self.log_path = log_path
def audit(self, user: str, question: str, sql: str, result_summary: str):
record = {
"timestamp": datetime.now().isoformat(),
"user": user,
"question": question,
"sql": sql,
"result_summary": result_summary[:200],
"risk_level": self._assess_risk(sql),
}
with open(self.log_path, "a") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
def _assess_risk(self, sql: str) -> str:
sql_upper = sql.upper()
if any(kw in sql_upper for kw in ["DROP", "DELETE", "UPDATE"]):
return "high"
if any(kw in sql_upper for kw in ["password", "secret", "token"]):
return "high"
if "LIMIT" not in sql_upper and "COUNT" not in sql_upper:
return "medium" # Unlimited query might return large amounts of data
return "low"
10.5 Result Interpretation: Turning Data Insights into Natural Language Reports
Structured Output of Analysis Results
class AnalysisReporter:
"""Convert analysis results into natural language reports"""
def __init__(self, llm):
self.llm = llm
def generate_report(self, question: str, query_result: str,
chart_path: str = None) -> str:
report_prompt = f"""
User question: {question}
Data query result:
{query_result}
Please generate a structured analysis report, including:
1. Core findings (1-2 sentences summarizing the most important insights)
2. Data details (key numbers and changes)
3. Trend analysis (if there is a time dimension)
4. Actionable recommendations (data-driven actionable suggestions)
Requirements:
- Use specific numbers, don't say "has increased", say "increased by 23.5%"
- Highlight any outliers
- Keep language concise, avoid data dumping
"""
return self.llm.invoke(report_prompt).content
def generate_summary_table(self, data: pd.DataFrame) -> str:
"""Convert DataFrame to Markdown summary table"""
if len(data) > 10:
summary = data.describe().to_markdown()
top5 = data.head(5).to_markdown()
return f"Statistical Summary:\n{summary}\n\nFirst 5 rows:\n{top5}"
return data.to_markdown()
Complete End-to-End Flow
class DataAnalysisAgent:
"""Complete Intelligent Data Analysis Agent"""
def __init__(self, db_path: str = "sales.db"):
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
self.db_path = db_path
self.sanitizer = OutputSanitizer()
self.auditor = QueryAuditor()
self.reporter = AnalysisReporter(self.llm)
# Build Agent
db_schema = get_condensed_schema(db_path)
system_prompt = f"""You are a professional data analysis assistant. Database structure: {db_schema}
Workflow: Analyze question -> Generate SQL -> Execute query -> Interpret results.
Only execute SELECT queries. Provide specific numbers."""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
tools = [execute_sql, generate_chart]
agent = create_openai_tools_agent(self.llm, tools, prompt)
self.executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=5)
def analyze(self, question: str, user: str = "anonymous") -> dict:
# 1. Agent execution
result = self.executor.invoke({"input": question})
# 2. Output sanitization
safe_output = self.sanitizer.sanitize(result["output"])
# 3. Audit record
self.auditor.audit(user, question, "auto-generated", safe_output[:200])
return {
"question": question,
"answer": safe_output,
"success": True,
}
Chapter Summary
| Module | Core Implementation | Key Points |
|---|---|---|
| Requirements Analysis | Intent recognition + Architecture design | Define capability boundaries first, then design architecture |
| Schema Mapping | Condensed description + Sample data | LLM needs sufficient but not redundant Schema information |
| SQL Generation & Execution | Function Calling + Security check | Only allow SELECT, prohibit modification operations |
| Security Fencing | Prompt + Tool layer + Output layer | Three-layer protection, output sanitization |
| Result Interpretation | Structured report + Specific numbers | Data insights backed by numbers |
In the next chapter, we will build the second practical project — an Automated R&D Operations Agent.