22.5 集成设计与可运行底座

本节目标：把 22.2–22.4 的组件按职责组装成一个智能数据分析 Agent，并说清它在仓库中的真实运行方式。

⚠️ 诚实说明（可运行性）：22.2–22.4 的 SafeDatabaseConnector、TextToSQL、DataAnalyzer、ChartGenerator、InsightGenerator、ReportGenerator 是在书中以代码片段形式给出的参考实现，依赖 langchain_openai 与你的 API Key，并非本仓库里可直接 import 的包。要让它们真正跑起来，你需要把各节代码保存为独立模块并自行安装依赖——本节展示的是组装逻辑，不是"开箱即跑"的文件。

想要已测试、默认离线即可运行的底座，请用仓库根目录的 reference-agent/。第 12–23 章的"实战"均以其为准。

架构设计理念

在将前面各节的组件整合为完整系统之前，我们先分析架构层面的关键设计决策。

管道式架构 vs Agent 循环架构

数据分析 Agent 可以采用两种截然不同的架构：

管道式架构（Pipeline）——本章的选择：

LLM 自主决定调用哪个工具、调用几次，可以根据中间结果追问或调整方向。优点是灵活智能；缺点是延迟不可控、调试困难、成本更高。

本章选择管道式架构的原因：

上表“单次请求成本”为示例量级估算，实际取决于模型、数据量与提供商账单，请以真实用量为准，勿作精确报价。

💡 实践建议：如果你的场景需要 Agent 自主探索（如"帮我找出数据中的异常"），建议结合第 13 章 LangGraph 构建 Agent 循环版本。管道式架构更适合流程明确的场景。

组件交互时序

完整的请求处理流程如下：

用户 → SmartDataAnalyst → TextToSQL（获取 Schema → LLM 生成 SQL）→ SafeDB（执行只读查询）→ Analyzer/Chart/Insight/Report（并行分析）→ 完整报告 → 用户

注意几个关键设计点：

1. Schema 预加载：TextToSQL 在初始化时就缓存了表结构，避免每次查询都读取数据库元数据
2. 分析与可视化并行：describe() 和 auto_chart() 理论上可以并行执行（当前为顺序执行，可优化）
3. 洞察依赖统计结果：generate_insights() 需要统计结果作为输入，帮助 LLM 基于数据而非猜测生成分析

完整实现（组装逻辑）

"""
智能数据分析 Agent —— 组装逻辑
依赖：22.2 SafeDatabaseConnector / TextToSQL
      22.3 DataAnalyzer / ChartGenerator / InsightGenerator
      22.4 ReportGenerator
前置：将上述类保存为独立模块并 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:
    """智能数据分析 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:
        """用自然语言提问，获取完整分析"""
        
        print(f"🤔 理解问题: {question}")
        
        # 1. 自然语言 → SQL
        print("📝 生成查询...")
        sql = await self.text2sql.convert(question)
        print(f"   SQL: {sql}")
        
        # 2. 执行查询
        print("🔍 查询数据...")
        try:
            data = self.db.execute_readonly(sql)
        except Exception as e:
            return f"❌ 查询出错: {e}"
        
        if not data:
            return "📭 查询没有返回结果，请换个问法试试。"
        
        print(f"   获得 {len(data)} 条数据")
        
        # 3. 统计分析
        print("📊 分析数据...")
        stats = self.analyzer.describe(data)
        
        # 4. 生成图表
        print("🎨 生成图表...")
        chart_path = self.chart_gen.auto_chart(data, question)
        
        # 5. 生成洞察
        print("💡 提取洞察...")
        insights = await self.insight_gen.generate_insights(
            data, stats, question
        )
        
        # 6. 生成报告
        print("📄 生成报告...")
        report = await self.report_gen.generate_report(
            question=question,
            sql_query=sql,
            data=data,
            stats=stats,
            insights=insights,
            chart_path=chart_path
        )
        
        # 保存报告
        filepath = self.report_gen.save_report(report)
        print(f"✅ 报告已保存: {filepath}")
        
        return report


