构建一个数据分析Agent：提升分析效率的实践

在上一篇文章中,我们讨论了如何构建一个智能客服Agent。今天,我想分享另一个实际项目:如何构建一个数据分析Agent。这个项目源于我们一个金融客户的真实需求 - 提升数据分析效率,加快决策速度。

从分析师的痛点说起

记得和分析师团队交流时的场景：

小张：每天要分析这么多数据,真的很耗时
小李：是啊,而且经常要写各种分析报告
我：主要在哪些环节比较耗时？
小张：数据清洗、指标计算、图表生成这些都很繁琐
我：这些正好可以用AI Agent来协助

经过需求分析,我们确定了几个核心功能：

智能数据清洗
自动特征分析
可视化生成
报告撰写

技术方案设计

首先是整体架构：

from typing import List, Dict, Any, Optional
from enum import Enum
from pydantic import BaseModel
import pandas as pd
import numpy as np

class AnalysisTask(Enum):
    CLEAN = "clean"
    ANALYZE = "analyze"
    VISUALIZE = "visualize"
    REPORT = "report"

class DataContext(BaseModel):
    data_path: str
    task_type: AnalysisTask
    requirements: Dict[str, Any]
    history: List[Dict[str, Any]]

class DataAnalyst:
    def __init__(
        self,
        config: Dict[str, Any]
    ):
        # 1. 初始化分析模型
        self.analysis_model = AnalysisLLM(
            model="gpt-4",
            temperature=0.1,
            context_length=8000
        )

        # 2. 初始化工具集
        self.tools = {
            "cleaner": DataCleaner(),
            "analyzer": DataAnalyzer(),
            "visualizer": DataVisualizer(),
            "reporter": ReportGenerator()
        }

        # 3. 初始化数据存储
        self.data_store = DataStore(
            cache_dir="./cache",
            max_size_gb=10
        )

    async def process_task(
        self,
        context: DataContext
    ) -> Dict[str, Any]:
        # 1. 加载数据
        data = await self._load_data(
            context.data_path
        )

        # 2. 理解需求
        requirements = await self._understand_requirements(
            context.requirements
        )

        # 3. 生成分析方案
        plan = await self._generate_plan(
            data,
            requirements
        )

        # 4. 执行分析
        result = await self._execute_analysis(
            data,
            plan
        )

        return result

    async def _understand_requirements(
        self,
        requirements: Dict[str, Any]
    ) -> Dict[str, Any]:
        # 1. 提取分析目标
        objectives = await self.analysis_model.extract_objectives(
            requirements
        )

        # 2. 识别关键指标
        metrics = await self._identify_metrics(
            objectives
        )

        # 3. 确定分析方法
        methods = await self._select_methods(
            objectives,
            metrics
        )

        return {
            "objectives": objectives,
            "metrics": metrics,
            "methods": methods
        }

数据清洗功能

首先实现数据清洗功能：

