Реализация AI-анализа Excel/CSV-файлов

Реализация AI-анализа Excel/CSV-файлов

AI-анализ Excel/CSV — инструмент для бизнес-пользователей, которые загружают файл и задают вопросы на естественном языке. Без SQL, без Python, без BI-инструмента. Загрузил файл с продажами — спросил "какой регион показал лучший рост за квартал".

Загрузка и профилирование файла

import pandas as pd
import io
from anthropic import Anthropic

class ExcelCSVAnalyzer:
    def __init__(self):
        self.llm = Anthropic()
        self.df = None
        self.profile = None

    def load(self, file_content: bytes, filename: str) -> dict:
        """Загрузка файла с автоопределением формата"""
        if filename.endswith('.csv'):
            # Автоопределение разделителя и кодировки
            self.df = self._smart_read_csv(file_content)
        elif filename.endswith(('.xlsx', '.xls')):
            # Чтение Excel с множеством листов
            xl = pd.ExcelFile(io.BytesIO(file_content))
            sheets = {}
            for sheet in xl.sheet_names:
                sheets[sheet] = pd.read_excel(xl, sheet_name=sheet)

            # Выбор основного листа
            self.df = max(sheets.values(), key=len)
            self.all_sheets = sheets

        self.profile = self._profile_dataframe(self.df)
        return self.profile

    def _smart_read_csv(self, content: bytes) -> pd.DataFrame:
        """Умное чтение CSV с определением параметров"""
        import chardet
        encoding = chardet.detect(content)['encoding'] or 'utf-8'

        for sep in [',', ';', '\t', '|']:
            try:
                df = pd.read_csv(
                    io.BytesIO(content),
                    sep=sep,
                    encoding=encoding,
                    thousands=',',
                    decimal='.'
                )
                if df.shape[1] > 1:  # Нашли правильный разделитель
                    return df
            except Exception:
                continue

        raise ValueError("Could not parse CSV file")

    def _profile_dataframe(self, df: pd.DataFrame) -> dict:
        """Автоматическое профилирование"""
        profile = {
            'shape': df.shape,
            'columns': {}
        }

        for col in df.columns:
            col_info = {
                'dtype': str(df[col].dtype),
                'null_count': int(df[col].isnull().sum()),
                'null_pct': float(df[col].isnull().mean()),
                'n_unique': int(df[col].nunique()),
            }

            if pd.api.types.is_numeric_dtype(df[col]):
                col_info.update({
                    'min': float(df[col].min()),
                    'max': float(df[col].max()),
                    'mean': float(df[col].mean()),
                    'std': float(df[col].std()),
                    'sample_values': df[col].dropna().head(3).tolist()
                })
            else:
                col_info['top_values'] = df[col].value_counts().head(5).to_dict()

            profile['columns'][col] = col_info

        # Автоопределение типов данных (даты, деньги, ID)
        profile['detected_types'] = self._detect_semantic_types(df)

        return profile

    def _detect_semantic_types(self, df: pd.DataFrame) -> dict:
        types = {}
        for col in df.columns:
            col_lower = col.lower()
            if any(kw in col_lower for kw in ['date', 'time', 'created', 'updated']):
                types[col] = 'datetime'
            elif any(kw in col_lower for kw in ['revenue', 'price', 'amount', 'cost', 'sum']):
                types[col] = 'currency'
            elif any(kw in col_lower for kw in ['id', 'code', 'number']):
                types[col] = 'identifier'
            elif df[col].dtype == 'object' and df[col].nunique() / len(df) < 0.05:
                types[col] = 'category'
        return types

    def ask(self, question: str) -> dict:
        """Анализ данных по вопросу"""
        schema_description = self._schema_to_text()

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=800,
            system=f"""You are a data analyst. Analyze a dataframe called 'df'.
Schema:
{schema_description}

Write Python pandas code to answer the question.
Use 'result' variable for the final answer.
Return ONLY code.""",
            messages=[{"role": "user", "content": question}]
        )

        code = response.content[0].text.strip().lstrip("```python").rstrip("```")

        local_vars = {'df': self.df, 'pd': pd, 'np': __import__('numpy')}
        exec(code, local_vars)
        result = local_vars.get('result')

        # Форматирование результата
        return {
            'result': self._format_result(result),
            'code': code,
            'chart': self._auto_visualize(result, question)
        }

Ключевая особенность: система понимает бизнес-контекст вопроса ("покажи топ клиентов") даже если колонка называется "client_name" или "company_id". LLM интерпретирует семантику вопроса и маппит на реальные названия колонок.