24.5 Integrated Design and Runnable Foundation

Section Goal: Assemble the components from 24.2–24.4 by responsibility into an intelligent data analysis Agent, and explain how it actually runs within the repository.

⚠️ Honest Disclaimer (Runnability): The SafeDatabaseConnector, TextToSQL, DataAnalyzer, ChartGenerator, InsightGenerator, and ReportGenerator from 24.2–24.4 are reference implementations given as code snippets in the book. They depend on langchain_openai and your API Key; they are not packages you can directly import from this repository. To actually run them, you need to save each section's code as independent modules and install dependencies yourself — this section demonstrates the assembly logic, not "ready-to-run" files.

For a tested, default offline-runnable foundation, use the reference-agent/ at the repository root. The "hands-on" content in Chapters 12–23 is based on it.

Architecture Design Philosophy

Before integrating the components from previous sections into a complete system, let's analyze the key design decisions at the architectural level.

Pipeline Architecture vs Agent Loop Architecture

A data analysis Agent can adopt two fundamentally different architectures:

Pipeline Architecture — the choice of this chapter:

The LLM autonomously decides which tool to call and how many times, and can ask follow-up questions or adjust direction based on intermediate results. The advantage is flexibility and intelligence; the disadvantage is unpredictable latency, difficult debugging, and higher cost.

Reasons for choosing pipeline architecture in this chapter:

The "per-request cost" in the table above is an example order-of-magnitude estimate. Actual costs depend on the model, data volume, and provider billing. Use real usage data as the reference, not as a precise quote.

💡 Practical Advice: If your scenario requires autonomous Agent exploration (e.g., "help me find anomalies in the data"), consider building an Agent loop version using Chapter 13's LangGraph. The pipeline architecture is better suited for scenarios with clear workflows.

Component Interaction Sequence

The complete request processing flow is as follows:

User → SmartDataAnalyst → TextToSQL (Get Schema → LLM generates SQL) → SafeDB (Execute read-only query) → Analyzer/Chart/Insight/Report (Parallel analysis) → Complete Report → User

Note several key design points:

1. Schema Preloading: TextToSQL caches table structures at initialization, avoiding reading database metadata for every query
2. Parallel Analysis and Visualization: describe() and auto_chart() can theoretically execute in parallel (currently sequential, optimizable)
3. Insights Depend on Statistical Results: generate_insights() requires statistical results as input, helping the LLM generate analysis based on data rather than guesses

Complete Implementation (Assembly Logic)

"""
Intelligent Data Analysis Agent — Assembly Logic
Dependencies: 24.2 SafeDatabaseConnector / TextToSQL
              24.3 DataAnalyzer / ChartGenerator / InsightGenerator
              24.4 ReportGenerator
Prerequisites: Save the above classes as independent modules and pip install langchain-openai
"""
import asyncio
from langchain_openai import ChatOpenAI

from db_connector import SafeDatabaseConnector
from text_to_sql import TextToSQL
from data_analyzer import DataAnalyzer
from chart_generator import ChartGenerator
from insight_generator import InsightGenerator
from report_generator import ReportGenerator