class DataCleaner:
    def __init__(
        self,
        model: AnalysisLLM
    ):
        self.model = model

    async def clean_data(
        self,
        data: pd.DataFrame
    ) -> Dict[str, Any]:
        # 1. 数据概览
        profile = await self._profile_data(
            data
        )

        # 2. 识别问题
        issues = await self._identify_issues(
            data,
            profile
        )

        # 3. 执行清洗
        cleaned_data = await self._perform_cleaning(
            data,
            issues
        )

        return {
            "cleaned_data": cleaned_data,
            "profile": profile,
            "issues": issues
        }

    async def _identify_issues(
        self,
        data: pd.DataFrame,
        profile: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        issues = []

        # 1. 检查缺失值
        missing = await self._check_missing_values(
            data
        )
        issues.extend(missing)

        # 2. 检查异常值
        outliers = await self._detect_outliers(
            data
        )
        issues.extend(outliers)

        # 3. 检查数据类型
        type_issues = await self._check_data_types(
            data
        )
        issues.extend(type_issues)

        return issues

    async def _perform_cleaning(
        self,
        data: pd.DataFrame,
        issues: List[Dict[str, Any]]
    ) -> pd.DataFrame:
        cleaned = data.copy()

        for issue in issues:
            # 1. 处理缺失值
            if issue["type"] == "missing":
                cleaned = await self._handle_missing(
                    cleaned,
                    issue
                )

            # 2. 处理异常值
            elif issue["type"] == "outlier":
                cleaned = await self._handle_outlier(
                    cleaned,
                    issue
                )

            # 3. 处理类型问题
            elif issue["type"] == "type":
                cleaned = await self._handle_type(
                    cleaned,
                    issue
                )

        return cleaned

特征分析功能

接下来是特征分析功能：

class DataAnalyzer:
    def __init__(
        self,
        model: AnalysisLLM
    ):
        self.model = model

    async def analyze_features(
        self,
        data: pd.DataFrame,
        requirements: Dict[str, Any]
    ) -> Dict[str, Any]:
        # 1. 统计分析
        stats = await self._statistical_analysis(
            data
        )

        # 2. 特征相关性
        correlations = await self._correlation_analysis(
            data
        )

        # 3. 时间趋势
        trends = await self._trend_analysis(
            data
        )

        return {
            "statistics": stats,
            "correlations": correlations,
            "trends": trends
        }

    async def _statistical_analysis(
        self,
        data: pd.DataFrame
    ) -> Dict[str, Any]:
        stats = {}

        # 1. 基础统计量
        basic_stats = await self._calculate_basic_stats(
            data
        )
        stats["basic"] = basic_stats

        # 2. 分布分析
        distribution = await self._analyze_distribution(
            data
        )
        stats["distribution"] = distribution

        # 3. 分组统计
        groupby = await self._group_statistics(
            data
        )
        stats["groupby"] = groupby

        return stats

    async def _correlation_analysis(
        self,
        data: pd.DataFrame
    ) -> Dict[str, Any]:
        # 1. 计算相关系数
        corr_matrix = await self._calculate_correlations(
            data
        )

        # 2. 特征重要性
        importance = await self._feature_importance(
            data
        )

        # 3. 共线性检测
        collinearity = await self._check_collinearity(
            data
        )

        return {
            "correlation_matrix": corr_matrix,
            "feature_importance": importance,
            "collinearity": collinearity
        }

可视化功能

再来实现可视化功能：

class DataVisualizer:
    def __init__(
        self,
        model: AnalysisLLM
    ):
        self.model = model

    async def create_visualizations(
        self,
        data: pd.DataFrame,
        analysis: Dict[str, Any]
    ) -> Dict[str, Any]:
        # 1. 选择图表类型
        chart_types = await self._select_charts(
            data,
            analysis
        )

        # 2. 生成图表
        charts = await self._generate_charts(
            data,
            chart_types
        )

        # 3. 优化展示
        optimized = await self._optimize_display(
            charts
        )

        return {
            "charts": charts,
            "layout": optimized
        }

    async def _select_charts(
        self,
        data: pd.DataFrame,
        analysis: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        charts = []

        # 1. 分布图表
        distribution_charts = await self._distribution_charts(
            data,
            analysis
        )
        charts.extend(distribution_charts)

        # 2. 关系图表
        relationship_charts = await self._relationship_charts(
            data,
            analysis
        )
        charts.extend(relationship_charts)

        # 3. 趋势图表
        trend_charts = await self._trend_charts(
            data,
            analysis
        )
        charts.extend(trend_charts)

        return charts

    async def _generate_charts(
        self,
        data: pd.DataFrame,
        chart_types: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        charts = []

        for chart_type in chart_types:
            # 1. 准备数据
            plot_data = await self._prepare_plot_data(
                data,
                chart_type
            )

            # 2. 设置样式
            style = await self._set_chart_style(
                chart_type
            )

            # 3. 生成图表
            chart = await self._plot_chart(
                plot_data,
                chart_type,
                style
            )

            charts.append({
                "type": chart_type,
                "data": plot_data,
                "style": style,
                "chart": chart
            })

        return charts

报告生成功能

最后是报告生成功能：

class ReportGenerator:
    def __init__(
        self,
        model: AnalysisLLM
    ):
        self.model = model

    async def generate_report(
        self,
        data: pd.DataFrame,
        analysis: Dict[str, Any],
        visualizations: Dict[str, Any]
    ) -> Dict[str, Any]:
        # 1. 提取要点
        key_points = await self._extract_key_points(
            analysis
        )

        # 2. 生成结构
        structure = await self._create_structure(
            key_points
        )

        # 3. 撰写内容
        content = await self._write_content(
            structure,
            analysis,
            visualizations
        )

        return {
            "key_points": key_points,
            "structure": structure,
            "content": content
        }

    async def _extract_key_points(
        self,
        analysis: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        points = []

        # 1. 统计发现
        statistical_points = await self._extract_statistical_points(
            analysis["statistics"]
        )
        points.extend(statistical_points)

        # 2. 相关性发现
        correlation_points = await self._extract_correlation_points(
            analysis["correlations"]
        )
        points.extend(correlation_points)

        # 3. 趋势发现
        trend_points = await self._extract_trend_points(
            analysis["trends"]
        )
        points.extend(trend_points)

        return points

    async def _write_content(
        self,
        structure: Dict[str, Any],
        analysis: Dict[str, Any],
        visualizations: Dict[str, Any]
    ) -> Dict[str, str]:
        content = {}

        # 1. 写摘要
        content["summary"] = await self._write_summary(
            structure,
            analysis
        )

        # 2. 写主体
        content["body"] = await self._write_body(
            structure,
            analysis,
            visualizations
        )

        # 3. 写结论
        content["conclusion"] = await self._write_conclusion(
            structure,
            analysis
        )

        return content