async def main():
    """交互式数据分析"""
    import sys
    
    db_path = sys.argv[1] if len(sys.argv) > 1 else "example.db"
    
    print("📊 智能数据分析助手")
    print("=" * 40)
    print("用自然语言描述你的分析需求")
    print("输入 'quit' 退出\n")
    
    analyst = SmartDataAnalyst(db_path)
    
    # 展示可用的表
    schemas = analyst.db.get_table_schemas()
    print(f"📁 数据库中有 {len(schemas)} 张表:")
    for table, info in schemas.items():
        cols = [c['name'] for c in info['columns']]
        print(f"   • {table}: {', '.join(cols)}")
    print()
    
    while True:
        question = input("你的问题: ").strip()
        
        if question.lower() in ('quit', 'exit', 'q'):
            print("👋 再见！")
            break
        
        if not question:
            continue
        
        result = await analyst.ask(question)
        print(f"\n{result}\n")


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

预期行为（示意，非真实运行日志）

下面是一段示意性交互，用于说明流程，不是本仓库直接运行产生的输出。真实结果取决于你的数据库、模型与提示词。

你的问题: 哪个区域的订单金额最高？按区域排序
→ TextToSQL 生成: SELECT region, SUM(amount) AS total
                   FROM orders GROUP BY region ORDER BY total DESC
→ SafeDB 执行只读查询 → 获得若干行聚合结果
→ Analyzer/Chart/Insight/Report 生成分析与报告

错误处理与降级策略

生产环境中，数据分析 Agent 的每个环节都可能失败。健壮的系统需要"优雅降级"——即使部分功能失败，仍能给出有价值的响应。

分层错误处理

class ResilientDataAnalyst(SmartDataAnalyst):
    """带降级能力的数据分析 Agent"""
    
    async def ask(self, question: str) -> str:
        """每个步骤独立 try-except，失败时降级而非中止"""
        
        # 步骤 1：Text-to-SQL（失败则终止，无法继续）
        try:
            sql = await self.text2sql.convert(question)
        except Exception as e:
            return f"❌ 无法理解您的问题，请换个表述试试。\n技术细节：{e}"
        
        # 步骤 2：执行查询（失败则尝试自我修正）
        data = None
        for attempt in range(3):
            try:
                data = self.db.execute_readonly(sql)
                break
            except Exception as e:
                if attempt < 2:
                    # 让 LLM 根据错误修正 SQL
                    sql = await self._fix_sql(sql, str(e))
                else:
                    return f"❌ 查询多次失败：{e}\n生成的 SQL：{sql}"
        
        if not data:
            return "📭 查询没有返回结果，请换个问法试试。"
        
        # 步骤 3：统计分析（失败则跳过）
        stats = None
        try:
            stats = self.analyzer.describe(data)
        except Exception:
            stats = {"error": "统计分析跳过"}
        
        # 步骤 4：图表生成（失败则跳过，不影响报告）
        chart_path = None
        try:
            chart_path = self.chart_gen.auto_chart(data, question)
        except Exception:
            chart_path = None  # 报告中将不包含图表
        
        # 步骤 5：洞察生成（失败则使用默认洞察）
        try:
            insights = await self.insight_gen.generate_insights(
                data, stats, question
            )
        except Exception:
            insights = "（洞察生成暂时不可用，以下为原始数据摘要）"
        
        # 步骤 6：报告生成（失败则返回原始数据）
        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:
            # 最低限度的降级输出
            report = f"## 查询结果\n\nSQL: `{sql}`\n\n数据（前 5 条）:\n"
            for row in data[:5]:
                report += f"- {row}\n"
        
        return report

降级层级示意

性能优化技巧

当数据分析 Agent 服务多用户时，性能优化至关重要：

1. Schema 缓存与增量更新

import time

class CachedSchemaManager:
    """带 TTL 缓存的 Schema 管理器"""
    
    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 调用优化

# 优化前：3 次串行 LLM 调用
sql = await text2sql.convert(question)       # ~2s
insights = await insight_gen.generate(...)    # ~3s
report = await report_gen.generate(...)       # ~3s
# 总计: ~8s

# 优化后：洞察和报告大纲并行生成
import asyncio
insights, report_outline = await asyncio.gather(
    insight_gen.generate(data, stats, question),
    report_gen.generate_outline(question, stats)  
)
# 节省 ~2-3s

3. 查询结果缓存

对于重复或相似的查询，可以缓存 SQL 和结果：

from functools import lru_cache
import hashlib

class QueryCache:
    """简单的查询结果缓存"""
    
    def __init__(self, max_size: int = 100):
        self._cache: dict[str, tuple[float, list]] = {}
        self.max_size = max_size
        self.ttl = 600  # 10 分钟过期
    
    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:
            # 淘汰最旧的缓存
            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)

