AI数据可视化与报告生成完全教程

1. AI数据可视化概述与工具链

数据可视化正在从"手动画图"走向"AI驱动的智能呈现"。传统流程中，分析师需要手动选择图表类型、调整配色、编写代码，而AI工具链能将这个过程压缩到秒级。

核心工具链全景：

环境准备：

pip install plotly echarts-python streamlit openai pandas jinja2 \
    python-pptx python-docx pdfkit duckdb

快速验证环境：

import plotly
import openai
import streamlit
print(f"Plotly: {plotly.__version__}")
print("AI可视化工具链就绪")

2. LLM驱动的图表自动推荐

让LLM根据数据特征自动推荐最合适的图表类型，是AI可视化的核心能力之一。

实现原理： 将数据的schema（字段名、类型、样本值、统计特征）作为prompt输入，让LLM输出图表类型建议和配置。

import pandas as pd
import json
from openai import OpenAI

def analyze_dataframe(df: pd.DataFrame) -> dict:
    """提取DataFrame的结构化特征"""
    schema = {
        "columns": [],
        "row_count": len(df),
        "summary": {}
    }
    for col in df.columns:
        col_info = {
            "name": col,
            "dtype": str(df[col].dtype),
            "null_count": int(df[col].isnull().sum()),
            "unique_count": int(df[col].nunique())
        }
        if pd.api.types.is_numeric_dtype(df[col]):
            col_info["min"] = float(df[col].min())
            col_info["max"] = float(df[col].max())
            col_info["mean"] = float(df[col].mean())
        else:
            col_info["top_values"] = df[col].value_counts().head(5).to_dict()
        schema["columns"].append(col_info)
    return schema

def recommend_chart(df: pd.DataFrame, user_intent: str = "") -> dict:
    """让LLM推荐最佳图表类型"""
    schema = analyze_dataframe(df)
    client = OpenAI()
    
    prompt = f"""你是一个数据可视化专家。根据以下数据特征，推荐最合适的图表类型。

数据Schema:
{json.dumps(schema, ensure_ascii=False, indent=2)}

用户意图: {user_intent or "探索数据分布和关系"}

请返回JSON格式：
{{
    "primary_chart": {{"type": "图表类型", "reason": "原因", "config": {{}}}},
    "alternative_charts": [...],
    "key_insights": ["潜在洞察1", "潜在洞察2"]
}}"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

使用示例：

df = pd.read_csv("sales_data.csv")
recommendation = recommend_chart(df, "查看各地区销售额趋势")
print(f"推荐图表: {recommendation['primary_chart']['type']}")
print(f"推荐理由: {recommendation['primary_chart']['reason']}")

3. 自然语言生成图表（Text-to-Chart）

Text-to-Chart让用户用自然语言描述需求，AI自动生成完整的图表代码。

import plotly.express as px
import plotly.graph_objects as go

def text_to_chart(df: pd.DataFrame, user_query: str):
    """自然语言转图表"""
    client = OpenAI()
    schema = analyze_dataframe(df)
    
    system_prompt = """你是Plotly代码生成专家。根据用户需求和数据结构，生成可直接执行的Plotly图表代码。

