AI 工具开发实战（7）：开发一个 AI 数据看板——用 Streamlit 上传文件自动分析

上传 Excel/CSV 文件
  ↓
1. 自动预览数据（前 10 行）
2. AI 分析数据特征（分布、异常值、趋势）
3. 生成可视化图表
4. 对话式数据问答："上个月销售额最高的品类是什么？"

aidash/
├── app.py               # Streamlit 主程序
├── analyzer.py           # AI 数据分析
├── requirements.txt
└── .env

# app.py
import streamlit as st
import pandas as pd
import plotly.express as px
from analyzer import AIAnalyzer

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

# 初始化 AI 分析器
if 'analyzer' not in st.session_state:
    st.session_state.analyzer = AIAnalyzer()

# 上传文件
uploaded_file = st.file_uploader("上传 Excel 或 CSV 文件", type=['csv', 'xlsx'])
if uploaded_file:
    # 读取数据
    if uploaded_file.name.endswith('.csv'):
        df = pd.read_csv(uploaded_file)
    else:
        df = pd.read_excel(uploaded_file)

    st.session_state.df = df

    # 数据预览
    col1, col2, col3, col4 = st.columns(4)
    col1.metric("行数", len(df))
    col2.metric("列数", len(df.columns))
    col3.metric("缺失值", df.isna().sum().sum())
    col4.metric("数值列", len(df.select_dtypes('number').columns))

    # 数据表格
    with st.expander("📋 数据预览", expanded=True):
        st.dataframe(df.head(20), use_container_width=True)

    # AI 分析
    if st.button("🤖 AI 分析数据", type="primary"):
        with st.spinner("AI 正在分析..."):
            analysis = st.session_state.analyzer.analyze(df)

            # 数据洞察
            if analysis.get("insights"):
                st.subheader("💡 关键洞察")
                for insight in analysis["insights"]:
                    st.info(insight)

            # 推荐图表
            if analysis.get("charts"):
                st.subheader("📈 推荐图表")
                for chart in analysis["charts"]:
                    col_name = chart["column"]
                    chart_type = chart["type"]

                    if chart_type == "bar" and col_name in df.columns:
                        counts = df[col_name].value_counts().head(10)
                        fig = px.bar(x=counts.index, y=counts.values,
                                     title=f"{col_name} 分布")
                        st.plotly_chart(fig, use_container_width=True)

                    elif chart_type == "line" and col_name in df.columns:
                        if pd.api.types.is_numeric_dtype(df[col_name]):
                            fig = px.line(df[col_name].dropna().head(200),
                                         title=f"{col_name} 趋势")
                            st.plotly_chart(fig, use_container_width=True)

    # 对话式问答
    st.subheader("💬 数据问答")
    question = st.text_input("输入你的数据问题", placeholder="例如：销售额最高的品类是哪个？")
    if question:
        with st.spinner("AI 正在回答..."):
            answer = st.session_state.analyzer.ask(df, question)
            st.markdown(answer)

# analyzer.py
from openai import OpenAI
import os
import json
from dotenv import load_dotenv
load_dotenv()

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

class AIAnalyzer:
    """AI 数据分析引擎。"""

    def analyze(self, df):
        """分析数据并返回洞察和图表建议。"""
        # 准备数据摘要
        summary = self._get_data_summary(df)

        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{
                "role": "system",
                "content": f"""你是一个数据分析专家。分析以下数据摘要，返回 JSON 格式：

{{
  "insights": ["洞察1", "洞察2", ...],
  "charts": [{{"column": "列名", "type": "bar|line|pie", "reason": "原因"}}, ...]
}}

数据摘要：{summary}"""
            }],
            temperature=0.3,
        )

        try:
            return json.loads(response.choices[0].message.content)
        except:
            return {"insights": [], "charts": []}

    def ask(self, df, question):
        """回答数据相关问题。"""
        summary = self._get_data_summary(df)
        sample = df.head(5).to_string()

        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{
                "role": "system",
                "content": f"基于以下数据回答用户问题。\n\n数据概览：{summary}\n\n数据样本：\n{sample}"
            }, {
                "role": "user",
                "content": question,
            }],
            temperature=0.3,
        )
        return response.choices[0].message.content

    def _get_data_summary(self, df):
        """生成数据摘要。"""
        return f"""
形状：{df.shape[0]} 行 x {df.shape[1]} 列
列名：{list(df.columns)}
数据类型：{dict(df.dtypes.astype(str))}
数值列统计：{df.describe().to_string() if len(df.select_dtypes('number').columns) > 0 else '无'}
缺失值：{dict(df.isna().sum())}
"""

pip install streamlit pandas plotly openai python-dotenv
streamlit run app.py