扩展方向

本章实现的是数据分析 Agent 的基础版本。以下是值得探索的进阶方向：

方向一：多轮对话式分析

当前系统每次问答独立，用户无法追问"按月份细分"或"只看华东地区"。可以引入对话上下文管理：

class ConversationalAnalyst:
    """支持多轮对话的数据分析 Agent"""
    
    def __init__(self, base_analyst: SmartDataAnalyst):
        self.analyst = base_analyst
        self.history: list[dict] = []
    
    async def ask(self, question: str) -> str:
        # 将历史对话注入 Prompt，帮助 LLM 理解上下文
        context = "\n".join(
            f"用户: {h['question']}\nSQL: {h['sql']}" 
            for h in self.history[-3:]  # 保留最近 3 轮
        )
        
        enhanced_question = f"对话历史:\n{context}\n\n当前问题: {question}"
        result = await self.analyst.ask(enhanced_question)
        
        self.history.append({"question": question, "sql": "...", "result": result})
        return result

方向二：自动异常检测

让 Agent 不仅回答问题，还主动发现数据异常：

• 检测数值列中的离群值（Z-score > 3）
• 发现时间序列中的突变点
• 标记不符合业务规则的数据（如负数订单金额）

方向三：与可视化前端集成

将 Agent 后端与前端仪表板（如 Streamlit、Gradio）结合，提供交互式体验：

• 自然语言问答 + 实时图表渲染
• 拖拽式数据探索
• 一键导出 PDF 报告

方向四：多 Agent 协作分析

对于复杂分析任务（如"对比三个季度的销售趋势并给出营销建议"），可以将任务拆分给多个专业 Agent：

• 数据查询 Agent：负责 Text-to-SQL 和数据获取
• 统计分析 Agent：负责趋势检测、回归分析
• 报告撰写 Agent：负责整合所有结果生成报告

这可以利用第 14 章介绍的 Multi-Agent 架构来实现。

小结

💡 延伸阅读：关于成本-质量权衡的模型路由评估方法，详见 18.8 模型路由评估。

🎓 本章总结：我们构建了一个"用自然语言做数据分析"的完整 Agent。从 Text-to-SQL 到自动可视化，展示了 Agent 在数据分析领域的强大应用。

📝 本章练习

读完本章，先合上书用自己的话回答下面的问题，再展开参考答案对照。

练习 1（概念）：本章 22.5 把整个数据分析 Agent 做成了"固定六步的管道式架构"，而不是让 LLM 自主决定下一步的"Agent 循环架构"。请说出管道式架构在本场景下的至少 3 个好处，并指出什么样的需求反而更适合 Agent 循环架构。

练习 2（辨析）：22.2 节为防止 LLM 生成危险 SQL 设计了"三层防护"：Prompt 约束、SQL 语法校验、数据库只读权限。有同学说："我已经在 Prompt 里明确写了'只许 SELECT'，这层就够了，后两层是多此一举。"请反驳，并解释为什么数据库只读权限被称为"最后一道安全网"。

练习 3（动手）：22.2 节提到了 Text-to-SQL 的"Self-Correction（自我修正）"策略——SQL 执行报错就把错误反馈给 LLM 重试。但本章的实现里，重试时只把"错误信息"喂回去了。请你改进 _fix_sql 的设计：写一个改进版，让修正时不仅带上报错，还带上表结构，并说明为什么加上表结构能提高修正成功率。

async def _fix_sql(self, broken_sql: str, error_msg: str) -> str:
    """带表结构上下文的 SQL 自我修正"""
    # 取出真实的表结构（复用已缓存的 schema）
    schema_desc = self.text2sql._format_schemas()

    fix_prompt = f"""你生成的 SQL 执行失败了，请修正它。

数据库真实表结构（请严格按此列名/表名）：
{schema_desc}

出错的 SQL：
{broken_sql}

数据库返回的错误：
{error_msg}

要求：
1. 对照上面的真实表结构，检查是不是用错了表名或列名
2. 仍然只能生成 SELECT 查询
3. 只返回修正后的 SQL，不要其他文字
"""
    response = await self.llm.ainvoke(fix_prompt)
    sql = response.content.strip()
    # 清理可能的 markdown 代码块标记
    if sql.startswith("```"):
        sql = sql.split("\n", 1)[1].rsplit("```", 1)[0]
    return sql.strip()

第23章 项目实战：多模态 Agent