class SmartDataAnalyst:
    """Intelligent Data Analysis Agent"""
    
    def __init__(self, db_path: str):
        self.llm = ChatOpenAI(model="gpt-4.1", temperature=0)
        self.db = SafeDatabaseConnector(db_path)
        self.text2sql = TextToSQL(self.llm, self.db)
        self.analyzer = DataAnalyzer()
        self.chart_gen = ChartGenerator()
        self.insight_gen = InsightGenerator(self.llm)
        self.report_gen = ReportGenerator(self.llm)
    
    async def ask(self, question: str) -> str:
        """Ask a question in natural language, get a complete analysis"""
        
        print(f"🤔 Understanding question: {question}")
        
        # 1. Natural Language → SQL
        print("📝 Generating query...")
        sql = await self.text2sql.convert(question)
        print(f"   SQL: {sql}")
        
        # 2. Execute query
        print("🔍 Querying data...")
        try:
            data = self.db.execute_readonly(sql)
        except Exception as e:
            return f"❌ Query error: {e}"
        
        if not data:
            return "📭 The query returned no results. Please try rephrasing your question."
        
        print(f"   Got {len(data)} records")
        
        # 3. Statistical analysis
        print("📊 Analyzing data...")
        stats = self.analyzer.describe(data)
        
        # 4. Generate chart
        print("🎨 Generating chart...")
        chart_path = self.chart_gen.auto_chart(data, question)
        
        # 5. Generate insights
        print("💡 Extracting insights...")
        insights = await self.insight_gen.generate_insights(
            data, stats, question
        )
        
        # 6. Generate report
        print("📄 Generating report...")
        report = await self.report_gen.generate_report(
            question=question,
            sql_query=sql,
            data=data,
            stats=stats,
            insights=insights,
            chart_path=chart_path
        )
        
        # Save report
        filepath = self.report_gen.save_report(report)
        print(f"✅ Report saved: {filepath}")
        
        return report


async def main():
    """Interactive data analysis"""
    import sys
    
    db_path = sys.argv[1] if len(sys.argv) > 1 else "example.db"
    
    print("📊 Intelligent Data Analysis Assistant")
    print("=" * 40)
    print("Describe your analysis needs in natural language")
    print("Type 'quit' to exit\n")
    
    analyst = SmartDataAnalyst(db_path)
    
    # Show available tables
    schemas = analyst.db.get_table_schemas()
    print(f"📁 Database has {len(schemas)} tables:")
    for table, info in schemas.items():
        cols = [c['name'] for c in info['columns']]
        print(f"   • {table}: {', '.join(cols)}")
    print()
    
    while True:
        question = input("Your question: ").strip()
        
        if question.lower() in ('quit', 'exit', 'q'):
            print("👋 Goodbye!")
            break
        
        if not question:
            continue
        
        result = await analyst.ask(question)
        print(f"\n{result}\n")


if __name__ == "__main__":
    asyncio.run(main())

Expected Behavior (Illustrative, Not Actual Runtime Logs)

The following is an illustrative interaction used to explain the workflow, not output produced by directly running this repository. Actual results depend on your database, model, and prompts.

Your question: Which region has the highest order amount? Sort by region
→ TextToSQL generates: SELECT region, SUM(amount) AS total
                        FROM orders GROUP BY region ORDER BY total DESC
→ SafeDB executes read-only query → Gets several rows of aggregated results
→ Analyzer/Chart/Insight/Report generates analysis and report

Error Handling and Degradation Strategy

In a production environment, every step of the data analysis Agent can fail. A robust system needs "graceful degradation" — even if some features fail, it should still provide a valuable response.

Layered Error Handling

