Реализация AI-генерации отчётов по данным

Реализация AI-генерации отчётов по данным

AI-генерация отчётов автоматически трансформирует данные в структурированные нарративы с выводами, рекомендациями и визуализациями. Вместо того чтобы аналитик тратил часы на написание очередного monthly report, AI генерирует черновик за минуты.

Архитектура системы генерации отчётов

from anthropic import Anthropic
import pandas as pd
from jinja2 import Template

class ReportGenerator:
    def __init__(self):
        self.llm = Anthropic()

    def generate_report(self, data: dict, report_type: str,
                         period: str) -> str:
        # 1. Вычисление ключевых метрик
        metrics = self._compute_metrics(data, report_type)

        # 2. Обнаружение аномалий и трендов
        insights = self._detect_insights(metrics)

        # 3. Генерация нарратива
        narrative = self._generate_narrative(metrics, insights, period, report_type)

        # 4. Сборка отчёта
        return self._assemble_report(narrative, metrics, data, period)

    def _compute_metrics(self, data: dict, report_type: str) -> dict:
        metrics = {}
        df = data.get('main_df')

        if report_type == 'sales':
            metrics = {
                'total_revenue': df['revenue'].sum(),
                'revenue_mom': self._mom_change(df, 'revenue'),
                'total_orders': df['order_id'].nunique(),
                'orders_mom': self._mom_change(df, 'order_id', agg='count'),
                'avg_order_value': df['revenue'].sum() / df['order_id'].nunique(),
                'top_products': df.groupby('product')['revenue'].sum().nlargest(5).to_dict(),
                'conversion_rate': df['converted'].mean(),
            }
        elif report_type == 'user_activity':
            metrics = {
                'dau': df[df['date'] == df['date'].max()]['user_id'].nunique(),
                'mau': df['user_id'].nunique(),
                'retention_rate': self._compute_retention(df),
                'churn_rate': 1 - self._compute_retention(df),
                'session_duration_avg': df['session_duration'].mean(),
            }

        return metrics

    def _generate_narrative(self, metrics: dict, insights: list,
                              period: str, report_type: str) -> str:
        metrics_str = '\n'.join([f"- {k}: {v}" for k, v in metrics.items()])
        insights_str = '\n'.join([f"- {i}" for i in insights])

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1500,
            messages=[{
                "role": "user",
                "content": f"""Write a professional {report_type} report for {period}.

Key Metrics:
{metrics_str}

Observations:
{insights_str}

Structure the report as:
1. Executive Summary (3-4 sentences)
2. Key Highlights (bullet points)
3. Areas of Concern (if any)
4. Recommendations (3-5 actionable items)

Use professional, concise business language. No bullet points in executive summary."""
            }]
        )
        return response.content[0].text

    def _detect_insights(self, metrics: dict) -> list[str]:
        insights = []
        for key, value in metrics.items():
            if key.endswith('_mom'):
                if isinstance(value, float):
                    if value > 0.1:
                        insights.append(f"{key.replace('_mom', '')} grew {value:.1%} vs last month")
                    elif value < -0.05:
                        insights.append(f"WARNING: {key.replace('_mom', '')} declined {abs(value):.1%} vs last month")
        return insights

Шаблонный рендеринг отчёта

REPORT_TEMPLATE = """
# {{ report_type|title }} Report — {{ period }}
*Generated: {{ generated_at }}*

{{ narrative }}

## Key Metrics

| Metric | Value | vs. Last Period |
|--------|-------|-----------------|
{% for metric, value in metrics.items() %}
| {{ metric }} | {{ value }} | {{ changes.get(metric, 'N/A') }} |
{% endfor %}

## Visualizations

{{ charts_html }}
"""

def render_report(narrative: str, metrics: dict,
                   charts_html: str, period: str) -> str:
    template = Template(REPORT_TEMPLATE)
    return template.render(
        narrative=narrative,
        metrics=metrics,
        charts_html=charts_html,
        period=period,
        generated_at=datetime.now().strftime('%Y-%m-%d %H:%M')
    )

Экспорт в разные форматы

def export_report(report_html: str, format: str = 'pdf') -> bytes:
    if format == 'pdf':
        import pdfkit
        return pdfkit.from_string(report_html, False)
    elif format == 'docx':
        from docx import Document
        from htmldocx import HtmlToDocx
        doc = Document()
        parser = HtmlToDocx()
        parser.add_html_to_document(report_html, doc)
        buffer = io.BytesIO()
        doc.save(buffer)
        return buffer.getvalue()
    elif format == 'html':
        return report_html.encode('utf-8')

Автоматизированная генерация ежемесячных отчётов: с 4-6 часов ручной работы до 5-10 минут при качестве черновика, требующего минимальной правки.