规则：
1. 变量df已存在，是pandas DataFrame
2. 只返回Python代码，不要包含```python标记
3. 最后一行必须是fig对象
4. 使用plotly.express优先，复杂图表用graph_objects
5. 添加合适的标题、轴标签、hover信息"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"数据结构:\n{json.dumps(schema, ensure_ascii=False)}\n\n需求: {user_query}"}
        ]
    )
    
    code = response.choices[0].message.content
    
    # 安全执行生成的代码
    local_vars = {"df": df, "px": px, "go": go, "pd": pd}
    exec(code, {}, local_vars)
    fig = local_vars.get("fig")
    
    return fig, code

# 使用
fig, code = text_to_chart(df, "按月份展示销售额变化趋势，按产品类别分组，使用面积图")
fig.show()

进阶：带数据预处理的Text-to-Chart

def smart_text_to_chart(df: pd.DataFrame, query: str):
    """带自动数据预处理的图表生成"""
    client = OpenAI()
    
    # 第一步：让LLM规划数据处理和可视化步骤
    plan_prompt = f"""用户需求: {query}
数据列: {list(df.columns)}
数据类型: {df.dtypes.to_dict()}

返回JSON格式的执行计划：
{{
    "data_prep": ["pandas操作1", "pandas操作2"],
    "chart_type": "图表类型",
    "chart_code": "完整的Plotly代码"
}}"""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": plan_prompt}],
        response_format={"type": "json_object"}
    )
    
    plan = json.loads(response.choices[0].message.content)
    
    # 执行数据预处理
    working_df = df.copy()
    for step in plan["data_prep"]:
        working_df = eval(step, {"df": working_df, "pd": pd})
    
    # 生成图表
    local_vars = {"df": working_df, "px": px, "go": go}
    exec(plan["chart_code"], {}, local_vars)
    return local_vars["fig"]

4. Plotly/ECharts AI增强交互图表

Plotly AI增强

import plotly.graph_objects as go

def ai_enhanced_plotly(df: pd.DataFrame, chart_type: str, title: str):
    """AI增强的Plotly图表，自动优化样式和交互"""
    client = OpenAI()
    
    # 让AI生成最佳配色和布局
    style_prompt = f"""为一个{chart_type}图表设计配色方案和布局。
标题: {title}
数据量: {len(df)}行
返回JSON: {{"color_palette": [...], "layout": {{}}, "annotations": [...]}}"""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": style_prompt}],
        response_format={"type": "json_object"}
    )
    style = json.loads(response.choices[0].message.content)
    
    # 构建图表
    if chart_type == "bar":
        fig = px.bar(df, x=df.columns[0], y=df.columns[1],
                     color_discrete_sequence=style["color_palette"])
    elif chart_type == "line":
        fig = px.line(df, x=df.columns[0], y=df.columns[1],
                      color_discrete_sequence=style["color_palette"])
    
    # 应用AI生成的布局
    fig.update_layout(
        title=dict(text=title, font=dict(size=20)),
        template="plotly_white",
        hovermode="x unified",
        **style.get("layout", {})
    )
    
    # 添加AI生成的注释
    for ann in style.get("annotations", []):
        fig.add_annotation(**ann)
    
    return fig

ECharts AI增强

def generate_echarts_option(df: pd.DataFrame, query: str) -> dict:
    """生成ECharts配置项"""
    client = OpenAI()
    
    prompt = f"""根据数据和需求生成ECharts option配置（JSON格式）。
数据列: {list(df.columns)}
样本数据: {df.head(3).to_dict()}
需求: {query}

只返回ECharts option JSON，不需要额外解释。"""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

# 嵌入Streamlit使用
def render_echarts(option: dict, height: str = "500px"):
    """在Streamlit中渲染ECharts"""
    import streamlit as st
    from streamlit_echarts import st_echarts
    st_echarts(option, height=height)

5. 自动化报告生成（PDF/PPT/Word）

Jinja2 + PDFKit 生成PDF报告

from jinja2 import Template
import pdfkit
import base64
from io import BytesIO

# HTML报告模板
REPORT_TEMPLATE = """
<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <style>
        body { font-family: "Microsoft YaHei", sans-serif; margin: 40px; }
        .header { border-bottom: 3px solid #2563eb; padding-bottom: 20px; }
        .kpi-grid { display: grid; grid-template-columns: repeat(3, 1fr); gap: 20px; }
        .kpi-card { background: #f8fafc; border-radius: 8px; padding: 20px; text-align: center; }
        .kpi-value { font-size: 36px; font-weight: bold; color: #2563eb; }
        .chart-container { margin: 30px 0; text-align: center; }
        .insight-box { background: #fffbeb; border-left: 4px solid #f59e0b; padding: 15px; margin: 15px 0; }
    </style>
</head>
<body>
    <div class="header">
        <h1>{{ title }}</h1>
        <p>生成时间: {{ generated_at }} | 数据范围: {{ date_range }}</p>
    </div>
    
    <h2>核心指标</h2>
    <div class="kpi-grid">
        {% for kpi in kpis %}
        <div class="kpi-card">
            <div class="kpi-value">{{ kpi.value }}</div>
            <div>{{ kpi.label }}</div>
            <div style="color: {{ 'green' if kpi.change > 0 else 'red' }}">
                {{ '+' if kpi.change > 0 else '' }}{{ kpi.change }}%
            </div>
        </div>
        {% endfor %}
    </div>
    
    <h2>趋势分析</h2>
    <div class="chart-container">
        <img src="data:image/png;base64,{{ chart_image }}" width="100%">
    </div>
    
    <h2>AI洞察</h2>
    {% for insight in insights %}
    <div class="insight-box">
        <strong>{{ insight.title }}</strong>
        <p>{{ insight.description }}</p>
    </div>
    {% endfor %}
</body>
</html>
"""

def generate_pdf_report(data: dict, output_path: str):
    """生成PDF报告"""
    template = Template(REPORT_TEMPLATE)
    html = template.render(**data)
    pdfkit.from_string(html, output_path, options={
        'encoding': 'UTF-8',
        'page-size': 'A4',
        'margin-top': '20mm'
    })
    return output_path

python-pptx 生成PPT报告

from pptx import Presentation
from pptx.util import Inches, Pt
from pptx.enum.chart import XL_CHART_TYPE

def generate_ppt_report(title: str, slides_data: list, output_path: str):
    """自动生成PPT报告"""
    prs = Presentation()
    
    # 标题页
    slide = prs.slides.add_slide(prs.slide_layouts[0])
    slide.shapes.title.text = title
    slide.placeholders[1].text = "AI自动生成报告"
    
    for slide_info in slides_data:
        slide = prs.slides.add_slide(prs.slide_layouts[1])
        slide.shapes.title.text = slide_info["title"]
        
        if slide_info["type"] == "text":
            body = slide.placeholders[1]
            body.text = slide_info["content"]
        
        elif slide_info["type"] == "chart":
            # 插入图表图片
            img_path = slide_info["chart_path"]
            slide.shapes.add_picture(img_path, Inches(1), Inches(2), Inches(8), Inches(4.5))
        
        elif slide_info["type"] == "table":
            rows, cols = len(slide_info["data"]), len(slide_info["data"][0])
            table = slide.shapes.add_table(rows, cols, Inches(1), Inches(2), Inches(8), Inches(4)).table
            for i, row in enumerate(slide_info["data"]):
                for j, cell in enumerate(row):
                    table.cell(i, j).text = str(cell)
    
    prs.save(output_path)
    return output_path

6. 数据叙事与洞察提取

数据叙事（Data Storytelling）是将冰冷的数字转化为有说服力的故事。

def generate_data_narrative(df: pd.DataFrame, context: str = "") -> dict:
    """从数据中提取洞察并生成叙事"""
    client = OpenAI()
    
    # 计算关键统计量
    stats = {
        "shape": df.shape,
        "numeric_summary": df.describe().to_dict(),
        "correlations": df.select_dtypes(include='number').corr().to_dict(),
        "missing": df.isnull().sum().to_dict()
    }
    
    prompt = f"""你是一位资深数据分析师。根据以下数据统计信息，生成一份数据叙事报告。

数据统计:
{json.dumps(stats, ensure_ascii=False, default=str)}

背景: {context}

请生成：
1. 标题（简洁有力）
2. 三个关键发现（每个包含：发现描述、数据支撑、业务含义）
3. 一个行动建议
4. 一段总结性叙述（150字以内）

返回JSON格式。"""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

def detect_anomalies(df: pd.DataFrame, column: str) -> list:
    """AI辅助异常检测"""
    series = df[column].dropna()
    q1, q3 = series.quantile(0.25), series.quantile(0.75)
    iqr = q3 - q1
    outliers = series[(series < q1 - 1.5 * iqr) | (series > q3 + 1.5 * iqr)]
    
    if len(outliers) > 0:
        return [{
            "column": column,
            "count": len(outliers),
            "range": [float(outliers.min()), float(outliers.max())],
            "normal_range": [float(q1 - 1.5 * iqr), float(q3 + 1.5 * iqr)],
            "sample_indices": outliers.index.tolist()[:5]
        }]
    return []

7. AI辅助数据清洗与预处理

def ai_clean_data(df: pd.DataFrame, instructions: str = "") -> tuple:
    """AI驱动的数据清洗"""
    client = OpenAI()
    
    # 分析数据质量问题
    quality_report = {
        "dtypes": df.dtypes.astype(str).to_dict(),
        "missing": df.isnull().sum().to_dict(),
        "duplicates": int(df.duplicated().sum()),
        "sample": df.head(5).to_dict()
    }
    
    prompt = f"""分析以下数据质量问题，生成Pandas清洗代码。

数据质量报告:
{json.dumps(quality_report, ensure_ascii=False)}

用户额外要求: {instructions or "自动清洗"}

返回JSON:
{{
    "issues_found": ["问题1", "问题2"],
    "cleaning_steps": ["pandas代码1", "pandas代码2"],
    "explanation": "清洗逻辑说明"
}}

规则：
- 变量名固定为df
- 每步代码必须能独立执行
- 只返回必要的清洗操作"""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"}
    )
    
    plan = json.loads(response.choices[0].message.content)
    
    # 执行清洗
    cleaned_df = df.copy()
    for step in plan["cleaning_steps"]:
        try:
            cleaned_df = eval(step, {"df": cleaned_df, "pd": pd})
        except:
            exec(step, {"df": cleaned_df, "pd": pd})
    
    return cleaned_df, plan

8. 实时数据看板搭建

使用Streamlit搭建AI驱动的实时数据看板：

import streamlit as st
import plotly.express as px
import pandas as pd
from datetime import datetime, timedelta
import time

st.set_page_config(page_title="AI数据看板", layout="wide")

# 侧边栏配置
st.sidebar.title("📊 AI数据看板")
refresh_interval = st.sidebar.slider("刷新间隔(秒)", 5, 60, 30)
chart_type = st.sidebar.selectbox("图表类型", ["折线图", "柱状图", "散点图", "热力图"])

# 模拟实时数据源
@st.cache_data(ttl=refresh_interval)
def fetch_data():
    """模拟实时数据获取"""
    now = datetime.now()
    dates = [now - timedelta(minutes=i) for i in range(100)]
    return pd.DataFrame({
        "时间": dates,
        "CPU使用率": [50 + 30 * (0.5 - abs(0.5 - i/100)) for i in range(100)],
        "内存使用率": [60 + 20 * (i % 20) / 20 for i in range(100)],
        "请求量": [1000 + 500 * (i % 10) / 10 for i in range(100)],
    })

df = fetch_data()

# KPI卡片区
col1, col2, col3, col4 = st.columns(4)
col1.metric("CPU", f"{df['CPU使用率'].iloc[-1]:.1f}%", 
            f"{df['CPU使用率'].iloc[-1] - df['CPU使用率'].iloc[-2]:.1f}%")
col2.metric("内存", f"{df['内存使用率'].iloc[-1]:.1f}%")
col3.metric("当前QPS", f"{df['请求量'].iloc[-1]:.0f}")
col4.metric("数据更新", datetime.now().strftime("%H:%M:%S"))

# AI图表区域
st.subheader("AI智能分析")
user_query = st.text_input("用自然语言描述你想看的图表:", "展示CPU和内存的趋势变化")

if user_query:
    fig = px.line(df, x="时间", y=["CPU使用率", "内存使用率"])
    fig.update_layout(hovermode="x unified")
    st.plotly_chart(fig, use_container_width=True)

# 自动刷新
time.sleep(refresh_interval)
st.rerun()

9. 多模态报告（图文结合）

def generate_multimodal_report(df: pd.DataFrame, title: str) -> str:
    """生成包含图表和文字的HTML报告"""
    import plotly.io as pio
    
    charts_html = []
    narratives = []
    
    # 为每个数值列生成图表和叙述
    numeric_cols = df.select_dtypes(include='number').columns
    
    for col in numeric_cols[:4]:  # 最多4个图表
        fig = px.histogram(df, x=col, title=f"{col} 分布")
        img_bytes = pio.to_image(fig, format="png", width=800, height=400)
        img_b64 = base64.b64encode(img_bytes).decode()
        
        charts_html.append(f'<img src="data:image/png;base64,{img_b64}" style="max-width:100%">')
        
        # AI生成该列的叙述
        stats = df[col].describe()
        narratives.append(f"**{col}**: 均值{stats['mean']:.2f}，标准差{stats['std']:.2f}，"
                         f"范围[{stats['min']:.2f}, {stats['max']:.2f}]")
    
    html = f"""
    <html><body>
    <h1>{title}</h1>
    <p>报告时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}</p>
    {''.join(f'<div><h3>{n}</h3>{c}</div>' for n, c in zip(narratives, charts_html))}
    </body></html>
    """
    
    output_path = f"report_{datetime.now().strftime('%Y%m%d_%H%M')}.html"
    with open(output_path, "w", encoding="utf-8") as f:
        f.write(html)
    return output_path

10. 实战案例：销售数据智能分析报告系统

将前述所有技术整合为一个完整的销售分析报告系统：

import pandas as pd
from openai import OpenAI
from datetime import datetime
import json

class SalesReportSystem:
    def __init__(self, data_path: str):
        self.df = pd.read_csv(data_path, parse_dates=["date"])
        self.client = OpenAI()
    
    def run_analysis(self) -> dict:
        """执行完整分析流程"""
        # 1. 数据清洗
        self.df, cleaning_plan = ai_clean_data(self.df)
        
        # 2. 关键指标计算
        kpis = self._calculate_kpis()
        
        # 3. 趋势分析
        trends = self._analyze_trends()
        
        # 4. AI洞察
        insights = generate_data_narrative(self.df, "销售数据分析")
        
        # 5. 生成图表
        charts = self._generate_charts()
        
        return {
            "kpis": kpis,
            "trends": trends,
            "insights": insights,
            "charts": charts,
            "cleaning_log": cleaning_plan
        }
    
    def _calculate_kpis(self) -> list:
        total_revenue = self.df["revenue"].sum()
        avg_order = self.df["revenue"].mean()
        order_count = len(self.df)
        
        return [
            {"label": "总营收", "value": f"¥{total_revenue:,.0f}", "change": 12.5},
            {"label": "平均客单价", "value": f"¥{avg_order:,.0f}", "change": 3.2},
            {"label": "订单数", "value": f"{order_count:,}", "change": 8.7}
        ]
    
    def _analyze_trends(self) -> dict:
        monthly = self.df.groupby(self.df["date"].dt.to_period("M"))["revenue"].sum()
        return {
            "monthly_revenue": monthly.to_dict(),
            "growth_rate": monthly.pct_change().iloc[-1] * 100
        }
    
    def _generate_charts(self) -> list:
        charts = []
        
        # 月度趋势
        fig1 = px.line(
            self.df.groupby(self.df["date"].dt.to_period("M"))["revenue"].sum().reset_index(),
            x="date", y="revenue", title="月度营收趋势"
        )
        charts.append({"title": "月度营收趋势", "fig": fig1})
        
        # 品类占比
        if "category" in self.df.columns:
            fig2 = px.pie(self.df, values="revenue", names="category", title="品类营收占比")
            charts.append({"title": "品类营收占比", "fig": fig2})
        
        return charts
    
    def export_report(self, output_format: str = "pdf"):
        results = self.run_analysis()
        
        report_data = {
            "title": f"销售分析报告 - {datetime.now().strftime('%Y年%m月')}",
            "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
            "date_range": f"{self.df['date'].min().date()} 至 {self.df['date'].max().date()}",
            "kpis": results["kpis"],
            "insights": results["insights"]["key_insights"],
            "chart_image": self._chart_to_base64(results["charts"][0]["fig"])
        }
        
        if output_format == "pdf":
            return generate_pdf_report(report_data, "sales_report.pdf")
        elif output_format == "pptx":
            slides = [
                {"title": "核心指标", "type": "text", "content": json.dumps(results["kpis"], ensure_ascii=False)},
                {"title": "趋势分析", "type": "chart", "chart_path": self._save_chart(results["charts"][0]["fig"])},
            ]
            return generate_ppt_report(report_data["title"], slides, "sales_report.pptx")
    
    def _chart_to_base64(self, fig) -> str:
        import plotly.io as pio
        img_bytes = pio.to_image(fig, format="png", width=1000, height=500)
        return base64.b64encode(img_bytes).decode()
    
    def _save_chart(self, fig, path: str = "temp_chart.png") -> str:
        fig.write_image(path)
        return path

# 使用
system = SalesReportSystem("sales_data.csv")
system.export_report("pdf")

11. 企业级BI系统集成

将AI可视化能力集成到企业级BI系统中：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional

app = FastAPI(title="AI Visualization API")

class ChartRequest(BaseModel):
    data_source: str  # 数据源标识
    query: str        # 自然语言查询
    chart_type: Optional[str] = None
    output_format: str = "html"  # html/png/json

class ReportRequest(BaseModel):
    data_source: str
    template: str = "default"
    title: str
    sections: list[str]
    output_format: str = "pdf"

@app.post("/api/chart")
async def create_chart(request: ChartRequest):
    """AI图表生成API"""
    # 加载数据
    df = load_data_source(request.data_source)
    
    if request.chart_type:
        fig = generate_chart(df, request.chart_type, request.query)
    else:
        fig, _ = text_to_chart(df, request.query)
    
    if request.output_format == "html":
        return {"html": fig.to_html()}
    elif request.output_format == "json":
        return {"echarts_option": generate_echarts_option(df, request.query)}
    else:
        img_bytes = fig.to_image(format="png")
        return {"image": base64.b64encode(img_bytes).decode()}

@app.post("/api/report")
async def generate_report(request: ReportRequest):
    """AI报告生成API"""
    df = load_data_source(request.data_source)
    system = SalesReportSystem.__new__(SalesReportSystem)
    system.df = df
    system.client = OpenAI()
    
    result = system.export_report(request.output_format)
    return {"report_path": result, "status": "success"}

@app.post("/api/insight")
async def extract_insights(data_source: str, context: str = ""):
    """AI洞察提取API"""
    df = load_data_source(data_source)
    narrative = generate_data_narrative(df, context)
    anomalies = []
    for col in df.select_dtypes(include='number').columns:
        anomalies.extend(detect_anomalies(df, col))
    
    return {"narrative": narrative, "anomalies": anomalies}

def load_data_source(source_id: str) -> pd.DataFrame:
    """从企业数据仓库加载数据"""
    # 实际实现中连接数据库或数据湖
    sources = {
        "sales": "SELECT * FROM sales WHERE date >= CURRENT_DATE - INTERVAL '90 days'",
        "inventory": "SELECT * FROM inventory_snapshot",
    }
    # import duckdb
    # conn = duckdb.connect("warehouse.duckdb")
    # return conn.execute(sources[source_id]).fetchdf()
    return pd.DataFrame()  # 占位

部署配置 (docker-compose.yml)：

version: '3.8'
services:
  ai-viz-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - DATABASE_URL=${DATABASE_URL}
    volumes:
      - ./reports:/app/reports
  
  redis:
    image: redis:alpine
    ports:
      - "6379:6379"

这套系统的核心价值在于：让每个业务人员都能用自然语言获取数据洞察，而不需要学习SQL或Python。AI不仅是工具，更是连接数据与决策之间的桥梁。