class ResilientDataAnalyst(SmartDataAnalyst):
    """Data analysis Agent with degradation capabilities"""
    
    async def ask(self, question: str) -> str:
        """Each step has independent try-except; degrade on failure instead of aborting"""
        
        # Step 1: Text-to-SQL (terminate on failure, cannot continue)
        try:
            sql = await self.text2sql.convert(question)
        except Exception as e:
            return f"❌ Unable to understand your question, please try a different wording.\nTechnical details: {e}"
        
        # Step 2: Execute query (self-correct on failure)
        data = None
        for attempt in range(3):
            try:
                data = self.db.execute_readonly(sql)
                break
            except Exception as e:
                if attempt < 2:
                    # Let LLM fix the SQL based on the error
                    sql = await self._fix_sql(sql, str(e))
                else:
                    return f"❌ Query failed multiple times: {e}\nGenerated SQL: {sql}"
        
        if not data:
            return "📭 The query returned no results. Please try rephrasing your question."
        
        # Step 3: Statistical analysis (skip on failure)
        stats = None
        try:
            stats = self.analyzer.describe(data)
        except Exception:
            stats = {"error": "Statistical analysis skipped"}
        
        # Step 4: Chart generation (skip on failure, does not affect report)
        chart_path = None
        try:
            chart_path = self.chart_gen.auto_chart(data, question)
        except Exception:
            chart_path = None  # Report will not include chart
        
        # Step 5: Insight generation (use default message on failure)
        try:
            insights = await self.insight_gen.generate_insights(
                data, stats, question
            )
        except Exception:
            insights = "(Insight generation is temporarily unavailable. Below is a raw data summary.)"
        
        # Step 6: Report generation (return raw data on failure)
        try:
            report = await self.report_gen.generate_report(
                question=question, sql_query=sql, data=data,
                stats=stats, insights=insights, chart_path=chart_path
            )
        except Exception:
            # Minimum viable degradation output
            report = f"## Query Results\n\nSQL: `{sql}`\n\nData (top 5 records):\n"
            for row in data[:5]:
                report += f"- {row}\n"
        
        return report

Degradation Level Overview

Performance Optimization Tips

When the data analysis Agent serves multiple users, performance optimization is crucial:

1. Schema Caching and Incremental Updates

import time

class CachedSchemaManager:
    """Schema manager with TTL caching"""
    
    def __init__(self, db: SafeDatabaseConnector, ttl_seconds: int = 300):
        self.db = db
        self.ttl = ttl_seconds
        self._cache = None
        self._cache_time = 0
    
    def get_schemas(self) -> dict:
        now = time.time()
        if self._cache is None or (now - self._cache_time) > self.ttl:
            self._cache = self.db.get_table_schemas()
            self._cache_time = now
        return self._cache

2. LLM Call Optimization

# Before optimization: 3 sequential LLM calls
sql = await text2sql.convert(question)       # ~2s
insights = await insight_gen.generate(...)    # ~3s
report = await report_gen.generate(...)       # ~3s
# Total: ~8s

# After optimization: insights and report outline generated in parallel
import asyncio
insights, report_outline = await asyncio.gather(
    insight_gen.generate(data, stats, question),
    report_gen.generate_outline(question, stats)  
)
# Saves ~2-3s

3. Query Result Caching

For repeated or similar queries, SQL and results can be cached:

from functools import lru_cache
import hashlib

class QueryCache:
    """Simple query result cache"""
    
    def __init__(self, max_size: int = 100):
        self._cache: dict[str, tuple[float, list]] = {}
        self.max_size = max_size
        self.ttl = 600  # 10 minute expiration
    
    def get(self, sql: str) -> list[dict] | None:
        key = hashlib.md5(sql.encode()).hexdigest()
        if key in self._cache:
            cached_time, results = self._cache[key]
            if time.time() - cached_time < self.ttl:
                return results
            del self._cache[key]
        return None
    
    def set(self, sql: str, results: list[dict]):
        if len(self._cache) >= self.max_size:
            # Evict the oldest cache entry
            oldest = min(self._cache, key=lambda k: self._cache[k][0])
            del self._cache[oldest]
        key = hashlib.md5(sql.encode()).hexdigest()
        self._cache[key] = (time.time(), results)

Extension Directions

This chapter implements a basic version of a data analysis Agent. Here are advanced directions worth exploring:

Direction 1: Multi-turn Conversational Analysis

Currently, each Q&A is independent — users cannot ask follow-ups like "break it down by month" or "only look at East China region". You can introduce conversation context management:

class ConversationalAnalyst:
    """Data analysis Agent supporting multi-turn conversation"""
    
    def __init__(self, base_analyst: SmartDataAnalyst):
        self.analyst = base_analyst
        self.history: list[dict] = []
    
    async def ask(self, question: str) -> str:
        # Inject conversation history into Prompt to help LLM understand context
        context = "\n".join(
            f"User: {h['question']}\nSQL: {h['sql']}" 
            for h in self.history[-3:]  # Keep the last 3 turns
        )
        
        enhanced_question = f"Conversation history:\n{context}\n\nCurrent question: {question}"
        result = await self.analyst.ask(enhanced_question)
        
        self.history.append({"question": question, "sql": "...", "result": result})
        return result

Direction 2: Automatic Anomaly Detection

Enable the Agent to not only answer questions but also proactively discover data anomalies:

• Detect outliers in numeric columns (Z-score > 3)
• Discover change points in time series
• Flag data that violates business rules (e.g., negative order amounts)

Direction 3: Integration with Visualization Frontend

Combine the Agent backend with frontend dashboards (e.g., Streamlit, Gradio) for an interactive experience:

• Natural language Q&A + real-time chart rendering
• Drag-and-drop data exploration
• One-click PDF report export

Direction 4: Multi-Agent Collaborative Analysis

For complex analysis tasks (e.g., "compare sales trends across three quarters and give marketing recommendations"), tasks can be split among multiple specialized Agents:

• Data Query Agent: Responsible for Text-to-SQL and data retrieval
• Statistical Analysis Agent: Responsible for trend detection, regression analysis
• Report Writing Agent: Responsible for integrating all results into a report

This can be achieved using the Multi-Agent architecture introduced in Chapter 14.

Summary

💡 Further Reading: For model routing evaluation methods regarding cost-quality trade-offs, see 18.8 Model Routing Evaluation.

🎓 Chapter Summary: We built a complete Agent that "does data analysis using natural language." From Text-to-SQL to automatic visualization, it demonstrates the powerful application of Agents in the data analysis domain.

📝 Chapter Exercises

After reading this chapter, close the book and answer the following questions in your own words first, then expand the reference answers to compare.

Exercise 1 (Concept): Section 24.5 built the entire data analysis Agent as a "fixed six-step pipeline architecture" rather than an "Agent loop architecture" where the LLM autonomously decides the next step. Name at least 3 benefits of the pipeline architecture in this scenario, and identify what kind of requirements would be better suited for the Agent loop architecture.

Exercise 2 (Analysis): Section 24.2 designed a "three-layer defense" to prevent LLMs from generating dangerous SQL: Prompt constraints, SQL syntax validation, and database read-only permissions. A student says: "I've already explicitly written 'only SELECT' in the Prompt — that one layer is enough; the other two are redundant." Refute this claim, and explain why database read-only permissions are called the "last line of defense."

Exercise 3 (Hands-on): Section 24.2 mentioned the "Self-Correction" strategy for Text-to-SQL — when SQL execution fails, feed the error back to the LLM for a retry. However, in this chapter's implementation, only the "error message" is fed back during retries. Please improve the _fix_sql design: write an improved version that includes not only the error but also the table schema during correction, and explain why including the table schema improves correction success rates.

async def _fix_sql(self, broken_sql: str, error_msg: str) -> str:
    """SQL self-correction with table schema context"""
    # Retrieve the actual table schema (reuse cached schema)
    schema_desc = self.text2sql._format_schemas()

    fix_prompt = f"""The SQL you generated failed to execute. Please correct it.

Actual database table schema (strictly follow these column/table names):
{schema_desc}

Erroneous SQL:
{broken_sql}

Database error returned:
{error_msg}

Requirements:
1. Compare against the actual table schema above — check if you used the wrong table or column names
2. Still only generate SELECT queries
3. Return only the corrected SQL, no other text
"""
    response = await self.llm.ainvoke(fix_prompt)
    sql = response.content.strip()
    # Clean up possible markdown code block markers
    if sql.startswith("```"):
        sql = sql.split("\n", 1)[1].rsplit("```", 1)[0]
    return sql.strip()

Chapter 23 Hands-on Project: Multimodal Agent