Разработка AI-системы для анализа данных

Разработка AI-системы для анализа данных

AI-система анализа данных — это инструмент, который позволяет бизнес-пользователям задавать вопросы к данным на естественном языке и получать аналитику без SQL и программирования. Под капотом: LLM для интерпретации запросов, код-генерация для анализа и автоматическая визуализация.

Архитектура системы

[User Query] → [Intent Classification]
                    ↓
          [Context Retrieval]
          (schema, examples, history)
                    ↓
          [Code Generation (LLM)]
          Python/SQL/Pandas code
                    ↓
          [Safe Code Execution]
          (sandboxed environment)
                    ↓
          [Result Formatting]
          (tables, charts, summary)
                    ↓
          [Natural Language Explanation]

Реализация Text-to-Code аналитики

from anthropic import Anthropic
import pandas as pd
import io

class AIDataAnalyst:
    def __init__(self, dataframes: dict[str, pd.DataFrame]):
        self.dfs = dataframes
        self.llm = Anthropic()
        self.schema = self._build_schema()

    def _build_schema(self) -> str:
        schema_parts = []
        for name, df in self.dfs.items():
            schema_parts.append(f"Table: {name}")
            schema_parts.append(f"Shape: {df.shape[0]} rows x {df.shape[1]} columns")
            schema_parts.append("Columns:")
            for col in df.columns:
                dtype = str(df[col].dtype)
                n_unique = df[col].nunique()
                sample = str(df[col].dropna().head(3).tolist())
                schema_parts.append(f"  - {col} ({dtype}, {n_unique} unique): {sample}")
            schema_parts.append("")
        return '\n'.join(schema_parts)

    def analyze(self, question: str) -> dict:
        """Анализ данных по вопросу на естественном языке"""
        system_prompt = f"""You are a data analyst. You have access to these dataframes:
{self.schema}

Write Python code using pandas to answer the user's question.
The dataframes are available as: {list(self.dfs.keys())}
Return ONLY the Python code, no explanations. Use variable 'result' for the final result."""

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1000,
            system=system_prompt,
            messages=[{"role": "user", "content": question}]
        )

        code = response.content[0].text.strip()
        if code.startswith("```python"):
            code = code[9:-3].strip()

        result = self._execute_safely(code)

        # Генерация объяснения
        explanation = self._generate_explanation(question, result, code)

        return {
            'question': question,
            'code': code,
            'result': result,
            'explanation': explanation
        }

    def _execute_safely(self, code: str) -> any:
        """Безопасное выполнение сгенерированного кода"""
        import builtins

        # Разрешённые функции
        safe_globals = {
            '__builtins__': {
                'len': builtins.len, 'range': builtins.range,
                'list': builtins.list, 'dict': builtins.dict,
                'str': builtins.str, 'int': builtins.int,
                'float': builtins.float, 'print': builtins.print,
                'sorted': builtins.sorted, 'sum': builtins.sum,
                'min': builtins.min, 'max': builtins.max,
                'round': builtins.round, 'abs': builtins.abs,
            },
            'pd': pd,
            'np': __import__('numpy'),
        }

        # Добавление датафреймов
        safe_globals.update(self.dfs)

        local_vars = {}
        exec(code, safe_globals, local_vars)

        return local_vars.get('result')

    def _generate_explanation(self, question: str, result, code: str) -> str:
        result_str = str(result)[:2000] if result is not None else "No result"

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Question: {question}
Analysis result: {result_str}

Provide a clear 2-3 sentence business explanation of this result."""
            }]
        )
        return response.content[0].text

Автоматическая визуализация

class AutoVisualizer:
    def create_chart(self, data, question: str) -> str:
        """Автоматический выбор и создание визуализации"""
        chart_type = self._suggest_chart_type(data, question)

        import plotly.express as px

        if isinstance(data, pd.DataFrame):
            if chart_type == 'bar':
                fig = px.bar(data, x=data.columns[0], y=data.columns[1],
                             title=question[:80])
            elif chart_type == 'line':
                fig = px.line(data, x=data.columns[0], y=data.columns[1:],
                              title=question[:80])
            elif chart_type == 'scatter':
                fig = px.scatter(data, x=data.columns[0], y=data.columns[1],
                                 title=question[:80])
            elif chart_type == 'pie':
                fig = px.pie(data, names=data.columns[0], values=data.columns[1],
                             title=question[:80])

            return fig.to_html(include_plotlyjs='cdn', full_html=False)

        return None

Типичный результат: аналитики тратят на рутинные ad-hoc запросы на 70% меньше времени. Бизнес-пользователи получают самообслуживание для 80% стандартных запросов без участия дата-инженеров.